From 7c81fb357f83aa2ef02a4a882ea1e10bfd35c3c1 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Wed, 26 Aug 2026 18:00:49 +0800 Subject: [PATCH] =?UTF-8?q?fix(plugin):=20=E6=94=B6=E6=95=9B=E5=B8=82?= =?UTF-8?q?=E5=9C=BA=E5=AE=89=E8=A3=85=E4=B8=8E=E8=BF=90=E8=A1=8C=E6=80=81?= =?UTF-8?q?=E9=94=99=E8=AF=AF=E5=A4=84=E7=90=86=20(#6470)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/adapters/external/market.py | 89 +++++---- app/application/plugin/install.py | 15 +- app/application/plugin/inventory.py | 2 +- app/application/plugin/source.py | 6 +- app/main.py | 7 + app/runtime/extensions/plugin/lifecycle.py | 45 +++-- .../architecture/dependency-baseline.json | 5 +- tests/test_plugin_candidate_inventory.py | 39 ++++ tests/test_plugin_helper.py | 148 +++++++++++++-- tests/test_plugin_install_command.py | 73 +++++++- tests/test_plugin_install_gateway.py | 4 +- tests/test_plugin_lifecycle_status.py | 14 ++ tests/test_plugin_market_index_policy.py | 174 +++++++++++++++++- tests/test_uvicorn_entrypoint.py | 38 ++++ 14 files changed, 579 insertions(+), 80 deletions(-) diff --git a/app/adapters/external/market.py b/app/adapters/external/market.py index 2c8e6f90c..f9acfe59b 100644 --- a/app/adapters/external/market.py +++ b/app/adapters/external/market.py @@ -861,7 +861,7 @@ class PluginHelper(metaclass=WeakSingleton): releases.extend(cls.__normalize_plugin_release_response(payload)) return len(payload) >= 100 - @cached(maxsize=128, ttl=1800) # type: ignore[misc] # 缓存装饰器暂未提供泛型签名 + @cached(maxsize=1024, ttl=1800, skip_none=False) # type: ignore[misc] def get_plugin_index_result( self, repo_url: str, @@ -884,7 +884,6 @@ class PluginHelper(metaclass=WeakSingleton): raise RuntimeError("插件索引响应格式无效") return payload - @cached(maxsize=128, ttl=1800) def get_plugins(self, repo_url: str, package_version: Optional[str] = None) -> Optional[Dict[str, dict]]: """ @@ -892,16 +891,13 @@ class PluginHelper(metaclass=WeakSingleton): :param repo_url: Github仓库地址 :param package_version: 首选插件版本 (如 "v2", "v3")，如果不指定则获取 v1 版本 """ - request = self._build_plugin_index_request(repo_url, package_version) - if request is None: + try: + payload = self.get_plugin_index_result(repo_url, package_version) + except (ValueError, RuntimeError): return None - package_url, headers = request - res = self.__request_with_fallback(package_url, headers=headers) - if res is None: - return None - return self._resolve_plugin_index_response(res.status_code, res.text) + return payload if payload is not None else {} - @cached(maxsize=32, ttl=1800, shared_key="get_plugin_repo_releases") + @cached(maxsize=256, ttl=1800, shared_key="get_plugin_repo_releases") def _get_plugin_repo_releases(self, repo_url: str) -> Optional[List[dict]]: """ 按仓库获取 GitHub Release 原始分页数据，供仓库内所有插件共享。 @@ -1137,12 +1133,12 @@ class PluginHelper(metaclass=WeakSingleton): return True, msg logger.warning(f"{pid} Release 安装失败，回退文件列表安装：{msg}") self.__remove_old_plugin(pid) - return self.__prepare_content_via_filelist_sync(pid.lower(), user_repo, package_version) + return self.__prepare_content_via_filelist_sync(pid, user_repo, package_version) return self.__install_flow_sync(pid, force_install, prepare_release, repo_url) # 未声明 release 打包的插件继续使用文件列表方式安装。 def prepare_filelist() -> Tuple[bool, str]: - return self.__prepare_content_via_filelist_sync(pid.lower(), user_repo, package_version) + return self.__prepare_content_via_filelist_sync(pid, user_repo, package_version) return self.__install_flow_sync(pid, force_install, prepare_filelist, repo_url) @@ -1236,6 +1232,8 @@ class PluginHelper(metaclass=WeakSingleton): timeout=30) if res is None: return None, "连接仓库失败" + elif res.status_code == 404: + return None, "插件源码目录不存在" elif res.status_code != 200: return None, f"连接仓库失败：{res.status_code} - " \ f"{'超出速率限制，请设置Github Token或稍后重试' if res.status_code == 403 else res.reason}" @@ -2385,8 +2383,12 @@ class PluginHelper(metaclass=WeakSingleton): timeout=30, is_api=True, ) - if rel_res is None or rel_res.status_code != 200: - return False, f"获取 Release 信息失败：{rel_res.status_code if rel_res else '连接失败'}" + if rel_res is None: + return False, "获取 Release 信息失败：连接失败" + if rel_res.status_code == 404: + return False, f"{release_tag} 插件发布包不存在" + if rel_res.status_code != 200: + return False, f"获取 Release 信息失败：{rel_res.status_code}" try: rel_json = rel_res.json() @@ -2550,7 +2552,7 @@ class PluginHelper(metaclass=WeakSingleton): logger.error(f"[GitHub] 所有策略均请求失败，URL: {url}，请检查网络连接或 GitHub 配置") return None - @cached(maxsize=128, ttl=1800) # type: ignore[misc] # 缓存装饰器暂未提供泛型签名 + @cached(maxsize=1024, ttl=1800, skip_none=False) # type: ignore[misc] async def async_get_plugin_index_result( self, repo_url: str, @@ -2576,7 +2578,6 @@ class PluginHelper(metaclass=WeakSingleton): raise RuntimeError("插件索引响应格式无效") return payload - @cached(maxsize=128, ttl=1800) async def async_get_plugins(self, repo_url: str, package_version: Optional[str] = None) -> Optional[Dict[str, dict]]: """ @@ -2584,19 +2585,16 @@ class PluginHelper(metaclass=WeakSingleton): :param repo_url: Github仓库地址 :param package_version: 首选插件版本 (如 "v2", "v3")，如果不指定则获取 v1 版本 """ - request = self._build_plugin_index_request(repo_url, package_version) - if request is None: + try: + payload = await self.async_get_plugin_index_result( + repo_url, + package_version, + ) + except (ValueError, RuntimeError): return None - package_url, headers = request - res = await self.__async_request_with_fallback( - package_url, - headers=headers, - ) - if res is None: - return None - return self._resolve_plugin_index_response(res.status_code, res.text) + return payload if payload is not None else {} - @cached(maxsize=32, ttl=1800, shared_key="get_plugin_repo_releases") + @cached(maxsize=256, ttl=1800, shared_key="get_plugin_repo_releases") async def _async_get_plugin_repo_releases(self, repo_url: str) -> Optional[List[dict]]: """ 异步按仓库获取 GitHub Release 原始分页数据。 @@ -2725,6 +2723,8 @@ class PluginHelper(metaclass=WeakSingleton): timeout=30) if res is None: return None, "连接仓库失败" + elif res.status_code == 404: + return None, "插件源码目录不存在" elif res.status_code != 200: return None, f"连接仓库失败：{res.status_code} - " \ f"{'超出速率限制，请设置Github Token或稍后重试' if res.status_code == 403 else res.text}" @@ -3287,12 +3287,12 @@ class PluginHelper(metaclass=WeakSingleton): 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.lower(), user_repo, package_version) + return await self.__prepare_content_via_filelist_async(pid, user_repo, package_version) return await self.__install_flow_async(pid, force_install, prepare_release, repo_url) # 未声明 release 打包的插件继续使用文件列表方式安装。 async def prepare_filelist() -> Tuple[bool, str]: - return await self.__prepare_content_via_filelist_async(pid.lower(), user_repo, package_version) + return await self.__prepare_content_via_filelist_async(pid, user_repo, package_version) return await self.__install_flow_async(pid, force_install, prepare_filelist, repo_url) @@ -3362,10 +3362,13 @@ class PluginHelper(metaclass=WeakSingleton): """ 同步准备插件内容，通过文件列表获取插件文件和依赖 """ - file_list, msg = self.__get_file_list(pid, user_repo, package_version) + runtime_pid = pid.lower() + file_list, msg = self.__get_file_list(runtime_pid, user_repo, package_version) if not file_list: + if msg == "插件源码目录不存在": + return False, f"{pid} {msg}" return False, msg - ok, m = self.__download_files(pid, file_list, user_repo, package_version) + ok, m = self.__download_files(runtime_pid, file_list, user_repo, package_version) if not ok: return False, m return True, "" @@ -3375,10 +3378,22 @@ class PluginHelper(metaclass=WeakSingleton): """ 异步准备插件内容，通过文件列表获取插件文件和依赖 """ - file_list, msg = await self.__async_get_file_list(pid, user_repo, package_version) + runtime_pid = pid.lower() + file_list, msg = await self.__async_get_file_list( + runtime_pid, + user_repo, + package_version, + ) if not file_list: + if msg == "插件源码目录不存在": + return False, f"{pid} {msg}" return False, msg - ok, m = await self.__async_download_files(pid, file_list, user_repo, package_version) + ok, m = await self.__async_download_files( + runtime_pid, + file_list, + user_repo, + package_version, + ) if not ok: return False, m return True, "" @@ -3399,8 +3414,12 @@ class PluginHelper(metaclass=WeakSingleton): timeout=30, is_api=True, ) - if rel_res is None or rel_res.status_code != 200: - return False, f"获取 Release 信息失败：{rel_res.status_code if rel_res else '连接失败'}" + if rel_res is None: + return False, "获取 Release 信息失败：连接失败" + if rel_res.status_code == 404: + return False, f"{release_tag} 插件发布包不存在" + if rel_res.status_code != 200: + return False, f"获取 Release 信息失败：{rel_res.status_code}" try: rel_json = rel_res.json() diff --git a/app/application/plugin/install.py b/app/application/plugin/install.py index b663ec88c..bec696101 100644 --- a/app/application/plugin/install.py +++ b/app/application/plugin/install.py @@ -23,11 +23,12 @@ from app.schemas.exception import ( PersistenceUnavailableError, PluginMutationRejectedError, ) +from app.schemas.plugin import PluginRuntimeStatus InstalledPluginsReader = Callable[[], list[str]] PluginIdsProvider = Callable[[], list[str]] InstallReporter = Callable[[str, str | None], Awaitable[object]] -PluginReloader = Callable[[str], Awaitable[object]] +PluginReloader = Callable[[str], Awaitable[PluginRuntimeStatus]] PluginRegistrationRefresher = Callable[[str], Awaitable[object]] PluginMutationAdmission = Callable[[str], ContextManager[None]] PluginPackageWriteGuard = Callable[[str], ContextManager[None]] @@ -380,7 +381,7 @@ class PluginInstallCommand: state.stage = "runtime_reload" state.runtime_touched = True - await self.__await_side_effect(self.__target_reloader(plugin_id)) + await self.__reload_active(plugin_id) state.stage = "registration_refresh" state.registrations_touched = True await self.__await_side_effect( @@ -511,7 +512,7 @@ class PluginInstallCommand: """刷新已经提交且载荷事实未变化的插件运行态。""" failure_stage = "runtime_reload" try: - await self.__await_side_effect(self.__target_reloader(plugin_id)) + await self.__reload_active(plugin_id) failure_stage = "registration_refresh" await self.__await_side_effect( self.__registration_refresher(plugin_id) @@ -542,6 +543,14 @@ class PluginInstallCommand: report_error=report_error, ) + async def __reload_active(self, plugin_id: str) -> None: + """重载只有进入 ACTIVE 才能作为安装或刷新成功继续提交。""" + runtime_status = await self.__await_side_effect( + self.__target_reloader(plugin_id) + ) + if runtime_status is not PluginRuntimeStatus.ACTIVE: + raise RuntimeError("插件加载失败，请查看插件日志") + async def __finish_committed(self, state: _InstallState) -> str: """幂等清理 COMMITTED 事务；失败时保留 journal 供启动回放。""" try: diff --git a/app/application/plugin/inventory.py b/app/application/plugin/inventory.py index 6cb9135c4..47aa04e51 100644 --- a/app/application/plugin/inventory.py +++ b/app/application/plugin/inventory.py @@ -53,7 +53,7 @@ class PluginCandidateInventoryReader: local_candidate_loader: LocalCandidateLoader | None = None, async_market_loader: AsyncMarketLoader | None = None, generations: Sequence[str] = PLUGIN_V3_GENERATIONS, - max_concurrency: int = 12, + max_concurrency: int = 24, ) -> None: """保存读取端口，并限制异步市场请求的进程内并发。""" normalized_generations = tuple( diff --git a/app/application/plugin/source.py b/app/application/plugin/source.py index 07d9929fd..f4c0302ef 100644 --- a/app/application/plugin/source.py +++ b/app/application/plugin/source.py @@ -689,7 +689,7 @@ def list_effective_online_candidates( plugin_id: str, generations: Sequence[str], ) -> tuple[PluginMarketCandidate, ...]: - """按来源列出当前运行代际实际可安装的最高版本候选。""" + """按来源列出当前运行代际实际可安装的最高版本候选，官方来源始终置顶。""" generation_order = _normalize_generation_order(generations) grouped: dict[ tuple[TrustedPluginSourceType, str], @@ -706,6 +706,10 @@ def list_effective_online_candidates( selected_candidate = _select_best(candidates, generation_order) if isinstance(selected_candidate, PluginMarketCandidate): selected.append(selected_candidate) + selected.sort( + key=lambda candidate: candidate.source_type + is not TrustedPluginSourceType.OFFICIAL + ) return tuple(selected) diff --git a/app/main.py b/app/main.py index b87e88454..fdd3360db 100644 --- a/app/main.py +++ b/app/main.py @@ -114,6 +114,13 @@ def run_api_server() -> None: host=get_runtime_setting('HOST'), port=get_runtime_setting('PORT'), reload=get_runtime_setting('DEV'), + # 运行插件及其恢复材料由插件生命周期管理，不属于宿主源码变更。 + reload_excludes=[ + str(get_runtime_setting('ROOT_PATH') / "app" / "plugins"), + str(get_runtime_setting('CONFIG_PATH')), + ] + if get_runtime_setting('DEV') + else None, workers=get_runtime_setting('API_WORKERS'), timeout_graceful_shutdown=60, ) diff --git a/app/runtime/extensions/plugin/lifecycle.py b/app/runtime/extensions/plugin/lifecycle.py index cfcef5b9b..2f9192034 100644 --- a/app/runtime/extensions/plugin/lifecycle.py +++ b/app/runtime/extensions/plugin/lifecycle.py @@ -108,15 +108,15 @@ class PluginLifecycle: plugins.sort(key=lambda item: getattr(item, "plugin_order", 0)) for plugin in plugins: current_id = plugin.__name__ - if plugin_id and current_id != plugin_id: + if plugin_id and current_id.casefold() != plugin_id.casefold(): continue try: if not self._auth_checker(plugin): if current_id in self._classes: self._classes[current_id] = plugin status = PluginRuntimeStatus.BLOCKED_BY_POLICY - self._runtime_status_writer(current_id, status) - results[current_id] = status + self._runtime_status_writer(plugin_id or current_id, status) + results[plugin_id or current_id] = status continue self._classes[current_id] = plugin instance = plugin() @@ -131,16 +131,19 @@ class PluginLifecycle: else: self._disable_events(plugin) status = PluginRuntimeStatus.ACTIVE - self._runtime_status_writer(current_id, status) - results[current_id] = status + self._runtime_status_writer(plugin_id or current_id, status) + results[plugin_id or current_id] = status except Exception as error: # noqa: BLE001 status = PluginRuntimeStatus.LOAD_FAILED - self._runtime_status_writer(current_id, status) - results[current_id] = status + self._runtime_status_writer(plugin_id or current_id, status) + results[plugin_id or current_id] = status self._logger.error( f"加载插件 {current_id} 出错：{error} - {traceback.format_exc()}" ) - if plugin_id and plugin_id not in results: + if plugin_id and not any( + result_id.casefold() == plugin_id.casefold() + for result_id in results + ): status = PluginRuntimeStatus.LOAD_FAILED self._runtime_status_writer(plugin_id, status) results[plugin_id] = status @@ -286,9 +289,10 @@ class PluginLifecycle: return False if plugin_id: - self._classes.pop(plugin_id, None) - self._running.pop(plugin_id, None) - self._quiesced_hooks.pop(plugin_id, None) + runtime_id = self._resolve_runtime_id(plugin_id) + self._classes.pop(runtime_id, None) + self._running.pop(runtime_id, None) + self._quiesced_hooks.pop(runtime_id, None) else: self._classes.clear() self._running.clear() @@ -300,14 +304,22 @@ class PluginLifecycle: """返回本阶段处理的稳定实例快照，并保持旧停机日志语义。""" if plugin_id: self._logger.info(f"正在停止插件 {plugin_id}...") - plugin = self._running.get(plugin_id) - plugins = {plugin_id: plugin} if plugin else {} + runtime_id = self._resolve_runtime_id(plugin_id) + plugin = self._running.get(runtime_id) + plugins = {runtime_id: plugin} if plugin else {} if not plugin: self._logger.debug(f"插件 {plugin_id} 不存在或未加载") return plugins self._logger.info("正在停止所有插件...") return dict(self._running) + def _resolve_runtime_id(self, plugin_id: str) -> str: + """按不区分大小写的插件 ID 找到运行时注册表键。""" + for runtime_id in (*self._running, *self._classes): + if runtime_id.casefold() == plugin_id.casefold(): + return runtime_id + return plugin_id + def _is_quiesced(self, plugin_id: str, plugin: Any) -> bool: """判断 handler 及当前实例声明的旧 ABI hooks 是否均已成功收敛。""" required = {self._EVENT_HANDLERS_QUIESCED} | { @@ -334,6 +346,11 @@ class PluginLifecycle: """重启指定插件并返回本次加载结果。""" self._runtime_status_writer(plugin_id, PluginRuntimeStatus.READY) self.stop(plugin_id) - status = self.start(plugin_id)[plugin_id] + results = self.start(plugin_id) + status = next( + status + for result_id, status in results.items() + if result_id.casefold() == plugin_id.casefold() + ) self._event_sender(reload_event, data={"plugin_id": plugin_id}) return status diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index bc572dd47..3601ec311 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6807, - "edge_sha256": "93c39ca5828e23fcb5e4b33ffea101cb1ebc9a4ef6fac1317e2ebd482592ecdd", + "edge_count": 6808, + "edge_sha256": "bde5dc465241463f8695da3a7140dfed1f05757fd3f57493fa91d20fd80b6847", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -2788,6 +2788,7 @@ "app.application.plugin.install -> app.runtime.log", "app.application.plugin.install -> app.schemas", "app.application.plugin.install -> app.schemas.exception", + "app.application.plugin.install -> app.schemas.plugin", "app.application.plugin.inventory -> app.application", "app.application.plugin.inventory -> app.application.plugin", "app.application.plugin.inventory -> app.application.plugin.identity", diff --git a/tests/test_plugin_candidate_inventory.py b/tests/test_plugin_candidate_inventory.py index d7816f10c..07a848c71 100644 --- a/tests/test_plugin_candidate_inventory.py +++ b/tests/test_plugin_candidate_inventory.py @@ -1,5 +1,7 @@ """插件市场候选库存读取测试。""" +import asyncio + import pytest from app.application.plugin.identity import TrustedPluginSourceType @@ -288,3 +290,40 @@ async def test_async_loader_preserves_generation_facts() -> None: assert [read.package_generation for read in inventory.market_reads] == [ "v3", "v2", "v1" ] + + +@pytest.mark.asyncio +async def test_async_inventory_bounds_large_market_working_set() -> None: + """大市场库存允许 24 个并发读取，但不能无界放大出站连接。""" + active = 0 + max_active = 0 + first_batch_started = asyncio.Event() + release = asyncio.Event() + + async def loader(_market: str, _package_version: str | None, _force: bool): + nonlocal active, max_active + active += 1 + max_active = max(max_active, active) + if active == 24: + first_batch_started.set() + await release.wait() + active -= 1 + return {} + + markets = [ + f"https://github.com/concurrency-owner/repository-{index}" + for index in range(10) + ] + reader = PluginCandidateInventoryReader( + market_loader=lambda *_args: {}, + async_market_loader=loader, + ) + inventory_task = asyncio.create_task(reader.async_load(markets)) + + await asyncio.wait_for(first_batch_started.wait(), timeout=1) + assert active == 24 + release.set() + inventory = await inventory_task + + assert max_active == 24 + assert inventory.complete diff --git a/tests/test_plugin_helper.py b/tests/test_plugin_helper.py index 02a2c68d8..4fbac8366 100644 --- a/tests/test_plugin_helper.py +++ b/tests/test_plugin_helper.py @@ -94,6 +94,8 @@ class _FakeResponse: def __init__(self, status_code: int, payload: dict | None = None): self.status_code = status_code self._payload = payload or {} + self.reason = self._payload.get("message", "") + self.text = self.reason def json(self): """返回构造时注入的 JSON payload。""" @@ -2160,7 +2162,7 @@ demo = { index = "private" } def test_install_reports_filelist_error_after_release_fallback_fails(self, monkeypatch): """ - release 和文件列表都不可用时返回最终文件列表错误，并在每次写入前后保持目录可回滚。 + release 和源码目录都不存在时返回稳定业务错误，并在每次写入前后保持目录可回滚。 """ try: from app.adapters.external.market import PluginHelper @@ -2172,14 +2174,14 @@ demo = { index = "private" } helper, monkeypatch, {"release": True, "version": "1.2.3"}, - (False, "未找到资产文件：demoplugin_v1.2.3.zip"), - (False, "获取文件列表失败"), + (False, "DemoPlugin_v1.2.3 插件发布包不存在"), + (False, "DemoPlugin 插件源码目录不存在"), ) success, message = helper._PluginHelper__install_package(PLUGIN_ID, REPO_URL, package_version="v2", force_install=True) assert not success - assert "获取文件列表失败" == message + assert "DemoPlugin 插件源码目录不存在" == message assert ["remove", "release", "remove", "filelist", "remove"] == calls def test_install_uses_filelist_when_release_flag_is_disabled(self, monkeypatch): @@ -2602,7 +2604,7 @@ demo = { index = "private" } def test_async_install_reports_filelist_error_after_release_fallback_fails(self, monkeypatch): """ - 异步安装路径在 release 与文件列表都失败时返回文件列表错误，并保持失败清理顺序稳定。 + 异步安装路径在 release 与源码目录都不存在时返回稳定业务错误，并保持失败清理顺序稳定。 """ try: from app.adapters.external.market import PluginHelper @@ -2614,8 +2616,8 @@ demo = { index = "private" } helper, monkeypatch, {"release": True, "version": "1.2.3"}, - (False, "未找到资产文件：demoplugin_v1.2.3.zip"), - (False, "获取文件列表失败"), + (False, "DemoPlugin_v1.2.3 插件发布包不存在"), + (False, "DemoPlugin 插件源码目录不存在"), ) success, message = asyncio.run( @@ -2623,12 +2625,12 @@ demo = { index = "private" } ) assert not success - assert "获取文件列表失败" == message + assert "DemoPlugin 插件源码目录不存在" == message assert calls == ["remove", "release", "remove", "filelist", "remove"] - def test_async_install_release_fallback_uses_lowercase_filelist_pid(self, monkeypatch): + def test_async_install_release_fallback_preserves_plugin_id(self, monkeypatch): """ - 异步 release 回退文件列表安装时使用小写插件 ID，保持 GitHub 目录查询与同步路径一致。 + 异步 release 回退保留清单中的插件 ID，供文件列表层生成可读错误。 """ try: from app.adapters.external.market import PluginHelper @@ -2657,11 +2659,11 @@ demo = { index = "private" } assert success assert "" == message - assert ["demoplugin"] == filelist_pids + assert ["DemoPlugin"] == filelist_pids - def test_async_install_non_release_uses_lowercase_filelist_pid(self, monkeypatch): + def test_async_install_non_release_preserves_plugin_id(self, monkeypatch): """ - 异步文件列表直装使用小写插件 ID，避免大小写插件 ID 影响远端目录匹配。 + 异步文件列表直装保留清单中的插件 ID，路径规范化由文件列表层负责。 """ try: from app.adapters.external.market import PluginHelper @@ -2690,7 +2692,7 @@ demo = { index = "private" } assert success assert "" == message - assert ["demoplugin"] == filelist_pids + assert ["DemoPlugin"] == filelist_pids def test_install_from_release_reports_missing_tag(self, monkeypatch): """ @@ -2707,7 +2709,30 @@ demo = { index = "private" } success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") assert not success - assert "获取 Release 信息失败：404" == message + assert "DemoPlugin_v1.2.3 插件发布包不存在" == message + + def test_get_file_list_reports_missing_plugin_directory(self, monkeypatch): + """索引存在但源码目录缺失时不向用户暴露底层 HTTP 404。""" + try: + from app.adapters.external.market import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + monkeypatch.setattr( + helper, + "_PluginHelper__request_with_fallback", + lambda *_args, **_kwargs: _FakeResponse(404), + ) + + file_list, message = helper._PluginHelper__get_file_list( + PLUGIN_ID, + "demo/repo", + "v2", + ) + + assert file_list is None + assert message == "插件源码目录不存在" def test_install_from_release_reports_missing_asset(self, monkeypatch): """ @@ -3157,6 +3182,36 @@ demo = { index = "private" } assert not success assert "list failed" == message + def test_prepare_content_via_filelist_sync_names_missing_plugin(self, monkeypatch): + """同步文件列表层规范化路径 ID，并在源码目录缺失时保留原插件 ID。""" + try: + from app.adapters.external.market import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + requested_ids = [] + + def fake_file_list(pid, *_args): + requested_ids.append(pid) + return None, "插件源码目录不存在" + + monkeypatch.setattr( + helper, + "_PluginHelper__get_file_list", + fake_file_list, + ) + + success, message = helper._PluginHelper__prepare_content_via_filelist_sync( + PLUGIN_ID, + "demo/repo", + "v2", + ) + + assert not success + assert message == "DemoPlugin 插件源码目录不存在" + assert requested_ids == ["demoplugin"] + def test_prepare_content_via_filelist_sync_returns_download_error(self, monkeypatch): """ 文件列表存在但文件下载失败时向上返回下载错误。 @@ -3231,6 +3286,38 @@ demo = { index = "private" } assert not success assert "list failed" == message + def test_async_prepare_content_via_filelist_names_missing_plugin(self, monkeypatch): + """异步文件列表层规范化路径 ID，并在源码目录缺失时保留原插件 ID。""" + try: + from app.adapters.external.market import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + requested_ids = [] + + async def fake_file_list(pid, *_args): + requested_ids.append(pid) + return None, "插件源码目录不存在" + + monkeypatch.setattr( + helper, + "_PluginHelper__async_get_file_list", + fake_file_list, + ) + + success, message = asyncio.run( + helper._PluginHelper__prepare_content_via_filelist_async( + PLUGIN_ID, + "demo/repo", + "v2", + ) + ) + + assert not success + assert message == "DemoPlugin 插件源码目录不存在" + assert requested_ids == ["demoplugin"] + def test_async_prepare_content_via_filelist_returns_download_error(self, monkeypatch): """ 异步文件列表下载失败时向上返回下载错误。 @@ -3374,7 +3461,36 @@ demo = { index = "private" } ) assert not success - assert "获取 Release 信息失败：404" == message + assert "DemoPlugin_v1.2.3 插件发布包不存在" == message + + def test_async_get_file_list_reports_missing_plugin_directory(self, monkeypatch): + """异步源码目录读取将 HTTP 404 收敛为稳定业务语义。""" + try: + from app.adapters.external.market import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + + async def fake_request(*_args, **_kwargs): + return _FakeResponse(404) + + monkeypatch.setattr( + helper, + "_PluginHelper__async_request_with_fallback", + fake_request, + ) + + file_list, message = asyncio.run( + helper._PluginHelper__async_get_file_list( + PLUGIN_ID, + "demo/repo", + "v2", + ) + ) + + assert file_list is None + assert message == "插件源码目录不存在" def test_async_install_from_release_reports_missing_asset_id(self, monkeypatch): """ diff --git a/tests/test_plugin_install_command.py b/tests/test_plugin_install_command.py index 62f62c970..b107892b3 100644 --- a/tests/test_plugin_install_command.py +++ b/tests/test_plugin_install_command.py @@ -38,6 +38,7 @@ from app.schemas.exception import ( PersistenceUnavailableError, PluginMutationRejectedError, ) +from app.schemas.plugin import PluginRuntimeStatus NOW = datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc) REPO_URL = "https://github.com/jxxghp/MoviePilot-Plugins" @@ -299,7 +300,8 @@ def _command( plugin_ids_provider=lambda: plugin_ids or [], packages=packages, install_reporter=reporter or default_reporter, - target_reloader=target_reloader or AsyncMock(), + target_reloader=target_reloader + or AsyncMock(return_value=PluginRuntimeStatus.ACTIVE), rollback_reloader=rollback_reloader or AsyncMock(), registration_refresher=registration_refresher or AsyncMock(), mutation=mutation or (lambda _operation: nullcontext()), @@ -361,6 +363,10 @@ async def test_success_commits_journal_before_report_and_cleans_package_snapshot return action + async def target_reload(_plugin_id): + calls.append("target_reload") + return PluginRuntimeStatus.ACTIVE + async def installer(**_kwargs): calls.append("package") return True, "installed" @@ -380,7 +386,7 @@ async def test_success_commits_journal_before_report_and_cleans_package_snapshot payload_receipt=receipt, package_stage_backup=mark("stage_backup"), package_activate_backup=mark("activate_backup"), - target_reloader=mark("target_reload"), + target_reloader=target_reload, registration_refresher=mark("registrations"), package_finalize_backup=mark("finalize_backup"), package_commit=mark("package_commit"), @@ -414,6 +420,67 @@ async def test_success_commits_journal_before_report_and_cleans_package_snapshot assert persistence.records == {} +@pytest.mark.asyncio +async def test_non_active_runtime_status_compensates_before_database_commit(): + """运行态重载未激活时，安装必须在数据库提交前补偿并返回失败。""" + package_restore = AsyncMock() + package_cleanup = AsyncMock() + rollback_reloader = AsyncMock(return_value=PluginRuntimeStatus.ACTIVE) + registration_refresher = AsyncMock() + target_reloader = AsyncMock(return_value=PluginRuntimeStatus.LOAD_FAILED) + reporter = AsyncMock() + + command, persistence, calls = _command( + package_restore=package_restore, + package_cleanup=package_cleanup, + target_reloader=target_reloader, + rollback_reloader=rollback_reloader, + registration_refresher=registration_refresher, + reporter=reporter, + ) + + result = await _execute(command) + + assert result.success is False + assert result.failure_stage == "runtime_reload" + assert result.package_installed is True + assert result.installed_list_persisted is False + assert result.runtime_reloaded is False + assert result.registrations_refreshed is False + assert result.reported is False + package_restore.assert_awaited_once() + rollback_reloader.assert_awaited_once_with("DemoPlugin") + registration_refresher.assert_awaited_once_with("DemoPlugin") + reporter.assert_not_awaited() + assert "journal_commit" not in calls + assert persistence.records == {} + + +@pytest.mark.asyncio +async def test_existing_plugin_with_non_active_runtime_status_is_not_reported_successfully(): + """已有载荷刷新失败时不得伪装成运行态成功或发送安装上报。""" + target_reloader = AsyncMock(return_value=PluginRuntimeStatus.LOAD_FAILED) + reporter = AsyncMock() + command, persistence, _ = _command( + installed=["DemoPlugin"], + plugin_ids=["DemoPlugin"], + target_reloader=target_reloader, + reporter=reporter, + ) + + result = await _execute(command, admission=_admission(identity=_identity())) + + assert result.success is False + assert result.refreshed_only is True + assert result.failure_stage == "runtime_reload" + assert result.runtime_reloaded is False + assert result.registrations_refreshed is False + assert result.reported is False + target_reloader.assert_awaited_once_with("DemoPlugin") + reporter.assert_not_awaited() + assert persistence.records == {} + + @pytest.mark.asyncio @pytest.mark.parametrize( "phase", @@ -726,7 +793,7 @@ async def test_existing_matching_payload_only_refreshes_runtime(): """同一来源、代际和版本已提交时只刷新运行态，不重复写包或 journal。""" checkpointer = AsyncMock() installer = AsyncMock() - reloader = AsyncMock() + reloader = AsyncMock(return_value=PluginRuntimeStatus.ACTIVE) refresher = AsyncMock() reporter = AsyncMock(return_value=True) command, persistence, _ = _command( diff --git a/tests/test_plugin_install_gateway.py b/tests/test_plugin_install_gateway.py index 6e5429a56..e96beed96 100644 --- a/tests/test_plugin_install_gateway.py +++ b/tests/test_plugin_install_gateway.py @@ -213,13 +213,13 @@ async def test_gateway_source_inspection_preserves_sources_and_hides_local_path( ) inventory = CandidateInventory( ( - MarketRead.present(REPO_URL, (official_v3,), package_generation="v3"), - MarketRead.present(REPO_URL, (official_v2,), package_generation="v2"), MarketRead.present( third_party.repo_url, (third_party,), package_generation="v3", ), + MarketRead.present(REPO_URL, (official_v3,), package_generation="v3"), + MarketRead.present(REPO_URL, (official_v2,), package_generation="v2"), ), (local,), local_read=LocalCandidateRead.present((local,)), diff --git a/tests/test_plugin_lifecycle_status.py b/tests/test_plugin_lifecycle_status.py index f4ead9d09..11a012cef 100644 --- a/tests/test_plugin_lifecycle_status.py +++ b/tests/test_plugin_lifecycle_status.py @@ -61,6 +61,20 @@ def test_lifecycle_records_active_result(): assert statuses["DemoPlugin"] is PluginRuntimeStatus.ACTIVE +def test_targeted_lifecycle_accepts_case_insensitive_directory_id(): + """市场目录 ID 通常为小写，不能因插件类名大小写不同而误报加载失败。""" + lifecycle, classes, running, statuses = _lifecycle( + plugins=[_plugin_class()], + ) + + result = lifecycle.start("demoplugin") + + assert result == {"demoplugin": PluginRuntimeStatus.ACTIVE} + assert "DemoPlugin" in classes + assert "DemoPlugin" in running + assert statuses["demoplugin"] is PluginRuntimeStatus.ACTIVE + + def test_lifecycle_records_policy_block_without_runtime_instance(): """类已发现但权限策略拒绝时进入 blocked_by_policy。""" lifecycle, _classes, running, statuses = _lifecycle( diff --git a/tests/test_plugin_market_index_policy.py b/tests/test_plugin_market_index_policy.py index b75398fc5..086f63eff 100644 --- a/tests/test_plugin_market_index_policy.py +++ b/tests/test_plugin_market_index_policy.py @@ -43,11 +43,11 @@ async def test_sync_and_async_plugin_indexes_share_request_and_result( async_request, ) - helper.get_plugins.cache_clear() - await helper.async_get_plugins.cache_clear() + helper.get_plugin_index_result.cache_clear() + await helper.async_get_plugin_index_result.cache_clear() sync_result = helper.get_plugins(repo_url, "v3") # 同步与异步装饰器按设计共享缓存区；清除后再验证异步 I/O 入口本身。 - await helper.async_get_plugins.cache_clear() + await helper.async_get_plugin_index_result.cache_clear() async_result = await helper.async_get_plugins(repo_url, "v3") assert sync_result == async_result == { @@ -60,6 +60,174 @@ async def test_sync_and_async_plugin_indexes_share_request_and_result( ) +@pytest.mark.asyncio +async def test_market_read_warms_source_inventory_cache(monkeypatch) -> None: + """市场目录成功读取后，来源库存不得再次请求同一仓库代际。""" + helper = PluginHelper() + repo_url = "https://github.com/policy-owner/shared-index-cache" + requests = 0 + + async def request(_url: str, *, headers: dict): + nonlocal requests + requests += 1 + return SimpleNamespace( + status_code=200, + text='{"DemoPlugin": {"version": "1.2.3"}}', + ) + + monkeypatch.setattr( + helper, + "_PluginHelper__async_request_with_fallback", + request, + ) + await helper.async_get_plugin_index_result.cache_clear() + + market_result = await helper.async_get_plugins(repo_url, "v3") + inventory_result = await helper.async_get_plugin_index_result(repo_url, "v3") + + assert market_result == inventory_result + assert requests == 1 + + +def test_sync_force_refresh_bypasses_index_cache( + monkeypatch, +) -> None: + """同步强刷必须绕过唯一索引缓存。""" + helper = PluginHelper() + client = PluginMarketClient(helper) + repo_url = "https://github.com/policy-owner/sync-force-refresh" + version = "1.0.0" + requests = 0 + + def request(_url: str, *, headers: dict): + nonlocal requests + requests += 1 + return SimpleNamespace( + status_code=200, + text=f'{{"DemoPlugin": {{"version": "{version}"}}}}', + ) + + monkeypatch.setattr( + helper, + "_PluginHelper__request_with_fallback", + request, + ) + helper.get_plugin_index_result.cache_clear() + + assert client.get_plugins(repo_url, "v3") == { + "DemoPlugin": {"version": "1.0.0"}, + } + assert client.get_plugins(repo_url, "v3") == { + "DemoPlugin": {"version": "1.0.0"}, + } + assert requests == 1 + + version = "2.0.0" + assert client.get_plugins(repo_url, "v3", force=True) == { + "DemoPlugin": {"version": "2.0.0"}, + } + assert requests == 2 + + +@pytest.mark.asyncio +async def test_async_force_refresh_bypasses_index_cache( + monkeypatch, +) -> None: + """异步强刷必须绕过唯一索引缓存。""" + helper = PluginHelper() + client = PluginMarketClient(helper) + repo_url = "https://github.com/policy-owner/async-force-refresh" + version = "1.0.0" + requests = 0 + + async def request(_url: str, *, headers: dict): + nonlocal requests + requests += 1 + return SimpleNamespace( + status_code=200, + text=f'{{"DemoPlugin": {{"version": "{version}"}}}}', + ) + + monkeypatch.setattr( + helper, + "_PluginHelper__async_request_with_fallback", + request, + ) + await helper.async_get_plugin_index_result.cache_clear() + + assert await client.async_get_plugins(repo_url, "v3") == { + "DemoPlugin": {"version": "1.0.0"}, + } + assert await client.async_get_plugins(repo_url, "v3") == { + "DemoPlugin": {"version": "1.0.0"}, + } + assert requests == 1 + + version = "2.0.0" + assert await client.async_get_plugins(repo_url, "v3", force=True) == { + "DemoPlugin": {"version": "2.0.0"}, + } + assert requests == 2 + + +@pytest.mark.asyncio +async def test_absent_plugin_generation_is_cached(monkeypatch) -> None: + """明确不存在的代际是稳定事实，后续来源检查不得重复请求。""" + helper = PluginHelper() + repo_url = "https://github.com/policy-owner/absent-index-cache" + requests = 0 + + async def request(_url: str, *, headers: dict): + nonlocal requests + requests += 1 + return SimpleNamespace(status_code=404, text="404: Not Found") + + monkeypatch.setattr( + helper, + "_PluginHelper__async_request_with_fallback", + request, + ) + await helper.async_get_plugin_index_result.cache_clear() + + first = await helper.async_get_plugin_index_result(repo_url, "v2") + second = await helper.async_get_plugin_index_result(repo_url, "v2") + + assert first is second is None + assert requests == 1 + + +@pytest.mark.asyncio +async def test_index_cache_retains_multi_market_generation_working_set( + monkeypatch, +) -> None: + """数十个市场的多代索引不能因缓存容量不足立即重复出站。""" + helper = PluginHelper() + requests = 0 + + async def request(_url: str, *, headers: dict): + nonlocal requests + requests += 1 + return SimpleNamespace(status_code=200, text='{"DemoPlugin": {}}') + + monkeypatch.setattr( + helper, + "_PluginHelper__async_request_with_fallback", + request, + ) + await helper.async_get_plugin_index_result.cache_clear() + targets = [ + (f"https://github.com/cache-owner/repository-{index}", generation) + for index in range(100) + for generation in ("v3", "v2", None) + ] + + for repo_url, generation in targets: + await helper.async_get_plugin_index_result(repo_url, generation) + await helper.async_get_plugin_index_result(*targets[0]) + + assert requests == len(targets) + + @pytest.mark.parametrize( ("status_code", "content", "expected"), [ diff --git a/tests/test_uvicorn_entrypoint.py b/tests/test_uvicorn_entrypoint.py index 9109df165..8afac8181 100644 --- a/tests/test_uvicorn_entrypoint.py +++ b/tests/test_uvicorn_entrypoint.py @@ -5,6 +5,8 @@ from pathlib import Path from unittest.mock import MagicMock import pytest +from uvicorn import Config +from uvicorn.supervisors.watchfilesreload import FileFilter from app import factory, main from app.runtime.topology import UnsupportedProcessTopologyError @@ -60,11 +62,47 @@ def test_development_reload_uses_import_string_factory(monkeypatch): host=main.settings.HOST, port=main.settings.PORT, reload=True, + reload_excludes=[ + str(main.settings.ROOT_PATH / "app" / "plugins"), + str(main.settings.CONFIG_PATH), + ], workers=1, timeout_graceful_shutdown=60, ) +def test_development_reload_ignores_plugin_runtime_state(monkeypatch): + """插件载荷与恢复快照由热加载管理，仅宿主源码变更触发进程重载。""" + root_path = main.settings.ROOT_PATH + plugin_path = root_path / "app" / "plugins" + config_path = main.settings.CONFIG_PATH + assert plugin_path.is_dir() + assert config_path.is_dir() + monkeypatch.setattr(main.settings, "DEV", True) + monkeypatch.setattr(main.settings, "API_WORKERS", 1) + uvicorn_run = MagicMock() + monkeypatch.setattr(main.uvicorn, "run", uvicorn_run) + + main.run_api_server() + + reload_excludes = uvicorn_run.call_args.kwargs["reload_excludes"] + reload_filter = FileFilter( + Config(main.APP_FACTORY, reload=True, reload_excludes=reload_excludes) + ) + + assert reload_filter(plugin_path / "autosignin" / "__init__.py") is False + assert reload_filter( + config_path + / "plugins_backup" + / ".autosignin.staging-transaction" + / "__init__.py" + ) is False + assert reload_filter( + config_path / "plugin_transactions" / "transaction" / "package" / "__init__.py" + ) is False + assert reload_filter(root_path / "app" / "main.py") is True + + def test_safe_mode_multi_worker_uses_import_string_factory(monkeypatch): """安全模式多 worker 由 Uvicorn supervisor 创建独立 ASGI factory 实例。""" monkeypatch.setattr(main.settings, "DEV", False)