From 10dcb3727e97796e06394b2ba9816856a2fd49d6 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Fri, 12 Jun 2026 13:26:35 +0800 Subject: [PATCH] fix(plugin): fall back when release package is unavailable (#5929) --- app/core/plugin.py | 1 - app/helper/plugin.py | 29 +- tests/test_plugin_helper.py | 1303 +++++++++++++++++++++++++++++++++-- 3 files changed, 1254 insertions(+), 79 deletions(-) diff --git a/app/core/plugin.py b/app/core/plugin.py index a4b4aaf7..9de72588 100644 --- a/app/core/plugin.py +++ b/app/core/plugin.py @@ -360,7 +360,6 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): if runtime_pid: last_sync_time = self._recent_local_sync.get(runtime_pid) if last_sync_time and time.time() - last_sync_time < 2: - logger.debug(f"忽略本地插件同步产生的运行目录变化：{runtime_pid}") continue # 运行目录变化只重载，不能反向触发本地同步。 plugins_to_reload.add(runtime_pid) diff --git a/app/helper/plugin.py b/app/helper/plugin.py index bd131119..3a333825 100644 --- a/app/helper/plugin.py +++ b/app/helper/plugin.py @@ -528,9 +528,10 @@ class PluginHelper(metaclass=WeakSingleton): else: logger.debug(f"{pid} 从 package.{package_version}.json 中找到适用于当前版本的插件") - # 2. 决定安装方式（release 或 文件列表）并执行统一安装流程 + # 2. 决定安装方式（release 或文件列表）并执行统一安装流程。 meta = self.__get_plugin_meta(pid, repo_url, package_version) - # 是否release打包 + # 是否使用 Release 打包。Release 缺失或资产不可用时仍保留文件列表兜底， + # 避免索引先发布、Actions 打包滞后导致插件短时间无法安装。 is_release = meta.get("release") # 插件版本号 plugin_version = meta.get("version") @@ -547,13 +548,15 @@ class PluginHelper(metaclass=WeakSingleton): # 使用 release 进行安装 def prepare_release() -> Tuple[bool, str]: - return self.__install_from_release( - pid, user_repo, release_tag - ) + ok, msg = self.__install_from_release(pid, user_repo, release_tag) + if ok: + return True, msg + logger.warn(f"{pid} Release 安装失败，回退文件列表安装：{msg}") + return self.__prepare_content_via_filelist_sync(pid.lower(), user_repo, package_version) return self.__install_flow_sync(pid, force_install, prepare_release, repo_url) else: - # 如果 release_tag 不存在，说明插件没有发布版本，使用文件列表方式安装 + # 未声明 release 打包的插件继续使用文件列表方式安装。 def prepare_filelist() -> Tuple[bool, str]: return self.__prepare_content_via_filelist_sync(pid.lower(), user_repo, package_version) @@ -2281,9 +2284,9 @@ class PluginHelper(metaclass=WeakSingleton): else: logger.debug(f"{pid} 从 package.{package_version}.json 中找到适用于当前版本的插件") - # 2. 统一异步安装流程（release 或 文件列表） + # 2. 统一异步安装流程（release 或文件列表）。 meta = await self.__async_get_plugin_meta(pid, repo_url, package_version) - # 是否release打包 + # 是否使用 Release 打包；失败时兜底文件列表，保持同步/异步安装语义一致。 is_release = meta.get("release") # 插件版本号 plugin_version = meta.get("version") @@ -2300,13 +2303,15 @@ class PluginHelper(metaclass=WeakSingleton): # 使用 release 进行安装 async def prepare_release() -> Tuple[bool, str]: - return await self.__async_install_from_release( - pid, user_repo, release_tag - ) + ok, msg = await self.__async_install_from_release(pid, user_repo, release_tag) + if ok: + return True, msg + logger.warn(f"{pid} Release 安装失败，回退文件列表安装：{msg}") + 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) else: - # 如果没有 release_tag，则使用文件列表安装方式 + # 未声明 release 打包的插件继续使用文件列表方式安装。 async def prepare_filelist() -> Tuple[bool, str]: return await self.__prepare_content_via_filelist_async(pid, user_repo, package_version) diff --git a/tests/test_plugin_helper.py b/tests/test_plugin_helper.py index c954c6f1..1c4a85fd 100644 --- a/tests/test_plugin_helper.py +++ b/tests/test_plugin_helper.py @@ -1,18 +1,123 @@ import asyncio +import io import sys import tempfile import threading import time +import zipfile from pathlib import Path -from types import ModuleType -from unittest import TestCase +from types import ModuleType, SimpleNamespace from unittest.mock import patch +import pytest + from packaging.requirements import Requirement from packaging.version import Version -class PluginHelperTest(TestCase): +PLUGIN_ID = "DemoPlugin" +REPO_URL = "https://github.com/demo/MoviePilot-Plugins" + + +class _FakeResponse: + """模拟 requests/httpx 响应对象，覆盖插件 release 安装分支读取的最小协议。""" + + def __init__(self, status_code: int, payload: dict | None = None): + self.status_code = status_code + self._payload = payload or {} + + def json(self): + """返回构造时注入的 JSON payload。""" + return self._payload + + +class _FakeContentResponse(_FakeResponse): + """带二进制正文的响应对象，用于模拟 GitHub release asset 下载。""" + + def __init__(self, status_code: int, content: bytes): + super().__init__(status_code) + self.content = content + + +def _build_zip(entries: dict[str, bytes]) -> bytes: + """构造内存 zip 包，键为包内路径、值为文件内容。""" + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as zf: + for name, content in entries.items(): + zf.writestr(name, content) + return buffer.getvalue() + + +def _patch_sync_remote_install(helper, monkeypatch, meta: dict, + release_result: tuple[bool, str], + filelist_result: tuple[bool, str] = (True, "")): + """隔离同步远端插件安装流程，只观察 release 与文件列表准备路径选择。""" + calls = [] + monkeypatch.setattr(helper, "get_plugin_package_version", lambda *_args: "v2") + monkeypatch.setattr(helper, "_PluginHelper__get_plugin_meta", lambda *_args: meta) + monkeypatch.setattr(helper, "_PluginHelper__backup_plugin", lambda _pid: None) + monkeypatch.setattr(helper, "_PluginHelper__remove_old_plugin", lambda _pid: calls.append("remove")) + monkeypatch.setattr(helper, "_PluginHelper__install_dependencies_if_required", lambda _pid: (False, True, "")) + monkeypatch.setattr(helper, "refresh_persistent_plugin_backup", lambda _pid: calls.append("refresh")) + + def fake_release(_pid, _user_repo, _release_tag): + calls.append("release") + return release_result + + def fake_filelist(_pid, _user_repo, _package_version): + calls.append("filelist") + return filelist_result + + monkeypatch.setattr(helper, "_PluginHelper__install_from_release", fake_release) + monkeypatch.setattr(helper, "_PluginHelper__prepare_content_via_filelist_sync", fake_filelist) + return calls + + +def _patch_async_remote_install(helper, monkeypatch, meta: dict, + release_result: tuple[bool, str], + filelist_result: tuple[bool, str] = (True, "")): + """隔离异步远端插件安装流程，只观察 release 与文件列表准备路径选择。""" + calls = [] + + async def fake_package_version(*_args): + return "v2" + + async def fake_meta(*_args): + return meta + + async def fake_backup(_pid): + return None + + async def fake_remove(_pid): + calls.append("remove") + + async def fake_dependencies(_pid): + return False, True, "" + + async def fake_release(_pid, _user_repo, _release_tag): + calls.append("release") + return release_result + + async def fake_filelist(_pid, _user_repo, _package_version): + calls.append("filelist") + return filelist_result + + async def fake_to_thread(func, *args, **kwargs): + calls.append(("to_thread", func, args, kwargs)) + return None + + monkeypatch.setattr(helper, "async_get_plugin_package_version", fake_package_version) + monkeypatch.setattr(helper, "_PluginHelper__async_get_plugin_meta", fake_meta) + monkeypatch.setattr(helper, "_PluginHelper__async_backup_plugin", fake_backup) + monkeypatch.setattr(helper, "_PluginHelper__async_remove_old_plugin", fake_remove) + monkeypatch.setattr(helper, "_PluginHelper__async_install_dependencies_if_required", fake_dependencies) + monkeypatch.setattr(helper, "_PluginHelper__async_install_from_release", fake_release) + monkeypatch.setattr(helper, "_PluginHelper__prepare_content_via_filelist_async", fake_filelist) + monkeypatch.setattr("app.helper.plugin.asyncio.to_thread", fake_to_thread) + return calls + + +class TestPluginHelper: def test_sanitize_plugin_repo_url_keeps_remote_url(self): """ @@ -21,9 +126,9 @@ class PluginHelperTest(TestCase): try: from app.helper.server import MoviePilotServerHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") repo_url = "https://github.com/InfinityPacer/MoviePilot-Plugins" - self.assertEqual(repo_url, MoviePilotServerHelper.sanitize_plugin_repo_url(repo_url)) + assert repo_url == MoviePilotServerHelper.sanitize_plugin_repo_url(repo_url) def test_sanitize_plugin_repo_url_strips_local_path(self): """ @@ -32,12 +137,9 @@ class PluginHelperTest(TestCase): try: from app.helper.server import MoviePilotServerHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") repo_url = "local://TestPlugin?path=/Users/InfinityPacer/GitHub/MoviePilot/MoviePilot-Plugins&version=v2" - self.assertEqual( - "local://TestPlugin?version=v2", - MoviePilotServerHelper.sanitize_plugin_repo_url(repo_url) - ) + assert "local://TestPlugin?version=v2" == MoviePilotServerHelper.sanitize_plugin_repo_url(repo_url) def test_append_cache_buster_only_during_fresh_context(self): """ @@ -47,19 +149,16 @@ class PluginHelperTest(TestCase): from app.core.cache import fresh from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") url = "https://raw.githubusercontent.com/user/repo/main/package.json" - self.assertEqual(url, PluginHelper._PluginHelper__append_cache_buster(url)) + assert url == PluginHelper._PluginHelper__append_cache_buster(url) with patch("app.helper.plugin.time.time_ns", return_value=1234567890): with fresh(True): refreshed_url = PluginHelper._PluginHelper__append_cache_buster(url) - self.assertEqual( - "https://raw.githubusercontent.com/user/repo/main/package.json?_refresh=1234567890", - refreshed_url, - ) + assert "https://raw.githubusercontent.com/user/repo/main/package.json?_refresh=1234567890" == refreshed_url def test_check_plugin_system_version_allows_missing_field(self): """ @@ -68,12 +167,12 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") success, message = PluginHelper.check_plugin_system_version({"version": "1.0.0"}) - self.assertTrue(success) - self.assertEqual("", message) + assert success + assert "" == message def test_check_plugin_system_version_rejects_out_of_range(self): """ @@ -82,13 +181,13 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") with patch.object(PluginHelper, "get_current_system_version", return_value=Version("2.12.2")): success, message = PluginHelper.check_plugin_system_version({"system_version": ">=2.13.0"}) - self.assertFalse(success) - self.assertIn("MoviePilot 版本 >=2.13.0", message) + assert not success + assert "MoviePilot 版本 >=2.13.0" in message def test_check_plugin_system_version_accepts_v_prefix_specifier(self): """ @@ -97,13 +196,13 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") with patch.object(PluginHelper, "get_current_system_version", return_value=Version("2.12.2")): success, message = PluginHelper.check_plugin_system_version({"system_version": ">=v2.12.0"}) - self.assertTrue(success) - self.assertEqual("", message) + assert success + assert "" == message def test_annotate_plugin_system_version_marks_incompatible(self): """ @@ -112,14 +211,14 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") plugin_info = {"system_version": ">=2.13.0"} with patch.object(PluginHelper, "get_current_system_version", return_value=Version("2.12.2")): annotated = PluginHelper.annotate_plugin_system_version(plugin_info) - self.assertFalse(annotated["system_version_compatible"]) - self.assertIn("当前版本", annotated["system_version_message"]) + assert not annotated["system_version_compatible"] + assert "当前版本" in annotated["system_version_message"] def test_pip_install_keeps_modules_imported_during_install(self): """ @@ -128,7 +227,7 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") module_names = ["app.plugins.dynamicwechat.helper", "Crypto.Cipher._mode_cbc"] @@ -146,10 +245,10 @@ class PluginHelperTest(TestCase): with patch("app.helper.plugin.SystemUtils.execute_with_subprocess", side_effect=fake_execute): success, message = PluginHelper.pip_install_with_fallback(requirements_file) - self.assertTrue(success) - self.assertEqual("ok", message) + assert success + assert "ok" == message for module_name in module_names: - self.assertIn(module_name, sys.modules) + assert module_name in sys.modules def test_pip_install_serializes_concurrent_calls(self): """ @@ -158,7 +257,7 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") thread_count = 2 active_installs = 0 @@ -202,8 +301,8 @@ class PluginHelperTest(TestCase): for thread in threads: thread.join() - self.assertEqual([], errors) - self.assertEqual(1, max_active_installs) + assert [] == errors + assert 1 == max_active_installs def test_get_protected_runtime_packages_only_keeps_main_dependency_graph(self): """ @@ -212,7 +311,7 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") installed_packages = { "passlib": Version("1.7.4"), @@ -237,10 +336,10 @@ class PluginHelperTest(TestCase): ): protected_packages = PluginHelper._PluginHelper__get_protected_runtime_packages(installed_packages) - self.assertEqual({ + assert { "passlib": Version("1.7.4"), "bcrypt": Version("4.0.1"), - }, protected_packages) + } == protected_packages def test_pip_install_rejects_conflicting_runtime_dependency(self): """ @@ -249,7 +348,7 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") with tempfile.TemporaryDirectory() as temp_dir: requirements_file = Path(temp_dir) / "requirements.txt" @@ -261,9 +360,9 @@ class PluginHelperTest(TestCase): ): success, message = PluginHelper.pip_install_with_fallback(requirements_file) - self.assertFalse(success) - self.assertIn("主程序核心依赖", message) - self.assertIn("fastapi", message) + assert not success + assert "主程序核心依赖" in message + assert "fastapi" in message def test_pip_install_allows_changing_non_runtime_dependency(self): """ @@ -272,14 +371,14 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") seen_install_commands = [] def fake_execute(cmd): if cmd[:4] == [sys.executable, "-m", "pip", "install"]: seen_install_commands.append(cmd) - self.assertNotIn("-c", cmd) + assert "-c" not in cmd return True, "ok" return True, "ok" @@ -299,9 +398,9 @@ class PluginHelperTest(TestCase): with patch("app.helper.plugin.SystemUtils.execute_with_subprocess", side_effect=fake_execute): success, message = PluginHelper.pip_install_with_fallback(requirements_file) - self.assertTrue(success) - self.assertEqual("ok", message) - self.assertEqual(1, len(seen_install_commands)) + assert success + assert "ok" == message + assert 1 == len(seen_install_commands) def test_pip_install_uses_runtime_constraints_file(self): """ @@ -310,7 +409,7 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") seen_constraints = [] @@ -319,8 +418,8 @@ class PluginHelperTest(TestCase): constraint_index = cmd.index("-c") + 1 constraint_file = Path(cmd[constraint_index]) seen_constraints.append(constraint_file) - self.assertTrue(constraint_file.exists()) - self.assertIn("fastapi==0.115.14", constraint_file.read_text(encoding="utf-8")) + assert constraint_file.exists() + assert "fastapi==0.115.14" in constraint_file.read_text(encoding="utf-8") return True, "ok" return True, "ok" @@ -335,10 +434,10 @@ class PluginHelperTest(TestCase): with patch("app.helper.plugin.SystemUtils.execute_with_subprocess", side_effect=fake_execute): success, message = PluginHelper.pip_install_with_fallback(requirements_file) - self.assertTrue(success) - self.assertEqual("ok", message) - self.assertEqual(1, len(seen_constraints)) - self.assertFalse(seen_constraints[0].exists()) + assert success + assert "ok" == message + assert 1 == len(seen_constraints) + assert not seen_constraints[0].exists() def test_pip_install_repairs_runtime_when_healthcheck_fails(self): """ @@ -347,7 +446,7 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") repair_commands = [] healthcheck_failed = False @@ -380,10 +479,10 @@ class PluginHelperTest(TestCase): with patch("app.helper.plugin.SystemUtils.execute_with_subprocess", side_effect=fake_execute): success, message = PluginHelper.pip_install_with_fallback(requirements_file) - self.assertFalse(success) - self.assertIn("已自动恢复主程序依赖", message) - self.assertEqual(1, len(repair_commands)) - self.assertIn("runtime-constraints-", repair_commands[0][-1]) + assert not success + assert "已自动恢复主程序依赖" in message + assert 1 == len(repair_commands) + assert "runtime-constraints-" in repair_commands[0][-1] def test_async_pip_install_runs_in_threadpool(self): """ @@ -392,7 +491,7 @@ class PluginHelperTest(TestCase): try: from app.helper.plugin import PluginHelper except ModuleNotFoundError as exc: - self.skipTest(f"missing dependency: {exc}") + pytest.skip(f"missing dependency: {exc}") helper = PluginHelper() requirements_file = Path("/tmp/demo-requirements.txt") @@ -412,9 +511,1081 @@ class PluginHelperTest(TestCase): with patch("app.helper.plugin.asyncio.to_thread", side_effect=fake_to_thread): success, message = asyncio.run(run_install()) - self.assertTrue(success) - self.assertEqual("ok", message) - self.assertEqual(1, len(calls)) - self.assertEqual(helper.pip_install_with_fallback, calls[0][0]) - self.assertEqual((requirements_file, find_links_dirs), calls[0][1]) - self.assertEqual({}, calls[0][2]) + assert success + assert "ok" == message + assert 1 == len(calls) + assert helper.pip_install_with_fallback == calls[0][0] + assert (requirements_file, find_links_dirs) == calls[0][1] + assert {} == calls[0][2] + + def test_install_uses_release_package_when_asset_is_available(self, monkeypatch): + """ + release 包可用时优先使用 zip 安装，不再额外访问文件列表。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = _patch_sync_remote_install( + helper, + monkeypatch, + {"release": True, "version": "1.2.3"}, + (True, ""), + ) + + success, message = helper.install(PLUGIN_ID, REPO_URL, package_version="v2", force_install=True) + + assert success + assert "" == message + assert ["remove", "release", "refresh"] == calls + + def test_install_falls_back_to_filelist_when_release_is_missing(self, monkeypatch): + """ + release 标记存在但 tag 或 zip 尚未生成时回退文件列表安装。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = _patch_sync_remote_install( + helper, + monkeypatch, + {"release": True, "version": "1.2.3"}, + (False, "获取 Release 信息失败：404"), + (True, ""), + ) + + success, message = helper.install(PLUGIN_ID, REPO_URL, package_version="v2", force_install=True) + + assert success + assert "" == message + assert ["remove", "release", "filelist", "refresh"] == calls + + def test_install_reports_filelist_error_after_release_fallback_fails(self, monkeypatch): + """ + release 和文件列表都不可用时返回最终文件列表错误，便于定位真实安装阻断点。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = _patch_sync_remote_install( + helper, + monkeypatch, + {"release": True, "version": "1.2.3"}, + (False, "未找到资产文件：demoplugin_v1.2.3.zip"), + (False, "获取文件列表失败"), + ) + + success, message = helper.install(PLUGIN_ID, REPO_URL, package_version="v2", force_install=True) + + assert not success + assert "获取文件列表失败" == message + assert ["remove", "release", "filelist", "remove"] == calls + + def test_install_uses_filelist_when_release_flag_is_disabled(self, monkeypatch): + """ + 未开启 release 标记的插件保持原有文件列表安装路径。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = _patch_sync_remote_install( + helper, + monkeypatch, + {"release": False, "version": "1.2.3"}, + (False, "release should not be called"), + (True, ""), + ) + + success, message = helper.install(PLUGIN_ID, REPO_URL, package_version="v2", force_install=True) + + assert success + assert "" == message + assert ["remove", "filelist", "refresh"] == calls + + def test_install_rejects_release_without_version(self, monkeypatch): + """ + release 安装必须有插件版本号，否则无法构造稳定 tag。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = _patch_sync_remote_install( + helper, + monkeypatch, + {"release": True}, + (True, ""), + ) + + success, message = helper.install(PLUGIN_ID, REPO_URL, package_version="v2", force_install=True) + + assert not success + assert f"未在插件清单中找到 {PLUGIN_ID} 的版本号" in message + assert [] == calls + + def test_install_rejects_incompatible_plugin_before_content_preparation(self, monkeypatch): + """ + 系统版本不兼容时不会删除旧插件，也不会尝试 release 或文件列表安装。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = _patch_sync_remote_install( + helper, + monkeypatch, + {"release": True, "version": "1.2.3", "system_version": ">=9.0.0"}, + (True, ""), + ) + monkeypatch.setattr(PluginHelper, "get_current_system_version", lambda: Version("2.0.0")) + + success, message = helper.install(PLUGIN_ID, REPO_URL, package_version="v2", force_install=True) + + assert not success + assert "MoviePilot 版本 >=9.0.0" in message + assert [] == calls + + def test_install_rejects_invalid_parameters_before_remote_lookup(self): + """ + 远端安装缺少插件 ID 或仓库地址时直接拒绝。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + success, message = PluginHelper().install("", REPO_URL) + + assert not success + assert "参数错误" == message + + def test_install_rejects_invalid_repo_url(self): + """ + 仓库地址无法解析出 owner/repo 时直接拒绝。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + success, message = PluginHelper().install(PLUGIN_ID, "not-a-repo-url") + + assert not success + assert "不支持的插件仓库地址格式" == message + + def test_install_rejects_missing_package_version(self, monkeypatch): + """ + 当前系统版本找不到匹配插件索引时直接返回兼容性错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + monkeypatch.setattr(helper, "get_plugin_package_version", lambda *_args: None) + + success, message = helper.install(PLUGIN_ID, REPO_URL) + + assert not success + assert f"{PLUGIN_ID} 没有找到适用于当前版本的插件" == message + + def test_install_uses_default_package_version_when_not_provided(self, monkeypatch): + """ + 调用方未指定索引版本时使用系统版本标记继续安装。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + seen_versions = [] + monkeypatch.setattr(helper, "get_plugin_package_version", lambda _pid, _repo, version: seen_versions.append(version) or "") + monkeypatch.setattr(helper, "_PluginHelper__get_plugin_meta", lambda *_args: {"release": False, "version": "1.2.3"}) + monkeypatch.setattr(helper, "_PluginHelper__backup_plugin", lambda _pid: None) + monkeypatch.setattr(helper, "_PluginHelper__remove_old_plugin", lambda _pid: None) + monkeypatch.setattr(helper, "_PluginHelper__install_dependencies_if_required", lambda _pid: (False, True, "")) + monkeypatch.setattr(helper, "refresh_persistent_plugin_backup", lambda _pid: None) + monkeypatch.setattr(helper, "_PluginHelper__prepare_content_via_filelist_sync", lambda *_args: (True, "")) + + success, message = helper.install(PLUGIN_ID, REPO_URL, force_install=True) + + assert success + assert "" == message + assert seen_versions + + def test_install_local_delegates_local_repo_url(self, monkeypatch): + """ + local:// 来源由本地插件安装路径处理，不访问远端仓库。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + monkeypatch.setattr(helper, "install_local", lambda pid, repo_url, force_install=False: (True, f"{pid}:{repo_url}")) + + success, message = helper.install(PLUGIN_ID, f"local://{PLUGIN_ID}?path=/tmp/plugins") + + assert success + assert message.startswith(f"{PLUGIN_ID}:local://{PLUGIN_ID}") + + def test_install_release_download_failure_falls_back_to_filelist(self, monkeypatch): + """ + release tag 存在但 zip 下载失败时仍可回退文件列表安装。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = _patch_sync_remote_install( + helper, + monkeypatch, + {"release": True, "version": "1.2.3"}, + (False, "下载资产失败：502"), + (True, ""), + ) + + success, message = helper.install(PLUGIN_ID, REPO_URL, package_version="v2", force_install=True) + + assert success + assert "" == message + assert ["remove", "release", "filelist", "refresh"] == calls + + def test_async_install_uses_release_package_when_asset_is_available(self, monkeypatch): + """ + 异步安装路径在 release 包可用时优先使用 zip 安装。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = _patch_async_remote_install( + helper, + monkeypatch, + {"release": True, "version": "1.2.3"}, + (True, ""), + ) + + success, message = asyncio.run( + helper.async_install(PLUGIN_ID, REPO_URL, package_version="v2", force_install=True) + ) + + assert success + assert "" == message + assert calls[:2] == ["remove", "release"] + assert calls[2][0] == "to_thread" + + def test_async_install_falls_back_to_filelist_when_release_is_missing(self, monkeypatch): + """ + 异步安装路径在 release tag 或 zip 未生成时回退文件列表安装。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = _patch_async_remote_install( + helper, + monkeypatch, + {"release": True, "version": "1.2.3"}, + (False, "获取 Release 信息失败：404"), + (True, ""), + ) + + success, message = asyncio.run( + helper.async_install(PLUGIN_ID, REPO_URL, package_version="v2", force_install=True) + ) + + assert success + assert "" == message + assert calls[:3] == ["remove", "release", "filelist"] + assert calls[3][0] == "to_thread" + + def test_async_install_reports_filelist_error_after_release_fallback_fails(self, monkeypatch): + """ + 异步安装路径在 release 与文件列表都失败时返回文件列表错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = _patch_async_remote_install( + helper, + monkeypatch, + {"release": True, "version": "1.2.3"}, + (False, "未找到资产文件：demoplugin_v1.2.3.zip"), + (False, "获取文件列表失败"), + ) + + success, message = asyncio.run( + helper.async_install(PLUGIN_ID, REPO_URL, package_version="v2", force_install=True) + ) + + assert not success + assert "获取文件列表失败" == message + assert calls == ["remove", "release", "filelist", "remove"] + + def test_install_from_release_reports_missing_tag(self, monkeypatch): + """ + release tag 不存在时返回可用于降级判断的失败消息。 + """ + try: + from app.helper.plugin 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)) + + success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + + assert not success + assert "获取 Release 信息失败：404" == message + + def test_install_from_release_reports_missing_asset(self, monkeypatch): + """ + release tag 存在但缺少规范 zip 资产时返回明确错误。 + """ + try: + from app.helper.plugin 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(200, {"assets": [{"name": "other.zip", "id": 1}]}), + ) + + success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + + assert not success + assert "未找到资产文件：demoplugin_v1.2.3.zip" == message + + def test_install_from_release_reports_missing_asset_id(self, monkeypatch): + """ + release 资产缺少 id 时无法使用 API 下载，返回明确错误。 + """ + try: + from app.helper.plugin 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(200, {"assets": [{"name": "demoplugin_v1.2.3.zip"}]}), + ) + + success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + + assert not success + assert "资产缺少ID信息" == message + + def test_install_from_release_reports_malformed_release_payload(self, monkeypatch): + """ + release API 返回无法解析的结构时返回解析错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + class BadResponse(_FakeResponse): + """json() 抛错的响应对象。""" + + def json(self): + """模拟响应体不是合法 JSON。""" + raise ValueError("bad json") + + helper = PluginHelper() + monkeypatch.setattr(helper, "_PluginHelper__request_with_fallback", lambda *_args, **_kwargs: BadResponse(200)) + + success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + + assert not success + assert "解析 Release 信息失败" in message + + def test_install_from_release_reports_asset_download_failure(self, monkeypatch): + """ + release asset API 下载失败时返回下载错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + responses = iter([ + _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]}), + _FakeResponse(502), + ]) + monkeypatch.setattr(helper, "_PluginHelper__request_with_fallback", lambda *_args, **_kwargs: next(responses)) + + success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + + assert not success + assert "下载资产失败：502" == message + + def test_install_from_release_extracts_zip_with_top_level_directory(self, monkeypatch, tmp_path): + """ + release zip 带顶层插件目录时剥离该层后写入运行目录。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + release_payload = {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]} + zip_content = _build_zip({ + "demoplugin/__init__.py": b"plugin", + "demoplugin/nested/config.json": b"{}", + }) + responses = iter([ + _FakeResponse(200, release_payload), + _FakeContentResponse(200, zip_content), + ]) + monkeypatch.setattr("app.helper.plugin.settings", SimpleNamespace( + ROOT_PATH=tmp_path, + REPO_GITHUB_HEADERS=lambda repo=None: {}, + )) + monkeypatch.setattr(helper, "_PluginHelper__request_with_fallback", lambda *_args, **_kwargs: next(responses)) + + success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + + assert success + assert "" == message + assert (tmp_path / "app" / "plugins" / "demoplugin" / "__init__.py").read_bytes() == b"plugin" + assert (tmp_path / "app" / "plugins" / "demoplugin" / "nested" / "config.json").read_bytes() == b"{}" + + def test_install_from_release_creates_directory_entries(self, monkeypatch, tmp_path): + """ + release zip 内显式目录项会被创建，并继续写入后续文件。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as zf: + zf.writestr("demoplugin/assets/", b"") + zf.writestr("demoplugin/assets/icon.png", b"icon") + responses = iter([ + _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]}), + _FakeContentResponse(200, buffer.getvalue()), + ]) + monkeypatch.setattr("app.helper.plugin.settings", SimpleNamespace( + ROOT_PATH=tmp_path, + REPO_GITHUB_HEADERS=lambda repo=None: {}, + )) + monkeypatch.setattr(helper, "_PluginHelper__request_with_fallback", lambda *_args, **_kwargs: next(responses)) + + success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + + assert success + assert "" == message + assert (tmp_path / "app" / "plugins" / "demoplugin" / "assets").is_dir() + assert (tmp_path / "app" / "plugins" / "demoplugin" / "assets" / "icon.png").read_bytes() == b"icon" + + def test_install_from_release_reports_empty_zip(self, monkeypatch): + """ + release zip 为空时返回明确错误，避免安装出空插件目录。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + responses = iter([ + _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]}), + _FakeContentResponse(200, _build_zip({})), + ]) + monkeypatch.setattr(helper, "_PluginHelper__request_with_fallback", lambda *_args, **_kwargs: next(responses)) + + success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + + assert not success + assert "压缩包内容为空" == message + + def test_install_from_release_reports_directory_only_zip(self, monkeypatch, tmp_path): + """ + release zip 只有目录项时返回无可写入文件错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as zf: + zf.writestr("demoplugin/assets/", b"") + responses = iter([ + _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]}), + _FakeContentResponse(200, buffer.getvalue()), + ]) + monkeypatch.setattr("app.helper.plugin.settings", SimpleNamespace( + ROOT_PATH=tmp_path, + REPO_GITHUB_HEADERS=lambda repo=None: {}, + )) + monkeypatch.setattr(helper, "_PluginHelper__request_with_fallback", lambda *_args, **_kwargs: next(responses)) + + success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + + assert not success + assert "压缩包中无可写入文件" == message + + def test_install_from_release_reports_bad_zip(self, monkeypatch): + """ + release asset 不是合法 zip 时返回解压错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + responses = iter([ + _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]}), + _FakeContentResponse(200, b"not a zip"), + ]) + monkeypatch.setattr(helper, "_PluginHelper__request_with_fallback", lambda *_args, **_kwargs: next(responses)) + + success, message = helper._PluginHelper__install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + + assert not success + assert "解压 Release 压缩包失败" in message + + def test_install_flow_sync_restores_backup_when_prepare_fails(self, monkeypatch): + """ + 内容准备失败时恢复备份，避免安装失败后留下半成品目录。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = [] + monkeypatch.setattr(helper, "_PluginHelper__backup_plugin", lambda _pid: "/backup") + monkeypatch.setattr(helper, "_PluginHelper__remove_old_plugin", lambda _pid: calls.append("remove")) + monkeypatch.setattr(helper, "_PluginHelper__restore_plugin", lambda _pid, _backup: calls.append("restore")) + + success, message = helper._PluginHelper__install_flow_sync( + PLUGIN_ID, False, lambda: (False, "prepare failed") + ) + + assert not success + assert "prepare failed" == message + assert ["remove", "restore"] == calls + + def test_install_flow_sync_restores_backup_when_dependency_install_fails(self, monkeypatch): + """ + 依赖安装失败时恢复备份，避免新插件内容破坏可用版本。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = [] + monkeypatch.setattr(helper, "_PluginHelper__backup_plugin", lambda _pid: "/backup") + monkeypatch.setattr(helper, "_PluginHelper__remove_old_plugin", lambda _pid: calls.append("remove")) + monkeypatch.setattr(helper, "_PluginHelper__restore_plugin", lambda _pid, _backup: calls.append("restore")) + monkeypatch.setattr( + helper, + "_PluginHelper__install_dependencies_if_required", + lambda _pid: (True, False, "dependency failed"), + ) + + success, message = helper._PluginHelper__install_flow_sync( + PLUGIN_ID, False, lambda: (True, "") + ) + + assert not success + assert "dependency failed" == message + assert ["remove", "restore"] == calls + + def test_prepare_content_via_filelist_sync_preinstalls_requirements_and_downloads(self, monkeypatch): + """ + 文件列表安装会先尝试 requirements 预安装，再下载插件文件。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = [] + requirements = {"name": "requirements.txt", "download_url": "https://example.com/requirements.txt"} + file_list = [requirements, {"name": "__init__.py", "download_url": "https://example.com/__init__.py"}] + monkeypatch.setattr(helper, "_PluginHelper__get_file_list", lambda *_args: (file_list, "")) + monkeypatch.setattr( + helper, + "_PluginHelper__download_and_install_requirements", + lambda *_args: calls.append("requirements") or (True, ""), + ) + monkeypatch.setattr( + helper, + "_PluginHelper__download_files", + lambda *_args: calls.append("download") or (True, ""), + ) + + success, message = helper._PluginHelper__prepare_content_via_filelist_sync("demoplugin", "demo/repo", "v2") + + assert success + assert "" == message + assert ["requirements", "download"] == calls + + def test_prepare_content_via_filelist_sync_continues_when_requirements_preinstall_fails(self, monkeypatch): + """ + requirements 预安装失败不阻断文件下载，最终依赖安装由统一流程兜底。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = [] + file_list = [{"name": "requirements.txt"}, {"name": "__init__.py"}] + monkeypatch.setattr(helper, "_PluginHelper__get_file_list", lambda *_args: (file_list, "")) + monkeypatch.setattr( + helper, + "_PluginHelper__download_and_install_requirements", + lambda *_args: calls.append("requirements") or (False, "preinstall failed"), + ) + monkeypatch.setattr( + helper, + "_PluginHelper__download_files", + lambda *_args: calls.append("download") or (True, ""), + ) + + success, message = helper._PluginHelper__prepare_content_via_filelist_sync("demoplugin", "demo/repo", "v2") + + assert success + assert "" == message + assert ["requirements", "download"] == calls + + def test_prepare_content_via_filelist_sync_reports_missing_file_list(self, monkeypatch): + """ + 文件列表为空时直接返回列表获取错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + monkeypatch.setattr(helper, "_PluginHelper__get_file_list", lambda *_args: ([], "list failed")) + + success, message = helper._PluginHelper__prepare_content_via_filelist_sync("demoplugin", "demo/repo", "v2") + + assert not success + assert "list failed" == message + + def test_prepare_content_via_filelist_sync_returns_download_error(self, monkeypatch): + """ + 文件列表存在但文件下载失败时向上返回下载错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + monkeypatch.setattr(helper, "_PluginHelper__get_file_list", lambda *_args: ([{"name": "__init__.py"}], "")) + monkeypatch.setattr(helper, "_PluginHelper__download_files", lambda *_args: (False, "download failed")) + + success, message = helper._PluginHelper__prepare_content_via_filelist_sync("demoplugin", "demo/repo", "v2") + + assert not success + assert "download failed" == message + + def test_async_prepare_content_via_filelist_preinstalls_requirements_and_downloads(self, monkeypatch): + """ + 异步文件列表安装会先尝试 requirements 预安装，再下载插件文件。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = [] + requirements = {"name": "requirements.txt", "download_url": "https://example.com/requirements.txt"} + file_list = [requirements, {"name": "__init__.py", "download_url": "https://example.com/__init__.py"}] + + async def fake_file_list(*_args): + return file_list, "" + + async def fake_requirements(*_args): + calls.append("requirements") + return True, "" + + async def fake_download(*_args): + calls.append("download") + return True, "" + + monkeypatch.setattr(helper, "_PluginHelper__async_get_file_list", fake_file_list) + monkeypatch.setattr(helper, "_PluginHelper__async_download_and_install_requirements", fake_requirements) + monkeypatch.setattr(helper, "_PluginHelper__async_download_files", fake_download) + + success, message = asyncio.run( + helper._PluginHelper__prepare_content_via_filelist_async("demoplugin", "demo/repo", "v2") + ) + + assert success + assert "" == message + assert ["requirements", "download"] == calls + + def test_async_prepare_content_via_filelist_reports_missing_file_list(self, monkeypatch): + """ + 异步文件列表为空时直接返回列表获取错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + + async def fake_file_list(*_args): + return [], "list failed" + + monkeypatch.setattr(helper, "_PluginHelper__async_get_file_list", fake_file_list) + + success, message = asyncio.run( + helper._PluginHelper__prepare_content_via_filelist_async("demoplugin", "demo/repo", "v2") + ) + + assert not success + assert "list failed" == message + + def test_async_prepare_content_via_filelist_returns_download_error(self, monkeypatch): + """ + 异步文件列表下载失败时向上返回下载错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + + async def fake_file_list(*_args): + return [{"name": "__init__.py"}], "" + + async def fake_download(*_args): + return False, "download failed" + + monkeypatch.setattr(helper, "_PluginHelper__async_get_file_list", fake_file_list) + monkeypatch.setattr(helper, "_PluginHelper__async_download_files", fake_download) + + success, message = asyncio.run( + helper._PluginHelper__prepare_content_via_filelist_async("demoplugin", "demo/repo", "v2") + ) + + assert not success + assert "download failed" == message + + def test_install_flow_async_restores_backup_when_prepare_fails(self, monkeypatch): + """ + 异步内容准备失败时恢复备份。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = [] + + async def backup(_pid): + return "/backup" + + async def remove(_pid): + calls.append("remove") + + async def restore(_pid, _backup): + calls.append("restore") + + async def prepare(): + return False, "prepare failed" + + monkeypatch.setattr(helper, "_PluginHelper__async_backup_plugin", backup) + monkeypatch.setattr(helper, "_PluginHelper__async_remove_old_plugin", remove) + monkeypatch.setattr(helper, "_PluginHelper__async_restore_plugin", restore) + + success, message = asyncio.run(helper._PluginHelper__install_flow_async(PLUGIN_ID, False, prepare)) + + assert not success + assert "prepare failed" == message + assert ["remove", "restore"] == calls + + def test_install_flow_async_restores_backup_when_dependency_install_fails(self, monkeypatch): + """ + 异步依赖安装失败时恢复备份。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + calls = [] + + async def backup(_pid): + return "/backup" + + async def remove(_pid): + calls.append("remove") + + async def restore(_pid, _backup): + calls.append("restore") + + async def prepare(): + return True, "" + + async def dependencies(_pid): + return True, False, "dependency failed" + + monkeypatch.setattr(helper, "_PluginHelper__async_backup_plugin", backup) + monkeypatch.setattr(helper, "_PluginHelper__async_remove_old_plugin", remove) + monkeypatch.setattr(helper, "_PluginHelper__async_restore_plugin", restore) + monkeypatch.setattr(helper, "_PluginHelper__async_install_dependencies_if_required", dependencies) + + success, message = asyncio.run(helper._PluginHelper__install_flow_async(PLUGIN_ID, False, prepare)) + + assert not success + assert "dependency failed" == message + assert ["remove", "restore"] == calls + + def test_async_install_from_release_reports_missing_asset(self, monkeypatch): + """ + 异步 release tag 存在但缺少规范 zip 资产时返回明确错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + + async def fake_request(*_args, **_kwargs): + return _FakeResponse(200, {"assets": [{"name": "other.zip", "id": 1}]}) + + monkeypatch.setattr(helper, "_PluginHelper__async_request_with_fallback", fake_request) + + success, message = asyncio.run( + helper._PluginHelper__async_install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + ) + + assert not success + assert "未找到资产文件：demoplugin_v1.2.3.zip" == message + + def test_async_install_from_release_reports_missing_tag(self, monkeypatch): + """ + 异步 release tag 不存在时返回获取 release 失败。 + """ + try: + from app.helper.plugin 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) + + success, message = asyncio.run( + helper._PluginHelper__async_install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + ) + + assert not success + assert "获取 Release 信息失败：404" == message + + def test_async_install_from_release_reports_missing_asset_id(self, monkeypatch): + """ + 异步 release 资产缺少 id 时返回明确错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + + async def fake_request(*_args, **_kwargs): + return _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip"}]}) + + monkeypatch.setattr(helper, "_PluginHelper__async_request_with_fallback", fake_request) + + success, message = asyncio.run( + helper._PluginHelper__async_install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + ) + + assert not success + assert "资产缺少ID信息" == message + + def test_async_install_from_release_reports_asset_download_failure(self, monkeypatch): + """ + 异步 release asset 下载失败时返回下载错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + responses = iter([ + _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]}), + _FakeResponse(502), + ]) + + async def fake_request(*_args, **_kwargs): + return next(responses) + + monkeypatch.setattr(helper, "_PluginHelper__async_request_with_fallback", fake_request) + + success, message = asyncio.run( + helper._PluginHelper__async_install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + ) + + assert not success + assert "下载资产失败：502" == message + + def test_async_install_from_release_extracts_zip_with_top_level_directory(self, monkeypatch, tmp_path): + """ + 异步 release zip 带顶层插件目录时剥离该层后写入运行目录。 + """ + try: + from app.core.config import settings + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + responses = iter([ + _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]}), + _FakeContentResponse(200, _build_zip({"demoplugin/__init__.py": b"plugin"})), + ]) + + async def fake_request(*_args, **_kwargs): + return next(responses) + + monkeypatch.setattr("app.helper.plugin.settings", SimpleNamespace( + ROOT_PATH=tmp_path, + REPO_GITHUB_HEADERS=lambda repo=None: {}, + )) + monkeypatch.setattr(helper, "_PluginHelper__async_request_with_fallback", fake_request) + + success, message = asyncio.run( + helper._PluginHelper__async_install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + ) + + assert success + assert "" == message + assert (tmp_path / "app" / "plugins" / "demoplugin" / "__init__.py").read_bytes() == b"plugin" + + def test_async_install_from_release_reports_empty_zip(self, monkeypatch): + """ + 异步 release zip 为空时返回明确错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + responses = iter([ + _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]}), + _FakeContentResponse(200, _build_zip({})), + ]) + + async def fake_request(*_args, **_kwargs): + return next(responses) + + monkeypatch.setattr(helper, "_PluginHelper__async_request_with_fallback", fake_request) + + success, message = asyncio.run( + helper._PluginHelper__async_install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + ) + + assert not success + assert "压缩包内容为空" == message + + def test_async_install_from_release_reports_bad_zip(self, monkeypatch): + """ + 异步 release asset 不是合法 zip 时返回解压错误。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + helper = PluginHelper() + responses = iter([ + _FakeResponse(200, {"assets": [{"name": "demoplugin_v1.2.3.zip", "id": 42}]}), + _FakeContentResponse(200, b"not a zip"), + ]) + + async def fake_request(*_args, **_kwargs): + return next(responses) + + monkeypatch.setattr(helper, "_PluginHelper__async_request_with_fallback", fake_request) + + success, message = asyncio.run( + helper._PluginHelper__async_install_from_release(PLUGIN_ID, "demo/repo", "DemoPlugin_v1.2.3") + ) + + assert not success + assert "解压 Release 压缩包失败" in message + + def test_install_local_rejects_mismatched_local_repo_id(self): + """ + 本地插件来源中的插件 ID 必须与安装目标一致。 + """ + try: + from app.helper.plugin import PluginHelper + except ModuleNotFoundError as exc: + pytest.skip(f"missing dependency: {exc}") + + success, message = PluginHelper().install("DemoPlugin", "local://OtherPlugin?path=/tmp/plugins") + + assert not success + assert "本地插件来源与插件ID不匹配" == message