import unittest import requests from unittest.mock import patch from cloudsearch_transfer.adapter.quark import QuarkAdapter from cloudsearch_transfer.adapter.uc import UcAdapter from cloudsearch_transfer.config import PlatformConfig, TransferConfig from cloudsearch_transfer.adapter.base import BaseCloudDriveAdapter from cloudsearch_transfer.errors import TransferError class DummyResponse: def __init__(self, payload): self._payload = payload def json(self): return self._payload def raise_for_status(self): return None def make_adapter(adapter_cls): return adapter_cls( PlatformConfig(enabled=True, cookie="k=" + "x" * 80, account_name="test"), TransferConfig(request_timeout=1, max_retries=0, ad_filter_enabled=False), ) class QuarkUcCapabilityTests(unittest.TestCase): def test_quark_declares_p0_drive_api_capabilities(self): adapter = make_adapter(QuarkAdapter) self.assertTrue(adapter.capabilities["ensure_dir"]) self.assertTrue(adapter.capabilities["save_files"]) self.assertTrue(adapter.capabilities["poll_task"]) self.assertTrue(adapter.capabilities["rename"]) self.assertTrue(adapter.capabilities["move_files"]) self.assertTrue(adapter.capabilities["delete_files"]) self.assertTrue(adapter.capabilities["cleanup_recycle"]) def test_uc_declares_p0_drive_api_capabilities_and_share_staging(self): adapter = make_adapter(UcAdapter) self.assertTrue(adapter.capabilities["ensure_dir"]) self.assertTrue(adapter.capabilities["save_files"]) self.assertTrue(adapter.capabilities["poll_task"]) self.assertTrue(adapter.capabilities["rename"]) self.assertTrue(adapter.capabilities["move_files"]) self.assertTrue(adapter.capabilities["delete_files"]) self.assertTrue(adapter.capabilities["share_staging_folder"]) def test_quark_ensure_dir_uses_path_lookup_before_mkdir(self): adapter = make_adapter(QuarkAdapter) mkdir_calls = [] adapter.get_fids = lambda paths: [{"file_path": "/Media", "fid": "fid-media"}] adapter.mkdir = lambda path: mkdir_calls.append(path) or {"code": 0, "data": {"fid": "new"}} self.assertEqual(adapter.ensure_dir("/Media"), "fid-media") self.assertEqual(mkdir_calls, []) def test_quark_ensure_dir_creates_missing_path(self): adapter = make_adapter(QuarkAdapter) adapter.get_fids = lambda paths: [] adapter.mkdir = lambda path: {"code": 0, "data": {"fid": "created-fid"}} self.assertEqual(adapter.ensure_dir("Media/Shows"), "created-fid") def test_uc_share_folder_created_when_missing(self): adapter = make_adapter(UcAdapter) adapter.ls_dir = lambda parent: {"code": 0, "data": {"list": []}} adapter.mkdir = lambda path: {"code": 0, "data": {"fid": "share-folder-fid"}} self.assertEqual(adapter.get_or_create_share_folder(), "share-folder-fid") def test_uc_move_files_polls_task_until_complete(self): adapter = make_adapter(UcAdapter) calls = [] def fake_post(url, json_data=None, params=None, headers=None): calls.append((url, json_data)) return DummyResponse({"code": 0, "status": 200, "data": {"task_id": "task-1"}}) adapter._post = fake_post adapter.query_task = lambda task_id: {"code": 0, "status": 200, "data": {"status": 2}} result = adapter.move_files(["fid-1", "fid-2"], "target-fid") self.assertEqual(result["code"], 0) self.assertEqual(calls[0][1]["filelist"], ["fid-1", "fid-2"]) self.assertEqual(calls[0][1]["to_pdir_fid"], "target-fid") def test_quark_save_files_resolves_path_to_fid(self): adapter = make_adapter(QuarkAdapter) adapter.ensure_dir = lambda path: "target-fid" adapter._transfer_engine._get_stoken = lambda pwd_id: "stoken" captured = {} adapter._transfer_engine._init_save = lambda pwd_id, stoken, detail, to_pdir_fid: captured.setdefault("to_pdir_fid", to_pdir_fid) or "task-1" adapter._transfer_engine._poll_save_task = lambda task_id: ["new-fid"] self.assertEqual(adapter._save_files("pwd", {"fid": "src"}, "/Media"), ["new-fid"]) self.assertEqual(captured["to_pdir_fid"], "target-fid") def test_uc_transfer_resolves_path_to_fid(self): adapter = make_adapter(UcAdapter) adapter._credential.validate = lambda: True adapter.ensure_dir = lambda path: "target-fid" adapter.get_or_create_share_folder = lambda: "target-fid" adapter._parse_share_url = lambda url: ("pwd", "pass") adapter._transfer_engine._get_stoken = lambda pwd_id, passcode="": "stoken" adapter._transfer_engine._get_detail = lambda pwd_id, stoken: {"title": "demo", "fid": "src"} captured = {} def fake_init_save(pwd_id, stoken, detail, to_pdir_fid): captured["save_dir"] = to_pdir_fid return "save-task" adapter._transfer_engine._init_save = fake_init_save adapter._transfer_engine._poll_save_task = lambda task_id: ["new-fid"] adapter._transfer_engine._init_share = lambda fids, title: "share-task" adapter._transfer_engine._poll_share_task = lambda task_id: "share-id" adapter._transfer_engine._set_password = lambda share_id, password: ("https://drive.uc.cn/s/new", password) result = adapter.transfer("https://drive.uc.cn/s/abcdef", save_dir="/Media", share_password="pw") self.assertTrue(result.success) self.assertEqual(captured["save_dir"], "target-fid") def test_quark_transfer_resolves_path_to_fid(self): adapter = make_adapter(QuarkAdapter) adapter.ensure_dir = lambda path: "target-fid" adapter._credential.validate = lambda: True captured = {} def fake_transfer(share_url, save_dir, share_password): captured["save_dir"] = save_dir return { "new_file_ids": ["new-fid"], "file_name": "demo", "share_url": "https://pan.quark.cn/s/new", "passcode": share_password, } adapter._transfer_engine.transfer = fake_transfer result = adapter._transfer("https://pan.quark.cn/s/abcdef", save_dir="/Media", share_password="pw") self.assertTrue(result.success) self.assertEqual(captured["save_dir"], "target-fid") def test_ensure_dir_creates_nested_paths_progressively(self): adapter = make_adapter(QuarkAdapter) adapter.get_fids = lambda paths: [] created = [] def fake_mkdir(path): created.append(path) return {"code": 0, "data": {"fid": "fid-" + path.rsplit("/", 1)[-1]}} adapter.mkdir = fake_mkdir self.assertEqual(adapter.ensure_dir("/A/B"), "fid-B") self.assertEqual(created, ["/A", "/A/B"]) def test_uc_transfer_uses_staging_folder_then_moves_to_target(self): adapter = make_adapter(UcAdapter) adapter._credential.validate = lambda: True adapter.ensure_dir = lambda path: "target-fid" adapter.get_or_create_share_folder = lambda: "staging-fid" adapter._parse_share_url = lambda url: ("pwd", "pass") adapter._transfer_engine._get_stoken = lambda pwd_id, passcode="": "stoken" adapter._transfer_engine._get_detail = lambda pwd_id, stoken: {"title": "demo", "fid": "src"} captured = {} def fake_init_save(pwd_id, stoken, detail, to_pdir_fid): captured["save_to"] = to_pdir_fid return "save-task" adapter._transfer_engine._init_save = fake_init_save adapter._transfer_engine._poll_save_task = lambda task_id: ["new-fid"] def fake_move(fids, to_pdir_fid): captured["move_to"] = to_pdir_fid return {"code": 0, "status": 200} adapter.move_files = fake_move adapter._transfer_engine._init_share = lambda fids, title: "share-task" adapter._transfer_engine._poll_share_task = lambda task_id: "share-id" adapter._transfer_engine._set_password = lambda share_id, password: ("https://drive.uc.cn/s/new", password) result = adapter.transfer("https://drive.uc.cn/s/abcdef", save_dir="/Media", share_password="pw") self.assertTrue(result.success) self.assertEqual(captured["save_to"], "staging-fid") self.assertEqual(captured["move_to"], "target-fid") def test_api_success_requires_explicit_success_code_or_status(self): adapter = make_adapter(QuarkAdapter) self.assertFalse(adapter._api_success({})) self.assertFalse(adapter._api_success({"message": "bad"})) self.assertTrue(adapter._api_success({"code": 0})) self.assertTrue(adapter._api_success({"status": 200})) def test_query_task_passes_retry_index_and_timestamp_params(self): adapter = make_adapter(QuarkAdapter) captured_params = [] attempts = {"count": 0} def fake_get(url, params=None, retry=None): captured_params.append(dict(params)) attempts["count"] += 1 if attempts["count"] == 1: return DummyResponse({"data": {"status": 1}}) return DummyResponse({"data": {"status": 2, "task_id": "task-1"}}) adapter._get = fake_get result = adapter.query_task("task-1") self.assertEqual(result["code"], 0) self.assertEqual(captured_params[0]["retry_index"], 0) self.assertEqual(captured_params[1]["retry_index"], 1) self.assertIn("__dt", captured_params[0]) self.assertIn("__t", captured_params[0]) def test_quark_cleanup_recycle_only_removes_matching_fids(self): adapter = make_adapter(QuarkAdapter) adapter.recycle_list = lambda: [ {"fid": "keep", "record_id": "r-keep"}, {"fid": "target", "record_id": "r-target"}, ] captured = {} def fake_recycle_remove(records): captured["records"] = records return {"code": 0} adapter.recycle_remove = fake_recycle_remove result = adapter.cleanup_recycle(["target"]) self.assertEqual(result["code"], 0) self.assertEqual(captured["records"], [{"fid": "target", "record_id": "r-target"}]) def test_drive_api_json_raises_transfer_error_on_http_error(self): adapter = make_adapter(QuarkAdapter) class ErrorResponse(DummyResponse): status_code = 429 text = "too many requests" def raise_for_status(self): raise requests.HTTPError("429 Too Many Requests") with self.assertRaises(Exception) as ctx: adapter._drive_api_json(ErrorResponse({"code": 0}), context="限流测试") self.assertIn("限流测试", str(ctx.exception)) def test_drive_api_json_raises_transfer_error_on_invalid_json(self): adapter = make_adapter(QuarkAdapter) class InvalidJsonResponse(DummyResponse): status_code = 200 text = "bad gateway" def json(self): raise ValueError("not json") with self.assertRaises(Exception) as ctx: adapter._drive_api_json(InvalidJsonResponse({}), context="JSON测试") self.assertIn("JSON测试", str(ctx.exception)) def test_quark_delete_files_capability_calls_drive_delete(self): adapter = make_adapter(QuarkAdapter) calls = [] adapter.delete = lambda fids: calls.append(list(fids)) or True self.assertEqual(adapter.delete_files(["fid-1", "fid-2"]), {"code": 0, "status": 200}) self.assertEqual(calls, [["fid-1", "fid-2"]]) def test_uc_delete_files_capability_calls_drive_delete(self): adapter = make_adapter(UcAdapter) calls = [] adapter.delete = lambda fids: calls.append(list(fids)) or True self.assertEqual(adapter.delete_files(["fid-1"]), {"code": 0, "status": 200}) self.assertEqual(calls, [["fid-1"]]) def test_uc_transfer_filters_ads_after_staging_save(self): adapter = make_adapter(UcAdapter) adapter.ensure_dir = lambda path: "target-dir" adapter.get_or_create_share_folder = lambda: "staging-dir" adapter._parse_share_url = lambda url: ("pwd-id", "") adapter._filter_ads = lambda fids: [fid for fid in fids if fid != "ad-fid"] adapter.transfer_config.ad_filter_enabled = True adapter._transfer_engine._get_stoken = lambda pwd_id, passcode: "stoken" adapter._transfer_engine._get_detail = lambda pwd_id, stoken: {"title": "Title"} adapter._transfer_engine._init_save = lambda pwd_id, stoken, detail, to_pdir_fid: "save-task" adapter._transfer_engine._poll_save_task = lambda task_id: ["keep-fid", "ad-fid"] adapter.move_files = lambda fids, target: {"code": 0} adapter._transfer_engine._init_share = lambda fids, title: "share-task" if fids == ["keep-fid"] else (_ for _ in ()).throw(AssertionError(f"unfiltered fids: {fids}")) adapter._transfer_engine._poll_share_task = lambda task_id: "share-id" adapter._transfer_engine._set_password = lambda share_id, pwd: ("https://share", pwd) result = adapter.transfer("https://drive.uc.cn/s/abc") self.assertEqual(result.new_file_id, "keep-fid") def test_uc_ls_dir_wraps_http_and_json_errors(self): adapter = make_adapter(UcAdapter) class BadResponse: text = "bad gateway" def raise_for_status(self): raise requests.HTTPError("HTTP 502") def json(self): raise AssertionError("json() should not be called after HTTP error") adapter._get = lambda *args, **kwargs: BadResponse() with self.assertRaises(TransferError): adapter.ls_dir("0") def test_base_declares_optional_drive_api_methods(self): required = [ "ensure_dir", "get_fids", "mkdir", "rename", "move_files", "delete_files", "query_task", "cleanup_recycle", ] for name in required: self.assertTrue(hasattr(BaseCloudDriveAdapter, name), name) def test_base_optional_drive_api_methods_raise_transfer_error(self): class MinimalAdapter(BaseCloudDriveAdapter): PLATFORM_KEY = "minimal" def verify(self, share_url): raise NotImplementedError def _get_share_detail(self, pwd_id, passcode=""): raise NotImplementedError def _save_files(self, pwd_id, detail, save_dir): raise NotImplementedError def _create_share(self, file_ids, title, password=""): raise NotImplementedError def get_files(self, parent_fid="0"): raise NotImplementedError def delete(self, file_ids): raise NotImplementedError adapter = MinimalAdapter(PlatformConfig(), TransferConfig()) with self.assertRaises(TransferError): adapter.ensure_dir("/Media") with self.assertRaises(TransferError): adapter.move_files(["fid"], "target") if __name__ == "__main__": unittest.main()