from contextlib import nullcontext from pathlib import Path from types import SimpleNamespace import httpx import pytest from backend.imagefind.openlist_native import ( OpenListNativeService, _remote_join, _remote_parent, normalize_task_state, ) from backend.imagefind.remote import AlistClient def test_remote_root_join_is_safe(): assert _remote_join(".", "", "/videos", "clip.mp4") == "videos/clip.mp4" assert _remote_parent("clip.mp4") == "" assert _remote_parent("videos/clip.mp4") == "videos" assert normalize_task_state("succeeded") == 2 assert normalize_task_state("errored") == 5 assert normalize_task_state("7.0") == 7 def test_native_mapping_survives_library_only_deletion_by_falling_back_to_upload_history(): class Connection: def execute(self, query, _parameters): row = None if "FROM uploads" in query: row = { "physical_path": "cloud/library/opaque.bin", "physical_size_bytes": 356_793, "size_bytes": 356_665, } return SimpleNamespace(fetchone=lambda: row) service = object.__new__(OpenListNativeService) service.db = SimpleNamespace(read=lambda: nullcontext(Connection())) service.sources = SimpleNamespace( get=lambda _source_id: {"config": {"storage_backend": "openlist_native"}} ) assert service.physical_object_for_key("source", "ingest/movie.mp4") == ( "cloud/library/opaque.bin", 356_793, 356_665, ) def test_alist_client_copy_and_application_errors(): calls: list[str] = [] def handler(request: httpx.Request) -> httpx.Response: calls.append(request.url.path) if request.url.path.endswith("/api/auth/login"): return httpx.Response(200, json={"code": "200", "data": {"token": "test-token"}}) if request.url.path.endswith("/api/fs/get"): body = request.content if b"missing" in body: return httpx.Response(200, json={"code": 404, "message": "object not found"}) return httpx.Response(200, json={"code": 200, "data": {"is_dir": False, "size": 4}}) if request.url.path.endswith("/api/fs/copy"): return httpx.Response(200, json={"code": 200, "data": {"tasks": [{"id": "copy-1"}]}}) if request.url.path.endswith("/api/task/copy/info"): return httpx.Response(200, json={"code": 200, "data": {"state": 2, "progress": 100}}) return httpx.Response(200, json={"code": 200, "data": {}}) client = AlistClient("http://openlist", "admin", "password") client.client.close() client.client = httpx.Client(transport=httpx.MockTransport(handler)) try: assert client.object_info("video.mp4")["size"] == 4 assert client.object_info("missing") is None assert client.copy_file("source/video.mp4", "stage/video.mp4") == ["copy-1"] assert client.copy_task_info("copy-1")["state"] == 2 finally: client.close() assert calls.count("/api/auth/login") == 1 def test_native_catalog_uses_physical_object_but_keeps_logical_media_identity(): class Sources: def get(self, source_id: str): assert source_id == "source" return {"config": {"mode": "direct"}} def remote_access(self, source_id: str, key: str): assert source_id == "source" assert key == "合集/中文标题.mp4" return "http://127.0.0.1:1234/%E5%90%88%E9%9B%86/video.mp4", "user", "secret", True class Client: def object_info(self, path: str): assert path == "cloud/library/opaque-name.bin" return {"is_dir": False, "size": 120, "modified": "2026-08-07T10:00:00Z", "sign": "etag"} service = object.__new__(OpenListNativeService) service.sources = Sources() service.client = lambda _source_id: Client() item = service.catalog_item( "source", "合集/中文标题.mp4", { "external_target_path": "cloud/library/opaque-name.bin", "external_size_bytes": 120, "size_bytes": 100, "content_sha256": "a" * 64, }, ) assert item.key == "合集/中文标题.mp4" assert item.display_name == "中文标题.mp4" assert item.size_bytes == 100 assert item.etag == "etag" assert item.fingerprint def test_native_prepare_and_cancel_preserve_original_and_local_recovery(tmp_path: Path): original = tmp_path / "upload.part" original.write_bytes(b"data") staging = tmp_path / "native" class Sources: rclone = SimpleNamespace() def get(self, _source_id): return {} service = object.__new__(OpenListNativeService) service.sources = Sources() service.configuration = lambda _source: SimpleNamespace( local_staging_path=staging, encrypted=False, source_path="local-stage", target_path="cloud", ) paths = service.prepare("source", "upload-id", "folder/movie.mp4", original) assert original.read_bytes() == b"data" assert Path(paths["local_path"]).read_bytes() == b"data" cancelled: list[str] = [] service.client = lambda _source_id: SimpleNamespace( cancel_copy_task=lambda task_id: cancelled.append(task_id) ) service.cancel("source", "upload-id", "task-id", paths["staged_path"]) assert cancelled == ["task-id"] assert original.is_file() assert Path(paths["local_path"]).is_file() class _MovingClient: def __init__(self, files: dict[str, int], *, fail_after_move: bool = False): self.files = dict(files) self.fail_after_move = fail_after_move def object_info(self, path: str): size = self.files.get(path) return None if size is None else {"is_dir": False, "size": size} def ensure_directory(self, _path: str): return None def move_file(self, source: str, target_directory: str, *, overwrite: bool = False): target = f"{target_directory}/{source.rsplit('/', 1)[-1]}" if not overwrite and target in self.files: raise RuntimeError("target exists") self.files[target] = self.files.pop(source) if self.fail_after_move: self.fail_after_move = False raise TimeoutError("control-plane timeout") def remove(self, path: str): self.files.pop(path, None) def _native_service(client: _MovingClient): service = object.__new__(OpenListNativeService) service.sources = SimpleNamespace(get=lambda _source_id: {}) service.configuration = lambda _source: SimpleNamespace(target_path="cloud/library", encrypted=False) service.client = lambda _source_id, _source=None: client return service @pytest.mark.parametrize("ambiguous_timeout", [False, True]) def test_native_trash_and_restore_verify_provider_state(ambiguous_timeout: bool): client = _MovingClient( {"cloud/library/opaque.bin": 123}, fail_after_move=ambiguous_timeout, ) service = _native_service(client) trash_path = service.trash_object("source", "trash-id", "cloud/library/opaque.bin", 123) assert trash_path == "cloud/library/.imagefind-native-trash/trash-id/opaque.bin" assert client.files == {trash_path: 123} client.fail_after_move = ambiguous_timeout service.restore_object("source", trash_path, "cloud/library/opaque.bin", 123) assert client.files == {"cloud/library/opaque.bin": 123} def test_native_trash_waits_for_eventually_consistent_provider(monkeypatch): class EventuallyConsistentClient(_MovingClient): stale_reads = 4 def object_info(self, path: str): trash_path = "cloud/library/.imagefind-native-trash/trash-id/opaque.bin" if self.stale_reads and path in {"cloud/library/opaque.bin", trash_path}: self.stale_reads -= 1 if path == "cloud/library/opaque.bin": return {"name": "opaque.bin", "size": 123, "is_dir": False} return None return super().object_info(path) monkeypatch.setattr("imagefind.openlist_native.time.sleep", lambda _seconds: None) client = EventuallyConsistentClient({"cloud/library/opaque.bin": 123}) service = _native_service(client) trash_path = service.trash_object("source", "trash-id", "cloud/library/opaque.bin", 123) assert trash_path == "cloud/library/.imagefind-native-trash/trash-id/opaque.bin" assert client.files == {trash_path: 123} def test_native_trash_restore_conflicts_are_non_destructive_and_purge_is_idempotent(): trash_path = "cloud/library/.imagefind-native-trash/trash-id/opaque.bin" client = _MovingClient({"cloud/library/opaque.bin": 123, trash_path: 123}) service = _native_service(client) with pytest.raises(RuntimeError, match="目标已存在"): service.trash_object("source", "trash-id", "cloud/library/opaque.bin", 123) with pytest.raises(FileExistsError, match="已被占用"): service.restore_object("source", trash_path, "cloud/library/opaque.bin", 123) assert client.files == {"cloud/library/opaque.bin": 123, trash_path: 123} client.files.pop("cloud/library/opaque.bin") service.purge_object("source", trash_path) service.purge_object("source", trash_path) assert client.files == {}