import unittest from unittest.mock import patch from cloudsearch_transfer.adapter.aliyun import AliyunAdapter from cloudsearch_transfer.adapter.baidu import BaiduAdapter from cloudsearch_transfer.adapter.xunlei import XunleiAdapter from cloudsearch_transfer.config import PlatformConfig, TransferConfig from cloudsearch_transfer.adapter.base import FileInfo class DummyCredential: cookie = "k=xxxxxxxx" refresh_token = "rt" def validate(self): return True def get_bdstoken(self): return "bdstoken" def get_drive_id(self): return "drive-id" def get_headers(self): return {} def get_auth_headers(self): return {} def make_config(**kwargs): data = dict(enabled=True, cookie="k=" + "x" * 80, refresh_token="rt", account_name="test", save_dir="/") data.update(kwargs) return PlatformConfig(**data) def make_transfer_config(): return TransferConfig(request_timeout=1, max_retries=0, ad_filter_enabled=False, task_poll_interval=0, task_poll_max_attempts=2) class P1DriveApiCapabilityTests(unittest.TestCase): def make_aliyun(self): with patch("cloudsearch_transfer.adapter.aliyun.AliyunCredentialManager", return_value=DummyCredential()): return AliyunAdapter(make_config(), make_transfer_config()) def make_baidu(self): with patch("cloudsearch_transfer.adapter.baidu.BaiduCredentialManager", return_value=DummyCredential()): return BaiduAdapter(make_config(), make_transfer_config()) def make_xunlei(self): with patch("cloudsearch_transfer.adapter.xunlei.XunleiCredentialManager", return_value=DummyCredential()): return XunleiAdapter(make_config(), make_transfer_config()) def test_p1_adapters_declare_callable_core_drive_api_methods(self): expectations = { "aliyun": (self.make_aliyun(), ["ensure_dir", "save_files", "rename", "move_files", "delete_files"]), "baidu": (self.make_baidu(), ["ensure_dir", "save_files", "rename", "move_files", "delete_files"]), "xunlei": (self.make_xunlei(), ["ensure_dir", "save_files", "poll_task", "rename", "move_files"]), } for platform, (adapter, required) in expectations.items(): with self.subTest(platform=platform): for capability in required: self.assertTrue(adapter.capabilities.get(capability), f"{platform}.{capability} capability not declared") for method in ["ensure_dir", "mkdir", "rename", "move_files"]: self.assertNotEqual(getattr(type(adapter), method), getattr(adapter.__class__.__mro__[1], method, None), f"{platform}.{method} not overridden") if adapter.capabilities.get("delete_files"): self.assertNotEqual(getattr(type(adapter), "delete_files"), getattr(adapter.__class__.__mro__[1], "delete_files", None), f"{platform}.delete_files not overridden") def test_aliyun_save_files_resolves_path_to_file_id_before_copy(self): adapter = self.make_aliyun() adapter.ensure_dir = lambda path: "target-folder-id" captured = {} transfer = adapter._get_transfer() def fake_batch_copy(share_id, share_token, file_ids, to_parent): captured["to_parent"] = to_parent return ["new-file"] transfer._batch_copy = fake_batch_copy detail = {"share_token": "share-token", "files": [{"file_id": "src-file"}]} self.assertEqual(adapter._save_files("share-id", detail, "/Media"), ["new-file"]) self.assertEqual(captured["to_parent"], "target-folder-id") def test_aliyun_ensure_dir_creates_nested_paths_progressively(self): adapter = self.make_aliyun() existing = {"/A": "fid-A"} created = [] adapter.get_fids = lambda paths: [{"file_path": p, "fid": existing[p]} for p in paths if p in existing] def fake_mkdir(path): created.append(path) existing[path] = "fid-" + path.rsplit("/", 1)[-1] return {"code": 0, "data": {"fid": existing[path]}} adapter.mkdir = fake_mkdir self.assertEqual(adapter.ensure_dir("/A/B"), "fid-B") self.assertEqual(created, ["/A/B"]) def test_baidu_save_files_ensures_path_before_transfer_and_matching(self): adapter = self.make_baidu() adapter.ensure_dir = lambda path: "/CloudSearch/Media" captured = {} adapter.credential.get_bdstoken = lambda: "bdstoken" def fake_transfer_files(shareid, uk, fs_ids, save_dir, token): captured["transfer_dir"] = save_dir adapter._transfer._transfer_files = fake_transfer_files def fake_list_and_match(save_dir, filenames, token): captured["list_dir"] = save_dir return ["1001"] adapter._transfer._list_and_match = fake_list_and_match detail = {"shareid": "sid", "uk": "uk", "fs_ids": ["old"], "filenames": ["demo.mkv"]} self.assertEqual(adapter._save_files("pwd", detail, "/CloudSearch/Media"), ["1001"]) self.assertEqual(captured["transfer_dir"], "/CloudSearch/Media") self.assertEqual(captured["list_dir"], "/CloudSearch/Media") def test_baidu_ensure_dir_creates_nested_paths_progressively(self): adapter = self.make_baidu() dirs = {"/A"} created = [] adapter.get_files = lambda parent="/": [FileInfo(fid=parent.rstrip("/") + "/A", name="A", is_dir=True)] if parent == "/" else [] def fake_mkdir(path): created.append(path) dirs.add(path) return {"code": 0, "data": {"path": path}} adapter.mkdir = fake_mkdir self.assertEqual(adapter.ensure_dir("/A/B"), "/A/B") self.assertEqual(created, ["/A/B"]) def test_xunlei_save_files_resolves_path_and_polls_restore_task(self): adapter = self.make_xunlei() adapter.ensure_dir = lambda path: "target-parent-id" transfer = adapter._transfer transfer._restore_files = lambda share_id, token, fids, parent_id="": "task-1" if parent_id == "target-parent-id" else (_ for _ in ()).throw(AssertionError(parent_id)) transfer._poll_restore_task = lambda task_id: {"old-1": "new-1"} detail = {"share_id": "share", "pass_code_token": "pct", "files": [{"id": "old-1", "name": "demo"}]} self.assertEqual(adapter._save_files("share", detail, "/Media"), ["new-1"]) def test_xunlei_query_task_delegates_to_restore_poll(self): adapter = self.make_xunlei() adapter._transfer._poll_restore_task = lambda task_id: {"old": "new"} self.assertEqual(adapter.query_task("task-1"), {"code": 0, "status": 200, "data": {"old": "new"}}) def test_xunlei_does_not_advertise_permanent_delete_as_drive_delete_capability(self): adapter = self.make_xunlei() self.assertFalse(adapter.capabilities.get("delete_files"), "迅雷 batchDelete is permanent and must not be advertised as generic delete_files") def test_xunlei_get_share_detail_passes_extraction_code_to_transfer_engine(self): adapter = self.make_xunlei() captured = {} def fake_get_share_info(share_id, passcode=""): captured["share_id"] = share_id captured["passcode"] = passcode return {"title": "demo", "files": [{"id": "old"}], "pass_code_token": "pct"} adapter._transfer._get_share_info = fake_get_share_info detail = adapter._get_share_detail("share-id", "abcd") self.assertEqual(detail["title"], "demo") self.assertEqual(captured, {"share_id": "share-id", "passcode": "abcd"}) def test_xunlei_create_share_forwards_requested_share_password(self): adapter = self.make_xunlei() captured = {} def fake_create_share(file_ids, password=""): captured["file_ids"] = file_ids captured["password"] = password return ("https://pan.xunlei.com/s/new", password) adapter._transfer._create_share = fake_create_share self.assertEqual(adapter._create_share(["fid-1"], "demo", "xy12"), ("https://pan.xunlei.com/s/new", "xy12")) self.assertEqual(captured, {"file_ids": ["fid-1"], "password": "xy12"}) def test_xunlei_poll_task_alias_matches_declared_capability(self): adapter = self.make_xunlei() adapter._transfer._poll_restore_task = lambda task_id: {"old": "new"} self.assertEqual(adapter.poll_task("task-1"), {"code": 0, "status": 200, "data": {"old": "new"}}) def test_xunlei_get_fids_is_read_only_and_does_not_create_missing_paths(self): adapter = self.make_xunlei() calls = [] def fake_get_files(parent="0"): calls.append(("get_files", parent)) if parent in ("0", ""): return [FileInfo(fid="fid-A", name="A", is_dir=True)] if parent == "fid-A": return [FileInfo(fid="fid-B", name="B", is_dir=False)] return [] def fail_mkdir(*args, **kwargs): raise AssertionError("get_fids must not create directories") adapter.get_files = fake_get_files adapter.mkdir = fail_mkdir self.assertEqual(adapter.get_fids(["/A/B", "/A/Missing"]), [{"file_path": "/A/B", "fid": "fid-B"}]) self.assertEqual(calls, [("get_files", "0"), ("get_files", "fid-A"), ("get_files", "0"), ("get_files", "fid-A")]) def test_baidu_get_fids_checks_existing_parent_listing(self): adapter = self.make_baidu() calls = [] def fake_get_files(parent="/"): calls.append(parent) if parent == "/A": return [FileInfo(fid="1001", name="movie.mkv", is_dir=False)] return [] adapter.get_files = fake_get_files self.assertEqual(adapter.get_fids(["/A/movie.mkv", "/A/missing.mkv"]), [{"file_path": "/A/movie.mkv", "fid": "1001", "path": "/A/movie.mkv"}]) self.assertEqual(calls, ["/A", "/A"]) if __name__ == "__main__": unittest.main()