227 lines
7.7 KiB
Python
227 lines
7.7 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from imagefind.config import Settings
|
|
from imagefind.resources import ResourceGovernor
|
|
from imagefind.scanner import Scanner
|
|
from imagefind.speech_quality import transcript_quality_state
|
|
from imagefind.uploads import UploadService
|
|
from imagefind.usage import StorageUsageService
|
|
|
|
|
|
def _upload(**values) -> dict:
|
|
return {
|
|
"id": "upload",
|
|
"status": "completed",
|
|
"phase": "ai_queued",
|
|
"storage_backend": "openlist_native",
|
|
"size_bytes": 100,
|
|
"bytes_received": 100,
|
|
"progress": 1,
|
|
"external_progress": 1,
|
|
"external_status": "provider-internal-state",
|
|
"external_error": "provider-secret-error",
|
|
"external_task_id": "provider-task-12345678",
|
|
"deduplicated": 0,
|
|
**values,
|
|
}
|
|
|
|
|
|
def test_upload_public_contract_reports_ai_queue_without_leaking_provider_state():
|
|
result = UploadService._public(_upload())
|
|
|
|
assert result["stage"] == "ai_queued"
|
|
assert result["stage_label"] == "已加入媒体库，AI 已排队"
|
|
assert result["external_status"] == "completed"
|
|
assert "external_state" not in result
|
|
assert "external_error" not in result
|
|
assert "provider-internal-state" not in str(result)
|
|
assert "provider-secret-error" not in str(result)
|
|
assert result["external_task_ref"] == "12345678"
|
|
assert [stage["state"] for stage in result["stages"]] == ["completed"] * 4
|
|
assert result["stages"][-1]["label"] == "加入媒体库并排队 AI"
|
|
|
|
historical = UploadService._public(_upload(phase="cataloging"))
|
|
assert historical["stage"] == "ai_queued"
|
|
assert historical["stages"][-1]["state"] == "completed"
|
|
|
|
|
|
def test_native_restore_refresh_preserves_the_physical_object_mapping():
|
|
restored = Scanner._native_refresh_record(
|
|
None,
|
|
{
|
|
"physical_path": "cloud/library/opaque-name.bin",
|
|
"physical_size_bytes": 356_793,
|
|
"logical_size_bytes": 356_665,
|
|
},
|
|
)
|
|
|
|
assert restored == {
|
|
"storage_backend": "openlist_native",
|
|
"external_target_path": "cloud/library/opaque-name.bin",
|
|
"external_size_bytes": 356_793,
|
|
"size_bytes": 356_665,
|
|
"content_sha256": "",
|
|
}
|
|
assert Scanner._native_refresh_record(None, {"physical_path": "", "physical_size_bytes": 1}) is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("score", "accepted", "rejected", "flags", "expected"),
|
|
[
|
|
(None, 0, 0, [], "empty"),
|
|
(0.82, 8, 2, [], "ready"),
|
|
(0.60, 5, 3, [], "filtered"),
|
|
(0.90, 8, 0, ["language_script_conflict"], "low_quality"),
|
|
(0.40, 2, 8, [], "low_quality"),
|
|
],
|
|
)
|
|
def test_transcript_quality_state_contract(score, accepted, rejected, flags, expected):
|
|
assert transcript_quality_state(score, accepted, rejected, flags) == expected
|
|
|
|
|
|
def test_running_lane_clears_stale_io_pressure_reason(monkeypatch, tmp_path: Path):
|
|
db = SimpleNamespace(setting=lambda _key, default=None: default)
|
|
governor = ResourceGovernor(db, Settings(data_dir=tmp_path))
|
|
governor._pressure_reasons["ai"] = "磁盘 I/O 持续繁忙，后台任务已让出资源"
|
|
monkeypatch.setattr(
|
|
governor,
|
|
"sample",
|
|
lambda: {
|
|
"cpu_percent": 10,
|
|
"memory_total_bytes": 8 * 1024**3,
|
|
"memory_available_bytes": 6 * 1024**3,
|
|
"disk_available_bytes": 100 * 1024**3,
|
|
"io": {"supported": False},
|
|
},
|
|
)
|
|
|
|
assert governor.pressure_reason(lane="ai", running=True) is None
|
|
assert governor._pressure_reasons["ai"] is None
|
|
|
|
|
|
class _Sources:
|
|
def __init__(self, source: dict, connector=None):
|
|
self.source = source
|
|
self._connector = connector
|
|
|
|
def get(self, source_id: str):
|
|
assert source_id == self.source["id"]
|
|
return self.source
|
|
|
|
def connector(self, source_id: str):
|
|
assert source_id == self.source["id"]
|
|
return self._connector
|
|
|
|
|
|
def _usage(source: dict, *, rclone=None, connector=None) -> StorageUsageService:
|
|
return StorageUsageService(
|
|
db=None,
|
|
settings=SimpleNamespace(),
|
|
rclone=rclone,
|
|
sources=_Sources(source, connector),
|
|
cache_seconds=30,
|
|
)
|
|
|
|
|
|
def test_local_capacity_uses_source_filesystem(monkeypatch, tmp_path: Path):
|
|
source = {"id": "local", "kind": "local", "config": {"path": str(tmp_path)}}
|
|
monkeypatch.setattr(
|
|
"imagefind.usage.shutil.disk_usage",
|
|
lambda path: SimpleNamespace(total=10_000, used=4_000, free=6_000)
|
|
if path == tmp_path
|
|
else None,
|
|
)
|
|
|
|
result = _usage(source).capacity("local")
|
|
assert (result["scope"], result["provider"], result["status"]) == (
|
|
"local",
|
|
"local",
|
|
"available",
|
|
)
|
|
assert result["available_bytes"] == 6_000
|
|
|
|
|
|
def test_openlist_capacity_uses_remote_quota_and_cache():
|
|
source = {
|
|
"id": "cloud",
|
|
"kind": "webdav",
|
|
"config": {"driver": "alist", "storage_backend": "openlist_native"},
|
|
}
|
|
calls = []
|
|
rclone = SimpleNamespace(
|
|
about=lambda value: calls.append(value["id"])
|
|
or {"total_bytes": 5_000, "used_bytes": 1_000, "available_bytes": 4_000}
|
|
)
|
|
usage = _usage(source, rclone=rclone)
|
|
|
|
first = usage.capacity("cloud")
|
|
second = usage.capacity("cloud")
|
|
assert first["scope"] == "remote"
|
|
assert first["provider"] == "rclone"
|
|
assert first["available_bytes"] == 4_000
|
|
assert second["available_bytes"] == 4_000
|
|
assert calls == ["cloud"]
|
|
|
|
|
|
def test_remote_capacity_manual_override_wins_without_querying_the_nas_or_provider():
|
|
total = 6 * 1024**4
|
|
available = 4 * 1024**4
|
|
source = {
|
|
"id": "cloud",
|
|
"kind": "webdav",
|
|
"config": {
|
|
"driver": "alist",
|
|
"storage_backend": "openlist_native",
|
|
"capacity_override_total_bytes": total,
|
|
"capacity_override_available_bytes": available,
|
|
},
|
|
}
|
|
rclone = SimpleNamespace(about=lambda _source: (_ for _ in ()).throw(AssertionError("must not query")))
|
|
|
|
result = _usage(source, rclone=rclone).capacity("cloud")
|
|
|
|
assert result["provider"] == "manual"
|
|
assert result["scope"] == "remote"
|
|
assert result["total_bytes"] == total
|
|
assert result["used_bytes"] == total - available
|
|
assert result["available_bytes"] == available
|
|
|
|
|
|
def test_webdav_capacity_supports_quota_and_distinguishes_unsupported_from_unreachable():
|
|
xml = b"""<?xml version='1.0'?>
|
|
<d:multistatus xmlns:d='DAV:'><d:response><d:propstat><d:prop>
|
|
<d:quota-used-bytes>300</d:quota-used-bytes>
|
|
<d:quota-available-bytes>700</d:quota-available-bytes>
|
|
</d:prop></d:propstat></d:response></d:multistatus>"""
|
|
|
|
class Connector:
|
|
base_url = "https://dav.example/root/"
|
|
|
|
def __init__(self, response):
|
|
self.client = SimpleNamespace(request=lambda *_args, **_kwargs: response)
|
|
self.closed = False
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
source = {"id": "dav", "kind": "webdav", "config": {"driver": "webdav"}}
|
|
response = SimpleNamespace(status_code=207, content=xml)
|
|
connector = Connector(response)
|
|
available = _usage(source, connector=connector).capacity("dav")
|
|
assert available["provider"] == "webdav"
|
|
assert available["total_bytes"] == 1_000
|
|
assert connector.closed is True
|
|
|
|
no_quota = Connector(SimpleNamespace(status_code=207, content=b"<d:multistatus xmlns:d='DAV:'/>"))
|
|
unsupported = _usage(source, connector=no_quota).capacity("dav")
|
|
assert unsupported["status"] == "unsupported"
|
|
|
|
failed = Connector(SimpleNamespace(status_code=503, content=b""))
|
|
unreachable = _usage(source, connector=failed).capacity("dav")
|
|
assert unreachable["status"] == "unreachable"
|
|
assert unreachable["total_bytes"] == 0
|