mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
510 lines
15 KiB
Python
510 lines
15 KiB
Python
"""
|
||
Managed Resource 与 Capability Runtime 的集成合同测试。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from pathlib import Path
|
||
from types import ModuleType
|
||
from unittest.mock import MagicMock
|
||
|
||
import pytest
|
||
|
||
from app.runtime import resources as managed_resource_facade
|
||
from app.runtime.capabilities.errors import (
|
||
CapabilityOperationError,
|
||
CapabilityRuntimeClosedError,
|
||
)
|
||
from app.runtime.capabilities.runtime import CapabilityRuntime
|
||
from app.runtime.extensions.resource import (
|
||
AsyncManagedResourceAdapter,
|
||
SyncManagedResourceAdapter,
|
||
build_managed_resource_registry,
|
||
)
|
||
from app.runtime.resources import (
|
||
MANAGED_RESOURCE_ASYNC_KIND,
|
||
MANAGED_RESOURCE_SYNC_KIND,
|
||
acquire_managed_resource,
|
||
acquire_managed_resource_async,
|
||
configure_managed_resource_runtime,
|
||
managed_resource_observations,
|
||
managed_resource_snapshot,
|
||
shutdown_managed_resource_runtime,
|
||
)
|
||
|
||
PROJECT_ROOT = Path(__file__).parents[1]
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def isolate_managed_resource_facade(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""
|
||
每个用例使用独立 Runtime，避免不可逆关闭态泄漏到后续测试。
|
||
"""
|
||
monkeypatch.setattr(
|
||
managed_resource_facade,
|
||
"_managed_resource_runtime",
|
||
None,
|
||
)
|
||
|
||
|
||
def _write_manifest(
|
||
root: Path, *, capability_id: str, kind: str, entrypoint: str
|
||
) -> None:
|
||
"""
|
||
写入一个最小 on-first-use 托管资源声明。
|
||
"""
|
||
resource_dir = root / capability_id.replace(".", "_")
|
||
resource_dir.mkdir(parents=True)
|
||
(resource_dir / "capability.toml").write_text(
|
||
"\n".join(
|
||
(
|
||
"schema_version = 1",
|
||
f'id = "{capability_id}"',
|
||
f'kind = "{kind}"',
|
||
f'entrypoint = "{entrypoint}"',
|
||
"depends_on = []",
|
||
"",
|
||
"[metadata]",
|
||
f'name = "{capability_id}"',
|
||
"",
|
||
"[activation]",
|
||
'policy = "on_first_use"',
|
||
"watch = []",
|
||
"",
|
||
)
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
def _runtime(root: Path) -> CapabilityRuntime:
|
||
"""
|
||
构造同时支持同步与异步资源的测试 Runtime。
|
||
"""
|
||
registry = build_managed_resource_registry((root,))
|
||
return CapabilityRuntime(
|
||
registry,
|
||
adapters={
|
||
MANAGED_RESOURCE_SYNC_KIND: SyncManagedResourceAdapter(),
|
||
MANAGED_RESOURCE_ASYNC_KIND: AsyncManagedResourceAdapter(),
|
||
},
|
||
)
|
||
|
||
|
||
def test_sync_managed_resource_is_single_flight(
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""
|
||
并发首用只能发布一个同步资源实例。
|
||
"""
|
||
module_name = "fixture_sync_managed_resource"
|
||
module = ModuleType(module_name)
|
||
|
||
class SyncResource:
|
||
"""
|
||
记录同步资源的创建、启动和停止次数。
|
||
"""
|
||
|
||
instances: list["SyncResource"] = []
|
||
|
||
def __init__(self) -> None:
|
||
self.started = 0
|
||
self.stopped = 0
|
||
type(self).instances.append(self)
|
||
|
||
def start(self) -> None:
|
||
"""
|
||
累计同步资源启动次数。
|
||
"""
|
||
self.started += 1
|
||
|
||
def stop(self) -> None:
|
||
"""
|
||
累计同步资源停止次数。
|
||
"""
|
||
self.stopped += 1
|
||
|
||
module.SyncResource = SyncResource
|
||
monkeypatch.setitem(sys.modules, module_name, module)
|
||
_write_manifest(
|
||
tmp_path,
|
||
capability_id="fixture.sync",
|
||
kind=MANAGED_RESOURCE_SYNC_KIND,
|
||
entrypoint=f"{module_name}:SyncResource",
|
||
)
|
||
runtime = _runtime(tmp_path)
|
||
configure_managed_resource_runtime(runtime)
|
||
|
||
barrier = threading.Barrier(8)
|
||
|
||
def activate() -> SyncResource:
|
||
barrier.wait(timeout=2)
|
||
return acquire_managed_resource("fixture.sync", reason="test")
|
||
|
||
with ThreadPoolExecutor(max_workers=8) as executor:
|
||
resources = list(executor.map(lambda _index: activate(), range(8)))
|
||
|
||
assert len({id(resource) for resource in resources}) == 1
|
||
assert len(SyncResource.instances) == 1
|
||
assert SyncResource.instances[0].started == 1
|
||
assert managed_resource_snapshot("fixture.sync").generation == 1
|
||
assert [
|
||
observation.outcome
|
||
for observation in managed_resource_observations("fixture.sync")
|
||
if observation.operation == "activate"
|
||
] == ["started", "succeeded"]
|
||
|
||
assert (
|
||
asyncio.run(shutdown_managed_resource_runtime(reason="test_shutdown"))
|
||
is True
|
||
)
|
||
|
||
assert SyncResource.instances[0].stopped == 1
|
||
with pytest.raises(CapabilityRuntimeClosedError):
|
||
acquire_managed_resource("fixture.sync", reason="after_shutdown")
|
||
|
||
|
||
def test_async_managed_resource_uses_async_adapter(
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""
|
||
异步资源通过异步 Runtime 入口启动并关闭。
|
||
"""
|
||
module_name = "fixture_async_managed_resource"
|
||
module = ModuleType(module_name)
|
||
|
||
class AsyncResource:
|
||
"""
|
||
记录异步资源生命周期调用。
|
||
"""
|
||
|
||
instances: list["AsyncResource"] = []
|
||
|
||
def __init__(self) -> None:
|
||
self.events: list[str] = []
|
||
type(self).instances.append(self)
|
||
|
||
async def start(self) -> None:
|
||
"""
|
||
记录异步启动事件，供生命周期顺序断言使用。
|
||
"""
|
||
self.events.append("start")
|
||
|
||
async def stop(self) -> None:
|
||
"""
|
||
记录异步停止事件，供生命周期顺序断言使用。
|
||
"""
|
||
self.events.append("stop")
|
||
|
||
module.AsyncResource = AsyncResource
|
||
monkeypatch.setitem(sys.modules, module_name, module)
|
||
_write_manifest(
|
||
tmp_path,
|
||
capability_id="fixture.async",
|
||
kind=MANAGED_RESOURCE_ASYNC_KIND,
|
||
entrypoint=f"{module_name}:AsyncResource",
|
||
)
|
||
runtime = _runtime(tmp_path)
|
||
configure_managed_resource_runtime(runtime)
|
||
|
||
async def exercise() -> AsyncResource:
|
||
resource = await acquire_managed_resource_async(
|
||
"fixture.async",
|
||
reason="test",
|
||
)
|
||
await shutdown_managed_resource_runtime(reason="test_shutdown")
|
||
return resource
|
||
|
||
resource = asyncio.run(exercise())
|
||
|
||
assert resource.events == ["start", "stop"]
|
||
assert AsyncResource.instances == [resource]
|
||
|
||
|
||
def test_async_managed_resource_recreates_owner_across_event_loops(
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""
|
||
停止后的下一次激活必须在新事件循环创建新资源，不能复用旧 owner。
|
||
"""
|
||
module_name = "fixture_loop_bound_managed_resource"
|
||
module = ModuleType(module_name)
|
||
|
||
class LoopBoundResource:
|
||
"""
|
||
记录资源实际启动和停止所在的事件循环。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self.loop = None
|
||
|
||
async def start(self) -> None:
|
||
"""
|
||
保存启动时的事件循环，供关闭阶段核对。
|
||
"""
|
||
self.loop = asyncio.get_running_loop()
|
||
|
||
async def stop(self) -> None:
|
||
"""
|
||
断言资源仍在启动时的事件循环中关闭。
|
||
"""
|
||
assert asyncio.get_running_loop() is self.loop
|
||
|
||
module.LoopBoundResource = LoopBoundResource
|
||
monkeypatch.setitem(sys.modules, module_name, module)
|
||
_write_manifest(
|
||
tmp_path,
|
||
capability_id="fixture.loop_bound",
|
||
kind=MANAGED_RESOURCE_ASYNC_KIND,
|
||
entrypoint=f"{module_name}:LoopBoundResource",
|
||
)
|
||
runtime = _runtime(tmp_path)
|
||
|
||
async def exercise() -> LoopBoundResource:
|
||
"""
|
||
在当前事件循环完成一次资源启动与停止。
|
||
"""
|
||
resource = await runtime.activate_async(
|
||
"fixture.loop_bound",
|
||
reason="loop_cycle",
|
||
)
|
||
await runtime.stop_async("fixture.loop_bound", reason="loop_cycle")
|
||
return resource
|
||
|
||
first = asyncio.run(exercise())
|
||
second = asyncio.run(exercise())
|
||
|
||
assert first is not second
|
||
assert first.loop is not second.loop
|
||
|
||
|
||
def test_shutdown_propagates_stop_failure_and_retains_owner_for_retry(
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""
|
||
托管资源关闭失败必须向 startup 传播，并保留同一 owner 重试。
|
||
"""
|
||
module_name = "fixture_retry_stop_managed_resource"
|
||
module = ModuleType(module_name)
|
||
|
||
class RetryStopResource:
|
||
"""
|
||
首次停止失败、第二次停止收敛的同步资源。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self.stop_calls = 0
|
||
self.fail_stop = True
|
||
|
||
def start(self) -> None:
|
||
"""
|
||
资源启动无需额外动作。
|
||
"""
|
||
|
||
def stop(self) -> None:
|
||
"""
|
||
按测试开关模拟资源释放失败。
|
||
"""
|
||
self.stop_calls += 1
|
||
if self.fail_stop:
|
||
raise RuntimeError("stop failed")
|
||
|
||
module.RetryStopResource = RetryStopResource
|
||
monkeypatch.setitem(sys.modules, module_name, module)
|
||
_write_manifest(
|
||
tmp_path,
|
||
capability_id="fixture.retry_stop",
|
||
kind=MANAGED_RESOURCE_SYNC_KIND,
|
||
entrypoint=f"{module_name}:RetryStopResource",
|
||
)
|
||
configure_managed_resource_runtime(_runtime(tmp_path))
|
||
resource = acquire_managed_resource("fixture.retry_stop", reason="test")
|
||
|
||
assert (
|
||
asyncio.run(shutdown_managed_resource_runtime(reason="test_shutdown"))
|
||
is False
|
||
)
|
||
assert resource.stop_calls == 1
|
||
|
||
resource.fail_stop = False
|
||
assert (
|
||
asyncio.run(shutdown_managed_resource_runtime(reason="shutdown_retry"))
|
||
is True
|
||
)
|
||
assert resource.stop_calls == 2
|
||
|
||
|
||
def test_failed_start_is_cleaned_before_explicit_retry(
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""
|
||
启动失败的候选必须先清理，显式 retry 才能发布下一代资源。
|
||
"""
|
||
module_name = "fixture_retry_managed_resource"
|
||
module = ModuleType(module_name)
|
||
|
||
class RetryResource:
|
||
"""
|
||
首个候选启动失败，后续候选正常启动。
|
||
"""
|
||
|
||
instances: list["RetryResource"] = []
|
||
|
||
def __init__(self) -> None:
|
||
self.events: list[str] = []
|
||
self.fail_start = not type(self).instances
|
||
type(self).instances.append(self)
|
||
|
||
def start(self) -> None:
|
||
"""
|
||
记录启动尝试，并让首个候选失败以验证重试。
|
||
"""
|
||
self.events.append("start")
|
||
if self.fail_start:
|
||
raise RuntimeError("start failed")
|
||
|
||
def stop(self) -> None:
|
||
"""
|
||
记录失败候选和正常实例的停止事件。
|
||
"""
|
||
self.events.append("stop")
|
||
|
||
module.RetryResource = RetryResource
|
||
monkeypatch.setitem(sys.modules, module_name, module)
|
||
_write_manifest(
|
||
tmp_path,
|
||
capability_id="fixture.retry",
|
||
kind=MANAGED_RESOURCE_SYNC_KIND,
|
||
entrypoint=f"{module_name}:RetryResource",
|
||
)
|
||
configure_managed_resource_runtime(_runtime(tmp_path))
|
||
|
||
with pytest.raises(CapabilityOperationError, match="start failed"):
|
||
acquire_managed_resource(
|
||
"fixture.retry",
|
||
reason="first_use",
|
||
retry=False,
|
||
)
|
||
|
||
resource = acquire_managed_resource(
|
||
"fixture.retry",
|
||
reason="retry",
|
||
retry=True,
|
||
)
|
||
|
||
assert RetryResource.instances[0].events == ["start", "stop"]
|
||
assert resource is RetryResource.instances[1]
|
||
assert resource.events == ["start"]
|
||
|
||
asyncio.run(shutdown_managed_resource_runtime(reason="test_shutdown"))
|
||
|
||
assert resource.events == ["start", "stop"]
|
||
|
||
|
||
def test_shutdown_does_not_materialize_unused_resource(
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""
|
||
关闭未激活 Runtime 时不得构造资源或调用 start。
|
||
"""
|
||
module_name = "fixture_unused_managed_resource"
|
||
module = ModuleType(module_name)
|
||
|
||
class UnusedResource:
|
||
"""
|
||
任何实例化都表示关闭路径发生反向激活。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
raise AssertionError("unused resource must not be materialized")
|
||
|
||
def start(self) -> None:
|
||
"""
|
||
拒绝启动从未被使用的资源。
|
||
"""
|
||
raise AssertionError("unused resource must not start")
|
||
|
||
def stop(self) -> None:
|
||
"""
|
||
拒绝为未创建的资源执行停止。
|
||
"""
|
||
raise AssertionError("unused resource must not stop")
|
||
|
||
module.UnusedResource = UnusedResource
|
||
monkeypatch.setitem(sys.modules, module_name, module)
|
||
_write_manifest(
|
||
tmp_path,
|
||
capability_id="fixture.unused",
|
||
kind=MANAGED_RESOURCE_SYNC_KIND,
|
||
entrypoint=f"{module_name}:UnusedResource",
|
||
)
|
||
configure_managed_resource_runtime(_runtime(tmp_path))
|
||
|
||
asyncio.run(shutdown_managed_resource_runtime(reason="test_shutdown"))
|
||
|
||
|
||
def test_startup_initializer_discovers_manifest_without_importing_resource() -> None:
|
||
"""
|
||
启动装配只能读取声明，不得提前导入或构造虚拟显示实现。
|
||
"""
|
||
script = """
|
||
import asyncio
|
||
import sys
|
||
from app.startup.initializers.resources import (
|
||
init_managed_resources,
|
||
stop_managed_resources,
|
||
)
|
||
|
||
runtime = init_managed_resources()
|
||
assert runtime.get_running("host.display") is None
|
||
assert "app.adapters.system.display.resource" not in sys.modules
|
||
assert "pyvirtualdisplay" not in sys.modules
|
||
asyncio.run(stop_managed_resources())
|
||
assert "app.adapters.system.display.resource" not in sys.modules
|
||
assert "pyvirtualdisplay" not in sys.modules
|
||
"""
|
||
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_startup_shutdown_without_init_does_not_build_registry(monkeypatch) -> None:
|
||
"""
|
||
未执行启动装配时，关闭入口不得通过发现声明反向初始化 Runtime。
|
||
"""
|
||
from app.startup.composition import resource as resource_composition
|
||
from app.startup.initializers import resources as managed_resources_initializer
|
||
|
||
build_registry = MagicMock(side_effect=AssertionError("must not discover"))
|
||
monkeypatch.setattr(
|
||
resource_composition,
|
||
"_managed_resource_runtime",
|
||
None,
|
||
)
|
||
monkeypatch.setattr(
|
||
resource_composition,
|
||
"build_managed_resource_registry",
|
||
build_registry,
|
||
)
|
||
|
||
assert asyncio.run(managed_resources_initializer.stop_managed_resources()) is True
|
||
|
||
build_registry.assert_not_called()
|