mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-31 12:24:39 +08:00
442 lines
16 KiB
Python
442 lines
16 KiB
Python
"""
|
|
验证插件兼容导入清单与受控能力边界。
|
|
"""
|
|
|
|
import ast
|
|
import importlib
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.runtime.compat.manifest import MODULE_ALIASES, PACKAGE_ALIASES
|
|
|
|
PROJECT_ROOT = Path(__file__).parents[1]
|
|
APP_ROOT = PROJECT_ROOT / "app"
|
|
OFFICIAL_PLUGIN_BASELINE = (
|
|
PROJECT_ROOT / "tests" / "fixtures" / "architecture" / "official-plugin-baseline.json"
|
|
)
|
|
FILENAME_POLICY = PROJECT_ROOT / "tests" / "fixtures" / "architecture" / "filename-policy.json"
|
|
PLUGIN_COMPONENT_PACKAGES = (
|
|
APP_ROOT / "adapters" / "external" / "plugin",
|
|
APP_ROOT / "adapters" / "system" / "plugin",
|
|
APP_ROOT / "application" / "plugin",
|
|
APP_ROOT / "runtime" / "extensions" / "plugin",
|
|
)
|
|
AUDITED_PLUGIN_MANAGER_MEMBERS = {
|
|
"get_plugin_config",
|
|
"get_plugin_ids",
|
|
"plugins",
|
|
"running_plugins",
|
|
}
|
|
DELEGATED_PLUGIN_CATALOG_METHODS = {
|
|
"async_get_online_plugin_candidates",
|
|
"async_get_online_plugins",
|
|
"async_get_plugins_from_market",
|
|
"get_installed_plugins",
|
|
"get_local_plugin_version",
|
|
"get_local_plugins",
|
|
"get_local_repo_plugins",
|
|
"get_online_plugins",
|
|
"get_plugins_from_market",
|
|
"is_plugin_exists",
|
|
"process_plugins_list",
|
|
}
|
|
|
|
|
|
def _parse(path: Path) -> ast.Module:
|
|
"""
|
|
解析 UTF-8 Python 源码，兼容历史插件文件可能携带的 BOM。
|
|
"""
|
|
return ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
|
|
|
|
|
|
def _module_imports(tree: ast.AST) -> set[str]:
|
|
"""
|
|
返回语法树中的绝对导入模块，不把导入符号误当成子模块。
|
|
"""
|
|
imports: set[str] = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
imports.update(alias.name for alias in node.names)
|
|
elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
|
|
imports.add(node.module)
|
|
return imports
|
|
|
|
|
|
def _public_names(tree: ast.AST) -> set[str]:
|
|
"""
|
|
返回模块定义或非私有导入到顶层的符号名。
|
|
"""
|
|
names: set[str] = set()
|
|
for node in tree.body:
|
|
if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
names.add(node.name)
|
|
elif isinstance(node, ast.ImportFrom):
|
|
names.update(alias.asname or alias.name for alias in node.names)
|
|
elif isinstance(node, ast.Import):
|
|
names.update(alias.asname or alias.name.split(".", 1)[0] for alias in node.names)
|
|
elif isinstance(node, (ast.Assign, ast.AnnAssign)):
|
|
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
|
|
names.update(target.id for target in targets if isinstance(target, ast.Name))
|
|
return {name for name in names if not name.startswith("_")}
|
|
|
|
|
|
def _camel_to_snake(value: str) -> str:
|
|
"""
|
|
把合同类名转换为对应的稳定模块标识。
|
|
"""
|
|
first_pass = re.sub(r"(.)([A-Z][a-z]+)", r"\1_\2", value)
|
|
return re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", first_pass).lower()
|
|
|
|
|
|
def _matches_filename_contract(path: Path, rule: dict[str, str]) -> bool:
|
|
"""
|
|
按真实发现合同判断多词文件是否属于有理由的语义命名。
|
|
"""
|
|
relative = path.relative_to(PROJECT_ROOT).as_posix()
|
|
root = rule["root"].rstrip("/")
|
|
if not relative.startswith(f"{root}/"):
|
|
return False
|
|
kind = rule["kind"]
|
|
if kind == "agent_tool_private_helper":
|
|
return path.stem.startswith("_")
|
|
|
|
tree = _parse(path)
|
|
class_names = {node.name for node in tree.body if isinstance(node, ast.ClassDef)}
|
|
if kind == "agent_tool_contract":
|
|
expected_name = path.stem
|
|
for node in tree.body:
|
|
if not isinstance(node, ast.ClassDef):
|
|
continue
|
|
for statement in node.body:
|
|
if not isinstance(statement, ast.AnnAssign):
|
|
continue
|
|
if not isinstance(statement.target, ast.Name) or statement.target.id != "name":
|
|
continue
|
|
value = statement.value
|
|
if isinstance(value, ast.Call) and isinstance(value.func, ast.Name) and value.func.id == "Field":
|
|
value = next(
|
|
(keyword.value for keyword in value.keywords if keyword.arg == "default"),
|
|
value.args[0] if value.args else None,
|
|
)
|
|
if isinstance(value, ast.Constant) and value.value == expected_name:
|
|
return True
|
|
return False
|
|
if kind == "workflow_action_contract":
|
|
return any(
|
|
name.endswith("Action")
|
|
and _camel_to_snake(name.removesuffix("Action")) == path.stem
|
|
for name in class_names
|
|
)
|
|
if kind == "indexer_parser_contract":
|
|
if any(
|
|
name.endswith("SiteUserInfo")
|
|
and _camel_to_snake(name.removesuffix("SiteUserInfo")) == path.stem
|
|
for name in class_names
|
|
):
|
|
return True
|
|
return any(
|
|
isinstance(statement, ast.Assign)
|
|
and any(isinstance(target, ast.Name) and target.id == "schema" for target in statement.targets)
|
|
and isinstance(statement.value, ast.Attribute)
|
|
and _camel_to_snake(statement.value.attr) == path.stem
|
|
for node in tree.body
|
|
if isinstance(node, ast.ClassDef)
|
|
for statement in node.body
|
|
)
|
|
raise AssertionError(f"未知文件命名合同：{kind}")
|
|
|
|
|
|
def _is_documentation_only_root(tree: ast.Module) -> bool:
|
|
"""
|
|
判断包根是否只保留包说明。
|
|
"""
|
|
return not tree.body or (ast.get_docstring(tree) is not None and len(tree.body) == 1)
|
|
|
|
|
|
def _is_curated_lazy_facade(tree: ast.Module) -> bool:
|
|
"""
|
|
判断包根是否为显式白名单驱动且不含公开实现的懒门面。
|
|
"""
|
|
assignments = {
|
|
target.id: node.value
|
|
for node in tree.body
|
|
if isinstance(node, ast.Assign)
|
|
for target in node.targets
|
|
if isinstance(target, ast.Name) and target.id in {"_EXPORTS", "__all__"}
|
|
}
|
|
if set(assignments) != {"_EXPORTS", "__all__"}:
|
|
return False
|
|
if any(
|
|
isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef))
|
|
and not node.name.startswith("_")
|
|
for node in tree.body
|
|
):
|
|
return False
|
|
exports = ast.literal_eval(assignments["_EXPORTS"])
|
|
public = ast.literal_eval(assignments["__all__"])
|
|
export_names = set(exports) if isinstance(exports, dict) else set(exports)
|
|
return export_names == set(public)
|
|
|
|
|
|
def test_official_plugin_legacy_imports_are_exactly_routed_or_retired() -> None:
|
|
"""
|
|
官方插件旧导入必须有精确 Compat 路由，或保持一条已审查的可选退役路径。
|
|
"""
|
|
baseline = json.loads(OFFICIAL_PLUGIN_BASELINE.read_text(encoding="utf-8"))
|
|
legacy_imports = {
|
|
module
|
|
for module in baseline["imports"]
|
|
if module == "app.log"
|
|
or module.startswith(("app.core.", "app.helper.", "app.utils."))
|
|
}
|
|
routed = set(MODULE_ALIASES) | set(PACKAGE_ALIASES)
|
|
|
|
# app.helper.subscribe 已由宿主删除，不应恢复源码或猜测映射。
|
|
assert legacy_imports - routed == set()
|
|
assert "app.helper.subscribe" not in routed
|
|
|
|
|
|
def test_plugin_manager_legacy_and_sdk_paths_share_canonical_identity() -> None:
|
|
"""
|
|
V2 旧路径与 V3 SDK 必须复用同一个 PluginManager 身份及已验证成员。
|
|
"""
|
|
canonical = importlib.import_module("app.runtime.extensions.plugin.manager")
|
|
legacy = importlib.import_module("app.core.plugin")
|
|
sdk = importlib.import_module("app.sdk.plugin.manager")
|
|
|
|
assert MODULE_ALIASES["app.core.plugin"].target == canonical.__name__
|
|
assert MODULE_ALIASES["app.core.plugin"].replacement == "app.sdk.plugin.manager"
|
|
assert legacy is canonical
|
|
assert legacy.PluginManager is canonical.PluginManager
|
|
assert sdk.PluginManager is canonical.PluginManager
|
|
assert AUDITED_PLUGIN_MANAGER_MEMBERS <= set(dir(canonical.PluginManager))
|
|
assert sdk.__all__ == ["ModuleManager", "PluginManager"]
|
|
assert not hasattr(sdk, "PluginHelper")
|
|
|
|
|
|
def test_plugin_namespace_root_only_documents_the_install_directory() -> None:
|
|
"""
|
|
插件安装命名空间包根只保留说明，契约基类由 SDK 拥有。
|
|
"""
|
|
tree = _parse(APP_ROOT / "plugins" / "__init__.py")
|
|
|
|
assert _is_documentation_only_root(tree)
|
|
assert _public_names(tree) == set()
|
|
|
|
|
|
def test_plugin_contract_base_depends_on_canonical_modules_only() -> None:
|
|
"""
|
|
插件契约基类直接依赖 canonical 模块，宿主源码不消费幻影旧路径。
|
|
"""
|
|
tree = _parse(APP_ROOT / "sdk" / "plugin" / "base.py")
|
|
legacy = {
|
|
module
|
|
for module in _module_imports(tree)
|
|
if module == "app.log" or module.startswith(("app.core.", "app.helper.", "app.utils."))
|
|
}
|
|
|
|
assert legacy == set()
|
|
|
|
|
|
def test_plugin_helper_stays_on_exact_legacy_adapter_route() -> None:
|
|
"""
|
|
PluginHelper 只保留精确旧 Helper 路由，不进入推荐插件 SDK。
|
|
"""
|
|
canonical = importlib.import_module("app.adapters.external.market")
|
|
legacy = importlib.import_module("app.helper.plugin")
|
|
alias = MODULE_ALIASES["app.helper.plugin"]
|
|
|
|
assert alias.target == canonical.__name__
|
|
assert alias.replacement == canonical.__name__
|
|
assert legacy is canonical
|
|
assert legacy.PluginHelper is canonical.PluginHelper
|
|
|
|
|
|
def test_plugin_entry_and_application_do_not_reimport_concrete_legacy_owners() -> None:
|
|
"""
|
|
Plugin 入口与应用用例不得重新依赖 PluginHelper、ServerHelper 或具体 PluginManager。
|
|
"""
|
|
forbidden_modules = {
|
|
"app.adapters.external.market",
|
|
"app.adapters.external.server",
|
|
"app.runtime.extensions.plugin.manager",
|
|
}
|
|
forbidden_names = {"MoviePilotServerHelper", "PluginHelper", "PluginManager"}
|
|
paths = [APP_ROOT / "api" / "endpoints" / "plugin.py"]
|
|
paths.extend((APP_ROOT / "application" / "plugin").rglob("*.py"))
|
|
violations: list[str] = []
|
|
|
|
for path in paths:
|
|
tree = _parse(path)
|
|
relative = path.relative_to(PROJECT_ROOT).as_posix()
|
|
violations.extend(
|
|
f"{relative}:import:{module}"
|
|
for module in sorted(_module_imports(tree) & forbidden_modules)
|
|
)
|
|
referenced = {
|
|
node.id
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, ast.Name) and node.id in forbidden_names
|
|
}
|
|
violations.extend(f"{relative}:name:{name}" for name in sorted(referenced))
|
|
|
|
assert violations == []
|
|
|
|
|
|
def test_plugin_manager_catalog_compat_methods_remain_thin_delegates() -> None:
|
|
"""
|
|
PluginManager 的旧市场目录方法只能委托，不得重新吸收应用目录算法。
|
|
"""
|
|
path = APP_ROOT / "runtime" / "extensions" / "plugin" / "manager.py"
|
|
tree = _parse(path)
|
|
manager = next(
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.ClassDef) and node.name == "PluginManager"
|
|
)
|
|
methods = {
|
|
node.name: node
|
|
for node in manager.body
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
|
|
and node.name in DELEGATED_PLUGIN_CATALOG_METHODS
|
|
}
|
|
|
|
assert set(methods) == DELEGATED_PLUGIN_CATALOG_METHODS
|
|
for method_name, method in methods.items():
|
|
statements = [
|
|
statement
|
|
for statement in method.body
|
|
if not (
|
|
isinstance(statement, ast.Expr)
|
|
and isinstance(statement.value, ast.Constant)
|
|
and isinstance(statement.value.value, str)
|
|
)
|
|
]
|
|
assert len(statements) == 1, method_name
|
|
assert isinstance(statements[0], ast.Return), method_name
|
|
value = statements[0].value
|
|
if isinstance(value, ast.Await):
|
|
value = value.value
|
|
assert isinstance(value, ast.Call), method_name
|
|
assert ast.unparse(value.func).startswith("self._plugin_catalog_view."), method_name
|
|
|
|
|
|
def test_plugin_capability_package_roots_do_not_duplicate_exports() -> None:
|
|
"""
|
|
多文件插件能力包根只保留说明，不得复制子模块实现或宿主导出。
|
|
"""
|
|
violations: list[str] = []
|
|
for package in PLUGIN_COMPONENT_PACKAGES:
|
|
path = package / "__init__.py"
|
|
tree = _parse(path)
|
|
if not ast.get_docstring(tree) or len(tree.body) != 1 or _public_names(tree):
|
|
violations.append(path.relative_to(PROJECT_ROOT).as_posix())
|
|
|
|
assert violations == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("declaration", "expected"),
|
|
[
|
|
('"tool_name"', True),
|
|
('"other_name"', False),
|
|
('Field("tool_name", description="工具名称")', True),
|
|
('Field(default="tool_name", description="工具名称")', True),
|
|
('Field(default="other_name", description="工具名称")', False),
|
|
('Field(description="工具名称")', False),
|
|
('unrelated("tool_name")', False),
|
|
],
|
|
)
|
|
def test_agent_tool_filename_contract_reads_declared_default(tmp_path, monkeypatch, declaration, expected):
|
|
"""
|
|
文件名契约读取常量或 Field 的默认值，不把描述或其他调用误当成工具名称。
|
|
"""
|
|
monkeypatch.setattr(__name__ + ".PROJECT_ROOT", tmp_path)
|
|
path = tmp_path / "app" / "agent" / "tools" / "impl" / "tool_name.py"
|
|
path.parent.mkdir(parents=True)
|
|
path.write_text(f"class Tool:\n name: str = {declaration}\n", encoding="utf-8")
|
|
|
|
assert _matches_filename_contract(
|
|
path,
|
|
{"root": "app/agent/tools/impl", "kind": "agent_tool_contract"},
|
|
) is expected
|
|
|
|
|
|
def test_new_host_production_modules_use_single_word_filenames() -> None:
|
|
"""
|
|
宿主多词文件必须满足稳定发现合同或带有可审查的逐项理由。
|
|
"""
|
|
policy = json.loads(FILENAME_POLICY.read_text(encoding="utf-8"))
|
|
assert policy["schema_version"] == 2
|
|
assert policy["scope"] == {
|
|
"root": "app",
|
|
"excluded": ["app/plugins"],
|
|
"special": ["__init__.py"],
|
|
}
|
|
single_word = re.compile(r"[a-z][a-z0-9]*")
|
|
actual = sorted(
|
|
path
|
|
for path in APP_ROOT.rglob("*.py")
|
|
if not path.is_relative_to(APP_ROOT / "plugins")
|
|
and path.name != "__init__.py"
|
|
and not single_word.fullmatch(path.stem)
|
|
)
|
|
|
|
contracts = policy["semantic_filename_contracts"]
|
|
reviewed = {item["path"]: item for item in policy["reviewed_multiword_files"]}
|
|
assert all(item["reason"].strip() for item in contracts)
|
|
assert all(item["reason"].strip() and item["evidence"].strip() for item in reviewed.values())
|
|
|
|
unmatched = []
|
|
contract_paths = set()
|
|
for path in actual:
|
|
relative = path.relative_to(PROJECT_ROOT).as_posix()
|
|
if any(_matches_filename_contract(path, rule) for rule in contracts):
|
|
contract_paths.add(relative)
|
|
elif relative not in reviewed:
|
|
unmatched.append(relative)
|
|
|
|
assert unmatched == []
|
|
assert set(reviewed) == {
|
|
path.relative_to(PROJECT_ROOT).as_posix() for path in actual
|
|
} - contract_paths
|
|
|
|
|
|
def test_package_roots_are_documentation_facades_or_reasoned_owners() -> None:
|
|
"""
|
|
全局包根只能为空、说明、精确懒门面、模块入口或有理由的既有 owner。
|
|
"""
|
|
policy = json.loads(FILENAME_POLICY.read_text(encoding="utf-8"))
|
|
exceptions = {item["path"]: item for item in policy["package_root_exceptions"]}
|
|
assert all(
|
|
item["mode"] in {"bootstrap", "implementation", "public_facade", "contained_vendor"}
|
|
and item["reason"].strip()
|
|
for item in exceptions.values()
|
|
)
|
|
|
|
used_exceptions: set[str] = set()
|
|
violations: list[str] = []
|
|
modules_root = APP_ROOT / "modules"
|
|
for path in APP_ROOT.rglob("__init__.py"):
|
|
if path.is_relative_to(APP_ROOT / "plugins"):
|
|
continue
|
|
relative = path.relative_to(PROJECT_ROOT).as_posix()
|
|
if relative in exceptions:
|
|
used_exceptions.add(relative)
|
|
continue
|
|
tree = _parse(path)
|
|
if _is_documentation_only_root(tree) or _is_curated_lazy_facade(tree):
|
|
continue
|
|
if path.parent.parent == modules_root and any(
|
|
isinstance(node, ast.ClassDef) and node.name.endswith("Module")
|
|
for node in tree.body
|
|
):
|
|
continue
|
|
violations.append(relative)
|
|
|
|
assert violations == []
|
|
assert used_exceptions == set(exceptions)
|