mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-28 19:06:36 +08:00
839 lines
26 KiB
Python
839 lines
26 KiB
Python
"""
|
||
验证 Agent 能力发现与首次加载相互隔离，并发首次使用只初始化一次。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import importlib
|
||
import sys
|
||
import threading
|
||
import types
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
|
||
import pytest
|
||
|
||
from app.runtime.capabilities.errors import (
|
||
CapabilityOperationError,
|
||
CapabilityRuntimeClosedError,
|
||
)
|
||
from app.runtime.capabilities.model import (
|
||
CapabilityLifecycleState,
|
||
CapabilityMaterializationState,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def runtime_loader(monkeypatch):
|
||
"""
|
||
为每个用例提供未构建、未关闭的 Agent Capability Runtime。
|
||
"""
|
||
from app.agent import loader as module
|
||
|
||
monkeypatch.setattr(module, "_agent_runtime", None)
|
||
for implementation_module in (
|
||
"app.agent.manager",
|
||
"app.agent.orchestrator",
|
||
"app.agent.tools.factory",
|
||
):
|
||
package_name, _, attribute = implementation_module.rpartition(".")
|
||
package = importlib.import_module(package_name)
|
||
# 重导入也会写入父包属性；与 sys.modules 一起恢复，避免后续补丁命中另一份模块。
|
||
# 先登记再移除属性，保证原本不存在的属性也能由 monkeypatch 完整撤销。
|
||
monkeypatch.setattr(package, attribute, vars(package).get(attribute), raising=False)
|
||
monkeypatch.delattr(package, attribute)
|
||
monkeypatch.delitem(sys.modules, implementation_module, raising=False)
|
||
return module
|
||
|
||
|
||
class _FakeManager:
|
||
"""
|
||
记录异步 service 生命周期，并支持测试控制初始化时序。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
"""
|
||
初始化可观察的生命周期计数和并发控制状态。
|
||
"""
|
||
self.initialize_calls = 0
|
||
self.close_calls = 0
|
||
self.fail_initialize = False
|
||
self.initialize_entered: asyncio.Event | None = None
|
||
self.initialize_release: asyncio.Event | None = None
|
||
self.close_converged: bool | None = None
|
||
|
||
async def initialize(self) -> None:
|
||
"""
|
||
按测试设定等待屏障或抛错，模拟可控制的初始化时序。
|
||
"""
|
||
self.initialize_calls += 1
|
||
if self.initialize_entered is not None:
|
||
self.initialize_entered.set()
|
||
if self.initialize_release is not None:
|
||
await self.initialize_release.wait()
|
||
if self.fail_initialize:
|
||
raise RuntimeError("service initialization failed")
|
||
|
||
async def close(self) -> bool | None:
|
||
"""
|
||
记录关闭次数并返回当前测试指定的收敛结果。
|
||
"""
|
||
self.close_calls += 1
|
||
return self.close_converged
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_shutdown_retains_nonconverged_agent_service_for_retry(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
Manager 返回未收敛时 Runtime 必须保留 owner，后续关闭可继续等待。
|
||
"""
|
||
manager = _FakeManager()
|
||
manager.close_converged = False
|
||
modules = _fake_agent_modules(manager)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
lambda name: (
|
||
monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name]
|
||
),
|
||
)
|
||
assert await runtime_loader.activate_agent_service() is manager
|
||
|
||
assert await runtime_loader.begin_agent_shutdown() is False
|
||
snapshot = runtime_loader._agent_runtime.snapshot("agent.service")
|
||
assert snapshot.lifecycle is CapabilityLifecycleState.FAILED
|
||
assert snapshot.visible is False
|
||
|
||
manager.close_converged = True
|
||
assert await runtime_loader.begin_agent_shutdown() is True
|
||
snapshot = runtime_loader._agent_runtime.snapshot("agent.service")
|
||
assert snapshot.lifecycle is CapabilityLifecycleState.STOPPED
|
||
assert manager.close_calls == 2
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_shutdown_propagates_runtime_wide_convergence(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
Agent 关闭不得用单个 service 快照覆盖 Runtime 的整体结果。
|
||
"""
|
||
|
||
class RuntimeWithIndependentShutdownResult:
|
||
"""
|
||
模拟其它 Agent 能力失败而 service 已停止的 Runtime。
|
||
"""
|
||
|
||
async def shutdown_async(self, *, reason: str) -> bool:
|
||
"""
|
||
记录关闭原因并返回 Runtime 级未收敛。
|
||
"""
|
||
assert reason == "application_shutdown"
|
||
return False
|
||
|
||
@staticmethod
|
||
def snapshot(_capability_id: str) -> types.SimpleNamespace:
|
||
"""
|
||
提供旧实现读取的已停止 service 快照。
|
||
"""
|
||
return types.SimpleNamespace(lifecycle=CapabilityLifecycleState.STOPPED)
|
||
|
||
runtime = RuntimeWithIndependentShutdownResult()
|
||
monkeypatch.setattr(runtime_loader, "_ensure_runtime", lambda: runtime)
|
||
|
||
assert await runtime_loader.begin_agent_shutdown() is False
|
||
|
||
|
||
def _fake_agent_modules(manager: object | None = None) -> dict[str, types.ModuleType]:
|
||
"""
|
||
构造包含 canonical 符号的模块替身，隔离真实 Agent 服务初始化。
|
||
"""
|
||
manager_module = types.ModuleType("app.agent.manager")
|
||
manager_module.agent_manager = manager if manager is not None else object()
|
||
orchestrator = types.ModuleType("app.agent.orchestrator")
|
||
orchestrator.MoviePilotAgent = type("MoviePilotAgent", (), {})
|
||
tools = types.ModuleType("app.agent.tools.factory")
|
||
tools.MoviePilotToolFactory = type("MoviePilotToolFactory", (), {})
|
||
return {
|
||
manager_module.__name__: manager_module,
|
||
orchestrator.__name__: orchestrator,
|
||
tools.__name__: tools,
|
||
}
|
||
|
||
|
||
def test_registry_discovery_does_not_import_agent_implementation(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
声明发现只能读取 TOML，不得导入 Agent、LLM 或工具实现。
|
||
"""
|
||
imported = []
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
lambda name: imported.append(name),
|
||
)
|
||
|
||
runtime = runtime_loader._ensure_runtime()
|
||
|
||
assert {spec.id for spec in runtime.list_specs()} == {
|
||
"agent.manager",
|
||
"agent.moviepilot_type",
|
||
"agent.service",
|
||
"agent.tool_factory",
|
||
}
|
||
assert imported == []
|
||
|
||
|
||
def test_agent_capability_entrypoints_resolve_canonical_symbols(runtime_loader) -> None:
|
||
"""
|
||
entrypoint 必须命中 canonical 模块的物理符号，Compat 惰性映射不满足强校验。
|
||
"""
|
||
runtime = runtime_loader._ensure_runtime()
|
||
for spec in runtime.list_specs():
|
||
module_name, symbol_name = spec.entrypoint.split(":", maxsplit=1)
|
||
module = importlib.import_module(module_name)
|
||
assert symbol_name in vars(module), spec.entrypoint
|
||
implementation = vars(module)[symbol_name]
|
||
assert runtime._canonical_implementation(spec, implementation) is implementation
|
||
|
||
|
||
def test_concurrent_manager_first_use_is_single_flight(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
并发首用必须只导入一次并向全部调用者发布同一 canonical 对象。
|
||
"""
|
||
modules = _fake_agent_modules()
|
||
import_calls = []
|
||
import_entered = threading.Event()
|
||
import_release = threading.Event()
|
||
|
||
def import_module(name: str):
|
||
"""
|
||
挂起首次导入，验证并发请求只会加载一个服务 owner。
|
||
"""
|
||
import_calls.append(name)
|
||
import_entered.set()
|
||
assert import_release.wait(timeout=5)
|
||
monkeypatch.setitem(sys.modules, name, modules[name])
|
||
return modules[name]
|
||
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
import_module,
|
||
)
|
||
with ThreadPoolExecutor(max_workers=8) as executor:
|
||
futures = [executor.submit(runtime_loader.get_agent_manager) for _ in range(8)]
|
||
assert import_entered.wait(timeout=5)
|
||
import_release.set()
|
||
results = [future.result(timeout=5) for future in futures]
|
||
|
||
assert all(
|
||
result is modules["app.agent.manager"].agent_manager for result in results
|
||
)
|
||
assert (
|
||
runtime_loader.get_moviepilot_agent_type()
|
||
is modules["app.agent.orchestrator"].MoviePilotAgent
|
||
)
|
||
manager_snapshot = runtime_loader._agent_runtime.snapshot("agent.manager")
|
||
assert manager_snapshot.materialization is CapabilityMaterializationState.RESOLVED
|
||
assert manager_snapshot.lifecycle is CapabilityLifecycleState.DISCOVERED
|
||
assert manager_snapshot.visible is False
|
||
assert import_calls == ["app.agent.manager", "app.agent.orchestrator"]
|
||
|
||
|
||
def test_tool_factory_has_independent_first_use_entrypoint(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
工具工厂可独立首用，不需要先解析完整 Agent 编排模块。
|
||
"""
|
||
modules = _fake_agent_modules()
|
||
import_calls = []
|
||
|
||
def import_module(name: str):
|
||
"""
|
||
登记当前能力的模块替身，避免触发真实服务的导入和初始化。
|
||
"""
|
||
import_calls.append(name)
|
||
monkeypatch.setitem(sys.modules, name, modules[name])
|
||
return modules[name]
|
||
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
import_module,
|
||
)
|
||
|
||
assert runtime_loader.is_tool_factory_materialized() is False
|
||
assert (
|
||
runtime_loader.get_tool_factory()
|
||
is modules["app.agent.tools.factory"].MoviePilotToolFactory
|
||
)
|
||
assert runtime_loader.is_tool_factory_materialized() is True
|
||
assert import_calls == ["app.agent.tools.factory"]
|
||
|
||
|
||
def test_materialization_query_does_not_construct_runtime(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
只读物化查询在 Runtime 未构建时必须直接返回 False。
|
||
"""
|
||
build_calls = []
|
||
monkeypatch.setattr(
|
||
runtime_loader,
|
||
"_build_agent_runtime",
|
||
lambda: build_calls.append(True),
|
||
)
|
||
|
||
assert runtime_loader.is_tool_factory_materialized() is False
|
||
assert runtime_loader.get_running_agent_manager() is None
|
||
assert build_calls == []
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_service_first_and_entrypoint_first_share_canonical_identity(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
资源轴和兼容物化轴无论谁先解析，都必须共享 canonical manager。
|
||
"""
|
||
manager = _FakeManager()
|
||
modules = _fake_agent_modules(manager)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
|
||
def import_module(name: str):
|
||
"""
|
||
登记当前能力的模块替身，避免触发真实服务的导入和初始化。
|
||
"""
|
||
monkeypatch.setitem(sys.modules, name, modules[name])
|
||
return modules[name]
|
||
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
import_module,
|
||
)
|
||
|
||
service_first = await runtime_loader.activate_agent_service()
|
||
compat_after = runtime_loader.get_agent_manager()
|
||
|
||
assert service_first is manager
|
||
assert compat_after is manager
|
||
assert runtime_loader.get_running_agent_manager() is manager
|
||
assert manager.initialize_calls == 1
|
||
snapshot = runtime_loader._agent_runtime.snapshot("agent.service")
|
||
assert snapshot.lifecycle is CapabilityLifecycleState.RUNNING
|
||
assert snapshot.visible is True
|
||
|
||
await runtime_loader.begin_agent_shutdown()
|
||
|
||
assert manager.close_calls == 1
|
||
assert runtime_loader.get_running_agent_manager() is None
|
||
with pytest.raises(CapabilityRuntimeClosedError):
|
||
runtime_loader.get_agent_manager()
|
||
with pytest.raises(CapabilityRuntimeClosedError):
|
||
runtime_loader.get_moviepilot_agent_type()
|
||
with pytest.raises(CapabilityRuntimeClosedError):
|
||
runtime_loader.get_tool_factory()
|
||
with pytest.raises(CapabilityRuntimeClosedError):
|
||
await runtime_loader.activate_agent_service()
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_entrypoint_first_then_service_initializes_same_manager_once(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
兼容 getter 先物化时不初始化，随后 service 只初始化同一对象一次。
|
||
"""
|
||
manager = _FakeManager()
|
||
modules = _fake_agent_modules(manager)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
|
||
def import_module(name: str):
|
||
"""
|
||
登记当前能力的模块替身，避免触发真实服务的导入和初始化。
|
||
"""
|
||
monkeypatch.setitem(sys.modules, name, modules[name])
|
||
return modules[name]
|
||
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
import_module,
|
||
)
|
||
|
||
assert runtime_loader.get_agent_manager() is manager
|
||
assert manager.initialize_calls == 0
|
||
assert await runtime_loader.activate_agent_service() is manager
|
||
assert await runtime_loader.activate_agent_service() is manager
|
||
assert manager.initialize_calls == 1
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_concurrent_service_first_use_initializes_once(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
并发 service 首启只能 initialize 一次并发布同一实例。
|
||
"""
|
||
manager = _FakeManager()
|
||
manager.initialize_entered = asyncio.Event()
|
||
manager.initialize_release = asyncio.Event()
|
||
modules = _fake_agent_modules(manager)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
lambda name: (
|
||
monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name]
|
||
),
|
||
)
|
||
|
||
tasks = [
|
||
asyncio.create_task(runtime_loader.activate_agent_service()) for _ in range(8)
|
||
]
|
||
await manager.initialize_entered.wait()
|
||
assert runtime_loader.get_agent_manager() is manager
|
||
assert runtime_loader.get_running_agent_manager() is None
|
||
manager.initialize_release.set()
|
||
results = await asyncio.gather(*tasks)
|
||
|
||
assert all(result is manager for result in results)
|
||
assert manager.initialize_calls == 1
|
||
assert runtime_loader.get_running_agent_manager() is manager
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_concurrent_service_and_entrypoint_first_use_share_identity(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
两个 spec 并发首解析时也只能引用同一个 canonical manager。
|
||
"""
|
||
manager = _FakeManager()
|
||
modules = _fake_agent_modules(manager)
|
||
import_barrier = threading.Barrier(2)
|
||
import_calls = []
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
|
||
def import_module(name: str):
|
||
"""
|
||
在共享屏障处等待，验证不同能力的导入不会互相串行阻塞。
|
||
"""
|
||
import_calls.append(name)
|
||
import_barrier.wait(timeout=5)
|
||
monkeypatch.setitem(sys.modules, name, modules[name])
|
||
return modules[name]
|
||
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
import_module,
|
||
)
|
||
|
||
service_task = asyncio.create_task(runtime_loader.activate_agent_service())
|
||
entrypoint_task = asyncio.create_task(
|
||
asyncio.to_thread(runtime_loader.get_agent_manager)
|
||
)
|
||
service, entrypoint = await asyncio.gather(service_task, entrypoint_task)
|
||
|
||
assert service is manager
|
||
assert entrypoint is manager
|
||
assert runtime_loader.get_running_agent_manager() is manager
|
||
assert manager.initialize_calls == 1
|
||
assert import_calls == ["app.agent.manager"] * 2
|
||
|
||
await runtime_loader.begin_agent_shutdown()
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_service_failure_is_not_published_and_requires_retry(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
初始化失败必须清理候选、保持不可见，并要求显式 retry。
|
||
"""
|
||
manager = _FakeManager()
|
||
manager.fail_initialize = True
|
||
modules = _fake_agent_modules(manager)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
lambda name: (
|
||
monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name]
|
||
),
|
||
)
|
||
|
||
with pytest.raises(CapabilityOperationError, match="initialization failed"):
|
||
await runtime_loader.activate_agent_service()
|
||
|
||
assert runtime_loader.get_running_agent_manager() is None
|
||
failed = runtime_loader._agent_runtime.snapshot("agent.service")
|
||
assert failed.lifecycle is CapabilityLifecycleState.FAILED
|
||
assert failed.visible is False
|
||
assert manager.close_calls == 1
|
||
with pytest.raises(CapabilityOperationError, match="retry=True"):
|
||
await runtime_loader.activate_agent_service()
|
||
|
||
manager.fail_initialize = False
|
||
assert await runtime_loader.activate_agent_service(retry=True) is manager
|
||
assert manager.initialize_calls == 2
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_shutdown_racing_service_first_use_fails_closed(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
关闭与 service 首启竞争时不得发布对象，并须清理已初始化候选。
|
||
"""
|
||
manager = _FakeManager()
|
||
manager.initialize_entered = asyncio.Event()
|
||
manager.initialize_release = asyncio.Event()
|
||
modules = _fake_agent_modules(manager)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
lambda name: (
|
||
monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name]
|
||
),
|
||
)
|
||
|
||
activation = asyncio.create_task(runtime_loader.activate_agent_service())
|
||
await manager.initialize_entered.wait()
|
||
shutdown = asyncio.create_task(runtime_loader.begin_agent_shutdown())
|
||
await asyncio.sleep(0)
|
||
manager.initialize_release.set()
|
||
|
||
with pytest.raises(CapabilityRuntimeClosedError):
|
||
await activation
|
||
await shutdown
|
||
|
||
assert runtime_loader.get_running_agent_manager() is None
|
||
assert manager.initialize_calls == 1
|
||
assert manager.close_calls == 1
|
||
snapshot = runtime_loader._agent_runtime.snapshot("agent.service")
|
||
assert snapshot.lifecycle is CapabilityLifecycleState.STOPPED
|
||
assert snapshot.visible is False
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_disabled_service_reconcile_stays_unmaterialized(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
selector 为 false 时协调结果为空且不得导入 orchestrator。
|
||
"""
|
||
imported = []
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
False,
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
lambda name: imported.append(name),
|
||
)
|
||
|
||
assert await runtime_loader.activate_agent_service() is None
|
||
assert runtime_loader.get_running_agent_manager() is None
|
||
snapshot = runtime_loader._agent_runtime.snapshot("agent.service")
|
||
assert snapshot.materialization is CapabilityMaterializationState.UNRESOLVED
|
||
assert snapshot.lifecycle is CapabilityLifecycleState.STOPPED
|
||
assert imported == []
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_empty_shutdown_does_not_import_agent_or_tools(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
空载关闭只解析 data-only manifests，不导入编排器或工具实现。
|
||
"""
|
||
imported = []
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
lambda name: imported.append(name),
|
||
)
|
||
|
||
await runtime_loader.begin_agent_shutdown()
|
||
|
||
assert imported == []
|
||
assert runtime_loader.get_running_agent_manager() is None
|
||
assert runtime_loader.is_tool_factory_materialized() is False
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_disable_reconcile_waits_for_concurrent_service_start(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
关闭配置与首启竞争时必须等待启动并最终撤销实例。
|
||
"""
|
||
manager = _FakeManager()
|
||
manager.initialize_entered = asyncio.Event()
|
||
manager.initialize_release = asyncio.Event()
|
||
modules = _fake_agent_modules(manager)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
lambda name: (
|
||
monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name]
|
||
),
|
||
)
|
||
|
||
activation = asyncio.create_task(runtime_loader.activate_agent_service())
|
||
await manager.initialize_entered.wait()
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
False,
|
||
)
|
||
disable = asyncio.create_task(
|
||
runtime_loader.reconcile_agent_service(
|
||
reason="config_changed",
|
||
changed_keys={"AI_AGENT_ENABLE"},
|
||
retry=True,
|
||
)
|
||
)
|
||
await asyncio.sleep(0)
|
||
manager.initialize_release.set()
|
||
|
||
assert await activation is manager
|
||
assert await disable is None
|
||
assert runtime_loader.get_running_agent_manager() is None
|
||
assert manager.initialize_calls == 1
|
||
assert manager.close_calls == 1
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_config_reconcile_hot_switches_service_generations(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
watch 命中的配置切换应停止并重启同一 canonical service。
|
||
"""
|
||
manager = _FakeManager()
|
||
modules = _fake_agent_modules(manager)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
lambda name: (
|
||
monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name]
|
||
),
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
False,
|
||
)
|
||
|
||
assert await runtime_loader.activate_agent_service() is None
|
||
assert (
|
||
await runtime_loader.reconcile_agent_service(
|
||
reason="config_changed",
|
||
changed_keys={"UNRELATED"},
|
||
retry=True,
|
||
)
|
||
is None
|
||
)
|
||
assert manager.initialize_calls == 0
|
||
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
assert (
|
||
await runtime_loader.reconcile_agent_service(
|
||
reason="config_changed",
|
||
changed_keys={"AI_AGENT_ENABLE"},
|
||
retry=True,
|
||
)
|
||
is manager
|
||
)
|
||
assert manager.initialize_calls == 1
|
||
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
False,
|
||
)
|
||
assert (
|
||
await runtime_loader.reconcile_agent_service(
|
||
reason="config_changed",
|
||
changed_keys={"AI_AGENT_ENABLE"},
|
||
retry=True,
|
||
)
|
||
is None
|
||
)
|
||
assert manager.close_calls == 1
|
||
assert runtime_loader.get_running_agent_manager() is None
|
||
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
assert (
|
||
await runtime_loader.reconcile_agent_service(
|
||
reason="config_changed",
|
||
changed_keys={"AI_AGENT_ENABLE"},
|
||
retry=True,
|
||
)
|
||
is manager
|
||
)
|
||
assert manager.initialize_calls == 2
|
||
assert runtime_loader._agent_runtime.snapshot("agent.service").generation == 3
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_config_reconcile_after_shutdown_cannot_restart_service(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
Runtime 关闭后即使 selector 再次为 true，配置协调也必须拒绝重启。
|
||
"""
|
||
manager = _FakeManager()
|
||
modules = _fake_agent_modules(manager)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.importlib.import_module",
|
||
lambda name: (
|
||
monkeypatch.setitem(sys.modules, name, modules[name]) or modules[name]
|
||
),
|
||
)
|
||
|
||
assert await runtime_loader.activate_agent_service() is manager
|
||
await runtime_loader.begin_agent_shutdown()
|
||
|
||
with pytest.raises(CapabilityRuntimeClosedError):
|
||
await runtime_loader.reconcile_agent_service(
|
||
reason="config_changed",
|
||
changed_keys={"AI_AGENT_ENABLE"},
|
||
retry=True,
|
||
)
|
||
assert manager.initialize_calls == 1
|
||
assert manager.close_calls == 1
|
||
assert runtime_loader.get_running_agent_manager() is None
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_real_agent_manager_can_restart_across_config_generations(
|
||
runtime_loader,
|
||
monkeypatch,
|
||
) -> None:
|
||
"""
|
||
真实 AgentManager 在配置热切换后必须重新建立运行代际。
|
||
"""
|
||
manager_module = importlib.import_module("app.agent.manager")
|
||
memory_module = importlib.import_module("app.agent.memory")
|
||
|
||
manager = manager_module.AgentManager()
|
||
memory_events = []
|
||
|
||
async def close_memory() -> None:
|
||
"""
|
||
记录记忆存储关闭顺序，核对生命周期退出的先后关系。
|
||
"""
|
||
memory_events.append("close")
|
||
|
||
monkeypatch.setattr(
|
||
memory_module.memory_manager,
|
||
"initialize",
|
||
lambda: memory_events.append("initialize"),
|
||
)
|
||
monkeypatch.setattr(memory_module.memory_manager, "close", close_memory)
|
||
monkeypatch.setattr(manager_module, "agent_manager", manager)
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
False,
|
||
)
|
||
|
||
assert await runtime_loader.activate_agent_service() is None
|
||
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
assert (
|
||
await runtime_loader.reconcile_agent_service(
|
||
reason="config_changed",
|
||
changed_keys={"AI_AGENT_ENABLE"},
|
||
retry=True,
|
||
)
|
||
is manager
|
||
)
|
||
assert manager._accepting_tasks is True
|
||
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
False,
|
||
)
|
||
assert (
|
||
await runtime_loader.reconcile_agent_service(
|
||
reason="config_changed",
|
||
changed_keys={"AI_AGENT_ENABLE"},
|
||
retry=True,
|
||
)
|
||
is None
|
||
)
|
||
assert manager._accepting_tasks is False
|
||
|
||
monkeypatch.setattr(
|
||
"app.agent.capabilities.adapter.settings.AI_AGENT_ENABLE",
|
||
True,
|
||
)
|
||
assert (
|
||
await runtime_loader.reconcile_agent_service(
|
||
reason="config_changed",
|
||
changed_keys={"AI_AGENT_ENABLE"},
|
||
retry=True,
|
||
)
|
||
is manager
|
||
)
|
||
assert manager._accepting_tasks is True
|
||
assert memory_events == ["initialize", "close", "initialize"]
|
||
|
||
await runtime_loader.begin_agent_shutdown()
|
||
|
||
assert manager._accepting_tasks is False
|
||
assert memory_events == ["initialize", "close", "initialize", "close"]
|