From 6766a0516ef1712a5004dd33ee9bd63800a21f3a Mon Sep 17 00:00:00 2001 From: jxxghp Date: Tue, 20 Oct 2026 17:33:51 +0800 Subject: [PATCH] fix(transfer): recover legacy 115 receipts and isolate cloud copies Validate old instant-upload receipts against frozen source bytes and the original cloud file ID before source deletion. Isolate native 115 copies before renaming, verify moved object IDs, and quarantine uncertain native operations instead of replaying them. Require explicit delete success. Refs #7099 --- app/modules/filemanager/storages/u115.py | 192 ++++++++++----- app/modules/filemanager/transhandler.py | 26 +++ .../fixtures/architecture/mypy-baseline.json | 7 +- tests/test_transfer_execution_runner.py | 113 +++++++++ tests/test_transfer_safe_overwrite.py | 220 ++++++++++++++++++ tests/test_u115_storage.py | 12 + 6 files changed, 505 insertions(+), 65 deletions(-) diff --git a/app/modules/filemanager/storages/u115.py b/app/modules/filemanager/storages/u115.py index 389815a02..8e3cf8eb0 100644 --- a/app/modules/filemanager/storages/u115.py +++ b/app/modules/filemanager/storages/u115.py @@ -1048,16 +1048,14 @@ class U115Pan(StorageBase, metaclass=WeakSingleton): """ 向 115 提交文件或目录删除请求。 - 当前实现未检查空响应或业务错误，因此 True 不保证目标已经删除。 - :param fileitem: 带远端文件标识的待删除项 - :return: 没有抛出 HTTP 请求异常时为 True，否则为 False + :return: 接口明确返回成功时为 True；请求失败或业务拒绝时为 False """ try: - self._request_api( + resp = self._request_api( "POST", "/open/ufile/delete", data={"file_ids": int(fileitem.fileid)} ) - return True + return bool(isinstance(resp, dict) and resp.get("state")) except HttpRequestError: return False @@ -1200,82 +1198,154 @@ class U115Pan(StorageBase, metaclass=WeakSingleton): def copy(self, fileitem: _SchemaFileItem, path: Path, new_name: str, **_kwargs: Any) -> bool: """ - 复制到已存在的目录，再重命名为目标名称。 - - 源标识补查失败的日志分支仍可能抛出异常；返回 False 也可能已经完成复制。 + 兼容布尔复制接口，使用隔离暂存目录避免操作目标目录中的同名旧文件。 :param fileitem: 源文件项；缺少远端标识时先按路径查询 :param path: 已存在的目标目录路径 :param new_name: 复制后的名称 - :return: 复制、查询新文件和重命名均成功时为 True，否则为 False + :return: 完成复制、改名和目标身份核验时为 True，否则为 False """ - if fileitem.fileid is None: - fileitem = self.get_item(Path(fileitem.path)) - if not fileitem: - logger.warn(f"【115】获取文件 {fileitem.path} 失败！") - return False - dest_fileitem = self.get_item(path) - if not dest_fileitem or dest_fileitem.type != "dir": - logger.warn(f"【115】目标路径 {path} 不是一个有效的目录！") + try: + return self.copy_item(fileitem, path, new_name) is not None + except StorageQueryError as error: + logger.warn(f"【115】复制未完成: {error}") return False + def copy_item( + self, fileitem: _SchemaFileItem, path: Path, new_name: str, + ) -> Optional[_SchemaFileItem]: + """ + 在独立目录内复制和改名，再按文件 ID 移入目标，返回严格确认的文件项。 + + 115 复制接口先保留源名；直接复制进媒体目录可能生成编号副本，按原名查询还 + 可能误改已有文件。暂存目录使用随机名称，异常时保留现场并交由人工复核。 + 复制请求不做 HTTP 层自动重放，空目录清理失败不撤销已经确认的传输成功。 + + :param fileitem: 带大小和远端文件 ID 的源文件 + :param path: 已存在的目标目录 + :param new_name: 最终文件名，已有同名目标时拒绝操作 + :return: 最终文件项；明确失败时为 None + :raises StorageQueryError: 复制结果缺少身份或无法确认，保留暂存文件 + """ + destination = self.get_item_strict(path) + if not destination or not destination.fileid or destination.type != "dir": + return None + if self.get_item_strict(path / new_name) is not None: + return None + if not fileitem.path: + raise StorageQueryError("【115】复制源缺少路径") + source = self.get_item_strict(Path(fileitem.path)) + if (not source or not source.fileid or not source.name + or source.type == "file" and source.size is None): + raise StorageQueryError(f"【115】复制源身份无法确认: {fileitem.path}") + if fileitem.fileid and fileitem.fileid != source.fileid: + raise StorageQueryError(f"【115】复制源已被替换: {fileitem.path}") + if source.size != fileitem.size or source.type != fileitem.type: + raise StorageQueryError(f"【115】复制源大小或类型已变化: {fileitem.path}") + staging = self.create_folder(destination, f"mp-transfer-{secrets.token_hex(16)}") + if not staging or not staging.fileid or not staging.path: + return None + if self.list(staging): + raise StorageQueryError(f"【115】复制暂存目录非空，保留现场: {staging.path}") resp = self._request_api( - "POST", - "/open/ufile/copy", - data={ - "file_id": int(fileitem.fileid), - "pid": int(dest_fileitem.fileid), - }, + "POST", "/open/ufile/copy", retry_limit=0, + data={"file_id": int(source.fileid), "pid": int(staging.fileid)}, ) - if not resp: - return False - if resp["state"]: - new_path = Path(path) / fileitem.name - new_item = self.get_item(new_path) - if not new_item: - return False - if self.rename(new_item, new_name): - return True - return False + if not isinstance(resp, dict) or not resp.get("state"): + raise StorageQueryError(f"【115】复制结果无法确认，保留暂存目录: {staging.path}") + copied = self.get_item_strict(Path(staging.path) / source.name) + if (not copied or not copied.fileid or copied.fileid == source.fileid + or copied.size != source.size or copied.type != source.type): + raise StorageQueryError(f"【115】复制文件身份无法确认，保留暂存目录: {staging.path}") + if copied.name != new_name: + if not self.rename(copied, new_name): + raise StorageQueryError(f"【115】复制后改名失败，保留暂存目录: {staging.path}") + copied = self.__confirm_transfer_item(copied, Path(staging.path) / new_name) + result = self.move_item(copied, path, new_name) + if result is None: + raise StorageQueryError(f"【115】复制后改名或移动未完成，保留暂存目录: {staging.path}") + self.__cleanup_transfer_directory(staging) + return result + + def __cleanup_transfer_directory(self, staging: _SchemaFileItem) -> None: + """ + 只清理已经移出成功副本的空目录，清理失败不得再次执行复制。 + + :param staging: 本次复制创建的独立暂存目录；非空或查询失败时保留 + """ + try: + if self.list(staging) or not self.delete(staging): + logger.warn(f"【115】文件整理完成，暂存目录未清理: {staging.path}") + except Exception as error: + logger.warn(f"【115】文件整理完成，暂存目录清理失败: {staging.path} - {error}") def move(self, fileitem: _SchemaFileItem, path: Path, new_name: str) -> bool: """ - 移动到已存在的目录，再重命名为目标名称。 - - 源标识补查失败的日志分支仍可能抛出异常；返回 False 也可能已经完成移动。 + 兼容布尔移动接口，始终按源文件 ID 操作并严格确认最终目标。 :param fileitem: 源文件项；缺少远端标识时先按路径查询 :param path: 已存在的目标目录路径 :param new_name: 移动后的名称 - :return: 移动、查询新文件和重命名均成功时为 True，否则为 False + :return: 移动、改名及目标身份核验成功时为 True，否则为 False """ - if fileitem.fileid is None: - fileitem = self.get_item(Path(fileitem.path)) - if not fileitem: - logger.warn(f"【115】获取文件 {fileitem.path} 失败！") - return False - dest_fileitem = self.get_item(path) - if not dest_fileitem or dest_fileitem.type != "dir": - logger.warn(f"【115】目标路径 {path} 不是一个有效的目录！") + try: + return self.move_item(fileitem, path, new_name) is not None + except StorageQueryError as error: + logger.warn(f"【115】移动未完成: {error}") return False + + def move_item( + self, fileitem: _SchemaFileItem, path: Path, new_name: str, + ) -> Optional[_SchemaFileItem]: + """ + 按 ID 移动和改名，避免查询原名时误操作目标目录中的旧文件。 + + 改名或移动失败可能已有外部副作用，持久整理必须进入人工复核而非自动重放。 + + :param fileitem: 待移动文件的路径和冻结身份 + :param path: 已存在的目标目录 + :param new_name: 最终文件名 + :return: 严格核验 ID、类型和大小后的目标；明确失败时为 None + :raises StorageQueryError: 源或目标身份变化，或无法取得严格查询结果 + """ + if not fileitem.path: + raise StorageQueryError("【115】移动源缺少路径") + source = self.get_item_strict(Path(fileitem.path)) + if not source or not source.fileid: + return None + if fileitem.fileid and fileitem.fileid != source.fileid: + raise StorageQueryError(f"【115】移动源已被替换: {fileitem.path}") + if source.size != fileitem.size or source.type != fileitem.type: + raise StorageQueryError(f"【115】移动源大小或类型已变化: {fileitem.path}") + dest_fileitem = self.get_item_strict(path) + if not dest_fileitem or not dest_fileitem.fileid or dest_fileitem.type != "dir": + return None + if self.get_item_strict(path / new_name) is not None: + return None resp = self._request_api( - "POST", - "/open/ufile/move", - data={ - "file_ids": int(fileitem.fileid), - "to_cid": int(dest_fileitem.fileid), - }, + "POST", "/open/ufile/move", retry_limit=0, + data={"file_ids": int(source.fileid), "to_cid": int(dest_fileitem.fileid)}, ) - if not resp: - return False - if resp["state"]: - new_path = Path(path) / fileitem.name - new_file = self.get_item(new_path) - if not new_file: - return False - if self.rename(new_file, new_name): - return True - return False + if not isinstance(resp, dict) or not resp.get("state"): + return None + if source.name != new_name and not self.rename(source, new_name): + return None + return self.__confirm_transfer_item(source, path / new_name) + + def __confirm_transfer_item(self, source: _SchemaFileItem, path: Path) -> _SchemaFileItem: + """ + 严格确认改名或移动后的对象仍具有原 ID、类型及精确大小。 + + :param source: 操作前已确认的对象身份 + :param path: 操作后应出现的路径 + :return: 严格查询确认的文件项 + :raises StorageQueryError: 目标缺失、被替换或大小类型发生变化 + """ + result = self.get_item_strict(path) + if (not result or result.fileid != source.fileid + or result.type != source.type or result.size != source.size): + raise StorageQueryError(f"【115】传输目标身份无法确认: {path}") + return result def link(self, fileitem: _SchemaFileItem, target_file: Path) -> bool: """ diff --git a/app/modules/filemanager/transhandler.py b/app/modules/filemanager/transhandler.py index b988b4340..8497ed125 100644 --- a/app/modules/filemanager/transhandler.py +++ b/app/modules/filemanager/transhandler.py @@ -827,6 +827,24 @@ class TransHandler: if expected.fileid != current.fileid: raise TransferExecutionConflictError(f"文件 {expected.path} 已被替换，禁止清理") + @staticmethod + def __restore_move_target_identity(source: FileItem, target: FileItem) -> FileItem: + """ + 恢复旧版 115 秒传回执的精确大小，保留原文件 ID 和持久账本不变。 + + 仅本地文件上传到 115 且回执包含大小和文件 ID 时，使用冻结源的字节数。 + 删除前仍须严格核对云端大小、原文件 ID 及本地源身份，不能凭路径接纳替换文件。 + + :param source: 上传前冻结的本地源文件 + :param target: 已持久成功的上传回执，旧版本可能记录舍入后的显示大小 + :return: 用于严格核验和结算的目标身份；证据不完整或其它存储沿用原值 + """ + if (source.storage == "local" and target.storage == "u115" + and source.type == target.type == "file" and target.fileid + and source.size is not None and target.size is not None): + return cast(FileItem, target.model_copy(update={"size": source.size})) + return target + @classmethod def __observe_safe_delete( cls, storage_oper: StorageBase, path: Path, *, @@ -978,6 +996,13 @@ class TransHandler: "target_exists": target_exists, "item": TransHandler.__serialize_step_item(target_item), }) + # 115 盘内操作包含复制/移动及改名。最终路径缺失不能排除中间副本已产生， + # 包括旧版适配器留下的原名或编号副本，禁止据此自动重复复制。 + if fileitem.storage == target_storage == "u115": + return TransferOperationObservation( + state=TransferOperationObservationState.UNKNOWN, + evidence=evidence, + ) if transfer_type == "move": if not source_exists and target_exists: return TransferOperationObservation( @@ -1135,6 +1160,7 @@ class TransHandler: return None, "整理步骤成功证据缺少目标文件" if not cross_storage_move: return new_item, str(materialized.payload.get("message") or "") + new_item = cls.__restore_move_target_identity(fileitem, new_item) def execute_source_delete() -> TransferStepResult: """ diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 0ad7082bc..b334a0d1e 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -662,13 +662,12 @@ "union-attr": 5 }, "app/modules/filemanager/storages/u115.py": { - "arg-type": 16, - "assignment": 5, - "call-overload": 17, + "arg-type": 10, + "assignment": 3, + "call-overload": 15, "empty-body": 2, "no-any-return": 6, "no-untyped-def": 1, - "operator": 2, "override": 1, "type-arg": 8, "union-attr": 32 diff --git a/tests/test_transfer_execution_runner.py b/tests/test_transfer_execution_runner.py index 62001446e..d4ada5903 100644 --- a/tests/test_transfer_execution_runner.py +++ b/tests/test_transfer_execution_runner.py @@ -14,6 +14,7 @@ from sqlalchemy.orm import sessionmaker from app.application.transfer.execution import ( TransferExecutionCommand, TransferExecutionState, + TransferManualReviewDecision, TransferOperationObservation, TransferOperationObservationState, TransferStepIntent, @@ -26,6 +27,7 @@ from app.application.transfer.workflow import ( TransferPlanningInput, ) from app.chain.transfer import execution as transfer_chain_module +from app.db.adapters.transfer.admission import TransactionalTransferAdmissionRepository from app.db.adapters.transfer.execution import ( TransactionalTransferExecutionRepository, ) @@ -35,6 +37,7 @@ from app.db.models.transferhistory import TransferHistory from app.db.models.transferpending import TransferPending from app.domain.context import MediaInfo from app.domain.meta.metabase import MetaBase +from app.modules.filemanager.storages.local import LocalStorage from app.modules.filemanager.transhandler import TransHandler from app.schemas.types import MediaType from app.schemas.workflow import FileItem @@ -455,6 +458,116 @@ def test_cross_storage_move_materializes_before_independent_source_delete(tmp_pa source_oper.delete.assert_called_once_with(live_source) +@pytest.mark.parametrize("change", [ + None, "target_size", "target_id", "target_missing", "source_size", + "receipt_id_missing", "receipt_size_missing", "source_size_missing", "other_storage", + "target_id_missing", "target_type", "target_query", +]) +def test_legacy_u115_receipt_resumes_after_manual_review_without_uploading_again( + tmp_path, execution_repository, change, +): + """ + 真实旧账本经人工确认后只恢复源删除；身份不足、文件变化或其它存储仍隔离。 + """ + source_path = tmp_path / "source.mkv" + source_path.write_bytes(b"movie" * 251) + local = object.__new__(LocalStorage) + source = local.get_item_strict(source_path).model_dump(mode="json") + if change == "source_size_missing": + source["size"] = None + storage = "remote" if change == "other_storage" else "u115" + target = FileItem( + storage=storage, path="/library/source.mkv", type="file", + name="source.mkv", size=1255, fileid="200", + ) + receipt = target.model_copy(update={ + "size": None if change == "receipt_size_missing" else 1254, + "fileid": None if change == "receipt_id_missing" else "200", + }) + planning_input = TransferPlanningInput(source_fileitem=source) + checkpoint = replace( + _runner_plan_checkpoint(), planning_input=planning_input, + target_storage=storage, final_target_path=target.path, resolved_transfer_type="move", + items=(TransferPlanItem( + sequence=0, source_fileitem=source, target_storage=storage, target_path=target.path, + ),), + ) + with execution_repository._session_factory() as session: + pending = session.query(TransferPending).one() + pending.src_path = source["path"] + pending.planning_input = planning_input.to_payload() + pending.input_fingerprint = planning_input.fingerprint + pending.checkpoint_payload = checkpoint.to_payload() + session.commit() + runner = transfer_chain_module._DurableTransferStepRunner( + task_id="task-runner", lease_token="lease", + checkpoint_fingerprint=checkpoint.fingerprint, repository=execution_repository, + ) + original_result = TransferStepResult(payload={"item": receipt.model_dump(mode="json")}) + runner.run( + phase="transfer", kind="materialize_target", + payload={"source": source, "target_storage": storage, + "target_path": target.path, "transfer_type": "copy"}, + execute=lambda: original_result, + observe=lambda: pytest.fail("成功上传回执不应重新探测"), + ) + command = TransferExecutionCommand(execution_repository) + prepared = command.prepare( + task_id="task-runner", lease_token="lease", + intent=TransferStepIntent.create( + task_id="task-runner", checkpoint_fingerprint=checkpoint.fingerprint, + ordinal=1, phase="transfer", kind="delete_move_source", + payload={"source": source, "target_storage": storage, "target_path": target.path}, + ), + ) + started = command.begin(task_id="task-runner", lease_token="lease", operation_id=prepared.operation_id) + command.manual_review( + task_id="task-runner", lease_token="lease", step=started, + error="旧版秒传大小凭据冲突", evidence=TransferStepResult(payload={"identity_error": "size 已变化"}), + ) + resolved = command.resolve_manual_review( + task_id="task-runner", operation_id=started.operation_id, + decision=TransferManualReviewDecision.NOT_APPLIED, actor="admin", reason="确认源文件仍存在", + ) + assert resolved.state is TransferExecutionState.RETRY_WAIT + claimed = TransactionalTransferAdmissionRepository(execution_repository._session_factory).claim_task( + task_id="task-runner", owner_id="recovery", lease_seconds=120, + ) + runner = transfer_chain_module._DurableTransferStepRunner( + task_id="task-runner", lease_token=claimed.lease_token, + checkpoint_fingerprint=checkpoint.fingerprint, repository=execution_repository, + ) + if change == "source_size": + source_path.write_bytes(b"changed source") + live_target = target.model_copy(update={ + "size": 999 if change == "target_size" else target.size, + "fileid": None if change == "target_id_missing" else "201" if change == "target_id" else target.fileid, + "type": "dir" if change == "target_type" else target.type, + }) + remote = Mock() + remote.get_item_strict.return_value = None if change == "target_missing" else live_target + if change == "target_query": + remote.get_item_strict.side_effect = TimeoutError("115 unavailable") + arguments = dict( + step_runner=runner, source_fileitem=source, target_storage=storage, + source_oper=local, target_oper=remote, target_file=Path(target.path), transfer_type="move", + ) + if change: + with pytest.raises(transfer_chain_module._TransferManualReviewRequired): + TransHandler._TransHandler__execute_transfer_with_steps(**arguments) + assert source_path.exists() + assert execution_repository.get_snapshot(task_id="task-runner").state is TransferExecutionState.MANUAL_REVIEW + else: + result, error = TransHandler._TransHandler__execute_transfer_with_steps(**arguments) + assert not error + assert result == target + assert not source_path.exists() + snapshot = execution_repository.get_snapshot(task_id="task-runner") + assert snapshot.steps[0].result == original_result + remote.upload.assert_not_called() + remote.delete.assert_not_called() + + def test_remote_to_local_transfer_creates_target_directory_before_download(tmp_path): """ 网盘到本地的下载开始前必须先创建缺失的目标目录。 diff --git a/tests/test_transfer_safe_overwrite.py b/tests/test_transfer_safe_overwrite.py index 0538a005e..4c29016fd 100644 --- a/tests/test_transfer_safe_overwrite.py +++ b/tests/test_transfer_safe_overwrite.py @@ -4,6 +4,7 @@ import errno import os +import shutil from dataclasses import replace from datetime import datetime, timezone from pathlib import Path @@ -16,6 +17,7 @@ from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.application.transfer.execution import ( + TransferExecutionLeaseLostError, TransferExecutionState, TransferStepState, build_transfer_overwrite_paths, @@ -41,6 +43,224 @@ from app.schemas.file import FileItem from app.schemas.types import MediaType +@pytest.fixture +def u115_filesystem(tmp_path, monkeypatch): + """ + 以临时文件系统回放 115 的原名复制、按 ID 改名、移动和删除接口。 + """ + local = object.__new__(LocalStorage) + remote = object.__new__(U115Pan) + identifiers = {} + state = SimpleNamespace(copies=0, fail=None, renames=[], deletes=[]) + + def query(path): + """ + 严格读取远端文件，使用 inode 模拟跨重命名保持不变的文件 ID。 + """ + path = Path(path) + item = local.get_item_strict(path) + if item is None: + return None + identifier = str(path.stat().st_ino) + identifiers[int(identifier)] = path + return item.model_copy(update={"storage": "u115", "fileid": identifier}) + + def request(method, endpoint, **kwargs): + """ + 回放 API 副作用；失败可发生在复制之后、移动之前或删除阶段。 + """ + assert method == "POST" + data = kwargs["data"] + if endpoint == "/open/folder/add": + directory = identifiers[int(data["pid"])] / data["file_name"] + directory.mkdir() + return {"state": True, "data": {"file_id": query(directory).fileid}} + if endpoint == "/open/ufile/copy": + assert kwargs["retry_limit"] == 0 + source = identifiers[data["file_id"]] + destination = identifiers[data["pid"]] / source.name + index = 0 + while destination.exists(): + index += 1 + destination = destination.with_name(f"{source.stem}({index}){source.suffix}") + if source.is_dir(): + shutil.copytree(source, destination) + else: + shutil.copyfile(source, destination) + state.copies += 1 + query(destination) + return None if state.fail == "copy_response" else {"state": True} + if endpoint == "/open/ufile/move": + assert kwargs["retry_limit"] == 0 + if state.fail == "move": + return {"state": False} + source = identifiers[data["file_ids"]] + destination = identifiers[data["to_cid"]] / source.name + index = 0 + while destination.exists(): + index += 1 + destination = destination.with_name(f"{source.stem}({index}){source.suffix}") + source.rename(destination) + query(destination) + return {"state": True} + if endpoint == "/open/ufile/update": + source = identifiers[data["file_id"]] + state.renames.append(source) + if state.fail == "rename": + return {"state": False} + destination = source.with_name(data["file_name"]) + assert not destination.exists() + source.rename(destination) + query(destination) + return {"state": True} + assert endpoint == "/open/ufile/delete" + source = identifiers[data["file_ids"]] + state.deletes.append(source) + if state.fail == "delete": + return {"state": False} + if source.is_dir(): + source.rmdir() + else: + source.unlink() + return {"state": True} + + monkeypatch.setattr(remote, "get_item_strict", query) + monkeypatch.setattr(remote, "get_item", query) + monkeypatch.setattr(remote, "get_folder", query) + monkeypatch.setattr(remote, "list", lambda item: [query(path) for path in Path(item.path).iterdir()]) + monkeypatch.setattr(remote, "_request_api", request) + return remote, state + + +@pytest.mark.parametrize("directory", [False, True]) +def test_u115_native_copy_supports_files_and_directories(tmp_path, u115_filesystem, directory): + """ + 文件管理的文件和目录复制都应保留源，并准确返回新对象的 ID。 + """ + source = tmp_path / ("source" if directory else "source.mkv") + if directory: + source.mkdir() + (source / "movie.mkv").write_bytes(b"movie") + else: + source.write_bytes(b"movie") + destination = tmp_path / "library" + destination.mkdir() + remote, state = u115_filesystem + original = remote.get_item_strict(source) + result = remote.copy_item(original, destination, source.name) + assert result.fileid != original.fileid + assert result.type == original.type + assert result.path.rstrip("/") == (destination / source.name).as_posix() + assert state.copies == 1 + assert source.exists() + assert not list(destination.glob("mp-*")) + + +def test_u115_native_move_renames_by_id_when_raw_name_already_exists(tmp_path, u115_filesystem): + """ + 移动接口生成编号临时名时，后续改名仍操作源 ID，不误改原名旧文件。 + """ + source = tmp_path / "source.mkv" + source.write_bytes(b"new-version") + destination = tmp_path / "library" + destination.mkdir() + existing = destination / source.name + existing.write_bytes(b"old-version") + remote, state = u115_filesystem + original = remote.get_item_strict(source) + result = remote.move_item(original, destination, "Movie.mkv") + assert result.fileid == original.fileid + assert existing.read_bytes() == b"old-version" + assert (destination / "Movie.mkv").read_bytes() == b"new-version" + assert not source.exists() + assert state.copies == 0 + assert existing not in state.renames + + +@pytest.mark.parametrize("transfer_type", ["copy", "move"]) +@pytest.mark.parametrize("mode", ["always", "latest", "never"]) +@pytest.mark.parametrize("same_name", [False, True]) +def test_u115_native_overwrite_preserves_existing_raw_name_and_copies_once( + tmp_path, monkeypatch, repository_factory, u115_filesystem, transfer_type, mode, same_name, +): + """ + 盘内覆盖使用隔离副本，原名旧文件不被误改，重启重放不会生成编号副本。 + """ + target = tmp_path / "library" / "Movie.2026.mkv" + source = tmp_path / "downloads" / (target.name if same_name else "source.mkv") + source.parent.mkdir() + target.parent.mkdir() + source.write_bytes(b"new-version") + target.write_bytes(b"old-version") + raw_existing = target.parent / source.name + if not same_name: + raw_existing.write_bytes(b"unrelated-file") + remote, state = u115_filesystem + item = remote.get_item_strict(source).model_dump(mode="json") + checkpoint = replace( + _checkpoint(source, target, transfer_type=transfer_type, mode=mode), + planning_input=TransferPlanningInput(source_fileitem=item), target_storage="u115", + items=(TransferPlanItem(sequence=0, source_fileitem=item, target_storage="u115", target_path=target.as_posix()),), + ) + repository = repository_factory(checkpoint) + handler = TransHandler() + monkeypatch.setattr(handler, "_TransHandler__intercept_transfer", lambda **_kwargs: (True, "")) + for _ in range(2): + result = handler.execute_transfer_plan( + checkpoint, meta=MetaBase("Movie.2026.mkv"), + mediainfo=MediaInfo(type=MediaType.MOVIE, title="Movie", year="2026"), + source_oper=remote, target_oper=remote, step_runner=_durable_runner(repository, checkpoint), + ) + assert result.success is (mode != "never") + if not same_name: + assert raw_existing.read_bytes() == b"unrelated-file" + assert raw_existing not in state.renames + assert state.copies == (0 if mode == "never" else 1) + assert target.read_bytes() == (b"old-version" if mode == "never" else b"new-version") + assert source.exists() is (mode == "never" or transfer_type == "copy") + assert not list(target.parent.glob("mp-*")) + assert not list(target.parent.glob("*(*)*")) + + +@pytest.mark.parametrize("failure", ["copy_response", "move", "rename"]) +def test_u115_partial_native_copy_is_isolated_instead_of_repeated( + tmp_path, monkeypatch, repository_factory, u115_filesystem, failure, +): + """ + 复制已发生但改名或响应失败时，真实持久 runner 保留现场且禁止第二次复制。 + """ + source = tmp_path / "downloads" / "source.mkv" + target = tmp_path / "library" / "Movie.2026.mkv" + source.parent.mkdir() + target.parent.mkdir() + source.write_bytes(b"new-version") + target.write_bytes(b"old-version") + remote, state = u115_filesystem + state.fail = failure + item = remote.get_item_strict(source).model_dump(mode="json") + checkpoint = replace( + _checkpoint(source, target), planning_input=TransferPlanningInput(source_fileitem=item), + target_storage="u115", + items=(TransferPlanItem(sequence=0, source_fileitem=item, target_storage="u115", target_path=target.as_posix()),), + ) + repository = repository_factory(checkpoint) + handler = TransHandler() + monkeypatch.setattr(handler, "_TransHandler__intercept_transfer", lambda **_kwargs: (True, "")) + for attempt in range(2): + error = execution_module._TransferManualReviewRequired if attempt == 0 else TransferExecutionLeaseLostError + with pytest.raises(error): + handler.execute_transfer_plan( + checkpoint, meta=MetaBase("Movie.2026.mkv"), + mediainfo=MediaInfo(type=MediaType.MOVIE, title="Movie", year="2026"), + source_oper=remote, target_oper=remote, step_runner=_durable_runner(repository, checkpoint), + ) + assert repository.get_snapshot(task_id="safe-overwrite").state is TransferExecutionState.MANUAL_REVIEW + assert state.copies == 1 + assert target.read_bytes() == b"old-version" + assert source.read_bytes() == b"new-version" + assert not state.deletes + + @pytest.mark.parametrize("filename", ["美人余.S01E23.mkv", "Movie.MP4", "archive.tar.gz", "Movie"]) @pytest.mark.parametrize("storage", ["u115", "local", "alist", "alipan", "smb", "rclone"]) def test_overwrite_paths_preserve_storage_filename_contract(filename: str, storage: str) -> None: diff --git a/tests/test_u115_storage.py b/tests/test_u115_storage.py index 7e23cee55..988a4dbeb 100644 --- a/tests/test_u115_storage.py +++ b/tests/test_u115_storage.py @@ -281,3 +281,15 @@ def test_expired_access_token_uses_refreshed_credentials(): assert saved[0]["refresh_token"] == "new-refresh" assert saved[0]["access_token"] == "new-access" assert headers == [{"Authorization": "Bearer new-access"}] + + +@pytest.mark.parametrize("response,expected", [ + (None, False), ({"state": False}, False), ({"state": True}, True), +]) +def test_delete_requires_explicit_api_success(response, expected): + """ + 115 删除被拒绝或请求失败时，不能向整理结算谎报源文件已经删除。 + """ + storage = _build_storage() + storage._request_api = lambda *_args, **_kwargs: response + assert storage.delete(FileItem(storage="u115", path="/source.mkv", fileid="200")) is expected