feat: integrate Quark and UC drive APIs
Add optional drive API capabilities for Quark and UC adapters, including directory lookup/creation, rename, move, delete, task polling, Quark recycle cleanup, and UC share staging folder support.\n\nAdd unittest coverage for capability declarations, production transfer path save_dir resolution, staging-folder flow, delete_files behavior, task polling params, and HTTP/JSON error handling.\n\nDocument the Hong Kong test server deployment boundary and verification commands.
This commit is contained in:
@@ -76,6 +76,18 @@ class BaseCloudDriveAdapter(ABC):
|
||||
# URL匹配正则（子类覆盖）
|
||||
URL_PATTERNS: List[str] = []
|
||||
|
||||
# 可选 Drive API 能力；子类按需覆盖为 True 并实现对应方法。
|
||||
capabilities: Dict[str, bool] = {
|
||||
"ensure_dir": False,
|
||||
"save_files": False,
|
||||
"poll_task": False,
|
||||
"rename": False,
|
||||
"move_files": False,
|
||||
"delete_files": False,
|
||||
"cleanup_recycle": False,
|
||||
"share_staging_folder": False,
|
||||
}
|
||||
|
||||
# 默认请求头
|
||||
DEFAULT_HEADERS: Dict[str, str] = {
|
||||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||||
@@ -235,6 +247,40 @@ class BaseCloudDriveAdapter(ABC):
|
||||
"""广告过滤（默认不实现，子类可覆盖）"""
|
||||
return file_ids
|
||||
|
||||
|
||||
# ─── Optional Drive API capability protocol ─────────────────────
|
||||
|
||||
def _unsupported_capability(self, capability: str) -> None:
|
||||
raise TransferError(
|
||||
TransferErrorCode.NETWORK_ERROR,
|
||||
message=f"{self.PLATFORM_KEY or self.PLATFORM_NAME} 不支持 Drive API 能力: {capability}",
|
||||
platform=self.PLATFORM_KEY,
|
||||
)
|
||||
|
||||
def ensure_dir(self, dir_path: str) -> str:
|
||||
self._unsupported_capability("ensure_dir")
|
||||
|
||||
def get_fids(self, file_paths: List[str]) -> List[Dict[str, Any]]:
|
||||
self._unsupported_capability("get_fids")
|
||||
|
||||
def mkdir(self, dir_path: str) -> Dict[str, Any]:
|
||||
self._unsupported_capability("mkdir")
|
||||
|
||||
def rename(self, fid: str, file_name: str) -> Dict[str, Any]:
|
||||
self._unsupported_capability("rename")
|
||||
|
||||
def move_files(self, fids: List[str], to_pdir_fid: str) -> Dict[str, Any]:
|
||||
self._unsupported_capability("move_files")
|
||||
|
||||
def delete_files(self, fids: List[str]) -> Dict[str, Any]:
|
||||
self._unsupported_capability("delete_files")
|
||||
|
||||
def query_task(self, task_id: str) -> Dict[str, Any]:
|
||||
self._unsupported_capability("poll_task")
|
||||
|
||||
def cleanup_recycle(self, fids: List[str]) -> Dict[str, Any]:
|
||||
self._unsupported_capability("cleanup_recycle")
|
||||
|
||||
# ─── HTTP 工具方法 ─────────────────────────────────────
|
||||
|
||||
def _get(self, url: str, params: dict = None, headers: dict = None,
|
||||
@@ -271,6 +317,28 @@ class BaseCloudDriveAdapter(ABC):
|
||||
raise TransferError(TransferErrorCode.NETWORK_ERROR,
|
||||
message=str(last_exc), platform=self.PLATFORM_KEY)
|
||||
|
||||
|
||||
def _drive_api_json(self, resp: requests.Response, context: str = "网盘 API") -> Dict[str, Any]:
|
||||
"""Validate HTTP response and decode JSON for drive helper APIs."""
|
||||
try:
|
||||
resp.raise_for_status()
|
||||
except requests.HTTPError as exc:
|
||||
raise TransferError(
|
||||
TransferErrorCode.NETWORK_ERROR,
|
||||
message=f"{context} HTTP错误: {exc}",
|
||||
platform=self.PLATFORM_KEY,
|
||||
details={"status_code": getattr(resp, "status_code", None)},
|
||||
) from exc
|
||||
try:
|
||||
return resp.json()
|
||||
except ValueError as exc:
|
||||
text = getattr(resp, "text", "") or ""
|
||||
raise TransferError(
|
||||
TransferErrorCode.NETWORK_ERROR,
|
||||
message=f"{context} JSON解析失败: {text[:200]}",
|
||||
platform=self.PLATFORM_KEY,
|
||||
) from exc
|
||||
|
||||
def _poll_task(self, task_url: str, task_id: str,
|
||||
status_field: str = "status",
|
||||
success_value: Any = 2,
|
||||
@@ -292,10 +360,12 @@ class BaseCloudDriveAdapter(ABC):
|
||||
details={"task_id": task_id})
|
||||
|
||||
try:
|
||||
params = query_params or {}
|
||||
base_params = query_params(attempt) if callable(query_params) else (query_params or {})
|
||||
params = dict(base_params)
|
||||
params["task_id"] = task_id
|
||||
resp = self._get(task_url, params=params, retry=1)
|
||||
data = resp.json().get("data", resp.json())
|
||||
payload = resp.json()
|
||||
data = payload.get("data", payload)
|
||||
|
||||
current_status = data.get(status_field)
|
||||
if current_status == success_value:
|
||||
|
||||
@@ -55,19 +55,21 @@ class QuarkAdapter(BaseCloudDriveAdapter):
|
||||
r"pan\.quark\.cn/s/(\w+)",
|
||||
]
|
||||
|
||||
|
||||
capabilities: Dict[str, bool] = {
|
||||
"ensure_dir": True,
|
||||
"save_files": True,
|
||||
"poll_task": True,
|
||||
"rename": True,
|
||||
"move_files": True,
|
||||
"delete_files": True,
|
||||
"cleanup_recycle": True,
|
||||
"share_staging_folder": False,
|
||||
}
|
||||
|
||||
def __init__(self, config: PlatformConfig, transfer_config: TransferConfig) -> None:
|
||||
"""初始化夸克适配器。
|
||||
|
||||
Args:
|
||||
config: 平台配置（含 Cookie 等）。
|
||||
transfer_config: 全局转存配置（超时、重试、轮询参数等）。
|
||||
"""
|
||||
super().__init__(config, transfer_config)
|
||||
|
||||
# 初始化三个子模块
|
||||
self._credential: QuarkCredentialManager = QuarkCredentialManager(
|
||||
cookie=config.cookie
|
||||
)
|
||||
"""初始化适配器。"""
|
||||
self._credential: QuarkCredentialManager = QuarkCredentialManager(cookie=config.cookie)
|
||||
self._transfer_engine: QuarkTransfer = QuarkTransfer(
|
||||
credential=self._credential,
|
||||
timeout=transfer_config.request_timeout,
|
||||
@@ -78,6 +80,7 @@ class QuarkAdapter(BaseCloudDriveAdapter):
|
||||
credential=self._credential,
|
||||
timeout=transfer_config.request_timeout,
|
||||
)
|
||||
super().__init__(config, transfer_config)
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════
|
||||
# 公开接口实现
|
||||
@@ -116,8 +119,9 @@ class QuarkAdapter(BaseCloudDriveAdapter):
|
||||
platform=self.PLATFORM_KEY,
|
||||
)
|
||||
|
||||
# 目标目录：默认根目录 "0"
|
||||
target_dir: str = save_dir or self.config.save_dir or "0"
|
||||
# 目标目录：支持 fid 或路径；路径会先创建/解析为 fid。
|
||||
requested_dir: str = save_dir or self.config.save_dir or "/"
|
||||
target_dir: str = requested_dir if requested_dir and not str(requested_dir).startswith("/") else self.ensure_dir(requested_dir)
|
||||
|
||||
# 分享密码
|
||||
pwd: str = share_password or self.config.share_password or ""
|
||||
@@ -250,18 +254,13 @@ class QuarkAdapter(BaseCloudDriveAdapter):
|
||||
def _save_files(self, pwd_id: str, detail: dict, save_dir: str) -> List[str]:
|
||||
"""转存文件到自己的夸克网盘（基类 transfer() 流程中的步骤③④）。
|
||||
|
||||
Args:
|
||||
pwd_id: 分享 ID。
|
||||
detail: 分享详情（来自 _get_share_detail）。
|
||||
save_dir: 目标目录 ID。
|
||||
|
||||
Returns:
|
||||
转存后的新文件 ID 列表。
|
||||
save_dir 可传 fid 或路径；路径会先通过 ensure_dir 创建/解析为 fid。
|
||||
"""
|
||||
# 需要 stoken，从 detail 间接获取（重新请求）
|
||||
stoken: str = self._transfer_engine._get_stoken(pwd_id)
|
||||
target_fid = save_dir if save_dir and not str(save_dir).startswith("/") else self.ensure_dir(save_dir or "/")
|
||||
task_id: str = self._transfer_engine._init_save(
|
||||
pwd_id, stoken, detail, to_pdir_fid=save_dir
|
||||
pwd_id, stoken, detail, to_pdir_fid=target_fid
|
||||
)
|
||||
return self._transfer_engine._poll_save_task(task_id)
|
||||
|
||||
@@ -351,6 +350,160 @@ class QuarkAdapter(BaseCloudDriveAdapter):
|
||||
logger.warning("[QuarkAdapter] Cannot fetch file list for ad filtering, skipping")
|
||||
return file_ids
|
||||
|
||||
|
||||
# ─── Drive API capability helpers ─────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _normalize_dir_path(dir_path: str) -> str:
|
||||
path = "/" + str(dir_path or "").strip().strip("/")
|
||||
return "/" if path == "/" else path
|
||||
|
||||
@staticmethod
|
||||
def _api_success(payload: Dict[str, Any]) -> bool:
|
||||
if not isinstance(payload, dict):
|
||||
return False
|
||||
code = payload.get("code")
|
||||
status = payload.get("status")
|
||||
return code == 0 or status in (0, 200)
|
||||
|
||||
def ensure_dir(self, dir_path: str) -> str:
|
||||
normalized = self._normalize_dir_path(dir_path)
|
||||
if normalized == "/":
|
||||
return "0"
|
||||
|
||||
parts = [part for part in normalized.strip("/").split("/") if part]
|
||||
prefixes = ["/" + "/".join(parts[:idx]) for idx in range(1, len(parts) + 1)]
|
||||
existing = {
|
||||
item.get("file_path"): str(item.get("fid"))
|
||||
for item in self.get_fids(prefixes)
|
||||
if item.get("file_path") and item.get("fid")
|
||||
}
|
||||
|
||||
leaf_fid = existing.get(normalized)
|
||||
if leaf_fid:
|
||||
return leaf_fid
|
||||
|
||||
for prefix in prefixes:
|
||||
if prefix in existing:
|
||||
continue
|
||||
result = self.mkdir(prefix)
|
||||
if self._api_success(result) and result.get("data", {}).get("fid"):
|
||||
existing[prefix] = str(result["data"]["fid"])
|
||||
continue
|
||||
raise TransferError(
|
||||
TransferErrorCode.NETWORK_ERROR,
|
||||
message=f"创建目录失败: {result.get('message', result)}",
|
||||
platform=self.PLATFORM_KEY,
|
||||
)
|
||||
return existing[normalized]
|
||||
|
||||
def mkdir(self, dir_path: str) -> Dict[str, Any]:
|
||||
url = "https://drive-pc.quark.cn/1/clouddrive/file"
|
||||
params = {"pr": "ucpro", "fr": "pc", "uc_param_str": ""}
|
||||
payload = {
|
||||
"pdir_fid": "0",
|
||||
"file_name": "",
|
||||
"dir_path": self._normalize_dir_path(dir_path),
|
||||
"dir_init_lock": False,
|
||||
}
|
||||
return self._drive_api_json(self._post(url, json_data=payload, params=params, headers=self._credential.get_headers()), context="写入网盘目录")
|
||||
|
||||
def rename(self, fid: str, file_name: str) -> Dict[str, Any]:
|
||||
url = "https://drive-pc.quark.cn/1/clouddrive/file/rename"
|
||||
params = {"pr": "ucpro", "fr": "pc", "uc_param_str": ""}
|
||||
payload = {"fid": fid, "file_name": file_name}
|
||||
return self._drive_api_json(self._post(url, json_data=payload, params=params, headers=self._credential.get_headers()), context="写入网盘目录")
|
||||
|
||||
def get_fids(self, file_paths: List[str]) -> List[Dict[str, Any]]:
|
||||
pending = [self._normalize_dir_path(p) for p in file_paths]
|
||||
result: List[Dict[str, Any]] = []
|
||||
while pending:
|
||||
batch, pending = pending[:50], pending[50:]
|
||||
url = "https://drive-pc.quark.cn/1/clouddrive/file/info/path_list"
|
||||
params = {"pr": "ucpro", "fr": "pc"}
|
||||
payload = {"file_path": batch, "namespace": "0"}
|
||||
data = self._drive_api_json(self._post(url, json_data=payload, params=params, headers=self._credential.get_headers()), context="按路径获取文件ID")
|
||||
if not self._api_success(data):
|
||||
raise TransferError(
|
||||
TransferErrorCode.NETWORK_ERROR,
|
||||
message=f"获取目录ID失败: {data.get('message', data)}",
|
||||
platform=self.PLATFORM_KEY,
|
||||
)
|
||||
result.extend(data.get("data", []))
|
||||
return result
|
||||
|
||||
def move_files(self, fids: List[str], to_pdir_fid: str) -> Dict[str, Any]:
|
||||
if not fids:
|
||||
return {"code": 0, "message": "无文件需要移动"}
|
||||
last: Dict[str, Any] = {"code": 0, "message": "success"}
|
||||
for offset in range(0, len(fids), 100):
|
||||
batch = fids[offset:offset + 100]
|
||||
url = "https://drive-pc.quark.cn/1/clouddrive/file/move"
|
||||
params = {"uc_param_str": "", "fr": "pc", "pr": "ucpro"}
|
||||
payload = {"filelist": batch, "to_pdir_fid": to_pdir_fid, "exclude_fids": [], "action_type": 1}
|
||||
last = self._drive_api_json(self._post(url, json_data=payload, params=params, headers=self._credential.get_headers()), context="移动网盘文件")
|
||||
if not self._api_success(last):
|
||||
return last
|
||||
task_id = last.get("data", {}).get("task_id")
|
||||
if task_id:
|
||||
task = self.query_task(task_id)
|
||||
if not self._api_success(task) or task.get("data", {}).get("status") == -1:
|
||||
return {"code": 1, "message": task.get("data", {}).get("message", task.get("message", "移动任务失败")), "data": task.get("data", {})}
|
||||
return {"code": 0, "message": "移动完成", "data": last.get("data", {})}
|
||||
|
||||
def _task_query_params(self, retry_index: int = 0) -> Dict[str, Any]:
|
||||
now_ms = int(time.time() * 1000)
|
||||
return {
|
||||
"pr": "ucpro",
|
||||
"fr": "pc",
|
||||
"uc_param_str": "",
|
||||
"retry_index": retry_index,
|
||||
"__dt": 300,
|
||||
"__t": now_ms,
|
||||
}
|
||||
|
||||
def delete_files(self, fids: List[str]) -> Dict[str, Any]:
|
||||
if not fids:
|
||||
return {"code": 0, "status": 200}
|
||||
if self.delete(fids):
|
||||
return {"code": 0, "status": 200}
|
||||
return {"code": 1, "status": 500, "message": "删除文件失败"}
|
||||
|
||||
def query_task(self, task_id: str) -> Dict[str, Any]:
|
||||
url = "https://drive-pc.quark.cn/1/clouddrive/task"
|
||||
try:
|
||||
data = self._poll_task(url, task_id, query_params=self._task_query_params)
|
||||
return {"code": 0, "status": 200, "data": data}
|
||||
except TransferError as exc:
|
||||
return {"code": 1, "status": 500, "message": str(exc), "data": {"status": -1}}
|
||||
|
||||
def recycle_list(self, page: int = 1, size: int = 30) -> List[Dict[str, Any]]:
|
||||
url = "https://drive-pc.quark.cn/1/clouddrive/file/recycle/list"
|
||||
params = {"_page": page, "_size": size, "pr": "ucpro", "fr": "pc", "uc_param_str": ""}
|
||||
data = self._drive_api_json(self._get(url, params=params, headers=self._credential.get_headers()), context="列出回收站")
|
||||
return data.get("data", {}).get("list", [])
|
||||
|
||||
def recycle_remove(self, record_list: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
url = "https://drive-pc.quark.cn/1/clouddrive/file/recycle/remove"
|
||||
params = {"uc_param_str": "", "fr": "pc", "pr": "ucpro"}
|
||||
payload = {"select_mode": 2, "record_list": record_list}
|
||||
return self._drive_api_json(self._post(url, json_data=payload, params=params, headers=self._credential.get_headers()), context="写入网盘目录")
|
||||
|
||||
def cleanup_recycle(self, fids: List[str]) -> Dict[str, Any]:
|
||||
target_fids = {str(fid) for fid in fids if fid}
|
||||
if not target_fids:
|
||||
return {"code": 0, "message": "无回收站记录需要清理", "data": {"removed": 0}}
|
||||
records = [
|
||||
item for item in self.recycle_list()
|
||||
if str(item.get("fid") or item.get("file_id") or "") in target_fids
|
||||
]
|
||||
if not records:
|
||||
return {"code": 0, "message": "未找到匹配的回收站记录", "data": {"removed": 0}}
|
||||
result = self.recycle_remove(records)
|
||||
if self._api_success(result):
|
||||
result.setdefault("data", {})["removed"] = len(records)
|
||||
return result
|
||||
|
||||
# ─── get_files / delete ────────────────────────────────────
|
||||
|
||||
def get_files(self, parent_fid: str = "0") -> List[FileInfo]:
|
||||
|
||||
@@ -56,19 +56,21 @@ class UcAdapter(BaseCloudDriveAdapter):
|
||||
r"drive\.uc\.cn/s/(\w+)",
|
||||
]
|
||||
|
||||
|
||||
capabilities: Dict[str, bool] = {
|
||||
"ensure_dir": True,
|
||||
"save_files": True,
|
||||
"poll_task": True,
|
||||
"rename": True,
|
||||
"move_files": True,
|
||||
"delete_files": True,
|
||||
"cleanup_recycle": False,
|
||||
"share_staging_folder": True,
|
||||
}
|
||||
|
||||
def __init__(self, config: PlatformConfig, transfer_config: TransferConfig) -> None:
|
||||
"""初始化 UC 适配器。
|
||||
|
||||
Args:
|
||||
config: 平台配置（含 Cookie 等）。
|
||||
transfer_config: 全局转存配置（超时、重试、轮询参数等）。
|
||||
"""
|
||||
super().__init__(config, transfer_config)
|
||||
|
||||
# 初始化三个子模块
|
||||
self._credential: UcCredentialManager = UcCredentialManager(
|
||||
cookie=config.cookie
|
||||
)
|
||||
"""初始化适配器。"""
|
||||
self._credential: UcCredentialManager = UcCredentialManager(cookie=config.cookie)
|
||||
self._transfer_engine: UcTransfer = UcTransfer(
|
||||
credential=self._credential,
|
||||
timeout=transfer_config.request_timeout,
|
||||
@@ -79,6 +81,7 @@ class UcAdapter(BaseCloudDriveAdapter):
|
||||
credential=self._credential,
|
||||
timeout=transfer_config.request_timeout,
|
||||
)
|
||||
super().__init__(config, transfer_config)
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════
|
||||
# 公开接口实现
|
||||
@@ -93,21 +96,8 @@ class UcAdapter(BaseCloudDriveAdapter):
|
||||
|
||||
def transfer(self, share_url: str, save_dir: str = "",
|
||||
share_password: str = "") -> TransferResult:
|
||||
"""执行转存的核心逻辑（覆盖基类实现 UC 专用流程）。
|
||||
|
||||
通过 UcTransfer 引擎执行完整的 7 步流程。
|
||||
|
||||
Args:
|
||||
share_url: UC 分享链接。
|
||||
save_dir: 目标目录，空则使用配置的默认目录。
|
||||
share_password: 新分享的密码。
|
||||
|
||||
Returns:
|
||||
TransferResult 包含转存结果。
|
||||
"""
|
||||
start: float = time.time()
|
||||
|
||||
# 凭证检查
|
||||
if not self._credential.validate():
|
||||
raise TransferError(
|
||||
TransferErrorCode.NOT_LOGIN,
|
||||
@@ -115,18 +105,36 @@ class UcAdapter(BaseCloudDriveAdapter):
|
||||
platform=self.PLATFORM_KEY,
|
||||
)
|
||||
|
||||
# 目标目录：默认根目录 "0"
|
||||
target_dir: str = save_dir or self.config.save_dir or "0"
|
||||
|
||||
# 分享密码
|
||||
requested_dir: str = save_dir or self.config.save_dir or "/"
|
||||
target_dir: str = requested_dir if requested_dir and not str(requested_dir).startswith("/") else self.ensure_dir(requested_dir)
|
||||
staging_dir: str = self.get_or_create_share_folder() or target_dir
|
||||
pwd: str = share_password or self.config.share_password or ""
|
||||
|
||||
try:
|
||||
result: Dict[str, Any] = self._transfer_engine.transfer(
|
||||
share_url=share_url,
|
||||
save_dir=target_dir,
|
||||
share_password=pwd,
|
||||
pwd_id, passcode = self._parse_share_url(share_url)
|
||||
stoken: str = self._transfer_engine._get_stoken(pwd_id, passcode)
|
||||
detail: Dict[str, Any] = self._transfer_engine._get_detail(pwd_id, stoken)
|
||||
task_id: str = self._transfer_engine._init_save(
|
||||
pwd_id, stoken, detail, to_pdir_fid=staging_dir
|
||||
)
|
||||
new_fids: List[str] = self._transfer_engine._poll_save_task(task_id)
|
||||
if not new_fids:
|
||||
raise RuntimeError("转存完成但未获取到文件ID")
|
||||
|
||||
if staging_dir != target_dir:
|
||||
move_result = self.move_files(new_fids, target_dir)
|
||||
if not self._api_success(move_result):
|
||||
raise RuntimeError(f"移动到目标目录失败: {move_result.get('message', move_result)}")
|
||||
|
||||
if self.transfer_config.ad_filter_enabled and new_fids:
|
||||
new_fids = self._filter_ads(new_fids)
|
||||
if not new_fids:
|
||||
raise RuntimeError("广告过滤后无可分享文件")
|
||||
|
||||
title: str = detail.get("title", "分享")
|
||||
share_task_id: str = self._transfer_engine._init_share(new_fids, title)
|
||||
share_id: str = self._transfer_engine._poll_share_task(share_task_id)
|
||||
share_url_new, passcode_new = self._transfer_engine._set_password(share_id, pwd)
|
||||
except ValueError as exc:
|
||||
raise TransferError(
|
||||
TransferErrorCode.URL_INVALID,
|
||||
@@ -148,24 +156,13 @@ class UcAdapter(BaseCloudDriveAdapter):
|
||||
) from exc
|
||||
|
||||
elapsed: int = int((time.time() - start) * 1000)
|
||||
|
||||
# 广告过滤
|
||||
new_fids: List[str] = result.get("new_file_ids", [])
|
||||
if self.transfer_config.ad_filter_enabled and new_fids:
|
||||
new_fids = self._filter_ads(new_fids)
|
||||
if not new_fids:
|
||||
raise TransferError(
|
||||
TransferErrorCode.RESOURCE_EMPTY,
|
||||
platform=self.PLATFORM_KEY,
|
||||
)
|
||||
|
||||
return TransferResult(
|
||||
success=True,
|
||||
platform=self.PLATFORM_KEY,
|
||||
new_file_id=",".join(new_fids),
|
||||
file_name=result.get("file_name", ""),
|
||||
share_url=result.get("share_url", ""),
|
||||
share_password=result.get("passcode", pwd),
|
||||
file_name=title,
|
||||
share_url=share_url_new,
|
||||
share_password=passcode_new,
|
||||
original_url=share_url,
|
||||
elapsed_ms=elapsed,
|
||||
)
|
||||
@@ -204,8 +201,8 @@ class UcAdapter(BaseCloudDriveAdapter):
|
||||
files=files,
|
||||
)
|
||||
|
||||
except TransferError:
|
||||
raise
|
||||
except TransferError as exc:
|
||||
return VerifyResult(valid=False, platform=self.PLATFORM_KEY, error=exc)
|
||||
except (ValueError, RuntimeError) as exc:
|
||||
return VerifyResult(
|
||||
valid=False,
|
||||
@@ -344,6 +341,154 @@ class UcAdapter(BaseCloudDriveAdapter):
|
||||
)
|
||||
return file_ids
|
||||
|
||||
|
||||
# ─── Drive API capability helpers ─────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _normalize_dir_path(dir_path: str) -> str:
|
||||
path = "/" + str(dir_path or "").strip().strip("/")
|
||||
return "/" if path == "/" else path
|
||||
|
||||
@staticmethod
|
||||
def _api_success(payload: Dict[str, Any]) -> bool:
|
||||
if not isinstance(payload, dict):
|
||||
return False
|
||||
code = payload.get("code")
|
||||
status = payload.get("status")
|
||||
return code == 0 or status in (0, 200)
|
||||
|
||||
def ensure_dir(self, dir_path: str) -> str:
|
||||
normalized = self._normalize_dir_path(dir_path)
|
||||
if normalized == "/":
|
||||
return "0"
|
||||
|
||||
parts = [part for part in normalized.strip("/").split("/") if part]
|
||||
prefixes = ["/" + "/".join(parts[:idx]) for idx in range(1, len(parts) + 1)]
|
||||
existing = {
|
||||
item.get("file_path"): str(item.get("fid"))
|
||||
for item in self.get_fids(prefixes)
|
||||
if item.get("file_path") and item.get("fid")
|
||||
}
|
||||
|
||||
leaf_fid = existing.get(normalized)
|
||||
if leaf_fid:
|
||||
return leaf_fid
|
||||
|
||||
for prefix in prefixes:
|
||||
if prefix in existing:
|
||||
continue
|
||||
result = self.mkdir(prefix)
|
||||
if self._api_success(result) and result.get("data", {}).get("fid"):
|
||||
existing[prefix] = str(result["data"]["fid"])
|
||||
continue
|
||||
raise TransferError(
|
||||
TransferErrorCode.NETWORK_ERROR,
|
||||
message=f"创建目录失败: {result.get('message', result)}",
|
||||
platform=self.PLATFORM_KEY,
|
||||
)
|
||||
return existing[normalized]
|
||||
|
||||
def mkdir(self, dir_path: str) -> Dict[str, Any]:
|
||||
url = "https://pc-api.uc.cn/1/clouddrive/file"
|
||||
params = {"pr": "UCBrowser", "fr": "pc", "uc_param_str": ""}
|
||||
payload = {
|
||||
"pdir_fid": "0",
|
||||
"file_name": "",
|
||||
"dir_path": self._normalize_dir_path(dir_path),
|
||||
"dir_init_lock": False,
|
||||
}
|
||||
return self._drive_api_json(self._post(url, json_data=payload, params=params, headers=self._credential.get_headers()), context="写入网盘目录")
|
||||
|
||||
def rename(self, fid: str, file_name: str) -> Dict[str, Any]:
|
||||
url = "https://pc-api.uc.cn/1/clouddrive/file/rename"
|
||||
params = {"pr": "UCBrowser", "fr": "pc", "uc_param_str": ""}
|
||||
payload = {"fid": fid, "file_name": file_name}
|
||||
return self._drive_api_json(self._post(url, json_data=payload, params=params, headers=self._credential.get_headers()), context="写入网盘目录")
|
||||
|
||||
def get_fids(self, file_paths: List[str]) -> List[Dict[str, Any]]:
|
||||
pending = [self._normalize_dir_path(p) for p in file_paths]
|
||||
result: List[Dict[str, Any]] = []
|
||||
while pending:
|
||||
batch, pending = pending[:50], pending[50:]
|
||||
url = "https://pc-api.uc.cn/1/clouddrive/file/info/path_list"
|
||||
params = {"pr": "UCBrowser", "fr": "pc"}
|
||||
payload = {"file_path": batch, "namespace": "0"}
|
||||
data = self._drive_api_json(self._post(url, json_data=payload, params=params, headers=self._credential.get_headers()), context="按路径获取文件ID")
|
||||
if not self._api_success(data):
|
||||
raise TransferError(
|
||||
TransferErrorCode.NETWORK_ERROR,
|
||||
message=f"获取目录ID失败: {data.get('message', data)}",
|
||||
platform=self.PLATFORM_KEY,
|
||||
)
|
||||
result.extend(data.get("data", []))
|
||||
return result
|
||||
|
||||
def move_files(self, fids: List[str], to_pdir_fid: str) -> Dict[str, Any]:
|
||||
if not fids:
|
||||
return {"code": 0, "message": "无文件需要移动"}
|
||||
last: Dict[str, Any] = {"code": 0, "message": "success"}
|
||||
for offset in range(0, len(fids), 100):
|
||||
batch = fids[offset:offset + 100]
|
||||
url = "https://pc-api.uc.cn/1/clouddrive/file/move"
|
||||
params = {"uc_param_str": "", "fr": "pc", "pr": "UCBrowser"}
|
||||
payload = {"filelist": batch, "to_pdir_fid": to_pdir_fid, "exclude_fids": [], "action_type": 1}
|
||||
last = self._drive_api_json(self._post(url, json_data=payload, params=params, headers=self._credential.get_headers()), context="移动网盘文件")
|
||||
if not self._api_success(last):
|
||||
return last
|
||||
task_id = last.get("data", {}).get("task_id")
|
||||
if task_id:
|
||||
task = self.query_task(task_id)
|
||||
if not self._api_success(task) or task.get("data", {}).get("status") == -1:
|
||||
return {"code": 1, "message": task.get("data", {}).get("message", task.get("message", "移动任务失败")), "data": task.get("data", {})}
|
||||
return {"code": 0, "message": "移动完成", "data": last.get("data", {})}
|
||||
|
||||
def _task_query_params(self, retry_index: int = 0) -> Dict[str, Any]:
|
||||
now_ms = int(time.time() * 1000)
|
||||
return {
|
||||
"pr": "UCBrowser",
|
||||
"fr": "pc",
|
||||
"uc_param_str": "",
|
||||
"retry_index": retry_index,
|
||||
"__dt": 300,
|
||||
"__t": now_ms,
|
||||
}
|
||||
|
||||
def delete_files(self, fids: List[str]) -> Dict[str, Any]:
|
||||
if not fids:
|
||||
return {"code": 0, "status": 200}
|
||||
if self.delete(fids):
|
||||
return {"code": 0, "status": 200}
|
||||
return {"code": 1, "status": 500, "message": "删除文件失败"}
|
||||
|
||||
def query_task(self, task_id: str) -> Dict[str, Any]:
|
||||
url = "https://pc-api.uc.cn/1/clouddrive/task"
|
||||
try:
|
||||
data = self._poll_task(url, task_id, query_params=self._task_query_params)
|
||||
return {"code": 0, "status": 200, "data": data}
|
||||
except TransferError as exc:
|
||||
return {"code": 1, "status": 500, "message": str(exc), "data": {"status": -1}}
|
||||
|
||||
def get_or_create_share_folder(self) -> Optional[str]:
|
||||
if getattr(self, "_share_folder_fid", None):
|
||||
return self._share_folder_fid
|
||||
root = self.ls_dir("0")
|
||||
if self._api_success(root):
|
||||
for item in root.get("data", {}).get("list", []):
|
||||
if item.get("file_name") == "来自：分享" and item.get("dir"):
|
||||
self._share_folder_fid = str(item["fid"])
|
||||
return self._share_folder_fid
|
||||
result = self.mkdir("/来自：分享")
|
||||
if self._api_success(result) and result.get("data", {}).get("fid"):
|
||||
self._share_folder_fid = str(result["data"]["fid"])
|
||||
return self._share_folder_fid
|
||||
return None
|
||||
|
||||
def ls_dir(self, pdir_fid: str) -> Dict[str, Any]:
|
||||
url = "https://pc-api.uc.cn/1/clouddrive/file/sort"
|
||||
params = {"pr": "UCBrowser", "fr": "pc", "pdir_fid": pdir_fid or "0", "_page": 1, "_size": 50, "_fetch_total": 1, "_fetch_sub_dirs": 0, "_sort": "file_type:asc,updated_at:desc"}
|
||||
return self._drive_api_json(self._get(url, params=params, headers=self._credential.get_headers()), context="列出网盘目录")
|
||||
|
||||
|
||||
# ─── get_files / delete ────────────────────────────────────
|
||||
|
||||
def get_files(self, parent_fid: str = "0") -> List[FileInfo]:
|
||||
|
||||
@@ -0,0 +1,308 @@
|
||||
|
||||
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 = "<html>bad gateway</html>"
|
||||
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 = "<html>bad gateway</html>"
|
||||
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()
|
||||
Reference in New Issue
Block a user