From 93a8dca5e9b4024215342ed51418a4eb1de150e2 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 29 Aug 2026 21:23:12 +0800 Subject: [PATCH] refactor(plugin): unify adapter decisions --- app/adapters/external/plugin/client.py | 56 +++-- app/adapters/system/plugin/dependency.py | 71 ++++-- app/adapters/system/plugin/package.py | 215 +++++++++++------ tests/test_plugin_adapter_parity.py | 293 +++++++++++++++++++++++ 4 files changed, 515 insertions(+), 120 deletions(-) create mode 100644 tests/test_plugin_adapter_parity.py diff --git a/app/adapters/external/plugin/client.py b/app/adapters/external/plugin/client.py index 386f6b560..40c047347 100644 --- a/app/adapters/external/plugin/client.py +++ b/app/adapters/external/plugin/client.py @@ -976,11 +976,35 @@ class PluginMarketTransport: content: str, ) -> Optional[PluginIndex]: """统一解释插件索引 HTTP 响应，保留不存在、失败和有效索引三态。""" - if status_code == 404: - return {} - if status_code != 200: + try: + payload = cls._resolve_plugin_index_result((status_code, content)) + except RuntimeError: return None - return cls.__parse_plugin_index_response(content) + return payload if payload is not None else {} + + @classmethod + def _resolve_plugin_index_result( + cls, + response: Optional[tuple[int, str]], + ) -> Optional[PluginIndex]: + """ + 把传输结果统一映射为索引读取合同。 + + :param response: 传输层返回的状态码和已受限响应文本；None 表示连接失败 + :return: 有效索引，或以 None 表示仓库明确不存在该索引 + :raises RuntimeError: 连接失败、非成功状态或响应格式无效 + """ + if response is None: + raise RuntimeError("插件索引请求失败：连接失败") + status_code, content = response + if status_code == 404: + return None + if status_code != 200: + raise RuntimeError(f"插件索引请求失败：HTTP {status_code}") + payload = cls.__parse_plugin_index_response(content) + if payload is None: + raise RuntimeError("插件索引响应格式无效") + return payload @staticmethod def __build_plugin_release_item( @@ -1120,17 +1144,7 @@ class PluginMarketTransport: package_url, headers=headers, ) - if response is None: - raise RuntimeError("插件索引请求失败：连接失败") - status_code, content = response - if status_code == 404: - return None - if status_code != 200: - raise RuntimeError(f"插件索引请求失败：HTTP {status_code}") - payload = self.__parse_plugin_index_response(content) - if payload is None: - raise RuntimeError("插件索引响应格式无效") - return payload + return self._resolve_plugin_index_result(response) def get_plugins(self, repo_url: str, package_version: Optional[str] = None) -> Optional[PluginIndex]: @@ -1459,17 +1473,7 @@ class PluginMarketTransport: package_url, headers=headers, ) - if response is None: - raise RuntimeError("插件索引请求失败：连接失败") - status_code, content = response - if status_code == 404: - return None - if status_code != 200: - raise RuntimeError(f"插件索引请求失败：HTTP {status_code}") - payload = self.__parse_plugin_index_response(content) - if payload is None: - raise RuntimeError("插件索引响应格式无效") - return payload + return self._resolve_plugin_index_result(response) async def async_get_plugins(self, repo_url: str, package_version: Optional[str] = None) -> Optional[PluginIndex]: diff --git a/app/adapters/system/plugin/dependency.py b/app/adapters/system/plugin/dependency.py index 368182fdd..593d0bda0 100644 --- a/app/adapters/system/plugin/dependency.py +++ b/app/adapters/system/plugin/dependency.py @@ -35,6 +35,14 @@ class _RequirementGroup: specifiers: set[str] = field(default_factory=set) # 待求交集的版本约束 +@dataclass(frozen=True, slots=True) +class _DependencyInstallRequest: + """封装一次插件依赖安装所需的清单和本地 wheel 来源。""" + + manifest_paths: list[Path] + wheels_dirs: list[Path] + + class PluginDependencyPackagePort(Protocol): """声明依赖聚合器所需的唯一 Python 包安装边界。""" @@ -360,21 +368,46 @@ class PluginDependencyInstaller: result.append(wheels_dir) return list(dict.fromkeys(result)) - def install(self, dependencies: list[str]) -> tuple[bool, str]: - """把已安装插件的原始清单交给一次统一包安装。""" + def _prepare_install_request( + self, dependencies: list[str] + ) -> tuple[Optional[_DependencyInstallRequest], Optional[tuple[bool, str]]]: + """统一校验依赖安装请求，并在进入包 I/O 前构造规范输入。""" if not dependencies: - return False, "没有传入需要安装的依赖项" + return None, (False, "没有传入需要安装的依赖项") try: manifest_paths = [manifest.path for manifest in self._plugin_manifests()] if not manifest_paths: - return False, "没有找到已安装插件的依赖清单" - return self._packages.install_packages_with_fallback( - manifest_paths, - self._wheels_dirs(), + return None, (False, "没有找到已安装插件的依赖清单") + return ( + _DependencyInstallRequest( + manifest_paths=manifest_paths, + wheels_dirs=self._wheels_dirs(), + ), + None, ) - except Exception as err: - logger.error(f"安装依赖项时发生错误：{err}") - return False, f"安装依赖项时发生错误：{err}" + except Exception as error: # noqa: BLE001 - 统一映射为公开安装结果 + return None, self._dependency_install_failure(error) + + @staticmethod + def _dependency_install_failure(error: Exception) -> tuple[bool, str]: + """统一记录同步与异步包安装异常，并保留既有错误文本合同。""" + message = f"安装依赖项时发生错误：{error}" + logger.error(message) + return False, message + + def install(self, dependencies: list[str]) -> tuple[bool, str]: + """把已安装插件的原始清单交给一次统一包安装。""" + request, error_result = self._prepare_install_request(dependencies) + if error_result is not None: + return error_result + assert request is not None + try: + return self._packages.install_packages_with_fallback( + request.manifest_paths, + request.wheels_dirs, + ) + except Exception as error: # noqa: BLE001 - 统一映射为公开安装结果 + return self._dependency_install_failure(error) async def async_find_missing(self) -> list[str]: """在线程池中扫描缺失依赖，避免阻塞事件循环。""" @@ -382,16 +415,14 @@ class PluginDependencyInstaller: async def async_install(self, dependencies: list[str]) -> tuple[bool, str]: """异步安装依赖，使用可取消的包安装子进程。""" - if not dependencies: - return False, "没有传入需要安装的依赖项" + request, error_result = self._prepare_install_request(dependencies) + if error_result is not None: + return error_result + assert request is not None try: - manifest_paths = [manifest.path for manifest in self._plugin_manifests()] - if not manifest_paths: - return False, "没有找到已安装插件的依赖清单" return await self._packages.async_install_packages_with_fallback( - manifest_paths, - self._wheels_dirs(), + request.manifest_paths, + request.wheels_dirs, ) - except Exception as err: - logger.error(f"安装依赖项时发生错误：{err}") - return False, f"安装依赖项时发生错误：{err}" + except Exception as error: # noqa: BLE001 - 统一映射为公开安装结果 + return self._dependency_install_failure(error) diff --git a/app/adapters/system/plugin/package.py b/app/adapters/system/plugin/package.py index 4d9e18857..90d4730aa 100644 --- a/app/adapters/system/plugin/package.py +++ b/app/adapters/system/plugin/package.py @@ -161,6 +161,16 @@ class PluginPackageCheckpoint: return self.transaction_dir / ".rollback-complete" +@dataclass(frozen=True, slots=True) +class _RemotePluginInstallSelection: + """冻结远端插件完成准入后的安装方式和索引事实。""" + + user_repo: str + package_version: str + release_tag: Optional[str] + fallback_to_filelist: bool + + class PluginPackageManager: """隔离插件包安装、本地同步、分身改写和文件补偿能力。""" @@ -760,6 +770,69 @@ class PluginPackageManager: current_version=get_app_version(), ) + def _resolve_remote_plugin_repository( + self, + pid: str, + repo_url: str, + ) -> tuple[Optional[str], Optional[tuple[bool, str]]]: + """统一校验远端安装前置条件并返回规范仓库标识。""" + if SystemUtils.is_frozen(): + return None, (False, "可执行文件模式下，只能安装本地插件") + if not pid or not repo_url: + return None, (False, "参数错误") + user, repo = self.get_repo_info(repo_url) + if not user or not repo: + return None, (False, "不支持的插件仓库地址格式") + return f"{user}/{repo}", None + + @staticmethod + def _accept_remote_package_version( + pid: str, + package_version: Optional[str], + ) -> Optional[tuple[bool, str]]: + """统一记录索引代际选择，并映射找不到兼容插件的失败结果。""" + if package_version is None: + message = f"{pid} 没有找到适用于当前版本的插件" + logger.debug(message) + return False, message + if package_version: + logger.debug( + f"{pid} 从 package.{package_version}.json 中找到适用于当前版本的插件" + ) + else: + logger.debug(f"{pid} 从 package.json 中找到适用于当前版本的插件") + return None + + @classmethod + def _select_remote_install( + cls, + *, + pid: str, + user_repo: str, + package_version: str, + meta: dict[str, Any], + release_version: Optional[str], + release_items: Sequence[dict[str, Any]], + ) -> tuple[Optional[_RemotePluginInstallSelection], str]: + """把元数据和 Release 列表统一映射为可执行的远端安装选择。""" + plan, message = cls._build_remote_plugin_install_plan( + pid=pid, + meta=meta, + release_version=release_version, + release_items=release_items, + ) + if plan is None: + return None, message + return ( + _RemotePluginInstallSelection( + user_repo=user_repo, + package_version=package_version, + release_tag=plan.release_tag, + fallback_to_filelist=plan.fallback_to_filelist, + ), + "", + ) + def install_packages_with_fallback( self, dependency_files: Path | Sequence[Path], @@ -851,58 +924,47 @@ class PluginPackageManager: before_dependency_install=before_dependency_install, ) - if SystemUtils.is_frozen(): - return False, "可执行文件模式下，只能安装本地插件" + user_repo, error_result = self._resolve_remote_plugin_repository( + pid, repo_url + ) + if error_result is not None: + return error_result + assert user_repo is not None - # 验证参数 - if not pid or not repo_url: - return False, "参数错误" - - # 从 GitHub 的 repo_url 获取用户和项目名 - user, repo = self.get_repo_info(repo_url) - if not user or not repo: - return False, "不支持的插件仓库地址格式" - - user_repo = f"{user}/{repo}" - - if not package_version: - package_version = get_runtime_setting('VERSION_FLAG') - - # 1. 优先检查指定版本的插件 - package_version = self.get_plugin_package_version(pid, repo_url, package_version) - # 如果 package_version 为None，说明没有找到匹配的插件 - if package_version is None: - msg = f"{pid} 没有找到适用于当前版本的插件" - logger.debug(msg) - return False, msg - # package_version 为空，表示从 package.json 中找到插件 - elif package_version == "": - logger.debug(f"{pid} 从 package.json 中找到适用于当前版本的插件") - else: - logger.debug(f"{pid} 从 package.{package_version}.json 中找到适用于当前版本的插件") + preferred_version = package_version or get_runtime_setting('VERSION_FLAG') + selected_version = self.get_plugin_package_version( + pid, repo_url, preferred_version + ) + if error_result := self._accept_remote_package_version( + pid, selected_version + ): + return error_result + assert selected_version is not None # 2. 决定安装方式（release 或文件列表）并执行统一安装流程。 - meta = self.__get_plugin_meta(pid, repo_url, package_version) + meta = self.__get_plugin_meta(pid, repo_url, selected_version) release_items = ( self.get_plugin_release_versions(pid, repo_url) if release_version else [] ) - plan, message = self._build_remote_plugin_install_plan( + selection, message = self._select_remote_install( pid=pid, + user_repo=user_repo, + package_version=selected_version, meta=meta, release_version=release_version, release_items=release_items, ) - if plan is None: + if selection is None: return False, message - release_tag = plan.release_tag - if release_tag and not plan.fallback_to_filelist: + release_tag = selection.release_tag + if release_tag and not selection.fallback_to_filelist: def prepare_selected_release() -> tuple[bool, str]: return self.__install_from_release( pid, - user_repo, + selection.user_repo, release_tag, ) @@ -919,14 +981,18 @@ class PluginPackageManager: def prepare_release() -> tuple[bool, str]: ok, msg = self.__install_from_release( pid, - user_repo, + selection.user_repo, release_tag, ) if ok: return True, msg logger.warning(f"{pid} Release 安装失败，回退文件列表安装：{msg}") self.__remove_old_plugin(pid) - return self.__prepare_content_via_filelist_sync(pid, user_repo, package_version) + return self.__prepare_content_via_filelist_sync( + pid, + selection.user_repo, + selection.package_version, + ) return self.__install_flow_sync( pid, @@ -937,7 +1003,11 @@ class PluginPackageManager: ) # 未声明 release 打包的插件继续使用文件列表方式安装。 def prepare_filelist() -> tuple[bool, str]: - return self.__prepare_content_via_filelist_sync(pid, user_repo, package_version) + return self.__prepare_content_via_filelist_sync( + pid, + selection.user_repo, + selection.package_version, + ) return self.__install_flow_sync( pid, @@ -1754,58 +1824,47 @@ class PluginPackageManager: ), ) - if SystemUtils.is_frozen(): - return False, "可执行文件模式下，只能安装本地插件" + user_repo, error_result = self._resolve_remote_plugin_repository( + pid, repo_url + ) + if error_result is not None: + return error_result + assert user_repo is not None - # 验证参数 - if not pid or not repo_url: - return False, "参数错误" - - # 从 GitHub 的 repo_url 获取用户和项目名 - user, repo = self.get_repo_info(repo_url) - if not user or not repo: - return False, "不支持的插件仓库地址格式" - - user_repo = f"{user}/{repo}" - - if not package_version: - package_version = get_runtime_setting('VERSION_FLAG') - - # 1. 优先检查指定版本的插件 - package_version = await self.async_get_plugin_package_version(pid, repo_url, package_version) - # 如果 package_version 为None，说明没有找到匹配的插件 - if package_version is None: - msg = f"{pid} 没有找到适用于当前版本的插件" - logger.debug(msg) - return False, msg - # package_version 为空，表示从 package.json 中找到插件 - elif package_version == "": - logger.debug(f"{pid} 从 package.json 中找到适用于当前版本的插件") - else: - logger.debug(f"{pid} 从 package.{package_version}.json 中找到适用于当前版本的插件") + preferred_version = package_version or get_runtime_setting('VERSION_FLAG') + selected_version = await self.async_get_plugin_package_version( + pid, repo_url, preferred_version + ) + if error_result := self._accept_remote_package_version( + pid, selected_version + ): + return error_result + assert selected_version is not None # 2. 统一异步安装流程（release 或文件列表）。 - meta = await self.__async_get_plugin_meta(pid, repo_url, package_version) + meta = await self.__async_get_plugin_meta(pid, repo_url, selected_version) release_items = ( await self.async_get_plugin_release_versions(pid, repo_url) if release_version else [] ) - plan, message = self._build_remote_plugin_install_plan( + selection, message = self._select_remote_install( pid=pid, + user_repo=user_repo, + package_version=selected_version, meta=meta, release_version=release_version, release_items=release_items, ) - if plan is None: + if selection is None: return False, message - release_tag = plan.release_tag - if release_tag and not plan.fallback_to_filelist: + release_tag = selection.release_tag + if release_tag and not selection.fallback_to_filelist: async def prepare_selected_release() -> tuple[bool, str]: return await self.__async_install_from_release( pid, - user_repo, + selection.user_repo, release_tag, ) @@ -1822,14 +1881,18 @@ class PluginPackageManager: async def prepare_release() -> tuple[bool, str]: ok, msg = await self.__async_install_from_release( pid, - user_repo, + selection.user_repo, release_tag, ) if ok: return True, msg logger.warning(f"{pid} Release 安装失败，回退文件列表安装：{msg}") await self.__async_remove_old_plugin(pid) - return await self.__prepare_content_via_filelist_async(pid, user_repo, package_version) + return await self.__prepare_content_via_filelist_async( + pid, + selection.user_repo, + selection.package_version, + ) return await self.__install_flow_async( pid, @@ -1840,7 +1903,11 @@ class PluginPackageManager: ) # 未声明 release 打包的插件继续使用文件列表方式安装。 async def prepare_filelist() -> tuple[bool, str]: - return await self.__prepare_content_via_filelist_async(pid, user_repo, package_version) + return await self.__prepare_content_via_filelist_async( + pid, + selection.user_repo, + selection.package_version, + ) return await self.__install_flow_async( pid, diff --git a/tests/test_plugin_adapter_parity.py b/tests/test_plugin_adapter_parity.py new file mode 100644 index 000000000..c3f09b978 --- /dev/null +++ b/tests/test_plugin_adapter_parity.py @@ -0,0 +1,293 @@ +"""插件市场、包安装和依赖安装同步异步一致性测试。""" + +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from app.adapters.external.plugin.client import PluginMarketTransport +from app.adapters.system.plugin.dependency import PluginDependencyInstaller +from app.adapters.system.plugin.package import PluginPackageManager + +REPO_URL = "https://github.com/example/moviepilot-plugins" +SYNC_INDEX_REQUEST = "_PluginMarketTransport__request_plugin_index_with_fallback" +ASYNC_INDEX_REQUEST = ( + "_PluginMarketTransport__async_request_plugin_index_with_fallback" +) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("response", "expected", "error"), + [ + ((404, ""), None, None), + ((200, '{"DemoPlugin": {"version": "1.2.3"}}'), + {"DemoPlugin": {"version": "1.2.3"}}, None), + (None, None, "插件索引请求失败：连接失败"), + ((500, "upstream failed"), None, "插件索引请求失败：HTTP 500"), + ((200, "not-json"), None, "插件索引响应格式无效"), + ], +) +async def test_plugin_index_sync_async_result_contract_is_identical( + monkeypatch, + response: tuple[int, str] | None, + expected: dict | None, + error: str | None, +) -> None: + """同步与异步索引传输必须经过同一响应分类和错误语义。""" + transport = PluginMarketTransport() + + def request(_url: str, *, headers: dict) -> tuple[int, str] | None: + return response + + async def async_request( + _url: str, *, headers: dict + ) -> tuple[int, str] | None: + return response + + monkeypatch.setattr(transport, SYNC_INDEX_REQUEST, request) + monkeypatch.setattr(transport, ASYNC_INDEX_REQUEST, async_request) + transport.get_plugin_index_result.cache_clear() + await transport.async_get_plugin_index_result.cache_clear() + + if error is not None: + with pytest.raises(RuntimeError, match=error): + transport.get_plugin_index_result(REPO_URL, "v3") + with pytest.raises(RuntimeError, match=error): + await transport.async_get_plugin_index_result(REPO_URL, "v3") + return + + sync_result = transport.get_plugin_index_result(REPO_URL, "v3") + async_result = await transport.async_get_plugin_index_result(REPO_URL, "v3") + + assert sync_result == async_result == expected + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("plugin_id", "repo_url", "frozen", "repo_info", "selected", "expected"), + [ + ("DemoPlugin", REPO_URL, True, ("example", "repo"), "v3", + (False, "可执行文件模式下，只能安装本地插件")), + ("", REPO_URL, False, ("example", "repo"), "v3", + (False, "参数错误")), + ("DemoPlugin", "invalid", False, (None, None), "v3", + (False, "不支持的插件仓库地址格式")), + ("DemoPlugin", REPO_URL, False, ("example", "repo"), None, + (False, "DemoPlugin 没有找到适用于当前版本的插件")), + ], +) +async def test_plugin_package_sync_async_preflight_failures_are_identical( + monkeypatch, + plugin_id: str, + repo_url: str, + frozen: bool, + repo_info: tuple[str | None, str | None], + selected: str | None, + expected: tuple[bool, str], +) -> None: + """远端包安装的环境、参数、仓库和代际失败必须保持同一结果。""" + manager = PluginPackageManager(source=Mock()) + monkeypatch.setattr(manager, "is_local_repo_url", lambda _repo_url: False) + monkeypatch.setattr( + "app.adapters.system.plugin.package.SystemUtils.is_frozen", + lambda: frozen, + ) + monkeypatch.setattr(manager, "get_repo_info", lambda _repo_url: repo_info) + monkeypatch.setattr( + manager, + "get_plugin_package_version", + lambda *_args: selected, + ) + + async def async_get_plugin_package_version(*_args) -> str | None: + return selected + + monkeypatch.setattr( + manager, + "async_get_plugin_package_version", + async_get_plugin_package_version, + ) + + sync_result = manager.install_raw(plugin_id, repo_url, package_version="v3") + async_result = await manager.async_install_raw( + plugin_id, repo_url, package_version="v3" + ) + + assert sync_result == async_result == expected + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("metadata", "release_version", "release_items", "expected_calls"), + [ + ({"version": "1.2.3", "release": False}, None, [], ["filelist"]), + ( + {"version": "1.2.3", "release": True}, + None, + [], + ["release", "remove", "filelist"], + ), + ( + {"version": "1.2.3", "release": True}, + "1.2.3", + [{"version": "1.2.3"}], + ["release"], + ), + ], +) +async def test_plugin_package_sync_async_execute_same_selected_strategy( + monkeypatch, + metadata: dict, + release_version: str | None, + release_items: list[dict], + expected_calls: list[str], +) -> None: + """共享安装选择必须让同步与异步外壳执行相同 Release 回退路径。""" + manager = PluginPackageManager(source=Mock()) + sync_calls: list[str] = [] + async_calls: list[str] = [] + monkeypatch.setattr(manager, "is_local_repo_url", lambda _repo_url: False) + monkeypatch.setattr( + "app.adapters.system.plugin.package.SystemUtils.is_frozen", + lambda: False, + ) + monkeypatch.setattr( + manager, "get_repo_info", lambda _repo_url: ("example", "repo") + ) + monkeypatch.setattr( + manager, "get_plugin_package_version", lambda *_args: "v3" + ) + monkeypatch.setattr( + manager, "get_plugin_release_versions", lambda *_args: release_items + ) + monkeypatch.setattr( + manager, "_PluginPackageManager__get_plugin_meta", lambda *_args: metadata + ) + + async def selected_version(*_args) -> str: + return "v3" + + async def releases(*_args) -> list[dict]: + return release_items + + async def meta(*_args) -> dict: + return metadata + + def install_release(*_args) -> tuple[bool, str]: + sync_calls.append("release") + return (True, "installed") if release_version else (False, "missing asset") + + async def async_install_release(*_args) -> tuple[bool, str]: + async_calls.append("release") + return (True, "installed") if release_version else (False, "missing asset") + + def prepare_filelist(*_args) -> tuple[bool, str]: + sync_calls.append("filelist") + return True, "installed" + + async def async_prepare_filelist(*_args) -> tuple[bool, str]: + async_calls.append("filelist") + return True, "installed" + + def remove(*_args) -> None: + sync_calls.append("remove") + + async def async_remove(*_args) -> None: + async_calls.append("remove") + + def install_flow(_pid, _force, prepare, _repo_url, _before): + return prepare() + + async def async_install_flow(_pid, _force, prepare, _repo_url, _before): + return await prepare() + + monkeypatch.setattr(manager, "async_get_plugin_package_version", selected_version) + monkeypatch.setattr(manager, "async_get_plugin_release_versions", releases) + monkeypatch.setattr(manager, "_PluginPackageManager__async_get_plugin_meta", meta) + monkeypatch.setattr(manager, "_PluginPackageManager__install_from_release", install_release) + monkeypatch.setattr(manager, "_PluginPackageManager__async_install_from_release", async_install_release) + monkeypatch.setattr(manager, "_PluginPackageManager__prepare_content_via_filelist_sync", prepare_filelist) + monkeypatch.setattr(manager, "_PluginPackageManager__prepare_content_via_filelist_async", async_prepare_filelist) + monkeypatch.setattr(manager, "_PluginPackageManager__remove_old_plugin", remove) + monkeypatch.setattr(manager, "_PluginPackageManager__async_remove_old_plugin", async_remove) + monkeypatch.setattr(manager, "_PluginPackageManager__install_flow_sync", install_flow) + monkeypatch.setattr(manager, "_PluginPackageManager__install_flow_async", async_install_flow) + + sync_result = manager.install_raw( + "DemoPlugin", + REPO_URL, + package_version="v3", + release_version=release_version, + ) + async_result = await manager.async_install_raw( + "DemoPlugin", + REPO_URL, + package_version="v3", + release_version=release_version, + ) + + assert sync_result == async_result == (True, "installed") + assert sync_calls == async_calls == expected_calls + + +@pytest.mark.asyncio +async def test_dependency_install_sync_async_share_request_and_result( + tmp_path: Path, monkeypatch +) -> None: + """依赖安装必须向同步与异步包端口提交同一份规范请求。""" + manifest_path = tmp_path / "requirements.txt" + wheels_dir = tmp_path / "wheels" + packages = SimpleNamespace( + install_packages_with_fallback=Mock(return_value=(True, "installed")), + async_install_packages_with_fallback=AsyncMock( + return_value=(True, "installed") + ), + ) + installer = PluginDependencyInstaller(packages, plugin_dir=tmp_path) + monkeypatch.setattr( + installer, + "_plugin_manifests", + lambda: [SimpleNamespace(path=manifest_path)], + ) + monkeypatch.setattr(installer, "_wheels_dirs", lambda: [wheels_dir]) + + sync_result = installer.install(["demo>=1"]) + async_result = await installer.async_install(["demo>=1"]) + + assert sync_result == async_result == (True, "installed") + packages.install_packages_with_fallback.assert_called_once_with( + [manifest_path], [wheels_dir] + ) + packages.async_install_packages_with_fallback.assert_awaited_once_with( + [manifest_path], [wheels_dir] + ) + + +@pytest.mark.asyncio +async def test_dependency_install_sync_async_share_error_mapping( + tmp_path: Path, monkeypatch +) -> None: + """清单准备和包端口异常必须映射为完全一致的公开失败结果。""" + packages = SimpleNamespace( + install_packages_with_fallback=Mock(side_effect=OSError("disk full")), + async_install_packages_with_fallback=AsyncMock( + side_effect=OSError("disk full") + ), + ) + installer = PluginDependencyInstaller(packages, plugin_dir=tmp_path) + monkeypatch.setattr( + installer, + "_plugin_manifests", + lambda: [SimpleNamespace(path=tmp_path / "requirements.txt")], + ) + monkeypatch.setattr(installer, "_wheels_dirs", lambda: []) + + sync_result = installer.install(["demo>=1"]) + async_result = await installer.async_install(["demo>=1"]) + + assert sync_result == async_result == ( + False, + "安装依赖项时发生错误：disk full", + )