mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 19:37:54 +08:00
411 lines
18 KiB
Python
411 lines
18 KiB
Python
"""
|
||
插件依赖安装同步、异步入口的恢复决策和资源清理回归。
|
||
"""
|
||
|
||
import asyncio
|
||
import threading
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from app.adapters.system.package import PackageInstallRequest, PackageInstallStrategy
|
||
from app.adapters.system.plugin import health as health_module
|
||
from app.adapters.system.plugin.health import PluginRuntimeHealth
|
||
|
||
_HEALTHY = {"核心依赖导入检查": (True, "ok")}
|
||
_BROKEN = {"核心依赖导入检查": (False, "new failure")}
|
||
_OLD_FAILURE = {"核心依赖导入检查": (False, "old failure")}
|
||
_REGRESSION = "核心依赖导入检查失败：new failure"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _RecoveryCase:
|
||
"""
|
||
一次安装在外部边界的固定应答与预期结果。
|
||
|
||
:param installs: 各安装策略返回的成功标志和输出
|
||
:param health: 安装前、安装后及修复后的有序健康快照
|
||
:param repairs: 各修复策略返回的成功标志和输出
|
||
:param result: 对外返回的完整成功标志和说明
|
||
:param events: 必须保持的安装、探测、修复和导入刷新顺序
|
||
"""
|
||
|
||
installs: tuple[tuple[bool, str], ...]
|
||
health: tuple[dict[str, tuple[bool, str]], ...]
|
||
repairs: tuple[tuple[bool, str], ...]
|
||
result: tuple[bool, str]
|
||
events: tuple[str, ...]
|
||
|
||
|
||
_CASES = [
|
||
pytest.param(_RecoveryCase(
|
||
((True, "installed"),), (_HEALTHY, _HEALTHY), (), (True, "installed"),
|
||
("health", "install", "health", "refresh"),
|
||
), id="healthy"),
|
||
pytest.param(_RecoveryCase(
|
||
((True, "installed"),), (_OLD_FAILURE, _OLD_FAILURE), (), (True, "installed"),
|
||
("health", "install", "health", "refresh"),
|
||
), id="existing-failure"),
|
||
pytest.param(_RecoveryCase(
|
||
((False, "offline"), (True, "installed")), (_HEALTHY, _HEALTHY, _HEALTHY), (),
|
||
(True, "installed"), ("health", "install", "health", "install", "health", "refresh"),
|
||
), id="fallback"),
|
||
pytest.param(_RecoveryCase(
|
||
((False, "offline"), (False, "still offline")), (_HEALTHY, _HEALTHY, _HEALTHY), (),
|
||
(False, "[UV] 所有策略均安装依赖失败：still offline"),
|
||
("health", "install", "health", "install", "health"),
|
||
), id="exhausted"),
|
||
pytest.param(_RecoveryCase(
|
||
((False, ""), (False, "")), (_HEALTHY, _HEALTHY, _HEALTHY), (),
|
||
(False, "[UV] 所有策略均安装依赖失败，请检查网络连接、包源配置或插件依赖约束"),
|
||
("health", "install", "health", "install", "health"),
|
||
), id="empty-errors"),
|
||
pytest.param(_RecoveryCase(
|
||
((True, "installed"),), (_HEALTHY, _BROKEN, _HEALTHY), ((True, "repaired"),),
|
||
(False, f"依赖安装后运行环境自检失败，已自动恢复主程序依赖：{_REGRESSION}"),
|
||
("health", "install", "health", "repair", "refresh", "health", "refresh"),
|
||
), id="successful-install-restored"),
|
||
pytest.param(_RecoveryCase(
|
||
((True, "installed"),), (_HEALTHY, _BROKEN), ((False, "repair failed"),),
|
||
(False, "依赖安装后运行环境自检失败，且自动恢复主程序依赖失败：repair failed"),
|
||
("health", "install", "health", "repair"),
|
||
), id="successful-install-repair-failed"),
|
||
pytest.param(_RecoveryCase(
|
||
((True, "installed"),), (_HEALTHY, _BROKEN, _BROKEN), ((True, "repaired"),),
|
||
(False, f"依赖安装后运行环境自检失败，恢复主程序依赖后仍异常：{_REGRESSION}"),
|
||
("health", "install", "health", "repair", "refresh", "health"),
|
||
), id="successful-install-still-broken"),
|
||
pytest.param(_RecoveryCase(
|
||
((False, "install failed"),), (_HEALTHY, _BROKEN, _HEALTHY), ((True, "repaired"),),
|
||
(False, "策略 install-0 安装依赖失败：install failed；主运行环境已恢复"),
|
||
("health", "install", "health", "repair", "refresh", "health"),
|
||
), id="failed-install-restored-stops-fallback"),
|
||
pytest.param(_RecoveryCase(
|
||
((False, "install failed"),), (_HEALTHY, _BROKEN), ((False, "repair failed"),),
|
||
(False, f"策略 install-0 安装依赖失败：install failed；插件依赖安装失败后主运行环境异常，且恢复失败：{_REGRESSION}; repair failed"),
|
||
("health", "install", "health", "repair"),
|
||
), id="failed-install-repair-failed"),
|
||
pytest.param(_RecoveryCase(
|
||
((False, "install failed"),), (_HEALTHY, _BROKEN, _BROKEN), ((True, "repaired"),),
|
||
(False, f"策略 install-0 安装依赖失败：install failed；插件依赖安装失败后主运行环境异常，恢复后仍异常：{_REGRESSION}"),
|
||
("health", "install", "health", "repair", "refresh", "health"),
|
||
), id="failed-install-still-broken"),
|
||
]
|
||
|
||
|
||
@pytest.fixture
|
||
def recovery_environment(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||
"""
|
||
隔离依赖清单、约束文件、安装锁及子进程，保留真实安装与恢复编排。
|
||
|
||
:param tmp_path: 本用例独占的临时目录
|
||
:param monkeypatch: 在用例结束时撤销边界替换的夹具
|
||
:return: 清单路径、约束路径、独占安装锁以及捕获的日志
|
||
"""
|
||
manifest = tmp_path / "requirements.txt"
|
||
manifest.write_text("demo\n", encoding="utf-8")
|
||
constraints = tmp_path / "constraints.txt"
|
||
lock = threading.Lock()
|
||
logs: list[tuple[str, str]] = []
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_package_install_lock", lock)
|
||
monkeypatch.setattr(health_module, "is_free_threaded_runtime", lambda: False)
|
||
monkeypatch.setattr(health_module, "get_runtime_setting", lambda key: {
|
||
"ROOT_PATH": tmp_path, "CONFIG_PATH": tmp_path, "TEMP_PATH": tmp_path,
|
||
"PACKAGE_CACHE_PATH": tmp_path, "PIP_PROXY": "", "PROXY_HOST": "",
|
||
}[key])
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__get_installed_packages", lambda: {})
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__get_protected_runtime_packages", lambda _: {"fastapi": "1"})
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__validate_runtime_dependency_conflicts", lambda *_: (True, ""))
|
||
|
||
def create_constraints(_packages):
|
||
"""
|
||
在临时目录生成约束，检查后续路径是否承担清理责任。
|
||
"""
|
||
constraints.write_text("fastapi==1\n", encoding="utf-8")
|
||
return constraints
|
||
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__create_runtime_constraints_file", create_constraints)
|
||
monkeypatch.setattr(health_module.logger, "warning", lambda message: logs.append(("warning", message)))
|
||
monkeypatch.setattr(health_module.logger, "error", lambda message: logs.append(("error", message)))
|
||
return manifest, constraints, lock, logs
|
||
|
||
|
||
def _patch_runtime(monkeypatch, environment, case: _RecoveryCase) -> list[str]:
|
||
"""
|
||
在子进程和健康探测边界重放固定应答，并记录持锁期间的操作顺序。
|
||
|
||
:param monkeypatch: 自动恢复边界方法的夹具
|
||
:param environment: 隔离的清单、约束、锁与日志
|
||
:param case: 完整应答序列及预期结果
|
||
:return: 按真实发生顺序追加的操作列表
|
||
"""
|
||
_, constraints, lock, _ = environment
|
||
events: list[str] = []
|
||
installs, repairs, health = iter(case.installs), iter(case.repairs), iter(case.health)
|
||
|
||
def strategies(request: PackageInstallRequest, dry_run: bool = False) -> list[PackageInstallStrategy]:
|
||
"""
|
||
返回离线命令，保持真实修复目标和安装策略切换可观察。
|
||
"""
|
||
assert not dry_run
|
||
kind = "repair" if request.purpose == "runtime-repair" else "install"
|
||
if kind == "repair":
|
||
assert request.dependency_files == (constraints,)
|
||
else:
|
||
assert request.constraints_file == constraints
|
||
return [PackageInstallStrategy(f"{kind}-{index}", [kind], {}, [kind])
|
||
for index in range(1 if kind == "repair" else 2)]
|
||
|
||
def execute(command, **kwargs):
|
||
"""
|
||
检查安装与恢复持有同一锁，禁止启动真实子进程。
|
||
"""
|
||
assert lock.locked() and constraints.exists()
|
||
assert kwargs.get("timeout") in (None, 300)
|
||
kind = command[0]
|
||
events.append(kind)
|
||
return next(repairs if kind == "repair" else installs)
|
||
|
||
async def execute_async(command, **kwargs):
|
||
"""
|
||
复用边界应答，同时要求异步执行保留原有超时。
|
||
"""
|
||
assert kwargs["timeout"] == 300
|
||
return execute(command, **kwargs)
|
||
|
||
def probe():
|
||
"""
|
||
重放健康快照，若多探测或少探测则由序列与事件断言识别。
|
||
"""
|
||
assert lock.locked()
|
||
events.append("health")
|
||
return next(health)
|
||
|
||
async def probe_async():
|
||
"""
|
||
通过相同快照验证异步恢复决策。
|
||
"""
|
||
return probe()
|
||
|
||
monkeypatch.setattr(health_module, "build_package_install_strategies", strategies)
|
||
monkeypatch.setattr(health_module.SystemUtils, "execute_with_subprocess", execute)
|
||
monkeypatch.setattr(health_module.SystemUtils, "execute_with_subprocess_async", execute_async)
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__run_runtime_healthcheck", probe)
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__async_run_runtime_healthcheck", probe_async)
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__refresh_import_system", lambda: events.append("refresh"))
|
||
return events
|
||
|
||
|
||
@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"])
|
||
@pytest.mark.parametrize("case", _CASES)
|
||
def test_install_recovery_contract(recovery_environment, monkeypatch, case, use_async) -> None:
|
||
"""
|
||
两种入口保持既有结果、恢复顺序、停止条件及诊断差异。
|
||
|
||
:param recovery_environment: 隔离的安装环境
|
||
:param monkeypatch: 自动恢复子进程和健康检查的夹具
|
||
:param case: 参数化的外部应答及预期合同
|
||
:param use_async: 是否通过异步公开入口执行
|
||
"""
|
||
manifest, constraints, lock, logs = recovery_environment
|
||
events = _patch_runtime(monkeypatch, recovery_environment, case)
|
||
if use_async:
|
||
result = asyncio.run(PluginRuntimeHealth().async_install_packages_with_fallback(manifest))
|
||
else:
|
||
result = PluginRuntimeHealth.install_packages_with_fallback(manifest)
|
||
assert result == case.result
|
||
assert tuple(events) == case.events
|
||
assert not constraints.exists()
|
||
assert not lock.locked()
|
||
remaining = [message for _, message in logs if "安装前已有的运行环境异常仍然存在" in message]
|
||
assert bool(remaining) == (not use_async and case.health == (_OLD_FAILURE, _OLD_FAILURE))
|
||
restored_errors = [message for _, message in logs if "主程序依赖恢复后仍未通过健康检查" in message]
|
||
assert bool(restored_errors) == (not use_async and case.installs[0][0] and case.health[-1] == _BROKEN and bool(case.repairs) and case.repairs[0][0])
|
||
|
||
|
||
@pytest.mark.parametrize("stage", ["baseline", "install", "post-install", "repair", "post-repair"])
|
||
@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"])
|
||
def test_install_exception_releases_resources(recovery_environment, monkeypatch, stage, use_async) -> None:
|
||
"""
|
||
探测、安装和恢复异常按原样传播，始终释放约束文件与安装锁。
|
||
|
||
:param recovery_environment: 隔离的安装环境
|
||
:param monkeypatch: 自动恢复边界方法的夹具
|
||
:param stage: 注入异常的外部操作阶段
|
||
:param use_async: 是否执行异步公开入口
|
||
"""
|
||
manifest, constraints, lock, _ = recovery_environment
|
||
case = _RecoveryCase(((True, "ok"),), (_HEALTHY, _BROKEN, _HEALTHY),
|
||
((True, "ok"),), (False, ""), ())
|
||
_patch_runtime(monkeypatch, recovery_environment, case)
|
||
failure = RuntimeError(f"{stage} failed")
|
||
probe = PluginRuntimeHealth._PluginRuntimeHealth__run_runtime_healthcheck
|
||
execute = health_module.SystemUtils.execute_with_subprocess
|
||
phases = iter(("baseline", "post-install", "post-repair"))
|
||
|
||
def failing_probe():
|
||
"""
|
||
在指定的健康探测阶段注入同一异常对象。
|
||
"""
|
||
if next(phases) == stage:
|
||
raise failure
|
||
return probe()
|
||
|
||
def failing_execute(command, **kwargs):
|
||
"""
|
||
在安装或修复子进程边界注入异常。
|
||
"""
|
||
if command[0] == stage:
|
||
raise failure
|
||
return execute(command, **kwargs)
|
||
|
||
async def failing_probe_async():
|
||
"""
|
||
保持异步探测入口的异常传播行为。
|
||
"""
|
||
return failing_probe()
|
||
|
||
async def failing_execute_async(command, **kwargs):
|
||
"""
|
||
保持异步子进程入口的异常传播行为。
|
||
"""
|
||
return failing_execute(command, **kwargs)
|
||
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__run_runtime_healthcheck", failing_probe)
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__async_run_runtime_healthcheck", failing_probe_async)
|
||
monkeypatch.setattr(health_module.SystemUtils, "execute_with_subprocess", failing_execute)
|
||
monkeypatch.setattr(health_module.SystemUtils, "execute_with_subprocess_async", failing_execute_async)
|
||
with pytest.raises(RuntimeError) as raised:
|
||
if use_async:
|
||
asyncio.run(PluginRuntimeHealth().async_install_packages_with_fallback(manifest))
|
||
else:
|
||
PluginRuntimeHealth.install_packages_with_fallback(manifest)
|
||
assert raised.value is failure
|
||
assert not constraints.exists()
|
||
assert not lock.locked()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize("stage", ["baseline", "install", "post-install", "repair", "post-repair"])
|
||
async def test_cancelled_recovery_releases_resources(recovery_environment, monkeypatch, stage) -> None:
|
||
"""
|
||
在真实等待点取消安装或恢复，取消不得转换成失败结果或继续下一策略。
|
||
|
||
:param recovery_environment: 隔离的安装环境
|
||
:param monkeypatch: 自动恢复子进程和探测的夹具
|
||
:param stage: 等待取消的外部操作阶段
|
||
"""
|
||
manifest, constraints, lock, _ = recovery_environment
|
||
case = _RecoveryCase(((True, "ok"),), (_HEALTHY, _BROKEN, _HEALTHY),
|
||
((True, "ok"),), (False, ""), ())
|
||
events = _patch_runtime(monkeypatch, recovery_environment, case)
|
||
probe = PluginRuntimeHealth._PluginRuntimeHealth__run_runtime_healthcheck
|
||
execute = health_module.SystemUtils.execute_with_subprocess
|
||
phases = iter(("baseline", "post-install", "post-repair"))
|
||
entered = asyncio.Event()
|
||
exited = asyncio.Event()
|
||
|
||
async def wait_for_cancel(phase: str) -> None:
|
||
"""
|
||
使用屏障精确定位取消时点，收尾标记用于验证子操作先退出。
|
||
"""
|
||
if phase == stage:
|
||
entered.set()
|
||
try:
|
||
await asyncio.Future()
|
||
finally:
|
||
exited.set()
|
||
|
||
async def blocked_probe():
|
||
"""
|
||
在目标健康探测之前暂停。
|
||
"""
|
||
await wait_for_cancel(next(phases))
|
||
return probe()
|
||
|
||
async def blocked_execute(command, **kwargs):
|
||
"""
|
||
在目标子进程边界暂停，不启动真实安装器。
|
||
"""
|
||
await wait_for_cancel(command[0])
|
||
return execute(command, **kwargs)
|
||
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__async_run_runtime_healthcheck", blocked_probe)
|
||
monkeypatch.setattr(health_module.SystemUtils, "execute_with_subprocess_async", blocked_execute)
|
||
task = asyncio.create_task(PluginRuntimeHealth().async_install_packages_with_fallback(manifest))
|
||
try:
|
||
await asyncio.wait_for(entered.wait(), timeout=5)
|
||
before_cancel = list(events)
|
||
task.cancel()
|
||
with pytest.raises(asyncio.CancelledError):
|
||
await asyncio.wait_for(task, timeout=5)
|
||
assert exited.is_set()
|
||
assert events == before_cancel
|
||
assert not constraints.exists()
|
||
assert not lock.locked()
|
||
finally:
|
||
if not task.done():
|
||
task.cancel()
|
||
await asyncio.gather(task, return_exceptions=True)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cancelled_install_waiting_for_lock_preserves_owner(recovery_environment, monkeypatch) -> None:
|
||
"""
|
||
排队时取消只清理自己的约束，不能释放其他安装者持有的锁。
|
||
|
||
:param recovery_environment: 隔离的安装环境
|
||
:param monkeypatch: 自动恢复预检边界的夹具
|
||
"""
|
||
manifest, constraints, lock, _ = recovery_environment
|
||
ready = asyncio.Event()
|
||
|
||
async def preflight(_request):
|
||
"""
|
||
在创建约束之后标记就绪，下一步将等待已被占用的锁。
|
||
"""
|
||
ready.set()
|
||
return None
|
||
|
||
monkeypatch.setattr(health_module, "build_package_install_strategies", lambda _request: [])
|
||
monkeypatch.setattr(PluginRuntimeHealth, "_PluginRuntimeHealth__async_free_threaded_preflight", preflight)
|
||
lock.acquire()
|
||
task = asyncio.create_task(PluginRuntimeHealth().async_install_packages_with_fallback(manifest))
|
||
try:
|
||
await asyncio.wait_for(ready.wait(), timeout=5)
|
||
assert constraints.exists()
|
||
task.cancel()
|
||
with pytest.raises(asyncio.CancelledError):
|
||
await asyncio.wait_for(task, timeout=5)
|
||
assert lock.locked()
|
||
assert not constraints.exists()
|
||
finally:
|
||
lock.release()
|
||
if not task.done():
|
||
task.cancel()
|
||
await asyncio.gather(task, return_exceptions=True)
|
||
|
||
|
||
@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"])
|
||
def test_empty_install_strategies_preserve_failure(recovery_environment, monkeypatch, use_async) -> None:
|
||
"""
|
||
无可用策略时保留原有失败说明，不启动安装或恢复。
|
||
|
||
:param recovery_environment: 隔离的安装环境
|
||
:param monkeypatch: 自动恢复策略工厂和边界方法的夹具
|
||
:param use_async: 是否执行异步公开入口
|
||
"""
|
||
manifest, constraints, lock, _ = recovery_environment
|
||
case = _RecoveryCase((), (_HEALTHY,), (), (False, ""), ())
|
||
events = _patch_runtime(monkeypatch, recovery_environment, case)
|
||
monkeypatch.setattr(health_module, "build_package_install_strategies", lambda _request: [])
|
||
if use_async:
|
||
result = asyncio.run(PluginRuntimeHealth().async_install_packages_with_fallback(manifest))
|
||
else:
|
||
result = PluginRuntimeHealth.install_packages_with_fallback(manifest)
|
||
assert result == (False, "[UV] 所有策略均安装依赖失败，请检查网络连接、包源配置或插件依赖约束")
|
||
assert events == ["health"]
|
||
assert not constraints.exists()
|
||
assert not lock.locked()
|