mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-08-29 03:56:43 +08:00
jxxghp
4cc06ce7ad
该用例以独立子进程加载 app.monitor 实现，链路最终 import app.application.site.sites；CI 无该动态资源模块且子进程不经过 conftest 引导，导致 ModuleNotFoundError。改为触发实现加载前显式 安装与 conftest 同源的 sites 垫片，并通过 CONFIG_DIR 指向临时 目录隔离子进程配置。
255 lines
8.9 KiB
Python
255 lines
8.9 KiB
Python
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from app.schemas.types import ChainEventType, EventType
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).parents[1]
|
|
BASELINE_ROOT = PROJECT_ROOT / "tests" / "fixtures" / "architecture"
|
|
|
|
|
|
def test_architecture_contract_baselines_match_current_source():
|
|
"""宿主依赖图和公开运行契约变化必须显式刷新基线。"""
|
|
result = subprocess.run(
|
|
[sys.executable, "scripts/architecture/baseline.py", "--check"],
|
|
cwd=PROJECT_ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
|
|
|
|
def test_official_plugin_baseline_records_external_source():
|
|
"""官方插件快照必须绑定独立仓提交，且不得引用宿主插件副本。"""
|
|
baseline_path = BASELINE_ROOT / "official-plugin-baseline.json"
|
|
baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
|
|
|
|
assert baseline["source"]["repository"] == "MoviePilot-Plugins"
|
|
assert len(baseline["source"]["head"]) == 40
|
|
assert baseline["source"]["roots"] == ["plugins.v2", "plugins.v3"]
|
|
assert all(
|
|
not path.startswith("app/plugins/")
|
|
for contract in (*baseline["imports"].values(), *baseline["hooks"].values())
|
|
for path in contract["files"]
|
|
)
|
|
assert all(
|
|
not path.startswith("app/plugins/")
|
|
for path in baseline["api_routes"]
|
|
)
|
|
|
|
|
|
def test_official_discovery_plugins_explicitly_keep_host_page_envelope():
|
|
"""宿主探索页消费的官方插件 API 不得依赖动态路由隐式包装。"""
|
|
baseline_path = BASELINE_ROOT / "official-plugin-baseline.json"
|
|
baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
|
|
routes = {
|
|
(path, route["path"]): route
|
|
for path, file_routes in baseline["api_routes"].items()
|
|
for route in file_routes
|
|
}
|
|
|
|
expected_paths = {
|
|
("plugins.v3/imdbsource/__init__.py", "/imdb-discover"),
|
|
("plugins.v3/imdbsource/__init__.py", "/imdb-top-250"),
|
|
("plugins.v3/imdbsource/__init__.py", "/imdb-trending"),
|
|
("plugins.v3/imdbsource/__init__.py", "/trending"),
|
|
("plugins.v3/tvdbdiscover/__init__.py", "/tvdb_discover"),
|
|
}
|
|
for route_key in expected_paths:
|
|
route = routes[route_key]
|
|
assert route["response_model"] == "schemas.Response[List[schemas.MediaInfo]]"
|
|
assert route["endpoint_return"] == "schemas.Response[List[schemas.MediaInfo]]"
|
|
|
|
|
|
def test_startup_performance_baseline_records_all_cold_import_targets():
|
|
"""启动性能基线必须包含关键入口的可比较冷导入采样。"""
|
|
baseline_path = BASELINE_ROOT / "startup-performance-baseline.json"
|
|
baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
|
|
|
|
assert baseline["repeat"] >= 3
|
|
assert set(baseline["targets"]) == {
|
|
"app.startup.lifecycle",
|
|
"app.factory",
|
|
"app.main",
|
|
}
|
|
for contract in baseline["targets"].values():
|
|
assert len(contract["samples_ms"]) == baseline["repeat"]
|
|
assert contract["min_ms"] <= contract["median_ms"] <= contract["max_ms"]
|
|
assert contract["loaded_module_count"] > 0
|
|
|
|
|
|
def test_startup_performance_baseline_records_normal_and_safe_lifecycle_resources():
|
|
"""非功能基线必须同时记录正常/安全模式和隔离资源增量。"""
|
|
baseline_path = BASELINE_ROOT / "startup-performance-baseline.json"
|
|
baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
|
|
lifecycle = baseline["lifecycle"]
|
|
|
|
assert "no-op" in lifecycle["scope"]
|
|
assert set(lifecycle["modes"]) == {"normal", "safe"}
|
|
normal = lifecycle["modes"]["normal"]
|
|
safe = lifecycle["modes"]["safe"]
|
|
assert normal["enabled_component_count"] > safe["enabled_component_count"]
|
|
for mode in (normal, safe):
|
|
assert len(mode["samples"]) == baseline["repeat"]
|
|
for sample in mode["samples"]:
|
|
assert sample["threads_after"] == sample["threads_before"]
|
|
assert sample["tasks_after"] == sample["tasks_before"]
|
|
assert sample["database_connections_started"] == 0
|
|
assert sample["stage_ms"]
|
|
|
|
|
|
def test_schema_export_manifest_matches_current_modules():
|
|
"""Schema 公开符号或冲突来源变化必须显式刷新生成清单。"""
|
|
result = subprocess.run(
|
|
[sys.executable, "scripts/schema/exports.py", "--check"],
|
|
cwd=PROJECT_ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
|
|
|
|
def test_schema_root_import_does_not_eagerly_load_schema_graph():
|
|
"""仅导入 schema 根包时不得加载任一业务 schema 子模块。"""
|
|
script = """
|
|
import sys
|
|
import app.schemas
|
|
|
|
loaded = sorted(
|
|
name
|
|
for name in sys.modules
|
|
if name.startswith('app.schemas.') and name != 'app.schemas.exports'
|
|
)
|
|
assert not loaded, loaded
|
|
assert len(app.schemas.__all__) >= 400
|
|
"""
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
cwd=PROJECT_ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
|
|
|
|
def test_agent_policy_root_import_does_not_eagerly_load_policy_submodules():
|
|
"""策略公共入口惰性导出，避免 orchestrator、registry 和 sanitizer 形成导入环。"""
|
|
script = """
|
|
import sys
|
|
import app.agent.policy
|
|
|
|
assert not any(
|
|
name.startswith('app.agent.policy.')
|
|
for name in sys.modules
|
|
)
|
|
from app.agent.policy import ToolOrigin, sanitize_for_host
|
|
|
|
assert ToolOrigin.AGENT_API.value == 'agent_api'
|
|
assert sanitize_for_host({'token': 'secret'}) == {'token': '***'}
|
|
"""
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
cwd=PROJECT_ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
|
|
|
|
def test_doctor_and_monitor_roots_are_lazy_identity_preserving_facades(tmp_path):
|
|
"""诊断和监控包根不得预载实现，旧路径仍需返回同一公开对象。"""
|
|
script = """
|
|
import sys
|
|
import app.doctor
|
|
import app.monitor
|
|
|
|
assert not any(name.startswith('app.doctor.') for name in sys.modules)
|
|
assert not any(name.startswith('app.monitor.') for name in sys.modules)
|
|
|
|
# CI 无 app.application.site.sites 资源模块，触发实现加载前先补 conftest 同源垫片；
|
|
# 独立子进程不经过 pytest 引导，必须在此显式安装，否则链式 import 会因缺模块失败。
|
|
from app.testing.bootstrap import ensure_sites_stub
|
|
ensure_sites_stub()
|
|
|
|
from app.doctor import DoctorRunner, run_doctor
|
|
from app.doctor.runner import DoctorRunner as DirectDoctorRunner
|
|
from app.monitor import LocalDirectoryWatcher, Monitor
|
|
from app.monitor.monitor import Monitor as DirectMonitor
|
|
from app.monitor.watcher import LocalDirectoryWatcher as DirectWatcher
|
|
|
|
assert DoctorRunner is DirectDoctorRunner
|
|
assert Monitor is DirectMonitor
|
|
assert LocalDirectoryWatcher is DirectWatcher
|
|
assert callable(run_doctor)
|
|
"""
|
|
env = {**os.environ, "CONFIG_DIR": str(tmp_path / "config")}
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
cwd=PROJECT_ROOT,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
|
|
|
|
def test_split_host_module_roots_keep_manifest_entrypoint_identity():
|
|
"""迁入 module.py 的宿主模块须保持 manifest 包级入口和历史反射路径。"""
|
|
script = """
|
|
from importlib import import_module
|
|
import sys
|
|
|
|
contracts = (
|
|
('app.modules.qqbot', 'QQBotModule'),
|
|
('app.modules.telegram', 'TelegramModule'),
|
|
('app.modules.trimemedia', 'TrimeMediaModule'),
|
|
('app.modules.ugreen', 'UgreenModule'),
|
|
)
|
|
for package_name, symbol_name in contracts:
|
|
package = import_module(package_name)
|
|
implementation_name = f'{package_name}.module'
|
|
assert implementation_name not in sys.modules
|
|
public_class = getattr(package, symbol_name)
|
|
direct_class = getattr(import_module(implementation_name), symbol_name)
|
|
assert public_class is direct_class
|
|
assert public_class.__module__ == package_name
|
|
"""
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
cwd=PROJECT_ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0, result.stderr
|
|
|
|
|
|
def test_event_contract_baseline_covers_every_public_event_enum() -> None:
|
|
"""事件生产者/消费者快照必须覆盖全部广播和链式事件枚举。"""
|
|
baseline_path = BASELINE_ROOT / "runtime-contract-baseline.json"
|
|
baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
|
|
events = baseline["events"]
|
|
expected = {
|
|
*(f"EventType.{member.name}" for member in EventType),
|
|
*(f"ChainEventType.{member.name}" for member in ChainEventType),
|
|
}
|
|
|
|
assert set(events["events"]) == expected
|
|
assert events["event_count"] == len(expected)
|
|
assert events["producer_count"] > 0
|
|
assert events["consumer_count"] > 0
|