mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-12 08:40:27 +08:00
240 lines
9.6 KiB
Python
240 lines
9.6 KiB
Python
"""
|
||
慢插件钩子的准入拒绝、停机预算和真实资源所有权回归测试。
|
||
"""
|
||
|
||
import asyncio
|
||
import threading
|
||
import time
|
||
from collections.abc import Iterator
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from concurrent.futures import TimeoutError as FutureTimeoutError
|
||
from contextlib import contextmanager
|
||
from types import SimpleNamespace
|
||
from typing import Optional
|
||
from unittest.mock import MagicMock
|
||
|
||
import pytest
|
||
|
||
from app.runtime.extensions.plugin.admission import PluginMutationAdmission
|
||
from app.runtime.extensions.plugin.lifecycle import PluginLifecycle
|
||
from app.runtime.extensions.plugin.manager import PluginManager
|
||
from app.schemas.exception import PluginMutationRejectedError
|
||
|
||
|
||
@contextmanager
|
||
def _blocked_plugin(
|
||
manager: Optional[PluginManager] = None, *, start: bool = True,
|
||
) -> Iterator[SimpleNamespace]:
|
||
"""
|
||
在线程中阻塞真实 close 钩子，退出时始终放行并回收线程。
|
||
"""
|
||
entered = threading.Event()
|
||
release = threading.Event()
|
||
database = MagicMock()
|
||
calls = []
|
||
|
||
class SlowPlugin:
|
||
"""
|
||
以可控事件模拟未返回的第三方资源关闭钩子。
|
||
"""
|
||
|
||
def close(self) -> None:
|
||
"""
|
||
等待测试释放；兜底预算防止旧实现阻塞整个测试进程。
|
||
"""
|
||
entered.set()
|
||
calls.append("close")
|
||
assert release.wait(timeout=3)
|
||
|
||
instance = SlowPlugin()
|
||
running = {"SlowPlugin": instance}
|
||
lifecycle = PluginLifecycle(
|
||
classes={"SlowPlugin": SlowPlugin}, running=running,
|
||
load_plugins=lambda *_args: [], loadable_plugins=lambda: [],
|
||
plugin_config=lambda _pid: {}, auth_checker=lambda _plugin: True,
|
||
clear_modules=MagicMock(), clear_tools=MagicMock(),
|
||
enable_events=MagicMock(), disable_events=MagicMock(),
|
||
runtime_status_writer=MagicMock(), runtime_compatible=lambda _pid: True,
|
||
database=lambda: database, log=MagicMock(), event_sender=MagicMock(),
|
||
)
|
||
if manager is not None:
|
||
manager._plugin_lifecycle = lifecycle
|
||
manager._running_plugins = running
|
||
with ThreadPoolExecutor(max_workers=2) as executor:
|
||
stop = manager.stop if manager is not None else lifecycle.stop
|
||
future = executor.submit(stop, "SlowPlugin") if start else None
|
||
try:
|
||
if start:
|
||
assert entered.wait(timeout=1)
|
||
yield SimpleNamespace(
|
||
lifecycle=lifecycle, executor=executor, future=future,
|
||
running=running, instance=instance, database=database,
|
||
entered=entered, release=release, calls=calls,
|
||
)
|
||
finally:
|
||
release.set()
|
||
if future is not None:
|
||
future.result(timeout=1)
|
||
|
||
|
||
def _manager() -> PluginManager:
|
||
"""
|
||
仅装配停机与准入所需字段，避免触碰单例和真实插件持久化服务。
|
||
"""
|
||
manager = object.__new__(PluginManager)
|
||
manager._plugin_quiesce_lock = threading.RLock()
|
||
manager._plugin_quiesce_future = None
|
||
manager._plugin_service_quiesce_future = None
|
||
manager._plugin_runtime_closed = False
|
||
manager._plugin_mutation_admission = PluginMutationAdmission(
|
||
activity=lambda: manager._plugin_lifecycle.activity(),
|
||
)
|
||
return manager
|
||
|
||
|
||
def test_slow_hook_rejects_other_lifecycle_without_waiting() -> None:
|
||
"""
|
||
A 的钩子未退出时，B 必须明确拒绝执行且不能等到 A 释放。
|
||
"""
|
||
with _blocked_plugin() as blocked:
|
||
contender = blocked.executor.submit(blocked.lifecycle.start, "OtherPlugin")
|
||
with pytest.raises(PluginMutationRejectedError, match="SlowPlugin.*close"):
|
||
# 旧实现会在全局锁内等到此 Future 超时，因而使本断言失败。
|
||
try:
|
||
contender.result(timeout=0.3)
|
||
except FutureTimeoutError:
|
||
pytest.fail("慢钩子使其他插件的生命周期操作一直等待全局锁")
|
||
assert blocked.running["SlowPlugin"] is blocked.instance
|
||
blocked.database.release.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_shutdown_budget_includes_contended_lifecycle_and_keeps_loop_alive(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""
|
||
停机预算包含准入等待，事件循环可继续调度且资源不能提前释放。
|
||
"""
|
||
manager = _manager()
|
||
with _blocked_plugin(manager) as blocked:
|
||
monkeypatch.setattr(
|
||
"app.runtime.extensions.plugin.manager.ThreadHelper",
|
||
lambda: SimpleNamespace(submit=blocked.executor.submit),
|
||
)
|
||
|
||
heartbeat = asyncio.Event()
|
||
asyncio.get_running_loop().call_later(0.01, heartbeat.set)
|
||
started = time.monotonic()
|
||
result = await manager.quiesce_plugins(timeout=0.05)
|
||
assert time.monotonic() - started < 0.5
|
||
assert result is False
|
||
assert heartbeat.is_set()
|
||
assert blocked.running["SlowPlugin"] is blocked.instance
|
||
assert manager.finalize_plugins() is False
|
||
blocked.database.release.assert_not_called()
|
||
assert blocked.future.done() is False
|
||
|
||
|
||
def test_busy_mutation_rejects_before_side_effects_and_preserves_owner() -> None:
|
||
"""
|
||
根事务在写入前快速拒绝，副作用、owner 和数据库均保持真实状态。
|
||
"""
|
||
manager = _manager()
|
||
changed = []
|
||
with _blocked_plugin(manager) as blocked:
|
||
with pytest.raises(PluginMutationRejectedError, match="SlowPlugin.*close"):
|
||
with manager.mutation("保存 OtherPlugin 配置"):
|
||
changed.append(True)
|
||
assert changed == []
|
||
assert manager._plugin_mutation_admission.active_count == 1
|
||
assert manager._plugin_mutation_admission.wait_until_idle(timeout=0.01) is False
|
||
activity = manager.get_plugin_lifecycle_activity()
|
||
assert activity.plugin_id == "SlowPlugin"
|
||
assert activity.hook == "close"
|
||
assert activity.elapsed_seconds >= activity.hook_elapsed_seconds > 0
|
||
assert blocked.future.done() is False
|
||
assert manager._plugin_mutation_admission.wait_until_idle(timeout=0) is True
|
||
assert manager.get_plugin_lifecycle_activity() is None
|
||
with manager.mutation("重试保存 OtherPlugin 配置"):
|
||
changed.append(True)
|
||
assert changed == [True]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cancelled_shutdown_keeps_real_hook_owner_and_reuses_future(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""
|
||
调用方取消或再次超时都不能重复执行钩子、释放实例或销毁数据库。
|
||
"""
|
||
manager = _manager()
|
||
with _blocked_plugin(manager, start=False) as blocked:
|
||
monkeypatch.setattr(
|
||
"app.runtime.extensions.plugin.manager.ThreadHelper",
|
||
lambda: SimpleNamespace(submit=blocked.executor.submit),
|
||
)
|
||
assert await manager.quiesce_plugins(timeout=1) is True
|
||
task = asyncio.create_task(manager.quiesce_plugin_services(timeout=1))
|
||
try:
|
||
assert await asyncio.to_thread(blocked.entered.wait, 1)
|
||
owner = manager._plugin_service_quiesce_future
|
||
task.cancel()
|
||
with pytest.raises(asyncio.CancelledError):
|
||
await task
|
||
assert owner is not None and not owner.done()
|
||
assert await manager.quiesce_plugin_services(timeout=0.01) is False
|
||
assert manager._plugin_service_quiesce_future is owner
|
||
assert blocked.calls == ["close"]
|
||
assert manager.finalize_plugins() is False
|
||
assert manager.reopen_plugins() is False
|
||
assert blocked.running["SlowPlugin"] is blocked.instance
|
||
blocked.database.release.assert_not_called()
|
||
activity = manager.get_plugin_lifecycle_activity()
|
||
assert activity.operation == "quiesce_services"
|
||
assert activity.plugin_id == "SlowPlugin"
|
||
assert activity.hook == "close"
|
||
blocked.release.set()
|
||
assert await asyncio.wrap_future(owner) is True
|
||
assert manager.finalize_plugins() is True
|
||
assert blocked.running == {}
|
||
blocked.database.release.assert_called_once_with("SlowPlugin")
|
||
finally:
|
||
blocked.release.set()
|
||
if not task.done():
|
||
await task
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_runtime_summary_exposes_real_hook_and_does_not_report_ready(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""
|
||
摘要在后台收敛标记为空闲时，仍能识别实际执行中的插件钩子。
|
||
"""
|
||
from app.api.endpoints import plugin as endpoint
|
||
from app.schemas.plugin import PluginRuntimeStatus
|
||
|
||
with _blocked_plugin() as blocked:
|
||
manager = SimpleNamespace(
|
||
get_plugin_runtime_statuses=lambda: {"SlowPlugin": PluginRuntimeStatus.READY},
|
||
get_plugin_restart_requirements=lambda: {},
|
||
get_plugin_gil_fallbacks=lambda: [],
|
||
get_plugin_runtime_generation=lambda: 3,
|
||
is_plugin_settling=lambda: False,
|
||
get_plugin_lifecycle_activity=blocked.lifecycle.activity,
|
||
)
|
||
monkeypatch.setattr(endpoint, "get_plugin_manager", lambda: manager)
|
||
monkeypatch.setattr(
|
||
endpoint, "get_configured_system_config",
|
||
lambda: SimpleNamespace(get=lambda _key: ["SlowPlugin"]),
|
||
)
|
||
summary = await endpoint.runtime_status(None)
|
||
assert summary.ready is False
|
||
assert summary.pending_count == 1
|
||
assert summary.active_operation.plugin_id == "SlowPlugin"
|
||
assert summary.active_operation.hook == "close"
|
||
assert summary.model_dump()["active_operation"]["elapsed_seconds"] >= 0
|
||
summary = await endpoint.runtime_status(None)
|
||
assert summary.ready is True
|
||
assert summary.active_operation is None
|