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