Kris
696f5f44ec
这一批提交包含： 1. 配置项敏感字段标记与测试用例修复 2. 路由绑定乐观锁支持与静态路径校验 3. 微信WOC插件能力适配与新增单元测试 4. 多个适配器的接口对齐与测试补全 5. 新增定时任务清理处理器与依赖注入容器测试 6. 错误码体系扩展与整合测试 7. 删除临时验证脚本与代码清理
762 lines
28 KiB
Python
762 lines
28 KiB
Python
"""yuxi.channels.infrastructure.host_bootstrap 单元测试。
|
||
|
||
覆盖 ``HostBootstrap`` 启动编排行为：
|
||
- ``bootstrap`` 入口守卫：重复调用 / 已 ready 时抛 ``InternalError``
|
||
- ``_loadConfig``：空 schema 抛 ``ConfigValidationError``、``DependencyError`` 透传、
|
||
超时抛 ``OperationTimeoutError``
|
||
- ``_initSchema``：失败透传、超时抛 ``OperationTimeoutError``
|
||
- ``_initDrivenAdapters``：``ping()`` 失败抛 ``DependencyError``、
|
||
非 ``PingablePort`` 适配器跳过、空列表返回
|
||
- ``_loadPlugins``：discover 失败告警并继续、空 manifest 提前返回
|
||
- ``_loadSinglePlugin``：加载失败告警并继续（不阻塞宿主启动）
|
||
- ``_rollback``：按 §15.2 逆序回滚已启动组件
|
||
- ``bootstrap`` 端到端：成功路径与失败回滚路径
|
||
|
||
不依赖运行中的 Docker 服务，纯单元测试。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
import pytest
|
||
from yuxi.channels.application.extension.config_source_registry import (
|
||
ConfigSourceRegistry,
|
||
)
|
||
from yuxi.channels.application.extension.event_bus import EventBus
|
||
from yuxi.channels.application.extension.event_subscription_registry import (
|
||
EventSubscriptionRegistry,
|
||
)
|
||
from yuxi.channels.application.extension.stage_slot_registry import StageSlotRegistry
|
||
from yuxi.channels.application.health.health_aggregator import HealthAggregator
|
||
from yuxi.channels.application.lifecycle.plugin_dependency_resolver import (
|
||
PluginDependencyResolver,
|
||
)
|
||
from yuxi.channels.application.lifecycle.plugin_lifecycle_manager import (
|
||
PluginLifecycleManager,
|
||
)
|
||
from yuxi.channels.application.pipeline.stage_slot_injector import StageSlotInjector
|
||
from yuxi.channels.application.transport import TransportManager
|
||
from yuxi.channels.contract.dtos.lifecycle import LifecycleResult
|
||
from yuxi.channels.contract.errors import (
|
||
ConfigValidationError,
|
||
DependencyError,
|
||
InternalError,
|
||
OperationTimeoutError,
|
||
)
|
||
from yuxi.channels.contract.ports.driven.config_port import ConfigPort
|
||
from yuxi.channels.contract.ports.driven.logger_port import LoggerPort
|
||
from yuxi.channels.contract.ports.driven.persistence_port import PersistencePort
|
||
from yuxi.channels.contract.ports.driven.pingable_port import PingablePort
|
||
from yuxi.channels.core.registry.plugin_registry import PluginRegistry
|
||
from yuxi.channels.core.registry.route_match_registry import RouteMatchRegistry
|
||
from yuxi.channels.core.service.degradation_manager import DegradationManager
|
||
from yuxi.channels.infrastructure.dependency_injection import (
|
||
DependencyInjectionContainer,
|
||
)
|
||
from yuxi.channels.infrastructure.host_bootstrap import HostBootstrap
|
||
|
||
pytestmark = pytest.mark.unit
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试辅助：构造 HostBootstrap 与桩依赖
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_manifest(plugin_id: str = "test-plugin"):
|
||
"""构造测试用 manifest 桩（仅需 ``id`` 属性）。
|
||
|
||
使用 ``MagicMock`` 替代真实 ``ChannelManifest``，避免构造 ``provides`` /
|
||
``capabilities`` / ``config_schema`` 等大量必填字段。测试仅需访问
|
||
``manifest.id`` 属性。
|
||
"""
|
||
manifest = MagicMock()
|
||
manifest.id = plugin_id
|
||
return manifest
|
||
|
||
|
||
def _make_persistence_port_stub() -> MagicMock:
|
||
"""构造 ``PersistencePort`` 桩，``listRouteBindings`` 返回空列表。
|
||
|
||
``_loadRouteBindingsFromDB`` 通过 ``RouteMatchRegistryLoader.loadAll``
|
||
调用 ``repo.listRouteBindings(...)``，需返回可 await 的空列表以跳过
|
||
规则加载。
|
||
"""
|
||
stub = MagicMock(spec=PersistencePort)
|
||
stub.listRouteBindings = AsyncMock(return_value=[])
|
||
return stub
|
||
|
||
|
||
def _make_bootstrap(
|
||
fake_logger,
|
||
*,
|
||
di_container: DependencyInjectionContainer | None = None,
|
||
config_port: ConfigPort | None = None,
|
||
plugin_lifecycle: PluginLifecycleManager | None = None,
|
||
plugin_registry: PluginRegistry | None = None,
|
||
route_match_registry: RouteMatchRegistry | None = None,
|
||
stage_slot_injector: StageSlotInjector | None = None,
|
||
transport_manager: TransportManager | None = None,
|
||
driving_adapter_registrar=None,
|
||
driving_adapter_unregistrar=None,
|
||
app_driven_adapters=None,
|
||
ensure_schema: object | None = None,
|
||
) -> HostBootstrap:
|
||
"""构造测试用 HostBootstrap，所有依赖默认为 MagicMock。"""
|
||
if di_container is None:
|
||
di_container = DependencyInjectionContainer()
|
||
# 预注册 _loadRouteBindingsFromDB / _registerCommonCommands 所需依赖
|
||
di_container.registerSingleton(
|
||
RouteMatchRegistry, route_match_registry or RouteMatchRegistry()
|
||
)
|
||
di_container.registerSingleton(PersistencePort, _make_persistence_port_stub())
|
||
di_container.registerSingleton(
|
||
PluginRegistry, plugin_registry or MagicMock(spec=PluginRegistry)
|
||
)
|
||
di_container.registerSingleton(
|
||
PluginLifecycleManager,
|
||
plugin_lifecycle or MagicMock(spec=PluginLifecycleManager),
|
||
)
|
||
from yuxi.channels.core.registry.command_registry import (
|
||
CommandRegistryTable,
|
||
)
|
||
from yuxi.channels.application.lifecycle.sensitive_field_registry import (
|
||
SensitiveFieldRegistry,
|
||
)
|
||
from yuxi.channels.application.lifecycle.config_scope_registry import (
|
||
ConfigScopeRegistry as LifecycleCSR,
|
||
)
|
||
di_container.registerSingleton(CommandRegistryTable, CommandRegistryTable())
|
||
di_container.registerSingleton(SensitiveFieldRegistry, MagicMock())
|
||
di_container.registerSingleton(LifecycleCSR, MagicMock())
|
||
|
||
if config_port is None:
|
||
config_port = MagicMock(spec=ConfigPort)
|
||
|
||
if ensure_schema is None:
|
||
ensure_schema = MagicMock()
|
||
|
||
return HostBootstrap(
|
||
ensure_schema=ensure_schema,
|
||
di_container=di_container,
|
||
plugin_dir="/nonexistent/plugins",
|
||
event_bus=MagicMock(spec=EventBus),
|
||
stage_slot_registry=MagicMock(spec=StageSlotRegistry),
|
||
event_subscription_registry=MagicMock(spec=EventSubscriptionRegistry),
|
||
config_source_registry=MagicMock(spec=ConfigSourceRegistry),
|
||
plugin_lifecycle_manager=plugin_lifecycle or MagicMock(spec=PluginLifecycleManager),
|
||
plugin_dependency_resolver=MagicMock(spec=PluginDependencyResolver),
|
||
stage_slot_injector=stage_slot_injector or MagicMock(spec=StageSlotInjector),
|
||
health_aggregator=MagicMock(spec=HealthAggregator),
|
||
transport_manager=transport_manager or MagicMock(spec=TransportManager),
|
||
logger=fake_logger,
|
||
config_port=config_port,
|
||
driving_adapter_registrar=driving_adapter_registrar,
|
||
driving_adapter_unregistrar=driving_adapter_unregistrar,
|
||
app_driven_adapters=app_driven_adapters,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 入口守卫
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestBootstrapEntryGuard:
|
||
"""bootstrap() 入口守卫测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bootstrap_when_already_ready_raises_internal_error(
|
||
self, fake_logger
|
||
):
|
||
# Arrange
|
||
bootstrap = _make_bootstrap(fake_logger)
|
||
bootstrap._ready = True
|
||
|
||
# Act / Assert
|
||
with pytest.raises(InternalError) as exc_info:
|
||
await bootstrap.bootstrap()
|
||
|
||
assert "already bootstrapped" in exc_info.value.message
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bootstrap_when_in_progress_raises_internal_error(
|
||
self, fake_logger
|
||
):
|
||
# Arrange
|
||
bootstrap = _make_bootstrap(fake_logger)
|
||
bootstrap._bootstrapping = True
|
||
|
||
# Act / Assert
|
||
with pytest.raises(InternalError) as exc_info:
|
||
await bootstrap.bootstrap()
|
||
|
||
assert "in progress" in exc_info.value.message
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bootstrap_failure_resets_in_progress_flag(self, fake_logger):
|
||
"""启动失败后 ``_bootstrapping`` 必须重置，允许调用方重试。"""
|
||
# Arrange
|
||
config_port = MagicMock(spec=ConfigPort)
|
||
# _loadConfig 抛 ConfigValidationError 触发失败路径
|
||
config_port.getSchema = AsyncMock(side_effect=ConfigValidationError(["boom"]))
|
||
|
||
bootstrap = _make_bootstrap(fake_logger, config_port=config_port)
|
||
|
||
# Act
|
||
with pytest.raises(ConfigValidationError):
|
||
await bootstrap.bootstrap()
|
||
|
||
# Assert: 失败后标志重置，ready 仍为 False
|
||
assert bootstrap._bootstrapping is False
|
||
assert bootstrap.isReady is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _loadConfig
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestLoadConfig:
|
||
"""_loadConfig 配置加载测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_schema_raises_config_validation_error(self, fake_logger):
|
||
# Arrange
|
||
config_port = MagicMock(spec=ConfigPort)
|
||
config_port.getSchema = AsyncMock(return_value=())
|
||
bootstrap = _make_bootstrap(fake_logger, config_port=config_port)
|
||
|
||
# Act / Assert
|
||
with pytest.raises(ConfigValidationError) as exc_info:
|
||
await bootstrap._loadConfig()
|
||
|
||
assert "empty" in exc_info.value.message.lower()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dependency_error_propagates_unchanged(self, fake_logger):
|
||
# Arrange
|
||
config_port = MagicMock(spec=ConfigPort)
|
||
original_err = DependencyError("Redis", RuntimeError("conn refused"))
|
||
config_port.getSchema = AsyncMock(side_effect=original_err)
|
||
bootstrap = _make_bootstrap(fake_logger, config_port=config_port)
|
||
|
||
# Act / Assert
|
||
with pytest.raises(DependencyError) as exc_info:
|
||
await bootstrap._loadConfig()
|
||
|
||
assert exc_info.value is original_err
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_other_exception_wrapped_as_config_validation_error(
|
||
self, fake_logger
|
||
):
|
||
# Arrange
|
||
config_port = MagicMock(spec=ConfigPort)
|
||
config_port.getSchema = AsyncMock(side_effect=RuntimeError("unexpected"))
|
||
bootstrap = _make_bootstrap(fake_logger, config_port=config_port)
|
||
|
||
# Act / Assert
|
||
with pytest.raises(ConfigValidationError):
|
||
await bootstrap._loadConfig()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_valid_schema_does_not_raise(self, fake_logger):
|
||
# Arrange
|
||
config_port = MagicMock(spec=ConfigPort)
|
||
config_port.getSchema = AsyncMock(return_value=("field1", "field2"))
|
||
bootstrap = _make_bootstrap(fake_logger, config_port=config_port)
|
||
|
||
# Act / Assert: 不抛异常即成功
|
||
await bootstrap._loadConfig()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _initSchema
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestInitSchema:
|
||
"""_initSchema schema 初始化测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_failure_propagates_exception(self, fake_logger):
|
||
# Arrange
|
||
ensure_schema = MagicMock()
|
||
ensure_schema.ensure_channel_schema = AsyncMock(
|
||
side_effect=RuntimeError("db down")
|
||
)
|
||
bootstrap = _make_bootstrap(fake_logger, ensure_schema=ensure_schema)
|
||
|
||
# Act / Assert
|
||
with pytest.raises(RuntimeError, match="db down"):
|
||
await bootstrap._initSchema()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_success_does_not_raise(self, fake_logger):
|
||
# Arrange
|
||
ensure_schema = MagicMock()
|
||
ensure_schema.ensure_channel_schema = AsyncMock(return_value=None)
|
||
bootstrap = _make_bootstrap(fake_logger, ensure_schema=ensure_schema)
|
||
|
||
# Act / Assert
|
||
await bootstrap._initSchema()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _initDrivenAdapters / _pingAdapter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestInitDrivenAdapters:
|
||
"""_initDrivenAdapters / _pingAdapter 连通性检查测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_adapters_does_not_raise(self, fake_logger):
|
||
# Arrange
|
||
bootstrap = _make_bootstrap(fake_logger, app_driven_adapters=())
|
||
|
||
# Act / Assert: 空列表直接返回（向后兼容）
|
||
await bootstrap._initDrivenAdapters()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_none_adapters_does_not_raise(self, fake_logger):
|
||
# Arrange
|
||
bootstrap = _make_bootstrap(fake_logger, app_driven_adapters=None)
|
||
|
||
# Act / Assert
|
||
await bootstrap._initDrivenAdapters()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_non_pingable_adapter_skipped(self, fake_logger):
|
||
"""未实现 PingablePort 契约的适配器显式跳过。"""
|
||
# Arrange
|
||
non_pingable = MagicMock() # 不实现 PingablePort
|
||
bootstrap = _make_bootstrap(
|
||
fake_logger, app_driven_adapters=[non_pingable]
|
||
)
|
||
|
||
# Act / Assert: 不抛异常即证明跳过逻辑生效
|
||
await bootstrap._initDrivenAdapters()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ping_returns_false_raises_dependency_error(self, fake_logger):
|
||
# Arrange
|
||
pingable_adapter = MagicMock(spec=PingablePort)
|
||
pingable_adapter.ping = AsyncMock(return_value=False)
|
||
# spec=PingablePort 让 isinstance 检查通过
|
||
bootstrap = _make_bootstrap(
|
||
fake_logger, app_driven_adapters=[pingable_adapter]
|
||
)
|
||
|
||
# Act / Assert
|
||
with pytest.raises(DependencyError):
|
||
await bootstrap._initDrivenAdapters()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ping_raises_exception_wrapped_as_dependency_error(
|
||
self, fake_logger
|
||
):
|
||
# Arrange
|
||
pingable_adapter = MagicMock(spec=PingablePort)
|
||
pingable_adapter.ping = AsyncMock(side_effect=RuntimeError("conn lost"))
|
||
bootstrap = _make_bootstrap(
|
||
fake_logger, app_driven_adapters=[pingable_adapter]
|
||
)
|
||
|
||
# Act / Assert
|
||
with pytest.raises(DependencyError):
|
||
await bootstrap._initDrivenAdapters()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ping_returns_true_passes(self, fake_logger):
|
||
# Arrange
|
||
pingable_adapter = MagicMock(spec=PingablePort)
|
||
pingable_adapter.ping = AsyncMock(return_value=True)
|
||
bootstrap = _make_bootstrap(
|
||
fake_logger, app_driven_adapters=[pingable_adapter]
|
||
)
|
||
|
||
# Act / Assert
|
||
await bootstrap._initDrivenAdapters()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _loadPlugins / _loadSinglePlugin
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestLoadPlugins:
|
||
"""_loadPlugins 插件加载测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_discover_failure_warns_and_returns(self, fake_logger):
|
||
"""discover 抛异常时记录告警并跳过插件加载（优雅降级）。"""
|
||
# Arrange
|
||
plugin_lifecycle = MagicMock(spec=PluginLifecycleManager)
|
||
plugin_lifecycle.discover = AsyncMock(side_effect=RuntimeError("disk error"))
|
||
bootstrap = _make_bootstrap(fake_logger, plugin_lifecycle=plugin_lifecycle)
|
||
# _loadPlugins 需要 resolve SensitiveFieldRegistry / ConfigScopeRegistry
|
||
# 已在 _make_bootstrap 中预注册
|
||
|
||
# Act: 不抛异常即证明 warn-and-continue
|
||
await bootstrap._loadPlugins()
|
||
|
||
# Assert: 插件未加载（_plugins_loaded 仍为 False，因 discover 失败提前 return）
|
||
assert bootstrap._plugins_loaded is False
|
||
# 告警日志被调用
|
||
fake_logger.warn.assert_called()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_manifests_returns_early(self, fake_logger):
|
||
# Arrange
|
||
plugin_lifecycle = MagicMock(spec=PluginLifecycleManager)
|
||
plugin_lifecycle.discover = AsyncMock(return_value=[])
|
||
bootstrap = _make_bootstrap(fake_logger, plugin_lifecycle=plugin_lifecycle)
|
||
|
||
# Act
|
||
await bootstrap._loadPlugins()
|
||
|
||
# Assert: 空 manifest 提前返回，_plugins_loaded 仍为 False
|
||
assert bootstrap._plugins_loaded is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_single_plugin_load_failure_warns_and_continues(
|
||
self, fake_logger
|
||
):
|
||
"""单个插件 load 失败记录告警并继续，不阻塞宿主启动。"""
|
||
# Arrange
|
||
manifest = _make_manifest("failed-plugin")
|
||
plugin_lifecycle = MagicMock(spec=PluginLifecycleManager)
|
||
plugin_lifecycle.discover = AsyncMock(return_value=[manifest])
|
||
plugin_lifecycle.load = AsyncMock(
|
||
return_value=LifecycleResult(
|
||
state="failed", error="init error"
|
||
)
|
||
)
|
||
# resolver.resolve 返回原列表（无依赖排序）
|
||
resolver = MagicMock(spec=PluginDependencyResolver)
|
||
resolver.resolve = MagicMock(return_value=[manifest])
|
||
bootstrap = _make_bootstrap(fake_logger, plugin_lifecycle=plugin_lifecycle)
|
||
bootstrap._resolver = resolver
|
||
|
||
# Act
|
||
await bootstrap._loadPlugins()
|
||
|
||
# Assert: _plugins_loaded 置位（步骤完成，含优雅降级）
|
||
assert bootstrap._plugins_loaded is True
|
||
# load 被调用一次
|
||
plugin_lifecycle.load.assert_awaited_once_with("failed-plugin")
|
||
# 告警日志记录失败
|
||
fake_logger.warn.assert_called()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_single_plugin_load_success_logs_info(self, fake_logger):
|
||
# Arrange
|
||
manifest = _make_manifest("ok-plugin")
|
||
plugin_lifecycle = MagicMock(spec=PluginLifecycleManager)
|
||
plugin_lifecycle.discover = AsyncMock(return_value=[manifest])
|
||
plugin_lifecycle.load = AsyncMock(
|
||
return_value=LifecycleResult(state="started")
|
||
)
|
||
resolver = MagicMock(spec=PluginDependencyResolver)
|
||
resolver.resolve = MagicMock(return_value=[manifest])
|
||
bootstrap = _make_bootstrap(fake_logger, plugin_lifecycle=plugin_lifecycle)
|
||
bootstrap._resolver = resolver
|
||
|
||
# Act
|
||
await bootstrap._loadPlugins()
|
||
|
||
# Assert
|
||
assert bootstrap._plugins_loaded is True
|
||
fake_logger.info.assert_called()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _rollback
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestRollback:
|
||
"""_rollback 逆序回滚测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_calls_driving_unregistrar_when_registered(self, fake_logger):
|
||
"""_driving_registered=True 且提供 unregistrar 时调用注销回调。"""
|
||
# Arrange
|
||
unregistrar = AsyncMock()
|
||
bootstrap = _make_bootstrap(
|
||
fake_logger, driving_adapter_unregistrar=unregistrar
|
||
)
|
||
bootstrap._driving_registered = True
|
||
|
||
# Act
|
||
await bootstrap._rollback()
|
||
|
||
# Assert
|
||
unregistrar.assert_awaited_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_skips_driving_unregistrar_when_not_registered(self, fake_logger):
|
||
"""_driving_registered=False 时不调用注销回调。"""
|
||
# Arrange
|
||
unregistrar = AsyncMock()
|
||
bootstrap = _make_bootstrap(
|
||
fake_logger, driving_adapter_unregistrar=unregistrar
|
||
)
|
||
bootstrap._driving_registered = False
|
||
|
||
# Act
|
||
await bootstrap._rollback()
|
||
|
||
# Assert
|
||
unregistrar.assert_not_awaited()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_skips_driving_unregistrar_when_none(self, fake_logger):
|
||
"""driving_adapter_unregistrar=None 时跳过注销步骤。"""
|
||
# Arrange
|
||
bootstrap = _make_bootstrap(fake_logger, driving_adapter_unregistrar=None)
|
||
bootstrap._driving_registered = True
|
||
|
||
# Act / Assert: 不抛异常即证明 None 路径安全
|
||
await bootstrap._rollback()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stops_plugins_when_loaded(self, fake_logger):
|
||
"""_plugins_loaded=True 时调用 stopAllStartedPlugins。"""
|
||
# Arrange
|
||
plugin_lifecycle = MagicMock(spec=PluginLifecycleManager)
|
||
plugin_lifecycle.stopAllStartedPlugins = AsyncMock()
|
||
bootstrap = _make_bootstrap(fake_logger, plugin_lifecycle=plugin_lifecycle)
|
||
bootstrap._plugins_loaded = True
|
||
|
||
# Act
|
||
await bootstrap._rollback()
|
||
|
||
# Assert
|
||
plugin_lifecycle.stopAllStartedPlugins.assert_awaited_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_skips_stop_plugins_when_not_loaded(self, fake_logger):
|
||
# Arrange
|
||
plugin_lifecycle = MagicMock(spec=PluginLifecycleManager)
|
||
plugin_lifecycle.stopAllStartedPlugins = AsyncMock()
|
||
bootstrap = _make_bootstrap(fake_logger, plugin_lifecycle=plugin_lifecycle)
|
||
bootstrap._plugins_loaded = False
|
||
|
||
# Act
|
||
await bootstrap._rollback()
|
||
|
||
# Assert
|
||
plugin_lifecycle.stopAllStartedPlugins.assert_not_awaited()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_driving_unregistrar_failure_does_not_block_subsequent_steps(
|
||
self, fake_logger
|
||
):
|
||
"""注销回调抛异常时记录告警并继续后续回滚步骤（INV-9）。"""
|
||
# Arrange
|
||
unregistrar = AsyncMock(side_effect=RuntimeError("unreg failed"))
|
||
plugin_lifecycle = MagicMock(spec=PluginLifecycleManager)
|
||
plugin_lifecycle.stopAllStartedPlugins = AsyncMock()
|
||
bootstrap = _make_bootstrap(
|
||
fake_logger,
|
||
plugin_lifecycle=plugin_lifecycle,
|
||
driving_adapter_unregistrar=unregistrar,
|
||
)
|
||
bootstrap._driving_registered = True
|
||
bootstrap._plugins_loaded = True
|
||
|
||
# Act: 不抛异常即证明后续步骤继续执行
|
||
await bootstrap._rollback()
|
||
|
||
# Assert: 后续 stopAllStartedPlugins 仍被调用
|
||
unregistrar.assert_awaited_once()
|
||
plugin_lifecycle.stopAllStartedPlugins.assert_awaited_once()
|
||
fake_logger.error.assert_called()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cancels_background_tasks(self, fake_logger):
|
||
"""回滚时取消已启动的后台任务。"""
|
||
# Arrange
|
||
bootstrap = _make_bootstrap(fake_logger)
|
||
task1 = asyncio.create_task(asyncio.sleep(100))
|
||
task2 = asyncio.create_task(asyncio.sleep(100))
|
||
bootstrap._plugin_reload_task = task1
|
||
bootstrap._transport_manager_task = task2
|
||
|
||
# Act
|
||
await bootstrap._rollback()
|
||
# _cancelBackgroundTasks 仅调度取消（同步），需让事件循环处理取消完成
|
||
await asyncio.sleep(0)
|
||
|
||
# Assert: 任务被取消
|
||
assert task1.cancelled() or task1.done()
|
||
assert task2.cancelled() or task2.done()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# bootstrap 端到端（happy path）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestBootstrapHappyPath:
|
||
"""bootstrap() 成功路径测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_full_bootstrap_marks_ready(self, fake_logger):
|
||
"""成功路径：所有步骤完成 → isReady=True。"""
|
||
# Arrange: 构造所有依赖的桩，配置为成功路径
|
||
config_port = MagicMock(spec=ConfigPort)
|
||
config_port.getSchema = AsyncMock(return_value=("field1",))
|
||
|
||
ensure_schema = MagicMock()
|
||
ensure_schema.ensure_channel_schema = AsyncMock(return_value=None)
|
||
|
||
plugin_lifecycle = MagicMock(spec=PluginLifecycleManager)
|
||
plugin_lifecycle.discover = AsyncMock(return_value=[])
|
||
# reloadLoop 返回的协程需长时间挂起以模拟后台任务
|
||
plugin_lifecycle.reloadLoop = MagicMock(return_value=self._endless_coro())
|
||
|
||
transport_manager = MagicMock(spec=TransportManager)
|
||
transport_manager.start = MagicMock(return_value=self._endless_coro())
|
||
|
||
stage_slot_injector = MagicMock(spec=StageSlotInjector)
|
||
stage_slot_injector.validateAnchors = MagicMock()
|
||
|
||
bootstrap = _make_bootstrap(
|
||
fake_logger,
|
||
config_port=config_port,
|
||
plugin_lifecycle=plugin_lifecycle,
|
||
stage_slot_injector=stage_slot_injector,
|
||
transport_manager=transport_manager,
|
||
app_driven_adapters=(),
|
||
)
|
||
bootstrap._ensure_schema = ensure_schema
|
||
|
||
# Act
|
||
await bootstrap.bootstrap()
|
||
|
||
# Assert
|
||
assert bootstrap.isReady is True
|
||
# 后台任务引用已填充
|
||
assert bootstrap.pluginReloadTask is not None
|
||
assert bootstrap.transportManagerTask is not None
|
||
|
||
# 清理：取消后台任务避免泄漏到后续测试
|
||
bootstrap.pluginReloadTask.cancel()
|
||
bootstrap.transportManagerTask.cancel()
|
||
try:
|
||
await bootstrap.pluginReloadTask
|
||
except (asyncio.CancelledError, Exception):
|
||
pass
|
||
try:
|
||
await bootstrap.transportManagerTask
|
||
except (asyncio.CancelledError, Exception):
|
||
pass
|
||
|
||
@staticmethod
|
||
def _endless_coro():
|
||
"""构造一个长时间挂起的协程，模拟后台任务。"""
|
||
return asyncio.sleep(1000)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bootstrap_failure_triggers_rollback(self, fake_logger):
|
||
"""关键步骤失败时触发 _rollback，异常向上抛出。"""
|
||
# Arrange: 让 _loadConfig 抛 ConfigValidationError
|
||
config_port = MagicMock(spec=ConfigPort)
|
||
config_port.getSchema = AsyncMock(
|
||
side_effect=ConfigValidationError(["bad config"])
|
||
)
|
||
bootstrap = _make_bootstrap(fake_logger, config_port=config_port)
|
||
|
||
# Act / Assert
|
||
with pytest.raises(ConfigValidationError):
|
||
await bootstrap.bootstrap()
|
||
|
||
# 失败后 ready 仍为 False
|
||
assert bootstrap.isReady is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 后台任务启动探测
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestStartBackgroundTask:
|
||
"""_startBackgroundTask 启动即失败探测测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_task_immediate_failure_raises_dependency_error(
|
||
self, fake_logger
|
||
):
|
||
# Arrange
|
||
bootstrap = _make_bootstrap(fake_logger)
|
||
|
||
async def _fail_immediately() -> None:
|
||
raise RuntimeError("startup boom")
|
||
|
||
# Act / Assert: 任务在探测窗口内失败，包装为 DependencyError
|
||
with pytest.raises(DependencyError):
|
||
await bootstrap._startBackgroundTask(
|
||
_fail_immediately(),
|
||
name="test_task",
|
||
log_message="should not see this",
|
||
)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_long_running_task_returns_successfully(self, fake_logger):
|
||
# Arrange
|
||
bootstrap = _make_bootstrap(fake_logger)
|
||
|
||
async def _long_running() -> None:
|
||
await asyncio.sleep(100)
|
||
|
||
# Act
|
||
task = await bootstrap._startBackgroundTask(
|
||
_long_running(),
|
||
name="test_task",
|
||
log_message="task started",
|
||
)
|
||
|
||
# Assert
|
||
assert not task.done()
|
||
task.cancel()
|
||
try:
|
||
await task
|
||
except (asyncio.CancelledError, Exception):
|
||
pass
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# isReady / pluginReloadTask / transportManagerTask 属性
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestProperties:
|
||
"""isReady / pluginReloadTask / transportManagerTask 属性测试。"""
|
||
|
||
def test_is_ready_default_false(self, fake_logger):
|
||
bootstrap = _make_bootstrap(fake_logger)
|
||
assert bootstrap.isReady is False
|
||
|
||
def test_plugin_reload_task_default_none(self, fake_logger):
|
||
bootstrap = _make_bootstrap(fake_logger)
|
||
assert bootstrap.pluginReloadTask is None
|
||
|
||
def test_transport_manager_task_default_none(self, fake_logger):
|
||
bootstrap = _make_bootstrap(fake_logger)
|
||
assert bootstrap.transportManagerTask is None
|