diff --git a/app/agent/policy/mcp.py b/app/agent/policy/mcp.py index 5326dc2c3..f37544942 100644 --- a/app/agent/policy/mcp.py +++ b/app/agent/policy/mcp.py @@ -391,7 +391,7 @@ FIELD_DESCRIPTIONS = { "min_rating": "筛选共享或热门订阅的最低评分。", "min_sample_rate": "最低可接受音频采样率，单位为赫兹。", "min_sub": "筛选热门订阅的最低订阅人数。", - "mode": "操作模式；music.explore 接受 chart 或 fresh，整理历史记录 move、copy、link 或 softlink。", + "mode": "操作模式；music.explore 接受 chart 或 fresh，整理历史记录 move、copy、link、softlink 或 reflink。", "modify_time": "存储项目的修改时间戳。", "mtype": "操作要求的媒体类型或订阅历史分类。", "music_cue_enable": "订阅级可选 CUE 整理设置，省略时继承系统默认值。", diff --git a/app/modules/filemanager/storages/local.py b/app/modules/filemanager/storages/local.py index e05d1dce3..9068dc876 100644 --- a/app/modules/filemanager/storages/local.py +++ b/app/modules/filemanager/storages/local.py @@ -545,9 +545,9 @@ class LocalStorage(StorageBase): :return: 系统操作返回成功时为 True，否则为 False """ if not fileitem.path: + logger.error("【本地】引用链接文件失败：源文件路径为空") return False - file_path = Path(fileitem.path) - code, message = SystemUtils.reflink(file_path, target_file) + code, message = SystemUtils.reflink(Path(fileitem.path), target_file) if code != 0: logger.error(f"【本地】引用链接文件失败：{message}") return False diff --git a/app/modules/filemanager/transhandler.py b/app/modules/filemanager/transhandler.py index c718325a1..933f14eab 100644 --- a/app/modules/filemanager/transhandler.py +++ b/app/modules/filemanager/transhandler.py @@ -2102,14 +2102,14 @@ class TransHandler: return None, f"【{target_storage}】{target_file.parent} 目录获取失败" action_name = "copy_item" if transfer_type == "copy" else "move_item" custom_item_action = getattr(source_oper, action_name, None) + fallback_action = getattr(source_oper, transfer_type, None) if callable(custom_item_action): new_item = custom_item_action( fileitem, Path(target_fileitem.path), target_file.name ) if new_item: return new_item, "" - fallback_action = getattr(source_oper, transfer_type, None) - if callable(fallback_action) and fallback_action( + elif callable(fallback_action) and fallback_action( fileitem, Path(target_fileitem.path), target_file.name ): return get_remote_targetitem(fileitem, target_file), "" diff --git a/tests/test_system_reflink.py b/tests/test_system_reflink.py index 601d9659b..854d381d9 100644 --- a/tests/test_system_reflink.py +++ b/tests/test_system_reflink.py @@ -6,108 +6,42 @@ import errno import os import shutil import sys -import warnings from pathlib import Path from unittest.mock import MagicMock -import psutil import pytest from app.adapters.system.host import SystemUtils -def _get_filesystem_type(path: Path) -> str: +def _reflink_available(directory: Path) -> bool: """ - 获取指定路径所在挂载点文件系统的格式类型。 + 探测目录所在文件系统是否支持底层引用链接。 - :param path: 目标路径 - :return: 小写文件系统名称（例如 apfs, btrfs, xfs, ext4） + 通过真实克隆一对探测文件判定能力，避免把平台判断当成文件系统判断。 + + :param directory: 用于探测的临时目录 + :return: 底层克隆成功返回 True """ - resolved = path.resolve() - best_mount = "" - best_fstype = "" + probe_source = directory / ".reflink-probe.source" + probe_target = directory / ".reflink-probe.target" + probe_source.write_bytes(b"reflink probe") try: - for part in psutil.disk_partitions(all=True): - try: - mount = Path(part.mountpoint).resolve() - if resolved == mount or mount in resolved.parents: - if len(str(mount)) > len(best_mount): - best_mount = str(mount) - best_fstype = (part.fstype or "").lower() - except Exception: - continue - except Exception: - pass - if best_fstype: - return best_fstype - - if sys.platform.startswith("linux"): - try: - with open("/proc/mounts", "r", encoding="utf-8") as stream: - for line in stream: - parts = line.split() - if len(parts) >= 3: - mount = Path(parts[1]).resolve() - if resolved == mount or mount in resolved.parents: - if len(str(mount)) > len(best_mount): - best_mount = str(mount) - best_fstype = parts[2].lower() - except Exception: - pass - if best_fstype: - return best_fstype - - if sys.platform == "darwin": - return "apfs" - return "unknown" - - -def _probe_reflink_support(path: Path) -> bool: - """ - 真实探测文件系统是否支持写时复制。 - - :param path: 探测临时目录 - :return: 支持写时复制时为 True - """ - src = path / ".probe_reflink_src" - dest = path / ".probe_reflink_dest" - try: - src.write_bytes(b"probe") - code, _ = SystemUtils.reflink(src, dest) + code, _ = SystemUtils.reflink(probe_source, probe_target) return code == 0 - except Exception: - return False finally: - src.unlink(missing_ok=True) - dest.unlink(missing_ok=True) - - -def _require_reflink_filesystem(tmp_path: Path) -> str: - """ - 检查测试目录文件系统格式，非写时复制格式发出警告并跳过测试。 - - :param tmp_path: pytest 临时测试目录 - :return: 当前被确认支持写时复制的文件系统格式 - """ - fstype = _get_filesystem_type(tmp_path) - supported_types = {"apfs", "btrfs", "xfs", "zfs"} - if fstype not in supported_types or not _probe_reflink_support(tmp_path): - warning_message = ( - f"当前测试目录所在文件系统格式为 [{fstype}]，" - f"非写时复制格式（需为 {supported_types} 且启用 CoW），已跳过对应引用链接测试" - ) - warnings.warn(warning_message, UserWarning, stacklevel=2) - pytest.skip(warning_message) - return fstype - + probe_source.unlink(missing_ok=True) + probe_target.unlink(missing_ok=True) + probe_target.with_suffix(probe_target.suffix + ".mp").unlink(missing_ok=True) def test_reflink_success_on_current_platform(tmp_path: Path): """ - 测试在当前支持引用链接的系统（如 macOS APFS）上成功创建 reflink， + 测试在当前支持引用链接的文件系统（如 macOS APFS、Linux Btrfs/XFS）上成功创建 reflink， 验证内容一致、不同 Inode、写时复制独立修改以及文件大小一致。 """ - _require_reflink_filesystem(tmp_path) + if not _reflink_available(tmp_path): + pytest.skip("当前文件系统不支持引用链接") src = tmp_path / "source.bin" dest = tmp_path / "cloned.bin" @@ -138,7 +72,8 @@ def test_reflink_cleans_preexisting_temp_file(tmp_path: Path): """ 测试当目标临时 .mp 文件预先存在时，reflink 能先清理临时文件并成功完成克隆。 """ - _require_reflink_filesystem(tmp_path) + if not _reflink_available(tmp_path): + pytest.skip("当前文件系统不支持引用链接") src = tmp_path / "source.txt" dest = tmp_path / "target.txt" diff --git a/tests/test_transfer_reflink.py b/tests/test_transfer_reflink.py index b4a4e0384..baf054784 100644 --- a/tests/test_transfer_reflink.py +++ b/tests/test_transfer_reflink.py @@ -4,12 +4,10 @@ import filecmp import os -import sys -import warnings +import shutil from pathlib import Path from unittest.mock import MagicMock -import psutil import pytest from app.adapters.system.host import SystemUtils @@ -36,89 +34,34 @@ def _make_local_storage() -> LocalStorage: return object.__new__(LocalStorage) -def _get_filesystem_type(path: Path) -> str: +def _patch_reflink_with_simulated_clone( + monkeypatch: pytest.MonkeyPatch, + calls: list[tuple[Path, Path]], +) -> None: """ - 获取指定路径所在挂载点文件系统的格式类型。 + 用同盘复制模拟底层引用链接，避免测试依赖宿主文件系统的 CoW 能力。 - :param path: 目标路径 - :return: 小写文件系统名称（例如 apfs, btrfs, xfs, ext4） + 真实克隆语义与不支持时的失败处理由 tests/test_system_reflink.py 覆盖， + 管道测试只验证整理流程是否把整理方式正确路由到存储适配器。 + + :param monkeypatch: pytest 猴子补丁夹具 + :param calls: 记录 (源路径, 目标路径) 调用的列表 """ - resolved = path.resolve() - best_mount = "" - best_fstype = "" - try: - for part in psutil.disk_partitions(all=True): - try: - mount = Path(part.mountpoint).resolve() - if resolved == mount or mount in resolved.parents: - if len(str(mount)) > len(best_mount): - best_mount = str(mount) - best_fstype = (part.fstype or "").lower() - except Exception: - continue - except Exception: - pass - if best_fstype: - return best_fstype - if sys.platform.startswith("linux"): - try: - with open("/proc/mounts", "r", encoding="utf-8") as stream: - for line in stream: - parts = line.split() - if len(parts) >= 3: - mount = Path(parts[1]).resolve() - if resolved == mount or mount in resolved.parents: - if len(str(mount)) > len(best_mount): - best_mount = str(mount) - best_fstype = parts[2].lower() - except Exception: - pass - if best_fstype: - return best_fstype + def _simulated_reflink(src: Path, dest: Path) -> tuple[int, str]: + """ + 复制内容并同步元数据，模拟克隆成功后的落地状态。 - if sys.platform == "darwin": - return "apfs" - return "unknown" + :param src: 源文件路径 + :param dest: 目标文件路径 + :return: 固定成功状态 + """ + shutil.copyfile(src, dest) + shutil.copystat(src, dest) + calls.append((src, dest)) + return 0, "" - -def _probe_reflink_support(path: Path) -> bool: - """ - 真实探测文件系统是否支持写时复制。 - - :param path: 探测临时目录 - :return: 支持写时复制时为 True - """ - src = path / ".probe_reflink_src" - dest = path / ".probe_reflink_dest" - try: - src.write_bytes(b"probe") - code, _ = SystemUtils.reflink(src, dest) - return code == 0 - except Exception: - return False - finally: - src.unlink(missing_ok=True) - dest.unlink(missing_ok=True) - - -def _require_reflink_filesystem(tmp_path: Path) -> str: - """ - 检查测试目录文件系统格式，非写时复制格式发出警告并跳过测试。 - - :param tmp_path: pytest 临时测试目录 - :return: 当前被确认支持写时复制的文件系统格式 - """ - fstype = _get_filesystem_type(tmp_path) - supported_types = {"apfs", "btrfs", "xfs", "zfs"} - if fstype not in supported_types or not _probe_reflink_support(tmp_path): - warning_message = ( - f"当前测试目录所在文件系统格式为 [{fstype}]，" - f"非写时复制格式（需为 {supported_types} 且启用 CoW），已跳过对应引用链接测试" - ) - warnings.warn(warning_message, UserWarning, stacklevel=2) - pytest.skip(warning_message) - return fstype + monkeypatch.setattr(SystemUtils, "reflink", _simulated_reflink) def test_transhandler_plan_and_execute_reflink( @@ -129,12 +72,12 @@ def test_transhandler_plan_and_execute_reflink( 测试 TransHandler 使用 transfer_type="reflink" 规划并执行本地文件整理。 验证整理执行成功，目标文件被创建，源文件与目标文件内容完全一致， - 且属于不同 Inode（非硬链接），也不是符号链接。 + 且属于不同 Inode（非硬链接），也不是符号链接；底层克隆调用被模拟， + 不依赖宿主文件系统是否支持 CoW。 :param tmp_path: pytest 临时测试目录 :param monkeypatch: pytest 猴子补丁夹具 """ - _require_reflink_filesystem(tmp_path) download_dir = tmp_path / "downloads" library_dir = tmp_path / "library" download_dir.mkdir(parents=True, exist_ok=True) @@ -156,6 +99,9 @@ def test_transhandler_plan_and_execute_reflink( monkeypatch.setattr(handler, "_TransHandler__intercept_transfer", lambda **_kwargs: (True, "")) monkeypatch.setattr(LocalStorage, "_LocalStorage__should_show_progress", lambda *_args: False) + cloned_calls: list[tuple[Path, Path]] = [] + _patch_reflink_with_simulated_clone(monkeypatch, cloned_calls) + meta = MetaBase("Test.Movie.2026.mkv") mediainfo = MediaInfo(type=MediaType.MOVIE, title="Test Movie", year="2026") @@ -192,6 +138,8 @@ def test_transhandler_plan_and_execute_reflink( assert transfer_info.success is True target_file = Path(checkpoint.final_target_path) + # 验证整理流程把引用链接路由到系统适配器 + assert cloned_calls == [(source_file, target_file)] # 验证目标文件已创建且为常规文件 assert target_file.exists() assert target_file.is_file() @@ -211,10 +159,11 @@ def test_transhandler_reflink_overwrite_always( """ 测试在覆盖模式为 always 时，reflink 也能安全覆盖已存在的目标文件。 + 底层克隆调用被模拟，不依赖宿主文件系统是否支持 CoW。 + :param tmp_path: pytest 临时测试目录 :param monkeypatch: pytest 猴子补丁夹具 """ - _require_reflink_filesystem(tmp_path) download_dir = tmp_path / "downloads" library_dir = tmp_path / "library" download_dir.mkdir(parents=True, exist_ok=True) @@ -240,6 +189,9 @@ def test_transhandler_reflink_overwrite_always( monkeypatch.setattr(handler, "_TransHandler__intercept_transfer", lambda **_kwargs: (True, "")) monkeypatch.setattr(LocalStorage, "_LocalStorage__should_show_progress", lambda *_args: False) + cloned_calls: list[tuple[Path, Path]] = [] + _patch_reflink_with_simulated_clone(monkeypatch, cloned_calls) + meta = MetaBase("Test.Movie.2026.mkv") mediainfo = MediaInfo(type=MediaType.MOVIE, title="Test Movie", year="2026") @@ -274,6 +226,13 @@ def test_transhandler_reflink_overwrite_always( assert transfer_info.success is True + # 验证整理流程把引用链接路由到系统适配器；覆盖模式先克隆到暂存路径再替换旧目标 + assert len(cloned_calls) == 1 + clone_source, clone_target = cloned_calls[0] + assert clone_source == source_file + assert clone_target != target_file + assert clone_target.parent == target_file.parent + assert not clone_target.exists() # 验证旧目标已被新版本的 reflink 内容原子替换 assert target_file.exists() assert target_file.is_file() @@ -283,12 +242,90 @@ def test_transhandler_reflink_overwrite_always( assert filecmp.cmp(source_file, target_file, shallow=False) +def test_transhandler_reflink_unsupported_filesystem_fails( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """ + 测试底层文件系统不支持引用链接时，整理严格失败且不隐式降级为全量复制。 + + :param tmp_path: pytest 临时测试目录 + :param monkeypatch: pytest 猴子补丁夹具 + """ + download_dir = tmp_path / "downloads" + library_dir = tmp_path / "library" + download_dir.mkdir(parents=True, exist_ok=True) + library_dir.mkdir(parents=True, exist_ok=True) + + source_file = download_dir / "Unsupported.2026.mkv" + source_file.write_bytes(b"reflink unsupported filesystem content") + + storage = _make_local_storage() + source_item = storage.get_item_strict(source_file) + assert source_item is not None + + planning_input = TransferPlanningInput( + source_fileitem=source_item.model_dump(mode="json"), + ) + + handler = TransHandler() + monkeypatch.setattr(handler, "_TransHandler__intercept_transfer", lambda **_kwargs: (True, "")) + monkeypatch.setattr(LocalStorage, "_LocalStorage__should_show_progress", lambda *_args: False) + + def _unsupported_reflink(_src: Path, _dest: Path) -> tuple[int, str]: + """ + 模拟不支持 FICLONE 的文件系统返回错误。 + + :param _src: 源文件路径 + :param _dest: 目标文件路径 + :return: 失败状态及错误信息 + """ + return -1, "Linux ioctl(FICLONE) 失败: [Errno 95] Operation not supported" + + monkeypatch.setattr(SystemUtils, "reflink", _unsupported_reflink) + + meta = MetaBase("Unsupported.2026.mkv") + mediainfo = MediaInfo(type=MediaType.MOVIE, title="Unsupported", year="2026") + + checkpoint = handler.plan_transfer( + planning_input, + meta=meta, + mediainfo=mediainfo, + source_oper=storage, + target_storage="local", + target_path=library_dir, + transfer_type="reflink", + need_scrape=False, + need_rename=False, + need_notify=False, + overwrite_mode=None, + episodes_info=None, + preview=False, + ) + + transfer_info = handler.execute_transfer_plan( + checkpoint, + meta=meta, + mediainfo=mediainfo, + source_oper=storage, + target_oper=storage, + ) + + # 克隆不可用时不复制、不降级，整理失败并保留源文件 + assert transfer_info.success is False + assert not Path(checkpoint.final_target_path).exists() + assert source_file.read_bytes() == b"reflink unsupported filesystem content" + + def test_transhandler_observe_reflink_operation_states( tmp_path: Path, ) -> None: """ 测试步骤观察器 __observe_transfer_operation 对 reflink 整理方式的状态判定。 + 观察器只判定目标链接类型与内容状态，测试用普通副本构造等价落地结果， + 不依赖宿主文件系统的 CoW 能力。 + 验证： 1. 目标正常独立且内容一致时返回 APPLIED。 2. 目标文件缺失时返回 NOT_APPLIED。 @@ -298,7 +335,6 @@ def test_transhandler_observe_reflink_operation_states( :param tmp_path: pytest 临时测试目录 """ - _require_reflink_filesystem(tmp_path) source_dir = tmp_path / "src" target_dir = tmp_path / "dst" source_dir.mkdir(parents=True, exist_ok=True) @@ -312,10 +348,9 @@ def test_transhandler_observe_reflink_operation_states( source_item = storage.get_item_strict(source_file) assert source_item is not None - # 1. 成功创建的真实 reflink 文件 -> APPLIED + # 1. 内容一致的独立常规文件（引用链接落地后的等价状态） -> APPLIED reflink_target = target_dir / "sample_reflink.mkv" - code, msg = SystemUtils.reflink(source_file, reflink_target) - assert code == 0, f"创建 reflink 失败: {msg}" + shutil.copyfile(source_file, reflink_target) obs_applied = TransHandler._TransHandler__observe_transfer_operation( fileitem=source_item,