diff --git a/app/adapters/system/plugin/package.py b/app/adapters/system/plugin/package.py index e1948be70..06dbfb163 100644 --- a/app/adapters/system/plugin/package.py +++ b/app/adapters/system/plugin/package.py @@ -993,11 +993,12 @@ class PluginPackageManager: release_tag = selection.release_tag if release_tag and not selection.fallback_to_filelist: - def prepare_selected_release() -> tuple[bool, str]: + def prepare_selected_release(staging_dir: Path) -> tuple[bool, str]: return self.__install_from_release( pid, selection.user_repo, release_tag, + staging_dir, ) return self.__install_flow_sync( @@ -1010,20 +1011,24 @@ class PluginPackageManager: if release_tag: # 当前索引 Release 失败时回退文件列表，避免发布产物短暂滞后阻断安装。 - def prepare_release() -> tuple[bool, str]: + def prepare_release(staging_dir: Path) -> tuple[bool, str]: ok, msg = self.__install_from_release( pid, selection.user_repo, release_tag, + staging_dir, ) if ok: return True, msg logger.warning(f"{pid} Release 安装失败，回退文件列表安装：{msg}") - self.__remove_old_plugin(pid) + # 回退前只清理本次暂存的半份 Release 产物；已安装的运行目录 + # 此刻还没有被触碰，清理它会让一次可恢复的回退变成数据丢失。 + shutil.rmtree(staging_dir, ignore_errors=True) return self.__prepare_content_via_filelist_sync( pid, selection.user_repo, selection.package_version, + staging_dir, ) return self.__install_flow_sync( @@ -1035,11 +1040,12 @@ class PluginPackageManager: ) # 未声明 release 打包的插件继续使用文件列表方式安装。 - def prepare_filelist() -> tuple[bool, str]: + def prepare_filelist(staging_dir: Path) -> tuple[bool, str]: return self.__prepare_content_via_filelist_sync( pid, selection.user_repo, selection.package_version, + staging_dir, ) return self.__install_flow_sync( @@ -1126,11 +1132,11 @@ class PluginPackageManager: except Exception: return False, "本地插件来源路径无效" - def prepare_local() -> tuple[bool, str]: + def prepare_local(staging_dir: Path) -> tuple[bool, str]: try: shutil.copytree( source_dir, - dest_dir, + staging_dir, dirs_exist_ok=True, ignore=shutil.ignore_patterns("__pycache__", "*.pyc", ".DS_Store", "node_modules") ) @@ -1206,8 +1212,9 @@ class PluginPackageManager: pid: str, remote_path: object, package_version: Optional[str], + dest_root: Path, ) -> Path: - """把市场文件路径限定到当前插件目录，拒绝绝对路径和目录穿越。""" + """把市场文件路径限定到给定落盘根目录之内，拒绝绝对路径和目录穿越。""" if not isinstance(remote_path, str) or not remote_path or "\\" in remote_path: raise ValueError("插件文件路径无效") pure_path = PurePosixPath(remote_path) @@ -1222,24 +1229,30 @@ class PluginPackageManager: or any(part in {"", ".", ".."} for part in parts[2:]) ): raise ValueError("插件文件路径无效") - plugin_dir = (self._plugins_root() / pid.lower()).resolve() - file_path = (plugin_dir / Path(*parts[2:])).resolve() - if not file_path.is_relative_to(plugin_dir): + resolved_root = dest_root.resolve() + file_path = (resolved_root / Path(*parts[2:])).resolve() + if not file_path.is_relative_to(resolved_root): raise ValueError("插件文件路径无效") return file_path def __download_files(self, pid: str, file_list: list[dict[str, Any]], user_repo: str, - package_version: Optional[str] = None) -> tuple[bool, str]: + package_version: Optional[str] = None, + dest_root: Optional[Path] = None) -> tuple[bool, str]: """ 下载插件文件 :param pid: 插件 ID :param file_list: 要下载的文件列表，包含文件的元数据（包括下载链接） :param user_repo: GitHub 仓库的 user/repo 路径 + :param dest_root: 文件落盘的根目录，未指定时直接写入插件运行目录 :return: (是否成功, 错误信息) """ if not file_list: return False, "文件列表为空" + download_root = ( + dest_root if dest_root is not None + else self._plugins_root() / pid.lower() + ) # 使用栈结构来替代递归调用，避免递归深度过大问题 stack = [(pid, file_list)] @@ -1254,6 +1267,7 @@ class PluginPackageManager: pid, item.get("path"), package_version, + download_root, ) except ValueError as error: return False, str(error) @@ -1426,56 +1440,152 @@ class PluginPackageManager: logger.error(f"获取插件 {pid} 元数据失败：{e}") return {} + def __new_install_staging_dir(self, pid: str) -> Path: + """分配一个全新的安装暂存目录，让新内容在触碰运行目录前先完整就位。""" + return ( + Path(get_runtime_setting('TEMP_PATH')) + / "plugin_install_staging" + / f"{pid.lower()}-{uuid.uuid4().hex}" + ) + + @staticmethod + def __swap_staged_plugin_content(staging_dir: Path, final_dir: Path) -> None: + """把已就位的暂存内容换入插件运行目录，失败时运行目录保持换入前的内容。 + + 手法与运行目录补偿替换一致（见 __restore_tree）：先把旧目录改名挪到同级 + 隐藏位置留作回滚材料，再把暂存目录改名落位，确认落位后才删除旧目录。同一 + 文件系统内两次改名都是原子的，因此进程在任意时刻被杀，磁盘上要么是旧内容 + 要么是新内容，唯一的中间态是"运行目录短暂缺失、旧内容完整躺在 previous + 目录里"，仍可人工恢复，不会出现任何一份内容被删到无处可取的时点。 + + 两个改名都可能因跨文件系统失败：overlayfs 会拒绝把镜像层的插件目录直接改 + 名到可写层，暂存目录又常落在独立的临时分区。两处都退化为复制，代价是复制 + 期间可能中途失败留下半份目录，因此复制失败一律先删掉半成品再把旧目录换 + 回；换回本身也失败时只记录旧目录的保留位置、不吞掉原始异常，避免出现 + "看起来只是安装失败"实则运行目录已空的假象。 + + :param staging_dir: 已就位的待安装内容目录 + :param final_dir: 插件运行目录，可能已存在旧内容 + :raise OSError: 换入失败，且已尽力把运行目录恢复或保留为换入前的内容 + """ + final_dir.parent.mkdir(parents=True, exist_ok=True) + previous = final_dir.parent / f".{final_dir.name}.previous-{uuid.uuid4().hex}" + previous_available = False + cleared = False + published = False + try: + if final_dir.exists(): + try: + final_dir.replace(previous) + except OSError as error: + if error.errno != errno.EXDEV: + raise + shutil.copytree(final_dir, previous, symlinks=True) + previous_available = True + PluginPackageManager.__remove_snapshot_path(final_dir) + else: + previous_available = True + # 旧内容已挪开或本来就不存在，此后才允许重建运行目录 + cleared = True + try: + staging_dir.replace(final_dir) + except OSError as error: + if error.errno != errno.EXDEV: + raise + logger.debug( + f"插件安装内容跨文件系统无法改名换入，退化为复制：{staging_dir}" + ) + shutil.copytree(staging_dir, final_dir, symlinks=True) + published = True + except Exception: + if cleared and not published: + try: + PluginPackageManager.__remove_snapshot_path(final_dir) + if previous_available: + previous.replace(final_dir) + previous_available = False + except Exception as rollback_error: + logger.error( + f"插件安装换入失败后恢复旧目录失败，已保留恢复材料 {previous}: " + f"{rollback_error}" + ) + raise + finally: + if previous.exists() and (published or not previous_available): + shutil.rmtree(previous, ignore_errors=True) + + async def __async_swap_staged_plugin_content( + self, staging_dir: Path, final_dir: Path + ) -> None: + """在线程中执行换入，避免目录改名与复制阻塞事件循环，语义与同步换入一致。""" + await _await_thread_operation( + self.__swap_staged_plugin_content, staging_dir, final_dir + ) + def __install_flow_sync( self, pid: str, force_install: bool, - prepare_content: Callable[[], tuple[bool, str]], + prepare_content: Callable[[Path], tuple[bool, str]], repo_url: Optional[str] = None, before_dependency_install: Optional[Callable[[], None]] = None, ) -> tuple[bool, str]: """ - 同步安装统一流程：备份→清理→准备内容→安装依赖→上报 - prepare_content 负责把插件文件放到 app/plugins/{pid} + 同步安装统一流程：暂存内容→备份→换入→安装依赖→上报 + prepare_content 负责把插件文件放到调用时给定的暂存目录；只有新内容在暂存 + 目录里完整就位后才会触碰运行目录，因此下载或解压失败时已装插件原样保留。 """ - backup_dir = None - if not force_install: - backup_dir = self.__backup_plugin(pid) + plugin_dir = self.__plugin_dir(pid) + staging_dir = self.__new_install_staging_dir(pid) + try: + success, message = prepare_content(staging_dir) + if not success: + # 运行目录此刻尚未被触碰，已装插件天然完整，无需还原也无需清理 + logger.error(f"{pid} 准备插件内容失败：{message}") + return False, message - self.__remove_old_plugin(pid) + backup_dir = None + if not force_install: + backup_dir = self.__backup_plugin(pid) - success, message = prepare_content() - if not success: - logger.error(f"{pid} 准备插件内容失败：{message}") - if backup_dir: - self.__restore_plugin(pid, backup_dir) - logger.warn(f"{pid} 插件安装失败，已还原备份插件") - else: - self.__remove_old_plugin(pid) - logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装") - return False, message + try: + self.__swap_staged_plugin_content(staging_dir, plugin_dir) + except OSError as error: + message = f"写入插件内容失败：{error}" + logger.error(f"{pid} {message}") + # 换入函数已保证失败时运行目录保持换入前的内容；只有连恢复都失败、 + # 运行目录确实缺失时，才动用本次的临时备份兜底。 + if backup_dir and not plugin_dir.exists(): + self.__restore_plugin(pid, backup_dir) + logger.warn(f"{pid} 插件安装失败，已还原备份插件") + elif backup_dir: + shutil.rmtree(backup_dir, ignore_errors=True) + return False, message - dependencies_exist, dep_ok, dep_msg = ( - self.__install_dependencies_if_required( - pid, - before_dependency_install, + dependencies_exist, dep_ok, dep_msg = ( + self.__install_dependencies_if_required( + pid, + before_dependency_install, + ) + if before_dependency_install is not None + else self.__install_dependencies_if_required(pid) ) - if before_dependency_install is not None - else self.__install_dependencies_if_required(pid) - ) - if dependencies_exist and not dep_ok: - logger.error(f"{pid} 依赖安装失败：{dep_msg}") - if backup_dir: - self.__restore_plugin(pid, backup_dir) - logger.warn(f"{pid} 插件安装失败，已还原备份插件") - else: - self.__remove_old_plugin(pid) - logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装") - return False, dep_msg + if dependencies_exist and not dep_ok: + logger.error(f"{pid} 依赖安装失败：{dep_msg}") + if backup_dir: + self.__restore_plugin(pid, backup_dir) + logger.warn(f"{pid} 插件安装失败，已还原备份插件") + else: + self.__remove_old_plugin(pid) + logger.warn(f"{pid} 已清理对应插件目录，请尝试重新安装") + return False, dep_msg - if backup_dir: - shutil.rmtree(backup_dir, ignore_errors=True) - return True, "" + if backup_dir: + shutil.rmtree(backup_dir, ignore_errors=True) + return True, "" + finally: + if staging_dir.exists(): + shutil.rmtree(staging_dir, ignore_errors=True) @staticmethod def __validate_release_zip_name(name: str) -> None: @@ -1566,11 +1676,12 @@ class PluginPackageManager: targets.append((info, dest_path, info.is_dir())) return targets - def __install_from_release(self, pid: str, user_repo: str, release_tag: str) -> tuple[bool, str]: + def __install_from_release(self, pid: str, user_repo: str, release_tag: str, + dest_root: Optional[Path] = None) -> tuple[bool, str]: """ 通过 GitHub Release 资产文件安装插件。 规范：release 中存在名为 "{pid}_v{version}.zip" 的资产，zip 根即插件文件； - 将其全部解压到 app/plugins/{pid} + 将其全部解压到 dest_root，未指定时直接解压到插件运行目录。 """ # 拼接资产文件名 asset_name = f"{release_tag.lower()}.zip" @@ -1616,7 +1727,10 @@ class PluginPackageManager: infos = zf.infolist() if not infos: return False, "压缩包内容为空" - dest_base = self._plugins_root() / pid.lower() + dest_base = ( + dest_root if dest_root is not None + else self._plugins_root() / pid.lower() + ) targets = self.__iter_release_zip_targets(zf, dest_base) wrote_any = False for info, dest_path, is_dir in targets: @@ -1671,17 +1785,23 @@ class PluginPackageManager: return None, "插件数据解析失败" async def __async_download_files(self, pid: str, file_list: list[dict[str, Any]], user_repo: str, - package_version: Optional[str] = None) -> tuple[bool, str]: + package_version: Optional[str] = None, + dest_root: Optional[Path] = None) -> tuple[bool, str]: """ 异步下载插件文件 :param pid: 插件 ID :param file_list: 要下载的文件列表，包含文件的元数据（包括下载链接） :param user_repo: GitHub 仓库的 user/repo 路径 + :param dest_root: 文件落盘的根目录，未指定时直接写入插件运行目录 :return: (是否成功, 错误信息) """ if not file_list: return False, "文件列表为空" + download_root = ( + dest_root if dest_root is not None + else self._plugins_root() / pid.lower() + ) # 使用栈结构来替代递归调用，避免递归深度过大问题 stack = [(pid, file_list)] @@ -1696,6 +1816,7 @@ class PluginPackageManager: pid, item.get("path"), package_version, + download_root, ) except ValueError as error: return False, str(error) @@ -1894,11 +2015,12 @@ class PluginPackageManager: release_tag = selection.release_tag if release_tag and not selection.fallback_to_filelist: - async def prepare_selected_release() -> tuple[bool, str]: + async def prepare_selected_release(staging_dir: Path) -> tuple[bool, str]: return await self.__async_install_from_release( pid, selection.user_repo, release_tag, + staging_dir, ) return await self.__install_flow_async( @@ -1911,20 +2033,24 @@ class PluginPackageManager: if release_tag: # 当前索引 Release 失败时回退文件列表，保持同步与异步安装一致。 - async def prepare_release() -> tuple[bool, str]: + async def prepare_release(staging_dir: Path) -> tuple[bool, str]: ok, msg = await self.__async_install_from_release( pid, selection.user_repo, release_tag, + staging_dir, ) if ok: return True, msg logger.warning(f"{pid} Release 安装失败，回退文件列表安装：{msg}") - await self.__async_remove_old_plugin(pid) + # 回退前只清理本次暂存的半份 Release 产物；已安装的运行目录 + # 此刻还没有被触碰，清理它会让一次可恢复的回退变成数据丢失。 + await aioshutil.rmtree(staging_dir, ignore_errors=True) return await self.__prepare_content_via_filelist_async( pid, selection.user_repo, selection.package_version, + staging_dir, ) return await self.__install_flow_async( @@ -1936,11 +2062,12 @@ class PluginPackageManager: ) # 未声明 release 打包的插件继续使用文件列表方式安装。 - async def prepare_filelist() -> tuple[bool, str]: + async def prepare_filelist(staging_dir: Path) -> tuple[bool, str]: return await self.__prepare_content_via_filelist_async( pid, selection.user_repo, selection.package_version, + staging_dir, ) return await self.__install_flow_async( @@ -2006,29 +2133,38 @@ class PluginPackageManager: self, pid: str, force_install: bool, - prepare_content: Callable[[], Awaitable[tuple[bool, str]]], + prepare_content: Callable[[Path], Awaitable[tuple[bool, str]]], repo_url: Optional[str] = None, before_dependency_install: Optional[Callable[[], None]] = None, ) -> tuple[bool, str]: """ - 异步安装流程，处理插件内容准备、依赖安装和注册 + 异步安装统一流程：暂存内容→备份→换入→安装依赖→上报 + prepare_content 负责把插件文件放到调用时给定的暂存目录；只有新内容在暂存 + 目录里完整就位后才会触碰运行目录，中断语义与同步流程一致。 """ + plugin_dir = self.__plugin_dir(pid) + staging_dir = self.__new_install_staging_dir(pid) backup_dir = None try: + success, message = await prepare_content(staging_dir) + if not success: + # 运行目录此刻尚未被触碰，已装插件天然完整，无需还原也无需清理 + logger.error(f"{pid} 准备插件内容失败：{message}") + return False, message + if not force_install: backup_dir = await self.__async_backup_plugin(pid) - await self.__async_remove_old_plugin(pid) - - success, message = await prepare_content() - if not success: - logger.error(f"{pid} 准备插件内容失败：{message}") - if backup_dir: + try: + await self.__async_swap_staged_plugin_content(staging_dir, plugin_dir) + except OSError as error: + message = f"写入插件内容失败：{error}" + logger.error(f"{pid} {message}") + # 换入函数已保证失败时运行目录保持换入前的内容；只有连恢复都失败、 + # 运行目录确实缺失时，才动用本次的临时备份兜底。 + if backup_dir and not plugin_dir.exists(): await self.__async_restore_plugin(pid, backup_dir) logger.warning(f"{pid} 插件安装失败，已还原备份插件") - else: - await self.__async_remove_old_plugin(pid) - logger.warning(f"{pid} 已清理对应插件目录，请尝试重新安装") return False, message dependencies_exist, dep_ok, dep_msg = ( @@ -2058,11 +2194,15 @@ class PluginPackageManager: finally: if backup_dir: await aioshutil.rmtree(backup_dir, ignore_errors=True) + if await AsyncPath(staging_dir).exists(): + await aioshutil.rmtree(staging_dir, ignore_errors=True) def __prepare_content_via_filelist_sync(self, pid: str, user_repo: str, - package_version: Optional[str]) -> tuple[bool, str]: + package_version: Optional[str], + dest_root: Optional[Path] = None) -> tuple[bool, str]: """ 同步准备插件内容，通过文件列表获取插件文件和依赖 + :param dest_root: 文件落盘的根目录，未指定时直接写入插件运行目录 """ runtime_pid = pid.lower() file_list, msg = self.__get_file_list(runtime_pid, user_repo, package_version) @@ -2070,15 +2210,19 @@ class PluginPackageManager: if msg == "插件源码目录不存在": return False, f"{pid} {msg}" return False, msg or "插件文件列表读取失败" - ok, m = self.__download_files(runtime_pid, file_list, user_repo, package_version) + ok, m = self.__download_files( + runtime_pid, file_list, user_repo, package_version, dest_root + ) if not ok: return False, m return True, "" async def __prepare_content_via_filelist_async(self, pid: str, user_repo: str, - package_version: Optional[str]) -> tuple[bool, str]: + package_version: Optional[str], + dest_root: Optional[Path] = None) -> tuple[bool, str]: """ 异步准备插件内容，通过文件列表获取插件文件和依赖 + :param dest_root: 文件落盘的根目录，未指定时直接写入插件运行目录 """ runtime_pid = pid.lower() file_list, msg = await self.__async_get_file_list( @@ -2095,16 +2239,18 @@ class PluginPackageManager: file_list, user_repo, package_version, + dest_root, ) if not ok: return False, m return True, "" - async def __async_install_from_release(self, pid: str, user_repo: str, release_tag: str) -> tuple[bool, str]: + async def __async_install_from_release(self, pid: str, user_repo: str, release_tag: str, + dest_root: Optional[Path] = None) -> tuple[bool, str]: """ 通过 GitHub Release 资产文件安装插件（异步）。 规范：release 中存在名为 "{pid}_v{version}.zip" 的资产，zip 根即插件文件； - 将其全部解压到 app/plugins/{pid} + 将其全部解压到 dest_root，未指定时直接解压到插件运行目录。 """ # 拼接资产文件名 asset_name = f"{release_tag.lower()}.zip" @@ -2152,7 +2298,10 @@ class PluginPackageManager: infos = zf.infolist() if not infos: return False, "压缩包内容为空" - dest_base = self._plugins_root() / pid.lower() + dest_base = ( + dest_root if dest_root is not None + else self._plugins_root() / pid.lower() + ) targets = self.__iter_release_zip_targets(zf, dest_base) wrote_any = False for info, dest_path, is_dir in targets: diff --git a/tests/fixtures/architecture/concurrency-baseline.json b/tests/fixtures/architecture/concurrency-baseline.json index 0c551a7c1..bc2a5efad 100644 --- a/tests/fixtures/architecture/concurrency-baseline.json +++ b/tests/fixtures/architecture/concurrency-baseline.json @@ -65,6 +65,7 @@ "app/adapters/system/plugin/package.py:app.runtime.execution.run_in_threadpool_to_completion": { "owners": { "PluginPackageManager.__async_install_dependencies_if_required": 1, + "PluginPackageManager.__async_swap_staged_plugin_content": 1, "PluginPackageManager.async_activate_persistent_backup": 1, "PluginPackageManager.async_checkpoint": 1, "PluginPackageManager.async_cleanup": 1, diff --git a/tests/test_plugin_adapter_parity.py b/tests/test_plugin_adapter_parity.py index c3f09b978..f6906f5b9 100644 --- a/tests/test_plugin_adapter_parity.py +++ b/tests/test_plugin_adapter_parity.py @@ -127,7 +127,7 @@ async def test_plugin_package_sync_async_preflight_failures_are_identical( {"version": "1.2.3", "release": True}, None, [], - ["release", "remove", "filelist"], + ["release", "filelist"], ), ( {"version": "1.2.3", "release": True}, @@ -139,6 +139,7 @@ async def test_plugin_package_sync_async_preflight_failures_are_identical( ) async def test_plugin_package_sync_async_execute_same_selected_strategy( monkeypatch, + tmp_path: Path, metadata: dict, release_version: str | None, release_items: list[dict], @@ -198,10 +199,10 @@ async def test_plugin_package_sync_async_execute_same_selected_strategy( async_calls.append("remove") def install_flow(_pid, _force, prepare, _repo_url, _before): - return prepare() + return prepare(tmp_path / "staging") async def async_install_flow(_pid, _force, prepare, _repo_url, _before): - return await prepare() + return await prepare(tmp_path / "staging") monkeypatch.setattr(manager, "async_get_plugin_package_version", selected_version) monkeypatch.setattr(manager, "async_get_plugin_release_versions", releases) diff --git a/tests/test_plugin_helper.py b/tests/test_plugin_helper.py index e64746316..9f86b19da 100644 --- a/tests/test_plugin_helper.py +++ b/tests/test_plugin_helper.py @@ -205,12 +205,18 @@ def _patch_sync_remote_install(helper, monkeypatch, meta: dict, monkeypatch.setattr(helper, "_PluginPackageManager__backup_plugin", lambda _pid: None) monkeypatch.setattr(helper, "_PluginPackageManager__remove_old_plugin", lambda _pid: calls.append("remove")) monkeypatch.setattr(helper, "_PluginPackageManager__install_dependencies_if_required", lambda _pid: (False, True, "")) + # 内容准备被替身接管、不会真的产出暂存内容，换入同样替身化，避免测试写入真实运行目录 + monkeypatch.setattr( + helper, + "_PluginPackageManager__swap_staged_plugin_content", + lambda _staging_dir, _final_dir: None, + ) - def fake_release(_pid, _user_repo, _release_tag): + def fake_release(_pid, _user_repo, _release_tag, _dest_root=None): calls.append("release") return release_result - def fake_filelist(_pid, _user_repo, _package_version): + def fake_filelist(_pid, _user_repo, _package_version, _dest_root=None): calls.append("filelist") return filelist_result @@ -240,14 +246,17 @@ def _patch_async_remote_install(helper, monkeypatch, meta: dict, async def fake_dependencies(_pid): return False, True, "" - async def fake_release(_pid, _user_repo, _release_tag): + async def fake_release(_pid, _user_repo, _release_tag, _dest_root=None): calls.append("release") return release_result - async def fake_filelist(_pid, _user_repo, _package_version): + async def fake_filelist(_pid, _user_repo, _package_version, _dest_root=None): calls.append("filelist") return filelist_result + async def fake_swap(_staging_dir, _final_dir): + """内容准备被替身接管，换入同样替身化，避免测试写入真实运行目录。""" + async def fake_to_thread(func, *args, **kwargs): calls.append(("to_thread", func, args, kwargs)) return None @@ -256,6 +265,9 @@ def _patch_async_remote_install(helper, monkeypatch, meta: dict, monkeypatch.setattr(helper, "_PluginPackageManager__async_get_plugin_meta", fake_meta) monkeypatch.setattr(helper, "_PluginPackageManager__async_backup_plugin", fake_backup) monkeypatch.setattr(helper, "_PluginPackageManager__async_remove_old_plugin", fake_remove) + monkeypatch.setattr( + helper, "_PluginPackageManager__async_swap_staged_plugin_content", fake_swap + ) monkeypatch.setattr(helper, "_PluginPackageManager__async_install_dependencies_if_required", fake_dependencies) monkeypatch.setattr(helper, "_PluginPackageManager__async_install_from_release", fake_release) monkeypatch.setattr(helper, "_PluginPackageManager__prepare_content_via_filelist_async", fake_filelist) @@ -2232,11 +2244,11 @@ demo = { index = "private" } assert success assert "" == message - assert ["remove", "release"] == calls + assert ["release"] == calls def test_install_falls_back_to_filelist_when_release_is_missing(self, monkeypatch): """ - release 标记存在但 tag 或 zip 尚未生成时，清理可能残留的安装目录后回退文件列表安装。 + release 标记存在但 tag 或 zip 尚未生成时，清理本次暂存的半份产物后回退文件列表安装。 """ try: from app.adapters.external.market import PluginHelper @@ -2256,11 +2268,11 @@ demo = { index = "private" } assert success assert "" == message - assert ["remove", "release", "remove", "filelist"] == calls + assert ["release", "filelist"] == calls def test_install_reports_filelist_error_after_release_fallback_fails(self, monkeypatch): """ - release 和源码目录都不存在时返回稳定业务错误，并在每次写入前后保持目录可回滚。 + release 和源码目录都不存在时返回稳定业务错误，全程不触碰已安装的运行目录。 """ try: from app.adapters.external.market import PluginHelper @@ -2280,7 +2292,7 @@ demo = { index = "private" } assert not success assert "DemoPlugin 插件源码目录不存在" == message - assert ["remove", "release", "remove", "filelist", "remove"] == calls + assert ["release", "filelist"] == calls def test_install_uses_filelist_when_release_flag_is_disabled(self, monkeypatch): """ @@ -2304,7 +2316,7 @@ demo = { index = "private" } assert success assert "" == message - assert ["remove", "filelist"] == calls + assert ["filelist"] == calls def test_install_rejects_release_without_version(self, monkeypatch): """ @@ -2414,7 +2426,7 @@ demo = { index = "private" } assert not success assert "未找到资产文件：demoplugin_v1.2.0.zip" == message - assert ["remove", "release", "remove"] == calls + assert ["release"] == calls def test_install_rejects_release_version_missing_from_release_list(self, monkeypatch): """ @@ -2508,6 +2520,11 @@ demo = { index = "private" } monkeypatch.setattr(helper, "_PluginPackageManager__remove_old_plugin", lambda _pid: None) monkeypatch.setattr(helper, "_PluginPackageManager__install_dependencies_if_required", lambda _pid: (False, True, "")) monkeypatch.setattr(helper, "_PluginPackageManager__prepare_content_via_filelist_sync", lambda *_args: (True, "")) + monkeypatch.setattr( + helper, + "_PluginPackageManager__swap_staged_plugin_content", + lambda _staging_dir, _final_dir: None, + ) success, message = helper.install_raw(PLUGIN_ID, REPO_URL, force_install=True) @@ -2561,7 +2578,7 @@ demo = { index = "private" } def test_install_release_download_failure_falls_back_to_filelist(self, monkeypatch): """ - release tag 存在但 zip 下载失败时清理可能残留的目录，再回退文件列表安装。 + release tag 存在但 zip 下载失败时清理本次暂存的半份产物，再回退文件列表安装。 """ try: from app.adapters.external.market import PluginHelper @@ -2581,7 +2598,7 @@ demo = { index = "private" } assert success assert "" == message - assert ["remove", "release", "remove", "filelist"] == calls + assert ["release", "filelist"] == calls def test_async_install_uses_release_package_when_asset_is_available(self, monkeypatch): """ @@ -2606,11 +2623,11 @@ demo = { index = "private" } assert success assert "" == message - assert calls == ["remove", "release"] + assert calls == ["release"] def test_async_install_falls_back_to_filelist_when_release_is_missing(self, monkeypatch): """ - 异步安装路径在 release tag 或 zip 未生成时，清理可能残留的安装目录后回退文件列表安装。 + 异步安装路径在 release tag 或 zip 未生成时，清理本次暂存的半份产物后回退文件列表安装。 """ try: from app.adapters.external.market import PluginHelper @@ -2632,7 +2649,7 @@ demo = { index = "private" } assert success assert "" == message - assert calls == ["remove", "release", "remove", "filelist"] + assert calls == ["release", "filelist"] def test_async_install_old_release_version_uses_release_asset_without_filelist_fallback(self, monkeypatch): """ @@ -2666,7 +2683,7 @@ demo = { index = "private" } assert not success assert "未找到资产文件：demoplugin_v1.2.0.zip" == message - assert calls[:3] == ["remove", "release", "remove"] + assert calls[:1] == ["release"] def test_async_install_rejects_release_version_missing_from_release_list(self, monkeypatch): """ @@ -2724,7 +2741,7 @@ demo = { index = "private" } assert not success assert "DemoPlugin 插件源码目录不存在" == message - assert calls == ["remove", "release", "remove", "filelist", "remove"] + assert calls == ["release", "filelist"] def test_async_install_release_fallback_preserves_plugin_id(self, monkeypatch): """ @@ -2745,7 +2762,7 @@ demo = { index = "private" } (True, ""), ) - async def fake_filelist(pid, _user_repo, _package_version): + async def fake_filelist(pid, _user_repo, _package_version, _dest_root=None): filelist_pids.append(pid) return True, "" @@ -2778,7 +2795,7 @@ demo = { index = "private" } (True, ""), ) - async def fake_filelist(pid, _user_repo, _package_version): + async def fake_filelist(pid, _user_repo, _package_version, _dest_root=None): filelist_pids.append(pid) return True, "" @@ -3093,9 +3110,9 @@ demo = { index = "private" } assert not success assert "解压 Release 压缩包失败" in message - def test_install_flow_sync_restores_backup_when_prepare_fails(self, monkeypatch): + def test_install_flow_sync_keeps_installed_plugin_when_prepare_fails(self, monkeypatch): """ - 内容准备失败时恢复备份，避免安装失败后留下半成品目录。 + 内容准备失败时运行目录尚未被触碰，既不需要还原也不该被清理。 """ try: from app.adapters.external.market import PluginHelper @@ -3109,12 +3126,12 @@ demo = { index = "private" } monkeypatch.setattr(helper, "_PluginPackageManager__restore_plugin", lambda _pid, _backup: calls.append("restore")) success, message = helper._PluginPackageManager__install_flow_sync( - PLUGIN_ID, False, lambda: (False, "prepare failed") + PLUGIN_ID, False, lambda _staging_dir: (False, "prepare failed") ) assert not success assert "prepare failed" == message - assert ["remove", "restore"] == calls + assert [] == calls def test_install_flow_sync_restores_backup_when_dependency_install_fails(self, monkeypatch): """ @@ -3130,6 +3147,11 @@ demo = { index = "private" } monkeypatch.setattr(helper, "_PluginPackageManager__backup_plugin", lambda _pid: "/backup") monkeypatch.setattr(helper, "_PluginPackageManager__remove_old_plugin", lambda _pid: calls.append("remove")) monkeypatch.setattr(helper, "_PluginPackageManager__restore_plugin", lambda _pid, _backup: calls.append("restore")) + monkeypatch.setattr( + helper, + "_PluginPackageManager__swap_staged_plugin_content", + lambda _staging_dir, _final_dir: calls.append("swap"), + ) monkeypatch.setattr( helper, "_PluginPackageManager__install_dependencies_if_required", @@ -3137,12 +3159,12 @@ demo = { index = "private" } ) success, message = helper._PluginPackageManager__install_flow_sync( - PLUGIN_ID, False, lambda: (True, "") + PLUGIN_ID, False, lambda _staging_dir: (True, "") ) assert not success assert "dependency failed" == message - assert ["remove", "restore"] == calls + assert ["swap", "restore"] == calls def test_install_flow_sync_restores_backup_for_invalid_modern_manifest(self, tmp_path, monkeypatch): """现代清单无效时恢复旧插件目录。""" @@ -3155,9 +3177,9 @@ demo = { index = "private" } monkeypatch.setattr(market_module, "PLUGIN_DIR", plugin_root) monkeypatch.setattr(market_module.settings, "CONFIG_DIR", str(tmp_path)) - def prepare_content(): - plugin_dir.mkdir(parents=True) - (plugin_dir / "pyproject.toml").write_text( + def prepare_content(staging_dir): + staging_dir.mkdir(parents=True) + (staging_dir / "pyproject.toml").write_text( "[project]\nname = 'demo'\n", encoding="utf-8", ) @@ -3271,7 +3293,7 @@ demo = { index = "private" } assert success assert "" == message - assert calls == [("demoplugin", file_list, "demo/repo", "v2")] + assert calls == [("demoplugin", file_list, "demo/repo", "v2", None)] def test_prepare_content_via_filelist_sync_reports_missing_file_list(self, monkeypatch): """ @@ -3369,7 +3391,7 @@ demo = { index = "private" } assert success assert "" == message - assert calls == [("demoplugin", file_list, "demo/repo", "v2")] + assert calls == [("demoplugin", file_list, "demo/repo", "v2", None)] def test_async_prepare_content_via_filelist_reports_missing_file_list(self, monkeypatch): """ @@ -3453,9 +3475,9 @@ demo = { index = "private" } assert not success assert "download failed" == message - def test_install_flow_async_restores_backup_when_prepare_fails(self, monkeypatch): + def test_install_flow_async_keeps_installed_plugin_when_prepare_fails(self, monkeypatch): """ - 异步内容准备失败时恢复备份。 + 异步内容准备失败时运行目录尚未被触碰，既不需要还原也不该被清理。 """ try: from app.adapters.external.market import PluginHelper @@ -3474,7 +3496,7 @@ demo = { index = "private" } async def restore(_pid, _backup): calls.append("restore") - async def prepare(): + async def prepare(_staging_dir): return False, "prepare failed" monkeypatch.setattr(helper, "_PluginPackageManager__async_backup_plugin", backup) @@ -3485,7 +3507,7 @@ demo = { index = "private" } assert not success assert "prepare failed" == message - assert ["remove", "restore"] == calls + assert [] == calls def test_install_flow_async_restores_backup_when_dependency_install_fails(self, monkeypatch): """ @@ -3508,22 +3530,26 @@ demo = { index = "private" } async def restore(_pid, _backup): calls.append("restore") - async def prepare(): + async def prepare(_staging_dir): return True, "" + async def swap(_staging_dir, _final_dir): + calls.append("swap") + async def dependencies(_pid): return True, False, "dependency failed" monkeypatch.setattr(helper, "_PluginPackageManager__async_backup_plugin", backup) monkeypatch.setattr(helper, "_PluginPackageManager__async_remove_old_plugin", remove) monkeypatch.setattr(helper, "_PluginPackageManager__async_restore_plugin", restore) + monkeypatch.setattr(helper, "_PluginPackageManager__async_swap_staged_plugin_content", swap) monkeypatch.setattr(helper, "_PluginPackageManager__async_install_dependencies_if_required", dependencies) success, message = asyncio.run(helper._PluginPackageManager__install_flow_async(PLUGIN_ID, False, prepare)) assert not success assert "dependency failed" == message - assert ["remove", "restore"] == calls + assert ["swap", "restore"] == calls def test_async_install_from_release_reports_missing_asset(self, monkeypatch): """ diff --git a/tests/test_plugin_install_atomic_swap.py b/tests/test_plugin_install_atomic_swap.py new file mode 100644 index 000000000..63d87c3ea --- /dev/null +++ b/tests/test_plugin_install_atomic_swap.py @@ -0,0 +1,356 @@ +"""插件安装的换入原子性：准备失败不得删掉已装插件，换入失败必须还原运行目录。""" + +import errno +import shutil +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from app.adapters.system.plugin.package import PluginPackageManager + +PLUGIN_ID = "DemoPlugin" +REPO_URL = "https://github.com/demo/MoviePilot-Plugins" +INSTALLED_MARK = "installed" +DOWNLOAD_ERROR = "文件 plugins.v2/demoplugin/__init__.py 下载失败！" + + +def _swap(staging_dir: Path, final_dir: Path) -> None: + """调用被测的换入实现，保持用例只表达行为约定。""" + PluginPackageManager._PluginPackageManager__swap_staged_plugin_content( # type: ignore[attr-defined] + staging_dir, final_dir + ) + + +def _installed_manager(monkeypatch, tmp_path: Path) -> tuple[PluginPackageManager, Path]: + """构造隔离目录的包 owner，并预置一个已安装且可用的插件运行目录。""" + plugin_root = tmp_path / "plugins" + settings = SimpleNamespace( + ROOT_PATH=tmp_path, + TEMP_PATH=tmp_path / "temp", + CONFIG_PATH=tmp_path / "config", + ) + monkeypatch.setattr( + "app.adapters.system.plugin.package.get_runtime_setting", + lambda key: getattr(settings, key), + ) + plugin_dir = plugin_root / PLUGIN_ID.lower() + plugin_dir.mkdir(parents=True) + (plugin_dir / "__init__.py").write_text(INSTALLED_MARK, encoding="utf-8") + manager = PluginPackageManager(source=Mock(), plugin_root=plugin_root) + return manager, plugin_dir + + +def _stub_market_lookup(manager: PluginPackageManager, monkeypatch) -> None: + """把远端市场查询固定下来，让内容准备的成败成为唯一变量。""" + monkeypatch.setattr( + "app.adapters.system.plugin.package.SystemUtils.is_frozen", lambda: False + ) + monkeypatch.setattr(manager, "is_local_repo_url", lambda _repo_url: False) + monkeypatch.setattr(manager, "get_repo_info", lambda _repo_url: ("demo", "repo")) + monkeypatch.setattr(manager, "get_plugin_package_version", lambda *_args: "v2") + monkeypatch.setattr( + manager, + "_PluginPackageManager__get_plugin_meta", + lambda *_args: {"release": False, "version": "1.0.0"}, + ) + + async def async_package_version(*_args) -> str: + """异步入口复用同一索引代际。""" + return "v2" + + async def async_meta(*_args) -> dict: + """异步入口复用同一插件元数据。""" + return {"release": False, "version": "1.0.0"} + + monkeypatch.setattr( + manager, "async_get_plugin_package_version", async_package_version + ) + monkeypatch.setattr( + manager, "_PluginPackageManager__async_get_plugin_meta", async_meta + ) + + +def _staging_root(tmp_path: Path) -> Path: + """返回安装暂存根目录。""" + return tmp_path / "temp" / "plugin_install_staging" + + +def _assert_no_staging_residue(tmp_path: Path) -> None: + """安装结束后不得残留暂存目录。""" + staging_root = _staging_root(tmp_path) + assert not staging_root.exists() or not list(staging_root.iterdir()) + + +def _assert_no_swap_residue(plugin_dir: Path) -> None: + """换入结束后不得在插件根目录残留换入用的回滚材料。""" + assert not list(plugin_dir.parent.glob(f".{plugin_dir.name}.previous-*")) + + +@pytest.mark.parametrize("force_install", [True, False]) +def test_sync_install_keeps_installed_plugin_when_content_preparation_fails( + monkeypatch, tmp_path: Path, force_install: bool +) -> None: + """同步安装在下载失败时必须原样保留已装插件，无论本次是否留了备份。""" + manager, plugin_dir = _installed_manager(monkeypatch, tmp_path) + _stub_market_lookup(manager, monkeypatch) + monkeypatch.setattr( + manager, + "_PluginPackageManager__prepare_content_via_filelist_sync", + lambda *_args: (False, DOWNLOAD_ERROR), + ) + + success, message = manager.install_raw( + PLUGIN_ID, REPO_URL, package_version="v2", force_install=force_install + ) + + assert not success + assert message == DOWNLOAD_ERROR + assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK + _assert_no_staging_residue(tmp_path) + _assert_no_swap_residue(plugin_dir) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("force_install", [True, False]) +async def test_async_install_keeps_installed_plugin_when_content_preparation_fails( + monkeypatch, tmp_path: Path, force_install: bool +) -> None: + """异步安装与同步保持一致的失败语义，下载失败不得删掉已装插件。""" + manager, plugin_dir = _installed_manager(monkeypatch, tmp_path) + _stub_market_lookup(manager, monkeypatch) + + async def failing_filelist(*_args) -> tuple[bool, str]: + """模拟资产瞬时不可用导致的内容准备失败。""" + return False, DOWNLOAD_ERROR + + monkeypatch.setattr( + manager, + "_PluginPackageManager__prepare_content_via_filelist_async", + failing_filelist, + ) + + success, message = await manager.async_install_raw( + PLUGIN_ID, REPO_URL, package_version="v2", force_install=force_install + ) + + assert not success + assert message == DOWNLOAD_ERROR + assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK + _assert_no_staging_residue(tmp_path) + _assert_no_swap_residue(plugin_dir) + + +def test_sync_install_replaces_installed_plugin_when_content_is_ready( + monkeypatch, tmp_path: Path +) -> None: + """同步安装成功时用新内容整体替换运行目录，不与旧文件混合。""" + manager, plugin_dir = _installed_manager(monkeypatch, tmp_path) + (plugin_dir / "stale.py").write_text("stale", encoding="utf-8") + _stub_market_lookup(manager, monkeypatch) + + def prepare(_pid, _user_repo, _package_version, dest_root: Path) -> tuple[bool, str]: + """把新版本内容写进本次分配的暂存目录。""" + dest_root.mkdir(parents=True, exist_ok=True) + (dest_root / "__init__.py").write_text("upgraded", encoding="utf-8") + return True, "" + + monkeypatch.setattr( + manager, "_PluginPackageManager__prepare_content_via_filelist_sync", prepare + ) + + success, message = manager.install_raw( + PLUGIN_ID, REPO_URL, package_version="v2", force_install=True + ) + + assert (success, message) == (True, "") + assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == "upgraded" + assert not (plugin_dir / "stale.py").exists() + _assert_no_staging_residue(tmp_path) + _assert_no_swap_residue(plugin_dir) + + +@pytest.mark.asyncio +async def test_async_install_replaces_installed_plugin_when_content_is_ready( + monkeypatch, tmp_path: Path +) -> None: + """异步安装成功时同样整体替换运行目录。""" + manager, plugin_dir = _installed_manager(monkeypatch, tmp_path) + (plugin_dir / "stale.py").write_text("stale", encoding="utf-8") + _stub_market_lookup(manager, monkeypatch) + + async def prepare( + _pid, _user_repo, _package_version, dest_root: Path + ) -> tuple[bool, str]: + """把新版本内容写进本次分配的暂存目录。""" + dest_root.mkdir(parents=True, exist_ok=True) + (dest_root / "__init__.py").write_text("upgraded", encoding="utf-8") + return True, "" + + monkeypatch.setattr( + manager, "_PluginPackageManager__prepare_content_via_filelist_async", prepare + ) + + success, message = await manager.async_install_raw( + PLUGIN_ID, REPO_URL, package_version="v2", force_install=True + ) + + assert (success, message) == (True, "") + assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == "upgraded" + assert not (plugin_dir / "stale.py").exists() + _assert_no_staging_residue(tmp_path) + _assert_no_swap_residue(plugin_dir) + + +def test_sync_install_keeps_installed_plugin_when_swap_fails( + monkeypatch, tmp_path: Path +) -> None: + """换入自身失败时运行目录必须仍是换入前那一份，并报出失败原因。""" + manager, plugin_dir = _installed_manager(monkeypatch, tmp_path) + _stub_market_lookup(manager, monkeypatch) + + def prepare(_pid, _user_repo, _package_version, dest_root: Path) -> tuple[bool, str]: + """内容准备成功，但随后的换入会被模拟为失败。""" + dest_root.mkdir(parents=True, exist_ok=True) + (dest_root / "__init__.py").write_text("upgraded", encoding="utf-8") + return True, "" + + def failing_swap(_staging_dir: Path, _final_dir: Path) -> None: + """模拟运行目录不可写导致的换入失败。""" + raise OSError(errno.EACCES, "permission denied") + + monkeypatch.setattr( + manager, "_PluginPackageManager__prepare_content_via_filelist_sync", prepare + ) + monkeypatch.setattr( + manager, "_PluginPackageManager__swap_staged_plugin_content", failing_swap + ) + + success, message = manager.install_raw( + PLUGIN_ID, REPO_URL, package_version="v2", force_install=True + ) + + assert not success + assert "写入插件内容失败" in message + assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK + _assert_no_staging_residue(tmp_path) + + +def _staged_pair(tmp_path: Path) -> tuple[Path, Path]: + """准备一份待换入内容和一个已有旧内容的目标目录。""" + staging_dir = tmp_path / "staging" + staging_dir.mkdir() + (staging_dir / "__init__.py").write_text("upgraded", encoding="utf-8") + final_dir = tmp_path / "plugins" / "demoplugin" + final_dir.mkdir(parents=True) + (final_dir / "__init__.py").write_text(INSTALLED_MARK, encoding="utf-8") + (final_dir / "stale.py").write_text("stale", encoding="utf-8") + return staging_dir, final_dir + + +def _break_replace(monkeypatch, *, source: Path, code: int) -> None: + """让指定源路径的原子改名失败，模拟跨文件系统或权限受限。""" + original = Path.replace + + def guarded(self: Path, target): + """只拦截被指定的源路径，其它改名仍走真实实现。""" + if self == source: + raise OSError(code, "simulated") + return original(self, target) + + monkeypatch.setattr(Path, "replace", guarded) + + +def test_swap_publishes_new_content_and_drops_old_content(tmp_path: Path) -> None: + """换入后运行目录只剩新内容，回滚材料被清理。""" + staging_dir, final_dir = _staged_pair(tmp_path) + + _swap(staging_dir, final_dir) + + assert (final_dir / "__init__.py").read_text(encoding="utf-8") == "upgraded" + assert not (final_dir / "stale.py").exists() + _assert_no_swap_residue(final_dir) + + +def test_swap_falls_back_to_copy_when_staging_crosses_filesystems( + monkeypatch, tmp_path: Path +) -> None: + """暂存目录与插件根目录跨文件系统时退化为复制，结果与改名一致。""" + staging_dir, final_dir = _staged_pair(tmp_path) + _break_replace(monkeypatch, source=staging_dir, code=errno.EXDEV) + + _swap(staging_dir, final_dir) + + assert (final_dir / "__init__.py").read_text(encoding="utf-8") == "upgraded" + assert not (final_dir / "stale.py").exists() + _assert_no_swap_residue(final_dir) + + +def test_swap_falls_back_to_copy_when_old_directory_cannot_be_renamed( + monkeypatch, tmp_path: Path +) -> None: + """overlayfs 拒绝改名旧目录时先复制出回滚材料，再继续换入新内容。""" + staging_dir, final_dir = _staged_pair(tmp_path) + _break_replace(monkeypatch, source=final_dir, code=errno.EXDEV) + + _swap(staging_dir, final_dir) + + assert (final_dir / "__init__.py").read_text(encoding="utf-8") == "upgraded" + assert not (final_dir / "stale.py").exists() + _assert_no_swap_residue(final_dir) + + +def test_swap_keeps_old_content_when_staging_is_missing(tmp_path: Path) -> None: + """待换入内容不存在时运行目录必须原样退回，而不是被清空。""" + staging_dir, final_dir = _staged_pair(tmp_path) + shutil.rmtree(staging_dir) + + with pytest.raises(OSError): + _swap(staging_dir, final_dir) + + assert (final_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK + assert (final_dir / "stale.py").exists() + _assert_no_swap_residue(final_dir) + + +def test_swap_keeps_old_content_when_old_directory_cannot_be_moved_aside( + monkeypatch, tmp_path: Path +) -> None: + """旧目录挪不开时必须整体放弃换入，绝不先删后写。""" + staging_dir, final_dir = _staged_pair(tmp_path) + _break_replace(monkeypatch, source=final_dir, code=errno.EACCES) + + with pytest.raises(OSError): + _swap(staging_dir, final_dir) + + assert (final_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK + assert (final_dir / "stale.py").exists() + assert (staging_dir / "__init__.py").exists() + _assert_no_swap_residue(final_dir) + + +def test_swap_rolls_back_when_cross_filesystem_copy_fails( + monkeypatch, tmp_path: Path +) -> None: + """跨文件系统复制中途失败时删掉半成品并把旧目录换回。""" + staging_dir, final_dir = _staged_pair(tmp_path) + _break_replace(monkeypatch, source=staging_dir, code=errno.EXDEV) + + def failing_copytree(src, dst, *_args, **_kwargs): + """模拟复制到一半磁盘写入失败，留下半份目标目录。""" + Path(dst).mkdir(parents=True, exist_ok=True) + (Path(dst) / "partial.py").write_text("partial", encoding="utf-8") + raise OSError(errno.ENOSPC, "no space left on device") + + monkeypatch.setattr( + "app.adapters.system.plugin.package.shutil.copytree", failing_copytree + ) + + with pytest.raises(OSError): + _swap(staging_dir, final_dir) + + assert (final_dir / "__init__.py").read_text(encoding="utf-8") == INSTALLED_MARK + assert (final_dir / "stale.py").exists() + assert not (final_dir / "partial.py").exists() + _assert_no_swap_residue(final_dir) diff --git a/tests/test_plugin_install_backup_cleanup.py b/tests/test_plugin_install_backup_cleanup.py index 5ea152932..2b3090ce4 100644 --- a/tests/test_plugin_install_backup_cleanup.py +++ b/tests/test_plugin_install_backup_cleanup.py @@ -33,6 +33,11 @@ async def test_successful_install_flows_remove_transient_backups( "_PluginPackageManager__remove_old_plugin", lambda _pid: None, ) + monkeypatch.setattr( + package, + "_PluginPackageManager__swap_staged_plugin_content", + lambda _staging_dir, _final_dir: None, + ) monkeypatch.setattr( package, "_PluginPackageManager__install_dependencies_if_required", @@ -47,7 +52,7 @@ async def test_successful_install_flows_remove_transient_backups( sync_result = package._PluginPackageManager__install_flow_sync( "DemoPlugin", False, - lambda: (True, ""), + lambda _staging_dir: (True, ""), ) async def backup_plugin(_pid: str) -> str: @@ -61,10 +66,13 @@ async def test_successful_install_flows_remove_transient_backups( """表示测试插件没有额外依赖。""" return False, False, "不存在依赖" - async def prepare_content() -> tuple[bool, str]: + async def prepare_content(_staging_dir: Path) -> tuple[bool, str]: """表示异步内容准备成功。""" return True, "" + async def swap_content(_staging_dir: Path, _final_dir: Path) -> None: + """隔离测试中的真实运行目录换入。""" + monkeypatch.setattr( package, "_PluginPackageManager__async_backup_plugin", @@ -75,6 +83,11 @@ async def test_successful_install_flows_remove_transient_backups( "_PluginPackageManager__async_remove_old_plugin", remove_plugin, ) + monkeypatch.setattr( + package, + "_PluginPackageManager__async_swap_staged_plugin_content", + swap_content, + ) monkeypatch.setattr( package, "_PluginPackageManager__async_install_dependencies_if_required",