diff --git a/astrbot/core/star/star_manager.py b/astrbot/core/star/star_manager.py index 0443dca54..23109e7d6 100644 --- a/astrbot/core/star/star_manager.py +++ b/astrbot/core/star/star_manager.py @@ -8,6 +8,7 @@ import json import keyword import logging import os +import shutil import sys import tempfile import traceback @@ -36,6 +37,7 @@ from astrbot.core.utils.astrbot_path import ( get_astrbot_config_path, get_astrbot_path, get_astrbot_plugin_path, + get_astrbot_system_tmp_path, get_astrbot_temp_path, ) from astrbot.core.utils.io import remove_dir @@ -292,6 +294,9 @@ class PluginManager: dirs = os.listdir(path) # 遍历文件夹，找到 main.py 或者和文件夹同名的文件 for d in dirs: + # Staged installs and rollback backups are not discoverable plugins. + if d.startswith((".plugin-install-", ".plugin-upload-", ".plugin-backup-")): + continue if os.path.isdir(os.path.join(path, d)): if os.path.exists(os.path.join(path, d, "main.py")): module_str = "main" @@ -1616,117 +1621,36 @@ class PluginManager: ignore_version_check: bool = False, download_url: str = "", ): - """从仓库 URL 安装插件 - - 从指定的仓库 URL 下载并安装插件，然后加载该插件到系统中 + """Install or update a plugin from a repository URL. Args: - repo_url (str): 要安装的插件仓库 URL - proxy (str, optional): 用于下载的代理服务器。默认为空字符串。 - download_url (str, optional): 插件压缩包下载地址。提供时优先从此地址下载安装包。 + repo_url: Plugin repository URL. + proxy: Optional proxy prefix for repository downloads. + ignore_version_check: Whether to bypass AstrBot version compatibility. + download_url: Optional archive URL to download instead of the repository. Returns: - dict | None: 安装成功时返回包含插件信息的字典: - - repo: 插件的仓库 URL - - readme: README.md 文件的内容(如果存在) - 如果找不到插件元数据则返回 None。 + Installed plugin repository, README, and name, if registered. + Raises: + Exception: If downloading, validation, or loading fails. """ - # this metric is for displaying plugins installation count in pages - asyncio.create_task( - Metric.upload( - et="install_star", - repo=repo_url, - ), - ) - + asyncio.create_task(Metric.upload(et="install_star", repo=repo_url)) async with self._pm_lock: - plugin_path = "" - dir_name = "" - try: - if download_url: - plugin_path = await self._updater.install( - repo_url, - proxy, - download_url=download_url, - ) - else: - plugin_path = await self._updater.install(repo_url, proxy) - - # reload the plugin - dir_name = os.path.basename(plugin_path) - metadata_dir_name = self._get_plugin_dir_name_from_metadata(plugin_path) - target_plugin_path = os.path.join( - self.plugin_store_path, - metadata_dir_name, + temp_root = Path(get_astrbot_system_tmp_path()) + temp_root.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory( + dir=temp_root, prefix=".plugin-install-" + ) as staging_dir: + plugin_path = await self._updater.install( + repo_url, + proxy, + download_url=download_url, + target_dir=Path(staging_dir) / "plugin", ) - if target_plugin_path != plugin_path and os.path.exists( - target_plugin_path - ): - raise Exception(f"安装失败：目录 {metadata_dir_name} 已存在。") - if target_plugin_path != plugin_path: - os.rename(plugin_path, target_plugin_path) - plugin_path = target_plugin_path - dir_name = metadata_dir_name - await self._ensure_plugin_requirements( - plugin_path, - dir_name, + return await self._install_plugin_from_directory( + plugin_path, ignore_version_check=ignore_version_check ) - success, error_message = await self.load( - specified_dir_name=dir_name, - ignore_version_check=ignore_version_check, - ) - if not success: - raise Exception( - error_message - or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。" - ) - - # Get the plugin metadata to return repo info - plugin = self.context.get_registered_star(dir_name) - if not plugin: - # Try to find by other name if directory name doesn't match plugin name - for star in self.context.get_all_stars(): - if star.root_dir_name == dir_name: - plugin = star - break - - # Extract README.md content if exists - readme_content = None - readme_path = os.path.join(plugin_path, "README.md") - if not os.path.exists(readme_path): - readme_path = os.path.join(plugin_path, "readme.md") - - if os.path.exists(readme_path): - try: - with open(readme_path, encoding="utf-8") as f: - readme_content = f.read() - except Exception as e: - logger.warning( - f"Failed to read README.md for plugin {dir_name}: {e!s}", - ) - - plugin_info = None - if plugin: - plugin_info = { - "repo": plugin.repo, - "readme": readme_content, - "name": plugin.name, - } - - return plugin_info - except Exception as e: - self._track_failed_install_dir( - dir_name=dir_name, - plugin_path=plugin_path, - error=e, - ) - if dir_name and plugin_path: - logger.warning( - f"Failed to install plugin {dir_name}; installation " - f"directory: {plugin_path}", - ) - raise async def inspect_plugin_repository( self, @@ -2074,45 +1998,195 @@ class PluginManager: async def install_plugin_from_file( self, zip_file_path: str, ignore_version_check: bool = False ): - dir_name = os.path.splitext(os.path.basename(zip_file_path))[0] - desti_dir = tempfile.mkdtemp( - dir=self.plugin_store_path, prefix="plugin_upload_" - ) - temp_desti_dir = desti_dir - skip_failed_tracking = False + """Install an uploaded archive or replace the same installed plugin. - try: - self._updater._extract_plugin_archive(zip_file_path, desti_dir) - metadata_dir_name = self._get_plugin_dir_name_from_metadata(desti_dir) - target_plugin_path = os.path.join( - self.plugin_store_path, - metadata_dir_name, - ) - if target_plugin_path != desti_dir and os.path.exists(target_plugin_path): - skip_failed_tracking = True - raise Exception(f"安装失败：目录 {metadata_dir_name} 已存在。") - if target_plugin_path != desti_dir: - os.rename(desti_dir, target_plugin_path) - dir_name = metadata_dir_name - desti_dir = target_plugin_path + Args: + zip_file_path: Path to the uploaded plugin archive. + ignore_version_check: Whether to bypass AstrBot version compatibility. - # remove the zip - try: - os.remove(zip_file_path) - except BaseException as e: - logger.warning(f"Failed to delete the plugin archive: {e!s}") - await self._ensure_plugin_requirements(desti_dir, dir_name) - # await self.reload() - success, error_message = await self.load( - specified_dir_name=dir_name, - ignore_version_check=ignore_version_check, - ) - if not success: - raise Exception( - error_message - or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。" + Returns: + Installed plugin repository, README, and name, if registered. + + Raises: + Exception: If validation, dependency installation, or loading fails. + """ + async with self._pm_lock: + temp_root = Path(get_astrbot_system_tmp_path()) + temp_root.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory( + dir=temp_root, prefix=".plugin-upload-" + ) as staging_dir: + plugin_path = Path(staging_dir) / "plugin" + self._updater._extract_plugin_archive(zip_file_path, str(plugin_path)) + plugin_info = await self._install_plugin_from_directory( + str(plugin_path), ignore_version_check=ignore_version_check ) + try: + Path(zip_file_path).unlink(missing_ok=True) + except OSError as exc: + logger.warning("Failed to delete the plugin archive: %s", exc) + if plugin_info and plugin_info.get("repo"): + asyncio.create_task( + Metric.upload(et="install_star_f", repo=plugin_info["repo"]) + ) + return plugin_info + async def _install_plugin_from_directory( + self, plugin_path: str, ignore_version_check: bool = False + ): + """Install a staged plugin directory, restoring old code on update failure. + + The caller holds the plugin manager lock and owns the staging directory. + Only failed first installs moved to a permanent directory are tracked. + + Args: + plugin_path: Extracted or cloned plugin in a temporary directory. + ignore_version_check: Whether to bypass AstrBot version compatibility. + + Returns: + Installed plugin repository, README, and name, if registered. + + Raises: + Exception: If validation, dependency installation, or loading fails. + """ + desti_dir = plugin_path + dir_name = Path(plugin_path).name + track_failed_install = False + try: + metadata_dir_name = self._get_plugin_dir_name_from_metadata(desti_dir) + plugin = next( + ( + star + for star in self.context.get_all_stars() + if star.name == metadata_dir_name + ), + None, + ) + target_plugin_path = Path(self.plugin_store_path) / ( + plugin.root_dir_name + if plugin and plugin.root_dir_name + else metadata_dir_name + ) + if plugin and plugin.reserved: + raise Exception("该插件是 AstrBot 保留插件，无法更新。") + if target_plugin_path.exists(): + # Only replace a directory whose metadata identifies the same plugin. + if ( + target_plugin_path.is_symlink() + or not target_plugin_path.is_dir() + or self._get_plugin_dir_name_from_metadata(str(target_plugin_path)) + != metadata_dir_name + ): + raise Exception( + f"安装失败：目录 {target_plugin_path.name} 已存在。" + ) + + metadata = self._load_plugin_metadata(desti_dir) + if metadata and not ignore_version_check: + is_valid, error_message = self._validate_astrbot_version_specifier( + metadata.astrbot_version + ) + if not is_valid: + raise PluginVersionUnsupportedError(error_message) + dir_name = target_plugin_path.name + await self._ensure_plugin_requirements(desti_dir, dir_name) + + # Finish copying the backup before removing any installed files. + # The system temporary directory may be on a different filesystem. + backup_dir = Path( + tempfile.mkdtemp( + dir=get_astrbot_system_tmp_path(), prefix=".plugin-backup-" + ) + ) + backup_path = backup_dir / dir_name + backup_complete = False + keep_backup = False + try: + if plugin: + try: + await self._terminate_plugin(plugin) + except Exception: + logger.warning( + "Plugin %s did not terminate cleanly", + dir_name, + exc_info=True, + ) + self._cleanup_plugin_state(dir_name) + shutil.copytree(target_plugin_path, backup_path, symlinks=True) + backup_complete = True + keep_backup = True + remove_dir(str(target_plugin_path)) + shutil.move(desti_dir, str(target_plugin_path)) + desti_dir = str(target_plugin_path) + success, error_message = await self.load( + specified_dir_name=dir_name, + ignore_version_check=ignore_version_check, + ) + if not success: + raise Exception(error_message or f"更新插件 {dir_name} 失败。") + keep_backup = False + except BaseException: + try: + self._cleanup_plugin_state(dir_name) + if backup_complete: + if target_plugin_path.exists(): + remove_dir(str(target_plugin_path)) + shutil.copytree( + backup_path, target_plugin_path, symlinks=True + ) + restored, restore_error = await self.load( + specified_dir_name=dir_name, + ignore_version_check=True, + ) + if not restored: + logger.error( + "Failed to reload restored plugin %s: %s", + dir_name, + restore_error, + ) + else: + keep_backup = False + self.failed_plugin_dict.pop(dir_name, None) + self._rebuild_failed_plugin_info() + except Exception: + logger.exception( + "Failed to restore plugin %s; backup: %s", + dir_name, + backup_path, + ) + raise + finally: + if not keep_backup: + try: + remove_dir(str(backup_dir)) + except Exception: + logger.warning( + "Failed to remove plugin backup %s", + backup_dir, + exc_info=True, + ) + else: + logger.warning( + "Retained plugin backup for recovery: %s", backup_path + ) + else: + track_failed_install = True + dir_name = target_plugin_path.name + desti_dir = str(target_plugin_path) + shutil.move(plugin_path, desti_dir) + await self._ensure_plugin_requirements(desti_dir, dir_name) + success, error_message = await self.load( + specified_dir_name=dir_name, + ignore_version_check=ignore_version_check, + ) + if not success: + raise Exception( + error_message + or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。" + ) + + self.failed_plugin_dict.pop(dir_name, None) + self._rebuild_failed_plugin_info() # Get the plugin metadata to return repo info plugin = self.context.get_registered_star(dir_name) if not plugin: @@ -2145,17 +2219,9 @@ class PluginManager: "name": plugin.name, } - if plugin.repo: - asyncio.create_task( - Metric.upload( - et="install_star_f", # install star - repo=plugin.repo, - ), - ) - return plugin_info except Exception as e: - if not skip_failed_tracking: + if track_failed_install: self._track_failed_install_dir( dir_name=dir_name, plugin_path=desti_dir, @@ -2166,14 +2232,3 @@ class PluginManager: f"{desti_dir}", ) raise - finally: - if (skip_failed_tracking or temp_desti_dir != desti_dir) and os.path.isdir( - temp_desti_dir - ): - try: - remove_dir(temp_desti_dir) - except Exception as e: - logger.warning( - "Failed to remove the temporary plugin extraction directory " - f"{temp_desti_dir}: {e!s}", - ) diff --git a/astrbot/core/star/updater.py b/astrbot/core/star/updater.py index c82a4546c..5a0ae4c51 100644 --- a/astrbot/core/star/updater.py +++ b/astrbot/core/star/updater.py @@ -197,11 +197,33 @@ class _PluginUpdater(_RepoZipUpdater): "repo": str(normalized_metadata.get("repo") or normalized_url), } - async def install(self, repo_url: str, proxy="", download_url: str = "") -> str: + async def install( + self, + repo_url: str, + proxy="", + download_url: str = "", + *, + target_dir: Path | None = None, + ) -> str: + """Download or clone a plugin into a new directory. + + Args: + repo_url: Plugin repository URL. + proxy: Optional proxy prefix for repository downloads. + download_url: Optional archive URL to use instead of the repository. + target_dir: Staging destination supplied by the plugin manager. Defaults + to the repository directory under the plugin store. + + Returns: + Path to the extracted or cloned plugin. + + Raises: + Exception: If the destination exists or downloading or validation fails. + """ normalized_url = normalize_repository_url(repo_url) repository = parse_repository_url(normalized_url) repo_name = self._format_name(repository.name) - plugin_path = os.path.join(self.plugin_store_path, repo_name) + plugin_path = str(target_dir or Path(self.plugin_store_path) / repo_name) if os.path.exists(plugin_path): raise Exception(f"安装失败：目录 {repo_name} 已存在。") if download_url: diff --git a/tests/fixtures/helpers.py b/tests/fixtures/helpers.py index de4aa30a7..64aa0cfe5 100644 --- a/tests/fixtures/helpers.py +++ b/tests/fixtures/helpers.py @@ -434,12 +434,15 @@ class MockPluginBuilder: def create( self, plugin_config: str | MockPluginConfig | None = None, + *, + target_dir: Path | None = None, **kwargs, ) -> Path: """创建模拟插件。 Args: plugin_config: 插件名称字符串、MockPluginConfig 对象或 None + target_dir: Optional staging directory instead of the installed path. **kwargs: 如果 plugin_config 是字符串或 None，这些参数用于构建 MockPluginConfig Returns: @@ -456,7 +459,11 @@ class MockPluginBuilder: raise TypeError(f"Invalid plugin_config type: {type(plugin_config)}") # 创建插件目录 - plugin_dir = self.plugin_store_path / config.name + plugin_dir = ( + target_dir + if target_dir is not None + else self.plugin_store_path / config.name + ) plugin_dir.mkdir(parents=True, exist_ok=True) # 创建 metadata.yaml @@ -548,8 +555,25 @@ def create_mock_updater_install( Callable: 异步函数，可用于 monkeypatch.setattr """ - async def mock_install(repo_url: str, proxy: str = "") -> str: - """Mock updater.install 方法。""" + async def mock_install( + repo_url: str, + proxy: str = "", + download_url: str = "", + *, + target_dir: Path | None = None, + ) -> str: + """Create a plugin at the updater's requested destination. + + Args: + repo_url: Repository URL used for plugin identity and metadata. + proxy: Unused download proxy. + download_url: Unused archive URL. + target_dir: Optional staging directory supplied by the manager. + + Returns: + Path to the prepared plugin directory. + """ + del proxy, download_url # 查找插件名称 plugin_name = None if repo_to_plugin: @@ -563,7 +587,7 @@ def create_mock_updater_install( # 创建插件目录 config = MockPluginConfig(name=plugin_name, repo=repo_url) - plugin_dir = plugin_builder.create(config) + plugin_dir = plugin_builder.create(config, target_dir=target_dir) return str(plugin_dir) return mock_install diff --git a/tests/test_plugin_manager.py b/tests/test_plugin_manager.py index 5cb8747eb..2bf27a2aa 100644 --- a/tests/test_plugin_manager.py +++ b/tests/test_plugin_manager.py @@ -1,10 +1,14 @@ import asyncio +import errno import functools import json import os +import sys +import zipfile from pathlib import Path from types import ModuleType from typing import Any, cast +from unittest.mock import AsyncMock import pytest import yaml @@ -384,6 +388,11 @@ def plugin_manager_pm(tmp_path, monkeypatch): "astrbot.core.star.star_manager.get_astrbot_plugin_path", lambda: str(plugin_dir), ) + monkeypatch.setattr( + star_manager_module, + "get_astrbot_system_tmp_path", + lambda: str(tmp_path / "system_temp"), + ) return pm @@ -401,18 +410,26 @@ def local_updater(plugin_manager_pm): @pytest.mark.asyncio @pytest.mark.parametrize("dependency_install_fails", [False, True]) +@pytest.mark.parametrize("cross_filesystem", [False, True]) async def test_install_plugin_dependency_install_flow( - plugin_manager_pm: PluginManager, monkeypatch, dependency_install_fails: bool + plugin_manager_pm: PluginManager, monkeypatch, dependency_install_fails: bool, + cross_filesystem: bool, ): plugin_path = Path(plugin_manager_pm.plugin_store_path) / TEST_PLUGIN_DIR events = [] _mock_missing_requirements(monkeypatch, {"networkx"}) + if cross_filesystem: + def cross_device_rename(*args, **kwargs): + raise OSError(errno.EXDEV, "Cross-device move") - async def mock_install(repo_url: str, proxy=""): + monkeypatch.setattr(star_manager_module.os, "rename", cross_device_rename) + + async def mock_install(repo_url: str, proxy="", *, download_url="", target_dir): assert repo_url == TEST_PLUGIN_REPO - _write_local_test_plugin(plugin_path, repo_url) - _write_requirements(plugin_path) - return str(plugin_path) + staged_path = Path(target_dir) + _write_local_test_plugin(staged_path, repo_url) + _write_requirements(staged_path) + return str(staged_path) monkeypatch.setattr(plugin_manager_pm._updater, "install", mock_install) monkeypatch.setattr( @@ -435,6 +452,8 @@ async def test_install_plugin_dependency_install_flow( expected_original_path=plugin_path / "requirements.txt", expected_content="networkx\n", ) + assert set(plugin_manager_pm.failed_plugin_dict) == {TEST_PLUGIN_DIR} + assert set(Path(plugin_manager_pm.plugin_store_path).iterdir()) == {plugin_path} else: await plugin_manager_pm.install_plugin(TEST_PLUGIN_REPO) assert len(events) == 2 @@ -501,7 +520,7 @@ async def test_install_plugin_from_file_conflict_keeps_failed_plugins_clean( zip_file_path = tmp_path / "plugin_upload_helloworld_v2.zip" zip_file_path.write_text("placeholder", encoding="utf-8") plugin_store_path = Path(plugin_manager_pm.plugin_store_path) - existing_upload_dirs = set(plugin_store_path.glob("plugin_upload_*")) + existing_dirs = set(plugin_store_path.iterdir()) def mock_unzip_file(zip_path: str, target_dir: str) -> None: assert zip_path == str(zip_file_path) @@ -512,6 +531,10 @@ async def test_install_plugin_from_file_conflict_keeps_failed_plugins_clean( ) assert local_updater.is_dir() + metadata_path = local_updater / "metadata.yaml" + metadata = yaml.safe_load(metadata_path.read_text(encoding="utf-8")) + metadata["name"] = "another_plugin" + metadata_path.write_text(yaml.safe_dump(metadata), encoding="utf-8") monkeypatch.setattr( plugin_manager_pm._updater, "_extract_plugin_archive", @@ -521,13 +544,453 @@ async def test_install_plugin_from_file_conflict_keeps_failed_plugins_clean( with pytest.raises(Exception, match=f"安装失败：目录 {TEST_PLUGIN_DIR} 已存在。"): await plugin_manager_pm.install_plugin_from_file(str(zip_file_path)) - new_upload_dirs = [ - upload_dir - for upload_dir in plugin_store_path.glob("plugin_upload_*") - if upload_dir not in existing_upload_dirs - ] assert plugin_manager_pm.failed_plugin_dict == {} - assert new_upload_dirs == [] + assert set(plugin_store_path.iterdir()) == existing_dirs + assert yaml.safe_load(metadata_path.read_text(encoding="utf-8")) == metadata + + +@pytest.mark.asyncio +@pytest.mark.parametrize("install_source", ["upload", "github", "url", "git"]) +@pytest.mark.parametrize("legacy_directory", [False, True]) +@pytest.mark.parametrize("failure", [None, "dependencies", "load", "cancel", "version"]) +async def test_install_updates_existing_plugin_and_restores_on_failure( + plugin_manager_pm: PluginManager, + local_updater: Path, + monkeypatch, + tmp_path: Path, + legacy_directory: bool, + failure: str | None, + install_source: str, +): + """Every install source replaces old code or restores it after a failed load.""" + _clear_star_runtime_state() + if legacy_directory: + local_updater = local_updater.rename(local_updater.with_name("legacy_plugin")) + system_temp = Path(star_manager_module.get_astrbot_system_tmp_path()) + original_rename = os.rename + + def cross_device_rename(source, destination, *args, **kwargs): + if Path(source).is_relative_to(system_temp) != Path(destination).is_relative_to(system_temp): + raise OSError(errno.EXDEV, "Cross-device move") + return original_rename(source, destination, *args, **kwargs) + + monkeypatch.setattr(star_manager_module.os, "rename", cross_device_rename) + dir_name = local_updater.name + module_path = f"data.plugins.{dir_name}.main" + old_plugin = star_manager_module.StarMetadata( + name=TEST_PLUGIN_NAME, + root_dir_name=dir_name, + module_path=module_path, + version="1.0.0", + reserved=False, + ) + star_manager_module.star_registry.append(old_plugin) + star_manager_module.star_map[module_path] = old_plugin + sys.modules[module_path] = ModuleType(module_path) + monkeypatch.setattr( + plugin_manager_pm.context, "stars", star_manager_module.star_registry + ) + (local_updater / "obsolete.py").write_text("old code", encoding="utf-8") + config_path = tmp_path / "config" / f"{dir_name}_config.json" + config_path.parent.mkdir() + config_path.write_text('{"custom": true}', encoding="utf-8") + monkeypatch.setattr( + plugin_manager_pm, "plugin_config_path", str(config_path.parent) + ) + data_path = tmp_path / "plugin_data" / dir_name / "digest.db" + data_path.parent.mkdir(parents=True) + data_path.write_bytes(b"saved digest") + + source_path = tmp_path / "new_version" + _write_local_test_plugin(source_path, TEST_PLUGIN_REPO, version="2.0.0") + metadata_path = source_path / "metadata.yaml" + metadata = yaml.safe_load(metadata_path.read_text(encoding="utf-8")) + metadata.pop("repo") + if failure == "version": + metadata["astrbot_version"] = ">=999.0" + metadata_path.write_text(yaml.safe_dump(metadata), encoding="utf-8") + (source_path / "README.md").write_text("Updated README", encoding="utf-8") + zip_path = tmp_path / "plugin-v2.zip" + with zipfile.ZipFile(zip_path, "w") as archive: + for path in source_path.iterdir(): + archive.write(path, f"release-v2/{path.name}") + + repo_url = f"https://github.com/AstrBotDevs/{TEST_PLUGIN_DIR}" + if install_source == "git": + repo_url = f"https://gitee.com/AstrBotDevs/{TEST_PLUGIN_DIR}.git" + proxy_url = "https://proxy.example" + download_url = "https://cdn.example/plugin-v2.zip" + + async def download_repository(plugin_path, url, proxy): + assert url == repo_url + assert proxy == proxy_url + Path(plugin_path + ".zip").write_bytes(zip_path.read_bytes()) + + async def download_file(url, path): + assert url == download_url + Path(path).write_bytes(zip_path.read_bytes()) + + async def clone_repository(url, target_path): + assert url == repo_url + target_path = Path(target_path) + assert not target_path.exists() + target_path.mkdir() + for path in source_path.iterdir(): + (target_path / path.name).write_bytes(path.read_bytes()) + + monkeypatch.setattr( + plugin_manager_pm._updater, "_download_repository", download_repository + ) + monkeypatch.setattr(plugin_manager_pm._updater, "_download_file", download_file) + monkeypatch.setattr( + plugin_manager_pm._updater, "_clone_repository", clone_repository + ) + monkeypatch.setattr(star_manager_module.Metric, "upload", AsyncMock()) + events = [] + + async def ensure_requirements(plugin_dir_path, plugin_label): + assert plugin_label == dir_name + assert (local_updater / "obsolete.py").exists() + assert Path(plugin_dir_path) != local_updater + assert Path(plugin_dir_path).is_relative_to(system_temp) + events.append("dependencies") + if failure == "dependencies": + raise RuntimeError("dependency failure") + + async def terminate(plugin): + assert plugin is old_plugin + assert (local_updater / "obsolete.py").exists() + events.append("terminate") + + async def load(specified_dir_name=None, ignore_version_check=False): + assert specified_dir_name == dir_name + assert module_path not in sys.modules + assert module_path not in star_manager_module.star_map + assert old_plugin not in star_manager_module.star_registry + current = plugin_manager_pm._load_plugin_metadata(str(local_updater)) + assert current is not None + events.append(current.version) + if current.version == "2.0.0": + assert list(system_temp.glob(".plugin-backup-*")) + assert not (local_updater / "obsolete.py").exists() + if failure == "load": + sys.modules[module_path] = ModuleType(module_path) + return False, "new plugin failed to load" + if failure == "cancel": + raise asyncio.CancelledError + else: + assert ignore_version_check is True + assert (local_updater / "obsolete.py").read_text() == "old code" + current.root_dir_name = dir_name + current.module_path = module_path + star_manager_module.star_registry.append(current) + star_manager_module.star_map[module_path] = current + return True, None + + monkeypatch.setattr( + plugin_manager_pm, "_ensure_plugin_requirements", ensure_requirements + ) + monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", terminate) + monkeypatch.setattr(plugin_manager_pm, "load", load) + try: + if install_source == "upload": + operation = plugin_manager_pm.install_plugin_from_file(str(zip_path)) + else: + operation = plugin_manager_pm.install_plugin( + repo_url, + proxy=proxy_url, + download_url=download_url if install_source == "url" else "", + ) + if failure: + exception_type = ( + asyncio.CancelledError if failure == "cancel" else Exception + ) + with pytest.raises(exception_type): + await operation + assert ( + plugin_manager_pm._load_plugin_metadata(str(local_updater)).version + == "1.0.0" + ) + if failure in {"load", "cancel"}: + assert events == ["dependencies", "terminate", "2.0.0", "1.0.0"] + else: + assert "terminate" not in events + assert star_manager_module.star_map[module_path] is old_plugin + else: + result = await operation + assert result == { + "name": TEST_PLUGIN_NAME, + "repo": None, + "readme": "Updated README", + } + assert events == ["dependencies", "terminate", "2.0.0"] + assert zip_path.exists() is (install_source != "upload") + assert config_path.read_text() == '{"custom": true}' + assert data_path.read_bytes() == b"saved digest" + assert set(Path(plugin_manager_pm.plugin_store_path).iterdir()) == { + local_updater + } + assert plugin_manager_pm.failed_plugin_dict == {} + assert list(system_temp.iterdir()) == [] + finally: + sys.modules.pop(module_path, None) + _clear_star_runtime_state() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure_stage", ["backup", "install", "restore"]) +async def test_install_copy_failure_preserves_complete_old_code( + plugin_manager_pm: PluginManager, + local_updater: Path, + monkeypatch, + tmp_path: Path, + failure_stage: str, +): + """A partial copy must never replace a complete recovery copy.""" + old_files = {path.name: path.read_bytes() for path in local_updater.iterdir()} + metadata = yaml.safe_load(old_files["metadata.yaml"]) + metadata["version"] = "2.0.0" + zip_path = tmp_path / "update.zip" + with zipfile.ZipFile(zip_path, "w") as archive: + archive.writestr("metadata.yaml", yaml.safe_dump(metadata)) + archive.writestr("main.py", "pass\n") + + original_copytree = star_manager_module.shutil.copytree + + def copytree(source, destination, *args, **kwargs): + source, destination = Path(source), Path(destination) + is_backup = destination.parent.name.startswith(".plugin-backup-") + is_restore = source.parent.name.startswith(".plugin-backup-") + if (failure_stage == "backup" and is_backup) or ( + failure_stage == "restore" and is_restore + ): + destination.mkdir() + (destination / "partial.py").write_text("partial", encoding="utf-8") + raise OSError("copy failed") + return original_copytree(source, destination, *args, **kwargs) + + original_move = star_manager_module.shutil.move + + def move(source, destination): + if failure_stage == "install": + destination = Path(destination) + destination.mkdir() + (destination / "partial.py").write_text("partial", encoding="utf-8") + raise OSError("copy failed") + return original_move(source, destination) + + versions_loaded = [] + + async def load(specified_dir_name=None, ignore_version_check=False): + version = plugin_manager_pm._load_plugin_metadata(str(local_updater)).version + versions_loaded.append(version) + return version == "1.0.0", "new plugin failed to load" + + monkeypatch.setattr(star_manager_module.shutil, "copytree", copytree) + monkeypatch.setattr(star_manager_module.shutil, "move", move) + monkeypatch.setattr(plugin_manager_pm, "load", load) + with pytest.raises(Exception, match="copy failed|new plugin failed to load"): + await plugin_manager_pm.install_plugin_from_file(str(zip_path)) + + system_temp = Path(star_manager_module.get_astrbot_system_tmp_path()) + if failure_stage == "restore": + backup_dirs = list(system_temp.iterdir()) + assert len(backup_dirs) == 1 + assert backup_dirs[0].name.startswith(".plugin-backup-") + backup = backup_dirs[0] / TEST_PLUGIN_DIR + assert {path.name: path.read_bytes() for path in backup.iterdir()} == old_files + assert versions_loaded == ["2.0.0"] + else: + assert {path.name: path.read_bytes() for path in local_updater.iterdir()} == old_files + assert versions_loaded == ["1.0.0"] + assert list(system_temp.iterdir()) == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("install_source", ["github", "url", "git"]) +@pytest.mark.parametrize("failure", ["download", "cancel", "metadata"]) +async def test_url_install_preparation_failure_preserves_existing_plugin( + plugin_manager_pm: PluginManager, + local_updater: Path, + monkeypatch, + install_source: str, + failure: str, +): + """Failed downloads and invalid packages leave the existing plugin untouched.""" + original_files = {path.name: path.read_bytes() for path in local_updater.iterdir()} + load = AsyncMock() + terminate = AsyncMock() + monkeypatch.setattr(plugin_manager_pm, "load", load) + monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", terminate) + monkeypatch.setattr(star_manager_module.Metric, "upload", AsyncMock()) + + async def prepare(*args): + if install_source == "git": + target = Path(args[1]) + target.mkdir() + (target / "main.py").write_text("pass\n", encoding="utf-8") + else: + target = Path(args[0] + ".zip" if install_source == "github" else args[1]) + with zipfile.ZipFile(target, "w") as archive: + archive.writestr("main.py", "pass\n") + if failure == "download": + raise RuntimeError("download interrupted") + if failure == "cancel": + raise asyncio.CancelledError + + monkeypatch.setattr(plugin_manager_pm._updater, "_download_repository", prepare) + monkeypatch.setattr(plugin_manager_pm._updater, "_download_file", prepare) + monkeypatch.setattr(plugin_manager_pm._updater, "_clone_repository", prepare) + repo_url = ( + f"https://gitee.com/AstrBotDevs/{TEST_PLUGIN_DIR}.git" + if install_source == "git" + else f"https://github.com/AstrBotDevs/{TEST_PLUGIN_DIR}" + ) + with pytest.raises(asyncio.CancelledError if failure == "cancel" else Exception): + await plugin_manager_pm.install_plugin( + repo_url, + download_url="https://cdn.example/update.zip" + if install_source == "url" + else "", + ) + assert { + path.name: path.read_bytes() for path in local_updater.iterdir() + } == original_files + assert set(Path(plugin_manager_pm.plugin_store_path).iterdir()) == {local_updater} + assert plugin_manager_pm.failed_plugin_dict == {} + load.assert_not_awaited() + terminate.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("install_source", ["upload", "url"]) +@pytest.mark.parametrize("disabled", [False, True]) +@pytest.mark.parametrize("initialization_fails", [False, True]) +async def test_upload_update_reloads_runtime_and_preserves_activation( + plugin_manager_pm: PluginManager, + local_updater: Path, + monkeypatch, + tmp_path: Path, + disabled: bool, + initialization_fails: bool, + install_source: str, +): + """Exercise real loading, registration, disabled state, and rollback.""" + _clear_star_runtime_state() + module_path = f"data.plugins.{TEST_PLUGIN_DIR}.main" + metadata_path = local_updater / "metadata.yaml" + metadata = yaml.safe_load(metadata_path.read_text(encoding="utf-8")) + metadata.pop("repo") + metadata_path.write_text(yaml.safe_dump(metadata), encoding="utf-8") + source = ( + "from astrbot.api.star import Star\n" + "from astrbot.api.event import filter\n" + "class Main(Star):\n" + " marker = 'old'\n" + " async def initialize(self):\n" + " pass\n" + " @filter.command('upload_update_test')\n" + " async def command(self, event):\n" + " pass\n" + ) + (local_updater / "main.py").write_text(source, encoding="utf-8") + monkeypatch.setattr( + plugin_manager_pm.context, "stars", star_manager_module.star_registry + ) + monkeypatch.setattr( + plugin_manager_pm, "reserved_plugin_path", str(tmp_path / "reserved") + ) + monkeypatch.setattr( + plugin_manager_pm, "plugin_config_path", str(tmp_path / "config") + ) + + async def global_get(key, default=None): + if key == "inactivated_plugins": + return [module_path] if disabled else [] + return default + + async def import_plugin(*, path, **kwargs): + if path not in sys.modules: + module = ModuleType(path) + sys.modules[path] = module + main_path = local_updater / "main.py" + exec( + compile(main_path.read_text(encoding="utf-8"), str(main_path), "exec"), + module.__dict__, + ) + return sys.modules[path] + + async def sync_command_configs(): + return None + + monkeypatch.setattr(star_manager_module.sp, "global_get", global_get) + monkeypatch.setattr( + star_manager_module, "sync_command_configs", sync_command_configs + ) + monkeypatch.setattr( + plugin_manager_pm, "_import_plugin_with_dependency_recovery", import_plugin + ) + try: + success, error = await plugin_manager_pm.load( + specified_dir_name=TEST_PLUGIN_DIR + ) + assert success, error + old_plugin = star_manager_module.star_map[module_path] + assert old_plugin.star_cls_type.marker == "old" + + metadata["version"] = "2.0.0" + metadata["astrbot_version"] = ">=999.0" + updated_source = source.replace("marker = 'old'", "marker = 'new'") + if initialization_fails: + updated_source = updated_source.replace( + " pass", " raise RuntimeError('init failed')", 1 + ) + zip_path = tmp_path / "update.zip" + with zipfile.ZipFile(zip_path, "w") as archive: + archive.writestr("metadata.yaml", yaml.safe_dump(metadata)) + archive.writestr("main.py", updated_source) + + async def download_file(url, path): + Path(path).write_bytes(zip_path.read_bytes()) + + monkeypatch.setattr(plugin_manager_pm._updater, "_download_file", download_file) + monkeypatch.setattr(star_manager_module.Metric, "upload", AsyncMock()) + if install_source == "upload": + operation = plugin_manager_pm.install_plugin_from_file( + str(zip_path), ignore_version_check=True + ) + else: + operation = plugin_manager_pm.install_plugin( + TEST_PLUGIN_REPO, + download_url="https://cdn.example/update.zip", + ignore_version_check=True, + ) + if initialization_fails and not disabled: + with pytest.raises(Exception, match="init failed"): + await operation + else: + await operation + current_plugin = star_manager_module.star_map[module_path] + assert current_plugin is not old_plugin + assert current_plugin.activated is (not disabled) + assert (current_plugin.star_cls is None) is disabled + assert current_plugin.star_cls_type.marker == ( + "old" if initialization_fails and not disabled else "new" + ) + assert len(star_manager_module.star_registry) == 1 + assert ( + len( + star_manager_module.star_handlers_registry.get_handlers_by_module_name( + module_path + ) + ) + == 1 + ) + assert plugin_manager_pm.failed_plugin_dict == {} + assert plugin_manager_pm.failed_plugin_info == "" + finally: + sys.modules.pop(module_path, None) + _clear_star_runtime_state() @pytest.mark.asyncio @@ -1561,10 +2024,11 @@ async def test_install_plugin_skips_dependency_install_when_no_requirements_miss events = [] _mock_missing_requirements(monkeypatch, set()) - async def mock_install(repo_url: str, proxy=""): - _write_local_test_plugin(plugin_path, repo_url) - _write_requirements(plugin_path) - return str(plugin_path) + async def mock_install(repo_url: str, proxy="", *, download_url="", target_dir): + staged_path = Path(target_dir) + _write_local_test_plugin(staged_path, repo_url) + _write_requirements(staged_path) + return str(staged_path) monkeypatch.setattr(plugin_manager_pm._updater, "install", mock_install) monkeypatch.setattr( @@ -1591,10 +2055,11 @@ async def test_install_plugin_runs_dependency_install_when_precheck_fails( plugin_path = Path(plugin_manager_pm.plugin_store_path) / TEST_PLUGIN_DIR events = [] - async def mock_install(repo_url: str, proxy=""): - _write_local_test_plugin(plugin_path, repo_url) - _write_requirements(plugin_path) - return str(plugin_path) + async def mock_install(repo_url: str, proxy="", *, download_url="", target_dir): + staged_path = Path(target_dir) + _write_local_test_plugin(staged_path, repo_url) + _write_requirements(staged_path) + return str(staged_path) _mock_precheck_fails(monkeypatch) monkeypatch.setattr(plugin_manager_pm._updater, "install", mock_install)