mirror of
https://github.com/AstrBotDevs/AstrBot.git
synced 2026-10-16 15:06:04 +08:00
Chen Zixing
97503bcdb5
* fix(agent): cancel pending tool reads with the runner * fix(agent): track background tasks by session * fix(agent): stop background tools and late wakeups * test(agent): align background wakeup event stub * test(plugin): align view metadata and bridge expectations * docs(api): regenerate public plugin view endpoints
2931 lines
102 KiB
Python
2931 lines
102 KiB
Python
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
|
|
|
|
from astrbot.core.star import star_manager as star_manager_module
|
|
from astrbot.core.star.star_handler import EventType, StarHandlerMetadata
|
|
from astrbot.core.star.star_manager import PluginDependencyInstallError, PluginManager
|
|
from astrbot.core.utils.pip_installer import PipInstallError
|
|
from astrbot.core.utils.requirements_utils import MissingRequirementsPlan
|
|
|
|
# --- Test Data & Helpers ---
|
|
|
|
TEST_PLUGIN_NAME = "helloworld"
|
|
TEST_PLUGIN_REPO = "https://github.com/AstrBotDevs/astrbot_plugin_helloworld"
|
|
TEST_PLUGIN_DIR = "helloworld"
|
|
|
|
|
|
def test_load_plugin_config_schema_accepts_utf8_bom(tmp_path: Path):
|
|
schema_path = tmp_path / "_conf_schema.json"
|
|
schema_path.write_bytes(b'\xef\xbb\xbf{"type": "object"}')
|
|
|
|
assert PluginManager._load_plugin_config_schema(str(schema_path)) == {
|
|
"type": "object"
|
|
}
|
|
|
|
|
|
def test_load_plugin_config_schema_accepts_utf8_without_bom(tmp_path: Path):
|
|
schema_path = tmp_path / "_conf_schema.json"
|
|
schema_path.write_text('{"type": "object"}', encoding="utf-8")
|
|
|
|
assert PluginManager._load_plugin_config_schema(str(schema_path)) == {
|
|
"type": "object"
|
|
}
|
|
|
|
|
|
def test_load_plugin_config_schema_reports_invalid_json(tmp_path: Path):
|
|
schema_path = tmp_path / "_conf_schema.json"
|
|
schema_path.write_text("{invalid", encoding="utf-8")
|
|
|
|
with pytest.raises(ValueError, match="不是有效的 JSON"):
|
|
PluginManager._load_plugin_config_schema(str(schema_path))
|
|
|
|
|
|
class MockStar:
|
|
def __init__(self):
|
|
self.root_dir_name = TEST_PLUGIN_DIR
|
|
self.name = TEST_PLUGIN_NAME
|
|
self.repo = TEST_PLUGIN_REPO
|
|
self.reserved = False
|
|
self.info = {"repo": TEST_PLUGIN_REPO, "readme": ""}
|
|
|
|
|
|
def _write_local_test_plugin(plugin_path: Path, repo_url: str, version: str = "1.0.0"):
|
|
"""Creates a minimal valid plugin structure."""
|
|
plugin_path.mkdir(parents=True, exist_ok=True)
|
|
metadata = {
|
|
"name": TEST_PLUGIN_NAME,
|
|
"repo": repo_url,
|
|
"version": version,
|
|
"author": "AstrBot Team",
|
|
"desc": "Local test plugin",
|
|
"short_desc": "Local test short description",
|
|
}
|
|
with open(plugin_path / "metadata.yaml", "w", encoding="utf-8") as f:
|
|
yaml.dump(metadata, f)
|
|
with open(plugin_path / "main.py", "w", encoding="utf-8") as f:
|
|
f.write("from astrbot.api.star import Star, Context, StarManager\n")
|
|
f.write("@StarManager.register\n")
|
|
f.write("class HelloWorld(Star):\n")
|
|
f.write(" def __init__(self, context: Context): ...\n")
|
|
|
|
|
|
def _write_requirements(plugin_path: Path):
|
|
"""Creates a requirements.txt file."""
|
|
with open(plugin_path / "requirements.txt", "w", encoding="utf-8") as f:
|
|
f.write("networkx\n")
|
|
|
|
|
|
def test_load_plugin_i18n_reads_locale_files(tmp_path: Path):
|
|
plugin_path = tmp_path / "plugin"
|
|
i18n_path = plugin_path / ".astrbot-plugin" / "i18n"
|
|
i18n_path.mkdir(parents=True)
|
|
(i18n_path / "zh-CN.json").write_bytes(
|
|
b"\xef\xbb\xbf"
|
|
+ json.dumps({"metadata": {"desc": "中文描述"}}, ensure_ascii=False).encode(
|
|
"utf-8"
|
|
),
|
|
)
|
|
(i18n_path / "en-US.json").write_text(
|
|
json.dumps({"metadata": {"desc": "English description"}}),
|
|
encoding="utf-8",
|
|
)
|
|
(i18n_path / "README.md").write_text("ignored", encoding="utf-8")
|
|
|
|
assert PluginManager._load_plugin_i18n(str(plugin_path)) == {
|
|
"zh-CN": {"metadata": {"desc": "中文描述"}},
|
|
"en-US": {"metadata": {"desc": "English description"}},
|
|
}
|
|
|
|
|
|
def test_load_plugin_i18n_ignores_legacy_directories(tmp_path: Path):
|
|
plugin_path = tmp_path / "plugin"
|
|
hidden_legacy_i18n_path = plugin_path / ".i18n"
|
|
legacy_i18n_path = plugin_path / "i18n"
|
|
hidden_legacy_i18n_path.mkdir(parents=True)
|
|
legacy_i18n_path.mkdir()
|
|
(hidden_legacy_i18n_path / "zh-CN.json").write_text(
|
|
json.dumps({"metadata": {"desc": "隐藏旧目录"}}, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
(legacy_i18n_path / "zh-CN.json").write_text(
|
|
json.dumps({"metadata": {"desc": "中文描述"}}, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assert PluginManager._load_plugin_i18n(str(plugin_path)) == {}
|
|
|
|
|
|
def test_load_plugin_metadata_includes_i18n(tmp_path: Path):
|
|
plugin_path = tmp_path / "helloworld"
|
|
_write_local_test_plugin(plugin_path, TEST_PLUGIN_REPO)
|
|
i18n_path = plugin_path / ".astrbot-plugin" / "i18n"
|
|
i18n_path.mkdir(parents=True)
|
|
(i18n_path / "zh-CN.json").write_text(
|
|
json.dumps({"metadata": {"display_name": "你好世界"}}, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
metadata = PluginManager._load_plugin_metadata(str(plugin_path))
|
|
|
|
assert metadata is not None
|
|
assert metadata.short_desc == "Local test short description"
|
|
assert metadata.views == []
|
|
assert metadata.i18n == {"zh-CN": {"metadata": {"display_name": "你好世界"}}}
|
|
|
|
|
|
def test_load_plugin_metadata_includes_views(tmp_path: Path):
|
|
plugin_path = tmp_path / "helloworld"
|
|
_write_local_test_plugin(plugin_path, TEST_PLUGIN_REPO)
|
|
metadata_path = plugin_path / "metadata.yaml"
|
|
metadata = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
|
|
metadata["views"] = [{"name": "dashboard", "title": "Dashboard"}]
|
|
metadata_path.write_text(yaml.dump(metadata), encoding="utf-8")
|
|
|
|
loaded_metadata = PluginManager._load_plugin_metadata(str(plugin_path))
|
|
|
|
assert loaded_metadata is not None
|
|
assert loaded_metadata.views == [{"name": "dashboard", "title": "Dashboard"}]
|
|
|
|
|
|
def test_load_plugin_metadata_pages_alias_maps_to_views(tmp_path: Path):
|
|
# "pages" stays a compatible alias for "views" in metadata.yaml.
|
|
plugin_path = tmp_path / "helloworld"
|
|
_write_local_test_plugin(plugin_path, TEST_PLUGIN_REPO)
|
|
metadata_path = plugin_path / "metadata.yaml"
|
|
metadata = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
|
|
metadata["pages"] = [{"name": "dashboard", "title": "Dashboard"}]
|
|
metadata_path.write_text(yaml.dump(metadata), encoding="utf-8")
|
|
|
|
loaded_metadata = PluginManager._load_plugin_metadata(str(plugin_path))
|
|
|
|
assert loaded_metadata is not None
|
|
assert loaded_metadata.views == [{"name": "dashboard", "title": "Dashboard"}]
|
|
|
|
|
|
def test_load_plugin_metadata_accepts_yml_suffix(tmp_path: Path):
|
|
plugin_path = tmp_path / "helloworld"
|
|
_write_local_test_plugin(plugin_path, TEST_PLUGIN_REPO)
|
|
metadata_path = plugin_path / "metadata.yaml"
|
|
yml_metadata_path = plugin_path / "metadata.yml"
|
|
yml_metadata_path.write_text(metadata_path.read_text(encoding="utf-8"))
|
|
metadata_path.unlink()
|
|
|
|
loaded_metadata = PluginManager._load_plugin_metadata(str(plugin_path))
|
|
|
|
assert loaded_metadata is not None
|
|
assert loaded_metadata.name == TEST_PLUGIN_NAME
|
|
assert PluginManager._get_plugin_dir_name_from_metadata(str(plugin_path)) == (
|
|
TEST_PLUGIN_NAME
|
|
)
|
|
|
|
|
|
def test_load_plugin_metadata_does_not_fallback_to_legacy_info(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
plugin_path = tmp_path / "helloworld"
|
|
plugin_path.mkdir()
|
|
info_called = False
|
|
|
|
class LegacyPlugin:
|
|
def info(self):
|
|
nonlocal info_called
|
|
info_called = True
|
|
return {
|
|
"name": TEST_PLUGIN_NAME,
|
|
"repo": TEST_PLUGIN_REPO,
|
|
"version": "1.0.0",
|
|
"author": "AstrBot Team",
|
|
"desc": "Legacy plugin",
|
|
}
|
|
|
|
loaded_metadata = PluginManager._load_plugin_metadata(
|
|
str(plugin_path),
|
|
plugin_obj=LegacyPlugin(),
|
|
)
|
|
|
|
assert loaded_metadata is None
|
|
assert info_called is False
|
|
|
|
|
|
def test_load_plugin_metadata_preserves_validation_error(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
plugin_path = tmp_path / "helloworld"
|
|
_write_local_test_plugin(plugin_path, TEST_PLUGIN_REPO)
|
|
metadata_path = plugin_path / "metadata.yaml"
|
|
metadata = yaml.safe_load(metadata_path.read_text(encoding="utf-8"))
|
|
metadata["version"] = ""
|
|
metadata_path.write_text(yaml.dump(metadata), encoding="utf-8")
|
|
|
|
with pytest.raises(Exception, match="version.*非空字符串"):
|
|
PluginManager._load_plugin_metadata(str(plugin_path))
|
|
|
|
|
|
def test_loaded_metadata_can_copy_i18n_into_existing_star_metadata(tmp_path: Path):
|
|
plugin_path = tmp_path / "helloworld"
|
|
_write_local_test_plugin(plugin_path, TEST_PLUGIN_REPO)
|
|
i18n_path = plugin_path / ".astrbot-plugin" / "i18n"
|
|
i18n_path.mkdir(parents=True)
|
|
(i18n_path / "zh-CN.json").write_text(
|
|
json.dumps({"metadata": {"desc": "中文描述"}}, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
existing_metadata = star_manager_module.StarMetadata(name="old")
|
|
loaded_metadata = PluginManager._load_plugin_metadata(str(plugin_path))
|
|
|
|
assert loaded_metadata is not None
|
|
existing_metadata.i18n = loaded_metadata.i18n
|
|
assert existing_metadata.i18n == {"zh-CN": {"metadata": {"desc": "中文描述"}}}
|
|
|
|
|
|
def _clear_module_cache():
|
|
"""Clear test-specific modules from sys.modules to allow reloading."""
|
|
import sys
|
|
|
|
to_del = [
|
|
m
|
|
for m in sys.modules
|
|
if m.startswith("data.plugins.helloworld")
|
|
or m.startswith("data.plugins.broken_plugin")
|
|
]
|
|
for m in to_del:
|
|
del sys.modules[m]
|
|
|
|
|
|
def _clear_star_runtime_state():
|
|
star_manager_module.star_map.clear()
|
|
star_manager_module.star_registry.clear()
|
|
star_manager_module.star_handlers_registry.clear()
|
|
|
|
|
|
def _build_load_mock(events):
|
|
async def mock_load(specified_dir_name=None, ignore_version_check=False):
|
|
del ignore_version_check
|
|
events.append(("load", specified_dir_name or TEST_PLUGIN_DIR))
|
|
return True, ""
|
|
|
|
return mock_load
|
|
|
|
|
|
def _build_reload_mock(events):
|
|
async def mock_reload(specified_dir_name=None):
|
|
events.append(("reload", specified_dir_name or TEST_PLUGIN_DIR))
|
|
return True, ""
|
|
|
|
return mock_reload
|
|
|
|
|
|
def _build_dependency_install_mock(
|
|
events,
|
|
fail: bool,
|
|
*,
|
|
capture_content: bool = False,
|
|
):
|
|
async def mock_install_requirements(
|
|
*,
|
|
requirements_path: str | None = None,
|
|
package_name: str | None = None,
|
|
**kwargs,
|
|
):
|
|
del kwargs
|
|
if requirements_path:
|
|
path = Path(requirements_path)
|
|
event = ("deps", str(path))
|
|
if capture_content:
|
|
event = (*event, path.read_text(encoding="utf-8"))
|
|
events.append(event)
|
|
if package_name:
|
|
events.append(("deps_pkg", package_name))
|
|
if fail:
|
|
raise Exception("pip failed")
|
|
|
|
return mock_install_requirements
|
|
|
|
|
|
def _mock_missing_requirements(monkeypatch, missing: set[str]):
|
|
_mock_missing_requirements_plan(monkeypatch, missing, sorted(missing))
|
|
|
|
|
|
def _mock_missing_requirements_plan(
|
|
monkeypatch,
|
|
missing_names,
|
|
install_lines,
|
|
*,
|
|
version_mismatch_names=(),
|
|
fallback_reason: str | None = None,
|
|
):
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.plan_missing_requirements_install",
|
|
lambda requirements_path: MissingRequirementsPlan(
|
|
missing_names=frozenset(missing_names),
|
|
version_mismatch_names=frozenset(version_mismatch_names),
|
|
install_lines=tuple(install_lines),
|
|
fallback_reason=fallback_reason,
|
|
),
|
|
)
|
|
|
|
|
|
def _mock_precheck_fails(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.plan_missing_requirements_install",
|
|
lambda requirements_path: None,
|
|
)
|
|
|
|
|
|
def _assert_dependency_install_event_matches(
|
|
event,
|
|
*,
|
|
expected_original_path: Path,
|
|
expected_content: str | None = None,
|
|
expect_filtered_tempfile: bool | None = None,
|
|
):
|
|
assert event[0] == "deps"
|
|
used_path = Path(event[1])
|
|
should_be_filtered = expected_content is not None
|
|
if expect_filtered_tempfile is not None:
|
|
should_be_filtered = expect_filtered_tempfile
|
|
|
|
if not should_be_filtered:
|
|
assert used_path == expected_original_path
|
|
else:
|
|
assert used_path != expected_original_path
|
|
assert used_path.name.endswith("_plugin_requirements.txt")
|
|
if expected_content is not None:
|
|
if len(event) >= 3:
|
|
assert event[2] == expected_content
|
|
|
|
|
|
# --- Fixtures ---
|
|
|
|
|
|
@pytest.fixture
|
|
def plugin_manager_pm(tmp_path, monkeypatch):
|
|
"""Provides a fully isolated PluginManager instance for testing."""
|
|
# Clear module cache before setup to ensure isolation
|
|
_clear_module_cache()
|
|
|
|
plugin_dir = tmp_path / "astrbot_root" / "data" / "plugins"
|
|
plugin_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
class MockContext:
|
|
def __init__(self):
|
|
self.stars = []
|
|
|
|
def get_all_stars(self):
|
|
return self.stars
|
|
|
|
def get_registered_star(self, name):
|
|
for s in self.stars:
|
|
if s.root_dir_name == name or s.name == name:
|
|
return s
|
|
return None
|
|
|
|
mock_context = MockContext()
|
|
mock_config = {}
|
|
pm = PluginManager(cast(Any, mock_context), cast(Any, mock_config))
|
|
|
|
# Patch paths to use tmp_path
|
|
monkeypatch.setattr(pm, "plugin_store_path", str(plugin_dir))
|
|
monkeypatch.setattr(
|
|
"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
|
|
|
|
|
|
@pytest.fixture
|
|
def local_updater(plugin_manager_pm):
|
|
"""Helper to setup a local plugin directory simulating a download."""
|
|
path = Path(plugin_manager_pm.plugin_store_path) / TEST_PLUGIN_DIR
|
|
_write_local_test_plugin(path, TEST_PLUGIN_REPO)
|
|
return path
|
|
|
|
|
|
# --- Tests ---
|
|
|
|
|
|
@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,
|
|
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")
|
|
|
|
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
|
|
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(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
_build_dependency_install_mock(events, dependency_install_fails),
|
|
)
|
|
|
|
def mock_load_and_register(*args, **kwargs):
|
|
cast(Any, plugin_manager_pm.context).stars.append(MockStar())
|
|
return _build_load_mock(events)(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(plugin_manager_pm, "load", mock_load_and_register)
|
|
|
|
if dependency_install_fails:
|
|
with pytest.raises(PluginDependencyInstallError, match="pip failed"):
|
|
await plugin_manager_pm.install_plugin(TEST_PLUGIN_REPO)
|
|
assert len(events) == 1
|
|
_assert_dependency_install_event_matches(
|
|
events[0],
|
|
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
|
|
_assert_dependency_install_event_matches(
|
|
events[0],
|
|
expected_original_path=plugin_path / "requirements.txt",
|
|
expected_content="networkx\n",
|
|
)
|
|
assert events[1] == ("load", TEST_PLUGIN_DIR)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("dependency_install_fails", [False, True])
|
|
async def test_install_plugin_from_file_dependency_install_flow(
|
|
plugin_manager_pm: PluginManager,
|
|
monkeypatch,
|
|
tmp_path,
|
|
dependency_install_fails: bool,
|
|
):
|
|
zip_file_path = tmp_path / f"{TEST_PLUGIN_DIR}.zip"
|
|
zip_file_path.write_text("placeholder", encoding="utf-8")
|
|
events = []
|
|
_mock_missing_requirements(monkeypatch, {"networkx"})
|
|
|
|
def mock_unzip_file(zip_path: str, target_dir: str) -> None:
|
|
assert zip_path == str(zip_file_path)
|
|
plugin_path = Path(target_dir)
|
|
_write_local_test_plugin(plugin_path, TEST_PLUGIN_REPO)
|
|
_write_requirements(plugin_path)
|
|
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm._updater,
|
|
"_extract_plugin_archive",
|
|
mock_unzip_file,
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
_build_dependency_install_mock(events, dependency_install_fails),
|
|
)
|
|
|
|
def mock_load_and_register(*args, **kwargs):
|
|
cast(Any, plugin_manager_pm.context).stars.append(MockStar())
|
|
return _build_load_mock(events)(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(plugin_manager_pm, "load", mock_load_and_register)
|
|
|
|
if dependency_install_fails:
|
|
with pytest.raises(PluginDependencyInstallError, match="pip failed"):
|
|
await plugin_manager_pm.install_plugin_from_file(str(zip_file_path))
|
|
assert any(e[0] == "deps" for e in events)
|
|
else:
|
|
await plugin_manager_pm.install_plugin_from_file(str(zip_file_path))
|
|
assert any(e[0] == "deps" for e in events)
|
|
assert ("load", TEST_PLUGIN_DIR) in events
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_plugin_from_file_conflict_keeps_failed_plugins_clean(
|
|
plugin_manager_pm: PluginManager,
|
|
local_updater: Path,
|
|
monkeypatch,
|
|
tmp_path: Path,
|
|
):
|
|
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_dirs = set(plugin_store_path.iterdir())
|
|
|
|
def mock_unzip_file(zip_path: str, target_dir: str) -> None:
|
|
assert zip_path == str(zip_file_path)
|
|
_write_local_test_plugin(
|
|
Path(target_dir),
|
|
TEST_PLUGIN_REPO,
|
|
version="2.0.0",
|
|
)
|
|
|
|
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",
|
|
mock_unzip_file,
|
|
)
|
|
|
|
with pytest.raises(Exception, match=f"安装失败：目录 {TEST_PLUGIN_DIR} 已存在。"):
|
|
await plugin_manager_pm.install_plugin_from_file(str(zip_file_path))
|
|
|
|
assert plugin_manager_pm.failed_plugin_dict == {}
|
|
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
|
|
@pytest.mark.parametrize("dependency_install_fails", [False, True])
|
|
async def test_reload_failed_plugin_dependency_install_flow(
|
|
plugin_manager_pm: PluginManager,
|
|
local_updater: Path,
|
|
monkeypatch,
|
|
dependency_install_fails: bool,
|
|
):
|
|
_write_requirements(local_updater)
|
|
plugin_manager_pm.failed_plugin_dict[TEST_PLUGIN_DIR] = {"error": "init fail"}
|
|
events = []
|
|
_mock_missing_requirements(monkeypatch, {"networkx"})
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
_build_dependency_install_mock(events, dependency_install_fails),
|
|
)
|
|
|
|
def mock_load_and_register(*args, **kwargs):
|
|
cast(Any, plugin_manager_pm.context).stars.append(MockStar())
|
|
return _build_load_mock(events)(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(plugin_manager_pm, "load", mock_load_and_register)
|
|
|
|
if dependency_install_fails:
|
|
with pytest.raises(PluginDependencyInstallError, match="pip failed"):
|
|
await plugin_manager_pm.reload_failed_plugin(TEST_PLUGIN_DIR)
|
|
assert len(events) == 1
|
|
_assert_dependency_install_event_matches(
|
|
events[0],
|
|
expected_original_path=local_updater / "requirements.txt",
|
|
expected_content="networkx\n",
|
|
)
|
|
else:
|
|
await plugin_manager_pm.reload_failed_plugin(TEST_PLUGIN_DIR)
|
|
assert len(events) == 2
|
|
_assert_dependency_install_event_matches(
|
|
events[0],
|
|
expected_original_path=local_updater / "requirements.txt",
|
|
expected_content="networkx\n",
|
|
)
|
|
assert events[1] == ("load", TEST_PLUGIN_DIR)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reload_all_unbinds_every_registered_plugin(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
_clear_star_runtime_state()
|
|
plugin_names = ["plugin_one", "plugin_two", "plugin_three"]
|
|
for plugin_name in plugin_names:
|
|
module_path = f"data.plugins.{plugin_name}.main"
|
|
metadata = star_manager_module.StarMetadata(
|
|
name=plugin_name,
|
|
root_dir_name=plugin_name,
|
|
module_path=module_path,
|
|
)
|
|
star_manager_module.star_map[module_path] = metadata
|
|
star_manager_module.star_registry.append(metadata)
|
|
|
|
terminated = []
|
|
unbound = []
|
|
|
|
async def mock_terminate(plugin):
|
|
terminated.append(plugin.name)
|
|
|
|
async def mock_unbind(plugin_name, plugin_module_path):
|
|
unbound.append(plugin_name)
|
|
star_manager_module.star_map.pop(plugin_module_path, None)
|
|
for index, metadata in enumerate(star_manager_module.star_registry):
|
|
if metadata.name == plugin_name:
|
|
del star_manager_module.star_registry[index]
|
|
break
|
|
|
|
async def mock_load(
|
|
specified_module_path=None,
|
|
specified_dir_name=None,
|
|
ignore_version_check=False,
|
|
):
|
|
del specified_module_path, specified_dir_name, ignore_version_check
|
|
return True, None
|
|
|
|
monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", mock_terminate)
|
|
monkeypatch.setattr(plugin_manager_pm, "_unbind_plugin", mock_unbind)
|
|
monkeypatch.setattr(plugin_manager_pm, "load", mock_load)
|
|
|
|
try:
|
|
await plugin_manager_pm.reload()
|
|
finally:
|
|
_clear_star_runtime_state()
|
|
|
|
assert terminated == plugin_names
|
|
assert unbound == plugin_names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_turn_plugin_toggles_llm_tools_from_plugin_child_module(
|
|
plugin_manager_pm: PluginManager,
|
|
monkeypatch,
|
|
):
|
|
plugin = star_manager_module.StarMetadata(
|
|
name="demo_plugin",
|
|
root_dir_name="demo_plugin",
|
|
module_path="data.plugins.demo_plugin.main",
|
|
)
|
|
cast(Any, plugin_manager_pm.context).stars.append(plugin)
|
|
plugin_tool = star_manager_module.FunctionTool(
|
|
name="plugin_search",
|
|
description="plugin search",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path="data.plugins.demo_plugin.main.tools.search",
|
|
)
|
|
other_tool = star_manager_module.FunctionTool(
|
|
name="other_search",
|
|
description="other search",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path="data.plugins.other_plugin.main.tools.search",
|
|
)
|
|
llm_tools = cast(Any, star_manager_module.llm_tools)
|
|
original_func_list = llm_tools.func_list
|
|
llm_tools.func_list = [plugin_tool, other_tool]
|
|
preferences = {
|
|
"inactivated_plugins": [],
|
|
"inactivated_llm_tools": [],
|
|
}
|
|
|
|
async def mock_global_get(key, default=None):
|
|
return preferences.get(key, default)
|
|
|
|
async def mock_global_put(key, value):
|
|
preferences[key] = value
|
|
|
|
async def mock_terminate(star_metadata):
|
|
assert star_metadata is plugin
|
|
|
|
async def mock_reload(plugin_name):
|
|
assert plugin_name == plugin.root_dir_name
|
|
return True, None
|
|
|
|
monkeypatch.setattr(star_manager_module.sp, "global_get", mock_global_get)
|
|
monkeypatch.setattr(star_manager_module.sp, "global_put", mock_global_put)
|
|
monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", mock_terminate)
|
|
monkeypatch.setattr(plugin_manager_pm, "reload", mock_reload)
|
|
|
|
try:
|
|
await plugin_manager_pm.turn_off_plugin(plugin.root_dir_name)
|
|
|
|
assert plugin_tool.active is False
|
|
assert other_tool.active is True
|
|
assert preferences["inactivated_plugins"] == [plugin.module_path]
|
|
assert preferences["inactivated_llm_tools"] == []
|
|
assert plugin.activated is False
|
|
|
|
await plugin_manager_pm.turn_on_plugin(plugin.root_dir_name)
|
|
|
|
assert plugin_tool.active is True
|
|
assert other_tool.active is True
|
|
assert preferences["inactivated_plugins"] == []
|
|
assert preferences["inactivated_llm_tools"] == []
|
|
finally:
|
|
llm_tools.func_list = original_func_list
|
|
cast(Any, plugin_manager_pm.context).stars.remove(plugin)
|
|
|
|
|
|
def test_is_plugin_llm_tool_requires_module_boundary():
|
|
plugin_module_path = "data.plugins.weather.main"
|
|
plugin_tool = star_manager_module.FunctionTool(
|
|
name="weather",
|
|
description="weather",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path=plugin_module_path,
|
|
)
|
|
child_module_tool = star_manager_module.FunctionTool(
|
|
name="weather_child",
|
|
description="weather child",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path="data.plugins.weather.main.tools.search",
|
|
)
|
|
prefixed_module_tool = star_manager_module.FunctionTool(
|
|
name="weather_prefixed",
|
|
description="weather prefixed",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path="data.plugins.weather.main_extra.tools.search",
|
|
)
|
|
|
|
assert PluginManager._is_plugin_llm_tool(plugin_tool, plugin_module_path) is True
|
|
assert (
|
|
PluginManager._is_plugin_llm_tool(child_module_tool, plugin_module_path) is True
|
|
)
|
|
assert (
|
|
PluginManager._is_plugin_llm_tool(prefixed_module_tool, plugin_module_path)
|
|
is False
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_turn_plugin_preserves_user_disabled_llm_tools(
|
|
plugin_manager_pm: PluginManager,
|
|
monkeypatch,
|
|
):
|
|
plugin = star_manager_module.StarMetadata(
|
|
name="demo_plugin",
|
|
root_dir_name="demo_plugin",
|
|
module_path="data.plugins.demo_plugin.main",
|
|
)
|
|
cast(Any, plugin_manager_pm.context).stars.append(plugin)
|
|
plugin_tool = star_manager_module.FunctionTool(
|
|
name="plugin_search",
|
|
description="plugin search",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path="data.plugins.demo_plugin.main.tools.search",
|
|
)
|
|
llm_tools = cast(Any, star_manager_module.llm_tools)
|
|
original_func_list = llm_tools.func_list
|
|
llm_tools.func_list = [plugin_tool]
|
|
preferences = {
|
|
"inactivated_plugins": [],
|
|
"inactivated_llm_tools": [plugin_tool.name],
|
|
}
|
|
|
|
async def mock_global_get(key, default=None):
|
|
return preferences.get(key, default)
|
|
|
|
async def mock_global_put(key, value):
|
|
preferences[key] = value
|
|
|
|
async def mock_terminate(star_metadata):
|
|
assert star_metadata is plugin
|
|
|
|
async def mock_reload(plugin_name):
|
|
assert plugin_name == plugin.root_dir_name
|
|
return True, None
|
|
|
|
monkeypatch.setattr(star_manager_module.sp, "global_get", mock_global_get)
|
|
monkeypatch.setattr(star_manager_module.sp, "global_put", mock_global_put)
|
|
monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", mock_terminate)
|
|
monkeypatch.setattr(plugin_manager_pm, "reload", mock_reload)
|
|
|
|
try:
|
|
await plugin_manager_pm.turn_off_plugin(plugin.root_dir_name)
|
|
await plugin_manager_pm.turn_on_plugin(plugin.root_dir_name)
|
|
|
|
assert plugin_tool.active is False
|
|
assert preferences["inactivated_plugins"] == []
|
|
assert preferences["inactivated_llm_tools"] == [plugin_tool.name]
|
|
finally:
|
|
llm_tools.func_list = original_func_list
|
|
cast(Any, plugin_manager_pm.context).stars.remove(plugin)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_migrate_legacy_plugin_tool_inactivation_state(
|
|
plugin_manager_pm: PluginManager,
|
|
monkeypatch,
|
|
):
|
|
active_plugin = star_manager_module.StarMetadata(
|
|
name="active_plugin",
|
|
module_path="data.plugins.active_plugin.main",
|
|
)
|
|
inactive_plugin = star_manager_module.StarMetadata(
|
|
name="inactive_plugin",
|
|
module_path="data.plugins.inactive_plugin.main",
|
|
)
|
|
star_manager_module.star_registry.extend([active_plugin, inactive_plugin])
|
|
active_tool = star_manager_module.FunctionTool(
|
|
name="active_plugin_search",
|
|
description="active plugin search",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path="data.plugins.active_plugin.main.tools.search",
|
|
)
|
|
inactive_tool = star_manager_module.FunctionTool(
|
|
name="inactive_plugin_search",
|
|
description="inactive plugin search",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path="data.plugins.inactive_plugin.main.tools.search",
|
|
)
|
|
user_tool = star_manager_module.FunctionTool(
|
|
name="user_disabled_builtin",
|
|
description="user disabled builtin",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path="astrbot.core.tools.user_disabled_builtin",
|
|
)
|
|
active_tool.active = False
|
|
inactive_tool.active = False
|
|
user_tool.active = False
|
|
llm_tools = cast(Any, star_manager_module.llm_tools)
|
|
original_func_list = llm_tools.func_list
|
|
llm_tools.func_list = [active_tool, inactive_tool, user_tool]
|
|
preferences = {
|
|
"inactivated_llm_tools": [
|
|
active_tool.name,
|
|
inactive_tool.name,
|
|
user_tool.name,
|
|
],
|
|
star_manager_module.PLUGIN_TOOL_STATE_MIGRATION_KEY: False,
|
|
}
|
|
|
|
async def mock_global_get(key, default=None):
|
|
return preferences.get(key, default)
|
|
|
|
async def mock_global_put(key, value):
|
|
preferences[key] = value
|
|
|
|
monkeypatch.setattr(star_manager_module.sp, "global_get", mock_global_get)
|
|
monkeypatch.setattr(star_manager_module.sp, "global_put", mock_global_put)
|
|
|
|
try:
|
|
updated_tools = (
|
|
await plugin_manager_pm._migrate_legacy_plugin_tool_inactivation_state(
|
|
preferences["inactivated_llm_tools"],
|
|
[inactive_plugin.module_path],
|
|
)
|
|
)
|
|
|
|
assert updated_tools == [user_tool.name]
|
|
assert preferences["inactivated_llm_tools"] == [user_tool.name]
|
|
assert preferences[star_manager_module.PLUGIN_TOOL_STATE_MIGRATION_KEY] is True
|
|
assert active_tool.active is True
|
|
assert inactive_tool.active is False
|
|
assert user_tool.active is False
|
|
finally:
|
|
llm_tools.func_list = original_func_list
|
|
_clear_star_runtime_state()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_migrate_legacy_plugin_tool_inactivation_state_defers_without_loaded_plugin_tools(
|
|
plugin_manager_pm: PluginManager,
|
|
monkeypatch,
|
|
):
|
|
llm_tools = cast(Any, star_manager_module.llm_tools)
|
|
original_func_list = llm_tools.func_list
|
|
llm_tools.func_list = []
|
|
preferences = {
|
|
"inactivated_llm_tools": ["legacy_plugin_tool"],
|
|
star_manager_module.PLUGIN_TOOL_STATE_MIGRATION_KEY: False,
|
|
}
|
|
|
|
async def mock_global_get(key, default=None):
|
|
return preferences.get(key, default)
|
|
|
|
async def mock_global_put(key, value):
|
|
preferences[key] = value
|
|
|
|
monkeypatch.setattr(star_manager_module.sp, "global_get", mock_global_get)
|
|
monkeypatch.setattr(star_manager_module.sp, "global_put", mock_global_put)
|
|
|
|
try:
|
|
updated_tools = (
|
|
await plugin_manager_pm._migrate_legacy_plugin_tool_inactivation_state(
|
|
preferences["inactivated_llm_tools"],
|
|
[],
|
|
)
|
|
)
|
|
|
|
assert updated_tools == ["legacy_plugin_tool"]
|
|
assert preferences["inactivated_llm_tools"] == ["legacy_plugin_tool"]
|
|
assert preferences[star_manager_module.PLUGIN_TOOL_STATE_MIGRATION_KEY] is False
|
|
finally:
|
|
llm_tools.func_list = original_func_list
|
|
_clear_star_runtime_state()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_applies_manual_inactivation_to_non_plugin_tools(
|
|
plugin_manager_pm: PluginManager,
|
|
monkeypatch,
|
|
):
|
|
manual_tool = star_manager_module.FunctionTool(
|
|
name="manual_tool",
|
|
description="manual tool",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path="external.tools.manual",
|
|
)
|
|
llm_tools = cast(Any, star_manager_module.llm_tools)
|
|
original_func_list = llm_tools.func_list
|
|
llm_tools.func_list = [manual_tool]
|
|
preferences = {
|
|
"inactivated_plugins": [],
|
|
"inactivated_llm_tools": [manual_tool.name],
|
|
"alter_cmd": {},
|
|
star_manager_module.PLUGIN_TOOL_STATE_MIGRATION_KEY: False,
|
|
}
|
|
|
|
async def mock_global_get(key, default=None):
|
|
return preferences.get(key, default)
|
|
|
|
async def mock_global_put(key, value):
|
|
preferences[key] = value
|
|
|
|
async def mock_sync_command_configs():
|
|
return None
|
|
|
|
monkeypatch.setattr(star_manager_module.sp, "global_get", mock_global_get)
|
|
monkeypatch.setattr(star_manager_module.sp, "global_put", mock_global_put)
|
|
monkeypatch.setattr(plugin_manager_pm, "_get_plugin_modules", lambda: [])
|
|
monkeypatch.setattr(
|
|
star_manager_module,
|
|
"sync_command_configs",
|
|
mock_sync_command_configs,
|
|
)
|
|
|
|
try:
|
|
success, error = await plugin_manager_pm.load()
|
|
|
|
assert success is True
|
|
assert error is None
|
|
assert manual_tool.active is False
|
|
assert preferences["inactivated_llm_tools"] == [manual_tool.name]
|
|
assert preferences[star_manager_module.PLUGIN_TOOL_STATE_MIGRATION_KEY] is False
|
|
finally:
|
|
llm_tools.func_list = original_func_list
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_reports_unregistered_plugin_without_index_error(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
_clear_star_runtime_state()
|
|
plugin_root = Path(plugin_manager_pm.plugin_store_path).parents[1]
|
|
plugin_name = "broken_plugin"
|
|
plugin_path = Path(plugin_manager_pm.plugin_store_path) / plugin_name
|
|
plugin_path.mkdir(parents=True)
|
|
(plugin_path / "metadata.yaml").write_text(
|
|
yaml.dump(
|
|
{
|
|
"name": plugin_name,
|
|
"author": "AstrBot Team",
|
|
"desc": "Broken test plugin",
|
|
"version": "1.0.0",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(plugin_path / "main.py").write_text("VALUE = 1\n", encoding="utf-8")
|
|
|
|
async def mock_global_get(key, default=None):
|
|
del key
|
|
return default
|
|
|
|
async def mock_sync_command_configs():
|
|
return None
|
|
|
|
monkeypatch.syspath_prepend(str(plugin_root))
|
|
monkeypatch.setattr(star_manager_module.sp, "global_get", mock_global_get)
|
|
monkeypatch.setattr(
|
|
star_manager_module,
|
|
"sync_command_configs",
|
|
mock_sync_command_configs,
|
|
)
|
|
|
|
try:
|
|
success, error = await plugin_manager_pm.load(specified_dir_name=plugin_name)
|
|
finally:
|
|
_clear_star_runtime_state()
|
|
_clear_module_cache()
|
|
|
|
assert success is False
|
|
assert error is not None
|
|
assert "未通过 Star 注册" in error
|
|
assert "list index out of range" not in error
|
|
assert plugin_name in plugin_manager_pm.failed_plugin_dict
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_plugin_requirements_reraises_cancelled_error(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
_write_requirements(local_updater)
|
|
_mock_missing_requirements(monkeypatch, {"networkx"})
|
|
|
|
async def mock_install_requirements(*args, **kwargs):
|
|
raise asyncio.CancelledError()
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
mock_install_requirements,
|
|
)
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await plugin_manager_pm._ensure_plugin_requirements(
|
|
str(local_updater),
|
|
TEST_PLUGIN_DIR,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_plugin_requirements_wraps_generic_dependency_install_failure(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
_write_requirements(local_updater)
|
|
_mock_missing_requirements(monkeypatch, {"networkx"})
|
|
|
|
async def mock_install_requirements(*args, **kwargs):
|
|
raise RuntimeError("pip failed")
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
mock_install_requirements,
|
|
)
|
|
|
|
with pytest.raises(PluginDependencyInstallError, match="pip failed") as exc_info:
|
|
await plugin_manager_pm._ensure_plugin_requirements(
|
|
str(local_updater),
|
|
TEST_PLUGIN_DIR,
|
|
)
|
|
|
|
assert exc_info.value.plugin_label == TEST_PLUGIN_DIR
|
|
assert exc_info.value.requirements_path == str(local_updater / "requirements.txt")
|
|
assert isinstance(exc_info.value.__cause__, RuntimeError)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_plugin_requirements_wraps_pip_install_error(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
_write_requirements(local_updater)
|
|
_mock_missing_requirements(monkeypatch, {"networkx"})
|
|
|
|
async def mock_install_requirements(*args, **kwargs):
|
|
raise PipInstallError("install failed", code=2)
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
mock_install_requirements,
|
|
)
|
|
|
|
with pytest.raises(
|
|
PluginDependencyInstallError, match="install failed"
|
|
) as exc_info:
|
|
await plugin_manager_pm._ensure_plugin_requirements(
|
|
str(local_updater),
|
|
TEST_PLUGIN_DIR,
|
|
)
|
|
|
|
assert isinstance(exc_info.value.__cause__, PipInstallError)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_plugin_requirements_logs_requirements_file_install_for_missing_dependencies(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
_write_requirements(local_updater)
|
|
_mock_missing_requirements(monkeypatch, {"networkx"})
|
|
logged_lines = []
|
|
|
|
async def mock_install_requirements(*args, **kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
mock_install_requirements,
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.logger.info",
|
|
lambda line, *args: logged_lines.append(line % args if args else line),
|
|
)
|
|
|
|
await plugin_manager_pm._ensure_plugin_requirements(
|
|
str(local_updater),
|
|
TEST_PLUGIN_DIR,
|
|
)
|
|
|
|
assert any("installing them from requirements.txt" in line for line in logged_lines)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("version_mismatch_names", "expected_allow_target_upgrade"),
|
|
[
|
|
(set(), False),
|
|
({"networkx"}, True),
|
|
],
|
|
)
|
|
async def test_ensure_plugin_requirements_sets_target_upgrade_based_on_version_mismatch(
|
|
plugin_manager_pm: PluginManager,
|
|
local_updater: Path,
|
|
monkeypatch,
|
|
version_mismatch_names,
|
|
expected_allow_target_upgrade: bool,
|
|
):
|
|
_write_requirements(local_updater)
|
|
_mock_missing_requirements_plan(
|
|
monkeypatch,
|
|
{"networkx"},
|
|
["networkx"],
|
|
version_mismatch_names=version_mismatch_names,
|
|
)
|
|
observed_calls = []
|
|
|
|
async def mock_install_requirements(*args, **kwargs):
|
|
observed_calls.append(kwargs)
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
mock_install_requirements,
|
|
)
|
|
|
|
await plugin_manager_pm._ensure_plugin_requirements(
|
|
str(local_updater),
|
|
TEST_PLUGIN_DIR,
|
|
)
|
|
|
|
assert len(observed_calls) == 1
|
|
assert observed_calls[0]["allow_target_upgrade"] is expected_allow_target_upgrade
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_plugin_prefers_installed_dependencies_before_first_import(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("networkx\n", encoding="utf-8")
|
|
events = []
|
|
sentinel_module = object()
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.prefer_installed_dependencies",
|
|
lambda *, requirements_path: events.append(("prefer", requirements_path)),
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.plan_missing_requirements_install",
|
|
lambda requirements_path: MissingRequirementsPlan(
|
|
missing_names=frozenset(),
|
|
install_lines=(),
|
|
version_mismatch_names=frozenset(),
|
|
),
|
|
)
|
|
|
|
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
|
|
del globals, locals, level
|
|
events.append(("import", name, tuple(fromlist)))
|
|
return sentinel_module
|
|
|
|
monkeypatch.setattr(star_manager_module, "__import__", fake_import, raising=False)
|
|
|
|
imported_module = await plugin_manager_pm._import_plugin_with_dependency_recovery(
|
|
path="data.plugins.helloworld.main",
|
|
module_str="main",
|
|
root_dir_name=TEST_PLUGIN_DIR,
|
|
requirements_path=str(requirements_path),
|
|
)
|
|
|
|
assert imported_module is sentinel_module
|
|
assert events == [
|
|
("prefer", str(requirements_path)),
|
|
("import", "data.plugins.helloworld.main", ("main",)),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_reserved_plugin_skips_preloading_user_site_dependencies(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("networkx\n", encoding="utf-8")
|
|
events = []
|
|
sentinel_module = object()
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.prefer_installed_dependencies",
|
|
lambda *, requirements_path: events.append(("prefer", requirements_path)),
|
|
)
|
|
|
|
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
|
|
del globals, locals, level
|
|
events.append(("import", name, tuple(fromlist)))
|
|
return sentinel_module
|
|
|
|
monkeypatch.setattr(star_manager_module, "__import__", fake_import, raising=False)
|
|
|
|
imported_module = await plugin_manager_pm._import_plugin_with_dependency_recovery(
|
|
path="astrbot.builtin_stars.web_searcher.main",
|
|
module_str="main",
|
|
root_dir_name="web_searcher",
|
|
requirements_path=str(requirements_path),
|
|
reserved=True,
|
|
)
|
|
|
|
assert imported_module is sentinel_module
|
|
assert events == [
|
|
("import", "astrbot.builtin_stars.web_searcher.main", ("main",)),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_plugin_skips_preloading_when_requirements_version_mismatch_detected(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("networkx>=3\n", encoding="utf-8")
|
|
events = []
|
|
sentinel_module = object()
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.prefer_installed_dependencies",
|
|
lambda *, requirements_path: events.append(("prefer", requirements_path)),
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.plan_missing_requirements_install",
|
|
lambda requirements_path: MissingRequirementsPlan(
|
|
missing_names=frozenset({"networkx"}),
|
|
install_lines=("networkx>=3",),
|
|
version_mismatch_names=frozenset({"networkx"}),
|
|
),
|
|
)
|
|
|
|
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
|
|
del globals, locals, level
|
|
events.append(("import", name, tuple(fromlist)))
|
|
return sentinel_module
|
|
|
|
monkeypatch.setattr(star_manager_module, "__import__", fake_import, raising=False)
|
|
|
|
imported_module = await plugin_manager_pm._import_plugin_with_dependency_recovery(
|
|
path="data.plugins.helloworld.main",
|
|
module_str="main",
|
|
root_dir_name=TEST_PLUGIN_DIR,
|
|
requirements_path=str(requirements_path),
|
|
)
|
|
|
|
assert imported_module is sentinel_module
|
|
assert events == [
|
|
("import", "data.plugins.helloworld.main", ("main",)),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_plugin_reinstalls_when_version_mismatch_import_fails(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("networkx>=3\n", encoding="utf-8")
|
|
events = []
|
|
sentinel_module = object()
|
|
import_attempts = {"count": 0}
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.prefer_installed_dependencies",
|
|
lambda *, requirements_path: events.append(("prefer", requirements_path)),
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.plan_missing_requirements_install",
|
|
lambda requirements_path: MissingRequirementsPlan(
|
|
missing_names=frozenset({"networkx"}),
|
|
install_lines=("networkx>=3",),
|
|
version_mismatch_names=frozenset({"networkx"}),
|
|
),
|
|
)
|
|
|
|
async def mock_check_plugin_dept_update(*, target_plugin=None):
|
|
events.append(("reinstall", target_plugin))
|
|
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm,
|
|
"_check_plugin_dept_update",
|
|
mock_check_plugin_dept_update,
|
|
)
|
|
|
|
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
|
|
del globals, locals, level
|
|
import_attempts["count"] += 1
|
|
events.append(("import", name, tuple(fromlist), import_attempts["count"]))
|
|
if import_attempts["count"] == 1:
|
|
raise ModuleNotFoundError("networkx")
|
|
return sentinel_module
|
|
|
|
monkeypatch.setattr(star_manager_module, "__import__", fake_import, raising=False)
|
|
|
|
imported_module = await plugin_manager_pm._import_plugin_with_dependency_recovery(
|
|
path="data.plugins.helloworld.main",
|
|
module_str="main",
|
|
root_dir_name=TEST_PLUGIN_DIR,
|
|
requirements_path=str(requirements_path),
|
|
)
|
|
|
|
assert imported_module is sentinel_module
|
|
assert events == [
|
|
("import", "data.plugins.helloworld.main", ("main",), 1),
|
|
("reinstall", TEST_PLUGIN_DIR),
|
|
("import", "data.plugins.helloworld.main", ("main",), 2),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_plugin_skips_preloading_when_requirement_precheck_is_unavailable(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("networkx\n", encoding="utf-8")
|
|
events = []
|
|
sentinel_module = object()
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.prefer_installed_dependencies",
|
|
lambda *, requirements_path: events.append(("prefer", requirements_path)),
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.plan_missing_requirements_install",
|
|
lambda requirements_path: None,
|
|
)
|
|
|
|
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
|
|
del globals, locals, level
|
|
events.append(("import", name, tuple(fromlist)))
|
|
return sentinel_module
|
|
|
|
monkeypatch.setattr(star_manager_module, "__import__", fake_import, raising=False)
|
|
|
|
imported_module = await plugin_manager_pm._import_plugin_with_dependency_recovery(
|
|
path="data.plugins.helloworld.main",
|
|
module_str="main",
|
|
root_dir_name=TEST_PLUGIN_DIR,
|
|
requirements_path=str(requirements_path),
|
|
)
|
|
|
|
assert imported_module is sentinel_module
|
|
assert events == [
|
|
("import", "data.plugins.helloworld.main", ("main",)),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_plugin_attempts_dependency_recovery_when_precheck_is_unavailable(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("networkx\n", encoding="utf-8")
|
|
events = []
|
|
sentinel_module = object()
|
|
import_attempts = {"count": 0}
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.prefer_installed_dependencies",
|
|
lambda *, requirements_path: events.append(("prefer", requirements_path)),
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.plan_missing_requirements_install",
|
|
lambda requirements_path: None,
|
|
)
|
|
|
|
async def unexpected_check_plugin_dept_update(*args, **kwargs):
|
|
raise AssertionError("dependency install fallback should not run")
|
|
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm,
|
|
"_check_plugin_dept_update",
|
|
unexpected_check_plugin_dept_update,
|
|
)
|
|
|
|
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
|
|
del globals, locals, level
|
|
import_attempts["count"] += 1
|
|
events.append(("import", name, tuple(fromlist), import_attempts["count"]))
|
|
if import_attempts["count"] == 1:
|
|
raise ModuleNotFoundError("networkx")
|
|
return sentinel_module
|
|
|
|
monkeypatch.setattr(star_manager_module, "__import__", fake_import, raising=False)
|
|
|
|
imported_module = await plugin_manager_pm._import_plugin_with_dependency_recovery(
|
|
path="data.plugins.helloworld.main",
|
|
module_str="main",
|
|
root_dir_name=TEST_PLUGIN_DIR,
|
|
requirements_path=str(requirements_path),
|
|
)
|
|
|
|
assert imported_module is sentinel_module
|
|
assert events == [
|
|
("import", "data.plugins.helloworld.main", ("main",), 1),
|
|
("prefer", str(requirements_path)),
|
|
("import", "data.plugins.helloworld.main", ("main",), 2),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_plugin_does_not_recover_from_plain_import_error(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("networkx\n", encoding="utf-8")
|
|
events = []
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.prefer_installed_dependencies",
|
|
lambda *, requirements_path: events.append(("prefer", requirements_path)),
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.plan_missing_requirements_install",
|
|
lambda requirements_path: MissingRequirementsPlan(
|
|
missing_names=frozenset(),
|
|
install_lines=(),
|
|
version_mismatch_names=frozenset(),
|
|
),
|
|
)
|
|
|
|
async def unexpected_check_plugin_dept_update(*args, **kwargs):
|
|
raise AssertionError("dependency install fallback should not run")
|
|
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm,
|
|
"_check_plugin_dept_update",
|
|
unexpected_check_plugin_dept_update,
|
|
)
|
|
|
|
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
|
|
del globals, locals, level
|
|
events.append(("import", name, tuple(fromlist)))
|
|
raise ImportError("plugin import error")
|
|
|
|
monkeypatch.setattr(star_manager_module, "__import__", fake_import, raising=False)
|
|
|
|
with pytest.raises(ImportError, match="plugin import error"):
|
|
await plugin_manager_pm._import_plugin_with_dependency_recovery(
|
|
path="data.plugins.helloworld.main",
|
|
module_str="main",
|
|
root_dir_name=TEST_PLUGIN_DIR,
|
|
requirements_path=str(requirements_path),
|
|
)
|
|
|
|
assert events == [
|
|
("prefer", str(requirements_path)),
|
|
("import", "data.plugins.helloworld.main", ("main",)),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_plugin_surfaces_unexpected_recovery_errors(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("networkx\n", encoding="utf-8")
|
|
events = []
|
|
|
|
def raising_prefer_installed_dependencies(*, requirements_path):
|
|
events.append(("prefer", requirements_path))
|
|
raise RuntimeError("unexpected recovery failure")
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.prefer_installed_dependencies",
|
|
raising_prefer_installed_dependencies,
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.plan_missing_requirements_install",
|
|
lambda requirements_path: None,
|
|
)
|
|
|
|
async def unexpected_check_plugin_dept_update(*args, **kwargs):
|
|
raise AssertionError("dependency install fallback should not run")
|
|
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm,
|
|
"_check_plugin_dept_update",
|
|
unexpected_check_plugin_dept_update,
|
|
)
|
|
|
|
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
|
|
del globals, locals, level
|
|
events.append(("import", name, tuple(fromlist)))
|
|
raise ModuleNotFoundError("networkx")
|
|
|
|
monkeypatch.setattr(star_manager_module, "__import__", fake_import, raising=False)
|
|
|
|
with pytest.raises(RuntimeError, match="unexpected recovery failure"):
|
|
await plugin_manager_pm._import_plugin_with_dependency_recovery(
|
|
path="data.plugins.helloworld.main",
|
|
module_str="main",
|
|
root_dir_name=TEST_PLUGIN_DIR,
|
|
requirements_path=str(requirements_path),
|
|
)
|
|
|
|
assert events == [
|
|
("import", "data.plugins.helloworld.main", ("main",)),
|
|
("prefer", str(requirements_path)),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("dependency_install_fails", [False, True])
|
|
async def test_update_plugin_dependency_install_flow(
|
|
plugin_manager_pm: PluginManager,
|
|
local_updater: Path,
|
|
monkeypatch,
|
|
dependency_install_fails: bool,
|
|
):
|
|
mock_star = MockStar()
|
|
cast(Any, plugin_manager_pm.context).stars.append(mock_star)
|
|
|
|
_write_requirements(local_updater)
|
|
events = []
|
|
_mock_missing_requirements(monkeypatch, {"networkx"})
|
|
|
|
async def mock_update(plugin, proxy="", download_url="", repo_url=""):
|
|
del proxy, download_url, repo_url
|
|
events.append(("update", plugin.name))
|
|
|
|
monkeypatch.setattr(plugin_manager_pm._updater, "update", mock_update)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
_build_dependency_install_mock(events, dependency_install_fails),
|
|
)
|
|
monkeypatch.setattr(plugin_manager_pm, "reload", _build_reload_mock(events))
|
|
|
|
if dependency_install_fails:
|
|
with pytest.raises(PluginDependencyInstallError, match="pip failed"):
|
|
await plugin_manager_pm.update_plugin(TEST_PLUGIN_NAME)
|
|
dep_event = next(event for event in events if event[0] == "deps")
|
|
_assert_dependency_install_event_matches(
|
|
dep_event,
|
|
expected_original_path=local_updater / "requirements.txt",
|
|
expected_content="networkx\n",
|
|
)
|
|
else:
|
|
await plugin_manager_pm.update_plugin(TEST_PLUGIN_NAME)
|
|
dep_event = next(event for event in events if event[0] == "deps")
|
|
_assert_dependency_install_event_matches(
|
|
dep_event,
|
|
expected_original_path=local_updater / "requirements.txt",
|
|
expected_content="networkx\n",
|
|
)
|
|
assert ("reload", TEST_PLUGIN_DIR) in events
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_plugin_skips_dependency_install_when_no_requirements_missing(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
events = []
|
|
_mock_missing_requirements(monkeypatch, set())
|
|
|
|
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(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
_build_dependency_install_mock(events, False),
|
|
)
|
|
|
|
def mock_load_and_register(*args, **kwargs):
|
|
cast(Any, plugin_manager_pm.context).stars.append(MockStar())
|
|
return _build_load_mock(events)(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(plugin_manager_pm, "load", mock_load_and_register)
|
|
|
|
await plugin_manager_pm.install_plugin(TEST_PLUGIN_REPO)
|
|
|
|
assert "deps" not in [e[0] for e in events]
|
|
assert ("load", TEST_PLUGIN_DIR) in events
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_install_plugin_runs_dependency_install_when_precheck_fails(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
plugin_path = Path(plugin_manager_pm.plugin_store_path) / TEST_PLUGIN_DIR
|
|
events = []
|
|
|
|
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)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
_build_dependency_install_mock(events, False),
|
|
)
|
|
|
|
def mock_load_and_register(*args, **kwargs):
|
|
cast(Any, plugin_manager_pm.context).stars.append(MockStar())
|
|
return _build_load_mock(events)(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(plugin_manager_pm, "load", mock_load_and_register)
|
|
|
|
await plugin_manager_pm.install_plugin(TEST_PLUGIN_REPO)
|
|
|
|
dep_event = next(event for event in events if event[0] == "deps")
|
|
_assert_dependency_install_event_matches(
|
|
dep_event,
|
|
expected_original_path=plugin_path / "requirements.txt",
|
|
)
|
|
assert ("load", TEST_PLUGIN_DIR) in events
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_plugin_requirements_installs_only_missing_requirement_lines(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text(
|
|
"aiohttp>=3.0\nboto3==1.2\nbotocore\n",
|
|
encoding="utf-8",
|
|
)
|
|
events = []
|
|
_mock_missing_requirements_plan(
|
|
monkeypatch, {"boto3", "botocore"}, ["boto3==1.2", "botocore"]
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
_build_dependency_install_mock(events, False, capture_content=True),
|
|
)
|
|
|
|
await plugin_manager_pm._ensure_plugin_requirements(
|
|
str(local_updater),
|
|
TEST_PLUGIN_DIR,
|
|
)
|
|
|
|
assert len(events) == 1
|
|
kind, used_path, content = events[0]
|
|
assert kind == "deps"
|
|
assert used_path != str(requirements_path)
|
|
assert content == "boto3==1.2\nbotocore\n"
|
|
assert not Path(used_path).exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_plugin_requirements_creates_temp_dir_before_filtered_install(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch, tmp_path
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("boto3\n", encoding="utf-8")
|
|
temp_dir = tmp_path / "missing-temp-dir"
|
|
events = []
|
|
_mock_missing_requirements_plan(monkeypatch, {"boto3"}, ["boto3"])
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.get_astrbot_temp_path",
|
|
lambda: str(temp_dir),
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
_build_dependency_install_mock(events, False, capture_content=True),
|
|
)
|
|
|
|
await plugin_manager_pm._ensure_plugin_requirements(
|
|
str(local_updater),
|
|
TEST_PLUGIN_DIR,
|
|
)
|
|
|
|
assert temp_dir.is_dir()
|
|
assert len(events) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_plugin_requirements_falls_back_when_missing_names_have_no_install_lines(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("boto3\n", encoding="utf-8")
|
|
events = []
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.plan_missing_requirements_install",
|
|
lambda path: MissingRequirementsPlan(
|
|
missing_names=frozenset({"botocore"}),
|
|
install_lines=(),
|
|
fallback_reason="unmapped missing requirement names",
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
_build_dependency_install_mock(events, False),
|
|
)
|
|
|
|
await plugin_manager_pm._ensure_plugin_requirements(
|
|
str(local_updater),
|
|
TEST_PLUGIN_DIR,
|
|
)
|
|
|
|
assert events == [("deps", str(requirements_path))]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_plugin_requirements_fallback_full_install_keeps_upgrade_for_version_mismatch(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("boto3>=2\n", encoding="utf-8")
|
|
observed_calls = []
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.plan_missing_requirements_install",
|
|
lambda path: MissingRequirementsPlan(
|
|
missing_names=frozenset({"boto3"}),
|
|
install_lines=(),
|
|
version_mismatch_names=frozenset({"boto3"}),
|
|
fallback_reason="unmapped missing requirement names",
|
|
),
|
|
)
|
|
|
|
async def mock_install_requirements(*args, **kwargs):
|
|
observed_calls.append(kwargs)
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
mock_install_requirements,
|
|
)
|
|
|
|
await plugin_manager_pm._ensure_plugin_requirements(
|
|
str(local_updater),
|
|
TEST_PLUGIN_DIR,
|
|
)
|
|
|
|
assert len(observed_calls) == 1
|
|
assert observed_calls[0]["requirements_path"] == str(requirements_path)
|
|
assert observed_calls[0]["allow_target_upgrade"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_plugin_requirements_does_not_mask_install_error_when_cleanup_fails(
|
|
plugin_manager_pm: PluginManager, local_updater: Path, monkeypatch, tmp_path
|
|
):
|
|
requirements_path = local_updater / "requirements.txt"
|
|
requirements_path.write_text("boto3\n", encoding="utf-8")
|
|
temp_dir = tmp_path / "cleanup-fails"
|
|
_mock_missing_requirements_plan(monkeypatch, {"boto3"}, ["boto3"])
|
|
warning_logs = []
|
|
|
|
async def mock_install_requirements(
|
|
*, requirements_path: str | None = None, **kwargs
|
|
):
|
|
del kwargs, requirements_path
|
|
raise RuntimeError("pip failed")
|
|
|
|
original_remove = os.remove
|
|
|
|
def flaky_remove(path):
|
|
if str(path).endswith("_plugin_requirements.txt"):
|
|
raise OSError("cleanup failed")
|
|
return original_remove(path)
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.get_astrbot_temp_path",
|
|
lambda: str(temp_dir),
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.pip_installer.install",
|
|
mock_install_requirements,
|
|
)
|
|
monkeypatch.setattr("astrbot.core.star.star_manager.os.remove", flaky_remove)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.logger.warning",
|
|
lambda line, *args: warning_logs.append(line % args if args else line),
|
|
)
|
|
|
|
with pytest.raises(PluginDependencyInstallError, match="pip failed"):
|
|
await plugin_manager_pm._ensure_plugin_requirements(
|
|
str(local_updater),
|
|
TEST_PLUGIN_DIR,
|
|
)
|
|
|
|
assert any(
|
|
"Failed to remove the temporary plugin requirements file" in log
|
|
for log in warning_logs
|
|
)
|
|
|
|
|
|
# --- Tests for plugin_id KV cleanup logic ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cleanup_plugin_optional_artifacts_clears_kv_when_plugin_id_present(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
cleared = []
|
|
|
|
async def clear_preferences(scope, scope_id):
|
|
cleared.append((scope, scope_id))
|
|
|
|
monkeypatch.setattr(star_manager_module.sp, "clear_async", clear_preferences)
|
|
|
|
await plugin_manager_pm._cleanup_plugin_optional_artifacts(
|
|
root_dir_name="test_plugin",
|
|
plugin_label="TestPlugin",
|
|
plugin_id="test_author/test_plugin",
|
|
delete_config=False,
|
|
delete_data=True,
|
|
)
|
|
|
|
assert cleared == [("plugin", "test_author/test_plugin")]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cleanup_plugin_optional_artifacts_skips_kv_when_plugin_id_none(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
cleared = []
|
|
|
|
async def clear_preferences(scope, scope_id):
|
|
cleared.append((scope, scope_id))
|
|
|
|
monkeypatch.setattr(star_manager_module.sp, "clear_async", clear_preferences)
|
|
|
|
await plugin_manager_pm._cleanup_plugin_optional_artifacts(
|
|
root_dir_name="test_plugin",
|
|
plugin_label="TestPlugin",
|
|
plugin_id=None,
|
|
delete_config=False,
|
|
delete_data=True,
|
|
)
|
|
|
|
assert cleared == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uninstall_plugin_reads_plugin_id_from_metadata(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
cleanup_calls = []
|
|
|
|
mock_star = MockStar()
|
|
mock_star.root_dir_name = TEST_PLUGIN_DIR
|
|
mock_star.name = TEST_PLUGIN_NAME
|
|
mock_star.module_path = "data.plugins.helloworld.main"
|
|
mock_star.reserved = False
|
|
mock_star.star_cls = None
|
|
mock_star.plugin_id = "mock_author/mock_name"
|
|
|
|
cast(Any, plugin_manager_pm.context).stars.append(mock_star)
|
|
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm, "_terminate_plugin", lambda p: asyncio.sleep(0)
|
|
)
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm, "_unbind_plugin", lambda n, m: asyncio.sleep(0)
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.remove_dir",
|
|
lambda p: None,
|
|
)
|
|
|
|
async def mock_cleanup(
|
|
*, root_dir_name, plugin_label, plugin_id, delete_config, delete_data
|
|
):
|
|
cleanup_calls.append(
|
|
{
|
|
"root_dir_name": root_dir_name,
|
|
"plugin_label": plugin_label,
|
|
"plugin_id": plugin_id,
|
|
}
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm, "_cleanup_plugin_optional_artifacts", mock_cleanup
|
|
)
|
|
|
|
await plugin_manager_pm.uninstall_plugin(
|
|
TEST_PLUGIN_NAME, delete_config=False, delete_data=True
|
|
)
|
|
|
|
assert len(cleanup_calls) == 1
|
|
assert cleanup_calls[0]["plugin_id"] == "mock_author/mock_name"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uninstall_plugin_handles_disabled_plugin_with_plugin_id(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
cleanup_calls = []
|
|
|
|
mock_star = MockStar()
|
|
mock_star.root_dir_name = TEST_PLUGIN_DIR
|
|
mock_star.name = TEST_PLUGIN_NAME
|
|
mock_star.module_path = "data.plugins.helloworld.main"
|
|
mock_star.star_cls = None
|
|
mock_star.plugin_id = "mock_author/mock_name"
|
|
|
|
cast(Any, plugin_manager_pm.context).stars.append(mock_star)
|
|
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm, "_terminate_plugin", lambda p: asyncio.sleep(0)
|
|
)
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm, "_unbind_plugin", lambda n, m: asyncio.sleep(0)
|
|
)
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.remove_dir",
|
|
lambda p: None,
|
|
)
|
|
|
|
async def mock_cleanup(
|
|
*, root_dir_name, plugin_label, plugin_id, delete_config, delete_data
|
|
):
|
|
cleanup_calls.append(
|
|
{
|
|
"root_dir_name": root_dir_name,
|
|
"plugin_label": plugin_label,
|
|
"plugin_id": plugin_id,
|
|
}
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm, "_cleanup_plugin_optional_artifacts", mock_cleanup
|
|
)
|
|
|
|
await plugin_manager_pm.uninstall_plugin(
|
|
TEST_PLUGIN_NAME, delete_config=False, delete_data=True
|
|
)
|
|
|
|
assert len(cleanup_calls) == 1
|
|
assert cleanup_calls[0]["plugin_id"] == "mock_author/mock_name"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uninstall_failed_plugin_passes_plugin_id_from_record(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
cleanup_calls = []
|
|
|
|
plugin_manager_pm.failed_plugin_dict[TEST_PLUGIN_DIR] = {
|
|
"name": TEST_PLUGIN_NAME,
|
|
"display_name": "Hello World",
|
|
"plugin_id": "astrbot_team/helloworld",
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.remove_dir",
|
|
lambda p: None,
|
|
)
|
|
|
|
async def mock_cleanup(
|
|
*, root_dir_name, plugin_label, plugin_id, delete_config, delete_data
|
|
):
|
|
cleanup_calls.append(
|
|
{
|
|
"root_dir_name": root_dir_name,
|
|
"plugin_label": plugin_label,
|
|
"plugin_id": plugin_id,
|
|
}
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm, "_cleanup_plugin_optional_artifacts", mock_cleanup
|
|
)
|
|
|
|
await plugin_manager_pm.uninstall_failed_plugin(
|
|
TEST_PLUGIN_DIR, delete_config=False, delete_data=True
|
|
)
|
|
|
|
assert len(cleanup_calls) == 1
|
|
assert cleanup_calls[0]["plugin_id"] == "astrbot_team/helloworld"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uninstall_failed_plugin_without_plugin_id_in_record(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
cleanup_calls = []
|
|
|
|
plugin_manager_pm.failed_plugin_dict[TEST_PLUGIN_DIR] = {
|
|
"name": TEST_PLUGIN_NAME,
|
|
"display_name": "Hello World",
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
"astrbot.core.star.star_manager.remove_dir",
|
|
lambda p: None,
|
|
)
|
|
|
|
async def mock_cleanup(
|
|
*, root_dir_name, plugin_label, plugin_id, delete_config, delete_data
|
|
):
|
|
cleanup_calls.append(
|
|
{
|
|
"root_dir_name": root_dir_name,
|
|
"plugin_label": plugin_label,
|
|
"plugin_id": plugin_id,
|
|
}
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm, "_cleanup_plugin_optional_artifacts", mock_cleanup
|
|
)
|
|
|
|
await plugin_manager_pm.uninstall_failed_plugin(
|
|
TEST_PLUGIN_DIR, delete_config=False, delete_data=True
|
|
)
|
|
|
|
assert len(cleanup_calls) == 1
|
|
assert cleanup_calls[0]["plugin_id"] is None
|
|
|
|
|
|
# --- reload + deactivated plugin regression tests ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("inactivated_plugins", "expected_activated"),
|
|
[
|
|
([], True),
|
|
(["data.plugins.demo_plugin.main"], False),
|
|
],
|
|
)
|
|
async def test_load_syncs_existing_metadata_activation_from_preferences(
|
|
plugin_manager_pm: PluginManager,
|
|
monkeypatch,
|
|
inactivated_plugins: list[str],
|
|
expected_activated: bool,
|
|
):
|
|
"""Existing plugin metadata activation follows persisted preferences."""
|
|
_clear_star_runtime_state()
|
|
plugin_name = "demo_plugin"
|
|
module_path = f"data.plugins.{plugin_name}.main"
|
|
metadata = star_manager_module.StarMetadata(
|
|
name=plugin_name,
|
|
author="AstrBot Team",
|
|
desc="Demo plugin",
|
|
version="1.0.0",
|
|
root_dir_name=plugin_name,
|
|
module_path=module_path,
|
|
activated=False,
|
|
)
|
|
star_manager_module.star_map[module_path] = metadata
|
|
star_manager_module.star_registry.append(metadata)
|
|
preferences = {
|
|
"inactivated_plugins": inactivated_plugins,
|
|
"inactivated_llm_tools": [],
|
|
"alter_cmd": {},
|
|
}
|
|
|
|
async def mock_global_get(key, default=None):
|
|
return preferences.get(key, default)
|
|
|
|
async def mock_import_plugin_with_dependency_recovery(
|
|
path,
|
|
module_str,
|
|
root_dir_name,
|
|
requirements_path,
|
|
*,
|
|
reserved=False,
|
|
):
|
|
del module_str, root_dir_name, requirements_path, reserved
|
|
assert path == module_path
|
|
return ModuleType(module_path)
|
|
|
|
async def mock_sync_command_configs():
|
|
return None
|
|
|
|
monkeypatch.setattr(star_manager_module.sp, "global_get", mock_global_get)
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm,
|
|
"_get_plugin_modules",
|
|
lambda: [{"pname": plugin_name, "module": "main"}],
|
|
)
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm,
|
|
"_import_plugin_with_dependency_recovery",
|
|
mock_import_plugin_with_dependency_recovery,
|
|
)
|
|
monkeypatch.setattr(plugin_manager_pm, "_load_plugin_metadata", lambda **_: None)
|
|
monkeypatch.setattr(
|
|
star_manager_module,
|
|
"sync_command_configs",
|
|
mock_sync_command_configs,
|
|
)
|
|
|
|
try:
|
|
success, error = await plugin_manager_pm.load(
|
|
specified_module_path=module_path,
|
|
)
|
|
|
|
assert success is True
|
|
assert error is None
|
|
assert metadata.activated is expected_activated
|
|
finally:
|
|
_clear_star_runtime_state()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reload_deactivated_plugin_preserves_tools(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
"""Specified reload of a deactivated plugin keeps its tools in func_list."""
|
|
_clear_star_runtime_state()
|
|
plugin_name = "demo_plugin"
|
|
module_path = f"data.plugins.{plugin_name}.main"
|
|
metadata = star_manager_module.StarMetadata(
|
|
name=plugin_name,
|
|
root_dir_name=plugin_name,
|
|
module_path=module_path,
|
|
activated=False,
|
|
)
|
|
star_manager_module.star_map[module_path] = metadata
|
|
star_manager_module.star_registry.append(metadata)
|
|
|
|
plugin_tool = star_manager_module.FunctionTool(
|
|
name="plugin_search",
|
|
description="plugin search",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path=f"data.plugins.{plugin_name}.main.tools.search",
|
|
)
|
|
llm_tools = cast(Any, star_manager_module.llm_tools)
|
|
original_func_list = llm_tools.func_list
|
|
llm_tools.func_list = [plugin_tool]
|
|
|
|
async def mock_terminate(smd):
|
|
pass # deactivated → no-op
|
|
|
|
async def mock_load(specified_module_path=None, **kwargs):
|
|
return True, None
|
|
|
|
monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", mock_terminate)
|
|
monkeypatch.setattr(plugin_manager_pm, "load", mock_load)
|
|
|
|
try:
|
|
await plugin_manager_pm.reload(plugin_name)
|
|
assert plugin_tool in llm_tools.func_list
|
|
finally:
|
|
llm_tools.func_list = original_func_list
|
|
_clear_star_runtime_state()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reload_activated_plugin_still_unbinds(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
"""Specified reload of an activated plugin still calls _unbind_plugin."""
|
|
_clear_star_runtime_state()
|
|
plugin_name = "demo_plugin"
|
|
module_path = f"data.plugins.{plugin_name}.main"
|
|
metadata = star_manager_module.StarMetadata(
|
|
name=plugin_name,
|
|
root_dir_name=plugin_name,
|
|
module_path=module_path,
|
|
activated=True,
|
|
)
|
|
star_manager_module.star_map[module_path] = metadata
|
|
star_manager_module.star_registry.append(metadata)
|
|
|
|
unbound = []
|
|
|
|
async def mock_terminate(smd):
|
|
pass
|
|
|
|
async def mock_unbind(name, path):
|
|
unbound.append(name)
|
|
|
|
async def mock_load(specified_module_path=None, **kwargs):
|
|
return True, None
|
|
|
|
monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", mock_terminate)
|
|
monkeypatch.setattr(plugin_manager_pm, "_unbind_plugin", mock_unbind)
|
|
monkeypatch.setattr(plugin_manager_pm, "load", mock_load)
|
|
|
|
try:
|
|
await plugin_manager_pm.reload(plugin_name)
|
|
assert unbound == [plugin_name]
|
|
finally:
|
|
_clear_star_runtime_state()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_reload_deactivated_plugin_stays_registered(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
"""Full reload keeps deactivated plugin in star_map with activated=False."""
|
|
_clear_star_runtime_state()
|
|
plugin_name = "demo_plugin"
|
|
module_path = f"data.plugins.{plugin_name}.main"
|
|
metadata = star_manager_module.StarMetadata(
|
|
name=plugin_name,
|
|
root_dir_name=plugin_name,
|
|
module_path=module_path,
|
|
activated=False,
|
|
)
|
|
star_manager_module.star_map[module_path] = metadata
|
|
star_manager_module.star_registry.append(metadata)
|
|
|
|
async def mock_terminate(smd):
|
|
pass
|
|
|
|
async def mock_unbind_full(name, path):
|
|
pass
|
|
|
|
async def mock_load(specified_module_path=None, **kwargs):
|
|
# In full reload, load() re-registers all plugins.
|
|
# Deactivated plugins get registered with activated=False.
|
|
re_registered = star_manager_module.StarMetadata(
|
|
name=plugin_name,
|
|
root_dir_name=plugin_name,
|
|
module_path=module_path,
|
|
activated=False,
|
|
)
|
|
star_manager_module.star_map[module_path] = re_registered
|
|
star_manager_module.star_registry.append(re_registered)
|
|
return True, None
|
|
|
|
monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", mock_terminate)
|
|
monkeypatch.setattr(plugin_manager_pm, "_unbind_plugin", mock_unbind_full)
|
|
monkeypatch.setattr(plugin_manager_pm, "load", mock_load)
|
|
|
|
try:
|
|
await plugin_manager_pm.reload()
|
|
assert module_path in star_manager_module.star_map
|
|
assert star_manager_module.star_map[module_path].activated is False
|
|
finally:
|
|
_clear_star_runtime_state()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_turn_on_plugin_after_deactivated_reload_reactivates_tools(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
"""turn_on_plugin reactivates tools after a deactivated plugin is reloaded."""
|
|
_clear_star_runtime_state()
|
|
plugin_name = "demo_plugin"
|
|
module_path = f"data.plugins.{plugin_name}.main"
|
|
plugin = star_manager_module.StarMetadata(
|
|
name=plugin_name,
|
|
root_dir_name=plugin_name,
|
|
module_path=module_path,
|
|
activated=False,
|
|
)
|
|
cast(Any, plugin_manager_pm.context).stars.append(plugin)
|
|
star_manager_module.star_map[module_path] = plugin
|
|
star_manager_module.star_registry.append(plugin)
|
|
|
|
plugin_tool = star_manager_module.FunctionTool(
|
|
name="plugin_search",
|
|
description="plugin search",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler_module_path=f"data.plugins.{plugin_name}.main.tools.search",
|
|
)
|
|
plugin_tool.active = False # simulate deactivated state
|
|
llm_tools = cast(Any, star_manager_module.llm_tools)
|
|
original_func_list = llm_tools.func_list
|
|
llm_tools.func_list = [plugin_tool]
|
|
preferences = {
|
|
"inactivated_plugins": [module_path],
|
|
"inactivated_llm_tools": [],
|
|
}
|
|
|
|
async def mock_global_get(key, default=None):
|
|
return preferences.get(key, default)
|
|
|
|
async def mock_global_put(key, value):
|
|
preferences[key] = value
|
|
|
|
async def mock_terminate(smd):
|
|
pass
|
|
|
|
async def mock_reload(plugin_name_arg):
|
|
assert plugin_name_arg == plugin_name
|
|
return True, None
|
|
|
|
monkeypatch.setattr(star_manager_module.sp, "global_get", mock_global_get)
|
|
monkeypatch.setattr(star_manager_module.sp, "global_put", mock_global_put)
|
|
monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", mock_terminate)
|
|
monkeypatch.setattr(plugin_manager_pm, "reload", mock_reload)
|
|
|
|
try:
|
|
await plugin_manager_pm.turn_on_plugin(plugin_name)
|
|
assert plugin_tool.active is True
|
|
assert module_path not in preferences["inactivated_plugins"]
|
|
assert plugin.activated is True
|
|
finally:
|
|
llm_tools.func_list = original_func_list
|
|
cast(Any, plugin_manager_pm.context).stars.remove(plugin)
|
|
_clear_star_runtime_state()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_repeated_deactivated_loads_bind_handlers_once_when_activated(
|
|
plugin_manager_pm: PluginManager, monkeypatch
|
|
):
|
|
"""Repeated disabled loads keep raw callables and activation binds once."""
|
|
_clear_star_runtime_state()
|
|
plugin_name = "demo_plugin"
|
|
module_path = f"data.plugins.{plugin_name}.main"
|
|
|
|
class DemoPlugin:
|
|
def __init__(self, context):
|
|
self.context = context
|
|
self.initialize_count = 0
|
|
|
|
async def initialize(self):
|
|
self.initialize_count += 1
|
|
|
|
stale_plugin = DemoPlugin(plugin_manager_pm.context)
|
|
|
|
metadata = star_manager_module.StarMetadata(
|
|
name=plugin_name,
|
|
author="AstrBot Team",
|
|
desc="Demo plugin",
|
|
version="1.0.0",
|
|
root_dir_name=plugin_name,
|
|
module_path=module_path,
|
|
star_cls_type=cast(Any, DemoPlugin),
|
|
star_cls=cast(Any, stale_plugin),
|
|
activated=False,
|
|
)
|
|
cast(Any, plugin_manager_pm.context).stars.append(metadata)
|
|
star_manager_module.star_map[module_path] = metadata
|
|
star_manager_module.star_registry.append(metadata)
|
|
|
|
async def raw_event_handler(plugin, event):
|
|
return plugin, event
|
|
|
|
async def raw_tool_handler(plugin, query):
|
|
return plugin, query
|
|
|
|
raw_event_handler.__module__ = module_path
|
|
raw_tool_handler.__module__ = module_path
|
|
event_handler = StarHandlerMetadata(
|
|
event_type=EventType.AdapterMessageEvent,
|
|
handler_full_name=f"{module_path}_raw_event_handler",
|
|
handler_name="raw_event_handler",
|
|
handler_module_path=module_path,
|
|
handler=functools.partial(raw_event_handler, stale_plugin),
|
|
event_filters=[],
|
|
)
|
|
star_manager_module.star_handlers_registry.append(event_handler)
|
|
|
|
plugin_tool = star_manager_module.FunctionTool(
|
|
name="plugin_search",
|
|
description="plugin search",
|
|
parameters={"type": "object", "properties": {}},
|
|
handler=functools.partial(raw_tool_handler, stale_plugin),
|
|
handler_module_path=module_path,
|
|
)
|
|
llm_tools = cast(Any, star_manager_module.llm_tools)
|
|
original_func_list = llm_tools.func_list
|
|
llm_tools.func_list = [plugin_tool]
|
|
preferences = {
|
|
"inactivated_plugins": [module_path],
|
|
"inactivated_llm_tools": [],
|
|
"alter_cmd": {},
|
|
}
|
|
|
|
async def mock_global_get(key, default=None):
|
|
return preferences.get(key, default)
|
|
|
|
async def mock_global_put(key, value):
|
|
preferences[key] = value
|
|
|
|
async def mock_import_plugin_with_dependency_recovery(
|
|
path,
|
|
module_str,
|
|
root_dir_name,
|
|
requirements_path,
|
|
*,
|
|
reserved=False,
|
|
):
|
|
del module_str, root_dir_name, requirements_path, reserved
|
|
assert path == module_path
|
|
return ModuleType(module_path)
|
|
|
|
async def mock_sync_command_configs():
|
|
return None
|
|
|
|
monkeypatch.setattr(star_manager_module.sp, "global_get", mock_global_get)
|
|
monkeypatch.setattr(star_manager_module.sp, "global_put", mock_global_put)
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm,
|
|
"_get_plugin_modules",
|
|
lambda: [{"pname": plugin_name, "module": "main"}],
|
|
)
|
|
monkeypatch.setattr(
|
|
plugin_manager_pm,
|
|
"_import_plugin_with_dependency_recovery",
|
|
mock_import_plugin_with_dependency_recovery,
|
|
)
|
|
monkeypatch.setattr(plugin_manager_pm, "_load_plugin_metadata", lambda **_: None)
|
|
monkeypatch.setattr(
|
|
star_manager_module,
|
|
"sync_command_configs",
|
|
mock_sync_command_configs,
|
|
)
|
|
|
|
try:
|
|
for _ in range(2):
|
|
success, error = await plugin_manager_pm.load(
|
|
specified_module_path=module_path,
|
|
)
|
|
assert success is True
|
|
assert error is None
|
|
assert event_handler.handler is raw_event_handler
|
|
assert plugin_tool.handler is raw_tool_handler
|
|
assert plugin_tool.active is False
|
|
assert metadata.star_cls is None
|
|
assert stale_plugin.initialize_count == 0
|
|
|
|
await plugin_manager_pm.turn_on_plugin(plugin_name)
|
|
|
|
assert isinstance(event_handler.handler, functools.partial)
|
|
assert event_handler.handler.func is raw_event_handler
|
|
assert event_handler.handler.args == (metadata.star_cls,)
|
|
assert isinstance(plugin_tool.handler, functools.partial)
|
|
assert plugin_tool.handler.func is raw_tool_handler
|
|
assert plugin_tool.handler.args == (metadata.star_cls,)
|
|
assert plugin_tool.active is True
|
|
assert metadata.star_cls.initialize_count == 1
|
|
assert await event_handler.handler("event") == (metadata.star_cls, "event")
|
|
assert await plugin_tool.handler("query") == (metadata.star_cls, "query")
|
|
|
|
success, error = await plugin_manager_pm.load(
|
|
specified_module_path=module_path,
|
|
)
|
|
assert success is True
|
|
assert error is None
|
|
assert isinstance(event_handler.handler, functools.partial)
|
|
assert event_handler.handler.func is raw_event_handler
|
|
assert event_handler.handler.args == (metadata.star_cls,)
|
|
assert isinstance(plugin_tool.handler, functools.partial)
|
|
assert plugin_tool.handler.func is raw_tool_handler
|
|
assert plugin_tool.handler.args == (metadata.star_cls,)
|
|
assert metadata.star_cls.initialize_count == 1
|
|
assert await event_handler.handler("event") == (metadata.star_cls, "event")
|
|
assert await plugin_tool.handler("query") == (metadata.star_cls, "query")
|
|
finally:
|
|
llm_tools.func_list = original_func_list
|
|
cast(Any, plugin_manager_pm.context).stars.remove(metadata)
|
|
_clear_star_runtime_state()
|