"""yuxi.channels.contract.plugin 模块单元测试。 覆盖插件协议包入口与各子模块： - ``__init__.py`` 的 ``__all__`` 导出完整性 - ``manifest.py`` 的枚举与 dataclass（FailurePolicy / CredentialType / AcquireMethod / PluginDependency / ResourceQuota / CredentialStrategy / ChannelManifest / PluginManifest），含 ChannelManifest 新增字段（capability_requirements / max_message_length / supports_credential_cloning / icon / transport_mode / max_restart_attempts） - ``lifecycle.py`` 的枚举与 Protocol（LifecycleState / LifecycleHook / LifecycleHookHandler），含默认方法体 ``...`` 返回 None 的行为验证 - ``extension_point.py`` 的枚举、dataclass 与 Protocol（ConflictStrategy / FailureStrategy / StageSlot / EventSubscription / ConfigSource / Stage） - ``capability.py`` 的 dataclass 与 Protocol（CapabilityDeclaration / CapabilityProof / CapabilityProver），含 proveCapability 默认抛 NotImplementedError 行为验证 - ``entry.py`` 的 Protocol 与类型别名（PluginHost / CHANNEL_ENTRY / PluginAdapter / Matcher） """ from __future__ import annotations import inspect from collections.abc import Callable from datetime import datetime from enum import StrEnum from typing import Any, Protocol, get_args, get_origin from unittest.mock import MagicMock import pytest from yuxi.channels.contract.dtos.capability import ChannelCapabilities from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.channels.contract.dtos.config import ConfigField from yuxi.channels.contract.errors import NotImplementedError from yuxi.channels.contract.plugin import ( CHANNEL_ENTRY, AcquireMethod, CapabilityDeclaration, CapabilityProof, CapabilityProver, ChannelManifest, ConfigSource, ConflictStrategy, CredentialStrategy, CredentialType, EventSubscription, FailurePolicy, FailureStrategy, LifecycleHook, LifecycleHookHandler, LifecycleState, PluginDependency, PluginHost, PluginManifest, ResourceQuota, Stage, StageSlot, ) pytestmark = pytest.mark.unit def _make_instance(protocol_cls: type[Protocol]) -> Any: """构造 Protocol 的最小实现实例。 Python 3.12 中 ``@runtime_checkable`` Protocol 的 ``isinstance`` 不再把 未配置属性的 ``MagicMock`` 视为实现类；本辅助函数为协议声明的数据属性 提供占位值并继承协议默认方法体，用于测试协议默认行为。 """ attrs: dict[str, Any] = {} for cls in protocol_cls.__mro__: if cls is object: continue for name in getattr(cls, "__annotations__", {}): if name.startswith("_") or name in attrs: continue if hasattr(cls, name) and callable(getattr(cls, name)): continue attrs[name] = None impl_name = f"_{protocol_cls.__name__}Impl" impl = type(impl_name, (protocol_cls,), attrs) return impl() # ─── 辅助工厂 ──────────────────────────────────────────────────────────── def _build_channel_manifest(**kwargs) -> ChannelManifest: """构造最小可用 ChannelManifest 实例，便于测试覆写字段。""" defaults: dict = dict( id="com.yuxi.channels.feishu", name="feishu", version="1.0.0", channel_type=ChannelType("feishu"), provides=("inbound", "outbound"), entry_module="yuxi.channels.plugins.feishu.entry", capabilities=ChannelCapabilities(), config_schema=(ConfigField(key="app_id", type="str"),), ) defaults.update(kwargs) return ChannelManifest(**defaults) # ─── __all__ 导出完整性 ────────────────────────────────────────────────── @pytest.mark.unit class TestPluginExports: """plugin 包 ``__all__`` 导出完整性。""" def test_all_contains_expected_core_names(self): # Arrange - 期望 __all__ 中包含的核心协议与值对象名称 from yuxi.channels.contract.plugin import __all__ as plugin_all expected = { # manifest "AcquireMethod", "ChannelManifest", "ConfigField", "CredentialStrategy", "CredentialType", "FailurePolicy", "PluginDependency", "PluginManifest", "ResourceQuota", # entry "CHANNEL_ENTRY", "PluginHost", # capability "CapabilityDeclaration", "CapabilityProof", "CapabilityProver", # lifecycle "LifecycleHook", "LifecycleHookHandler", "LifecycleState", # extension_point "ConflictStrategy", "ConfigSource", "EventSubscription", "FailureStrategy", "Stage", "StageSlot", } # Act # Assert assert expected.issubset(set(plugin_all)) def test_all_contains_adapter_names(self): # Arrange - 期望 __all__ 中包含的适配器协议名称（抽样） from yuxi.channels.contract.plugin import __all__ as plugin_all expected_adapters = { "InboundAdapter", "OutboundAdapter", "SessionAdapter", "RichMessageAdapter", "StreamingAdapter", "DirectoryAdapter", "CommandAdapter", "WizardAdapter", "DoctorAdapter", "WhitelistAdapter", "ToolsAdapter", "MessageOpsAdapter", "StatusAdapter", "MentionAdapter", "IdentityResolverAdapter", "LifecycleAdapter", "LoginAdapter", "ProbeableAdapter", "ContentModerationAdapter", "PullerAdapter", "StreamConnectorAdapter", "WebhookTestAdapter", "AttachmentUploadAdapter", } # Act # Assert assert expected_adapters.issubset(set(plugin_all)) def test_all_names_importable(self): # Arrange from yuxi.channels.contract import plugin as mod # Act # Assert - __all__ 中每个名称必须可在模块属性中找到 for name in mod.__all__: assert hasattr(mod, name), f"{name} 在 __all__ 但未定义" def test_all_list_meets_minimum_count(self): # Arrange - 当前 __all__ 至少 40 个名称（23 适配器 + 17 协议/枚举/值对象） from yuxi.channels.contract.plugin import __all__ as plugin_all # Act # Assert assert len(plugin_all) >= 40 def test_all_names_unique(self): # Arrange from yuxi.channels.contract.plugin import __all__ as plugin_all # Act # Assert assert len(plugin_all) == len(set(plugin_all)) # ─── manifest.py 枚举 ─────────────────────────────────────────────────── @pytest.mark.unit class TestManifestEnums: """manifest.py 的 FailurePolicy / CredentialType / AcquireMethod 枚举。""" def test_failure_policy_inherits_strenum(self): # Arrange # Act # Assert assert issubclass(FailurePolicy, StrEnum) def test_failure_policy_values(self): # Arrange # Act # Assert - 三个取值，字符串值与枚举名一致（小写） assert FailurePolicy.DEGRADE == "degrade" assert FailurePolicy.CIRCUIT_BREAK == "circuit_break" assert FailurePolicy.ISOLATE == "isolate" def test_failure_policy_member_count(self): # Arrange # Act # Assert assert len(list(FailurePolicy)) == 3 def test_failure_policy_str_value(self): # Arrange - StrEnum 的 str() 返回值本身 # Act # Assert assert str(FailurePolicy.DEGRADE) == "degrade" def test_credential_type_inherits_strenum(self): # Arrange # Act # Assert assert issubclass(CredentialType, StrEnum) def test_credential_type_values(self): # Arrange # Act # Assert assert CredentialType.STATIC == "static" assert CredentialType.DYNAMIC == "dynamic" assert CredentialType.HYBRID == "hybrid" def test_credential_type_member_count(self): # Arrange # Act # Assert assert len(list(CredentialType)) == 3 def test_acquire_method_inherits_strenum(self): # Arrange # Act # Assert assert issubclass(AcquireMethod, StrEnum) def test_acquire_method_values(self): # Arrange # Act # Assert assert AcquireMethod.MANUAL == "manual" assert AcquireMethod.QR_LOGIN == "qr_login" assert AcquireMethod.OAUTH == "oauth" assert AcquireMethod.WEBHOOK_VERIFY == "webhook_verify" def test_acquire_method_member_count(self): # Arrange # Act # Assert assert len(list(AcquireMethod)) == 4 # ─── PluginDependency ─────────────────────────────────────────────────── @pytest.mark.unit class TestPluginDependency: """PluginDependency 不可变值对象。""" def test_required_fields_set(self): # Arrange # Act dep = PluginDependency(plugin_id="p-1", version_range="^1.0.0") # Assert assert dep.plugin_id == "p-1" assert dep.version_range == "^1.0.0" def test_is_frozen_dataclass(self): # Arrange dep = PluginDependency(plugin_id="p-1", version_range="^1.0.0") # Act & Assert - frozen dataclass 字段不可变 with pytest.raises(Exception): dep.plugin_id = "p-2" # type: ignore[misc] def test_version_range_immutable(self): # Arrange dep = PluginDependency(plugin_id="p-1", version_range="^1.0.0") # Act & Assert with pytest.raises(Exception): dep.version_range = "^2.0.0" # type: ignore[misc] # ─── ResourceQuota ────────────────────────────────────────────────────── @pytest.mark.unit class TestResourceQuota: """ResourceQuota 不可变值对象，所有字段可选。""" def test_all_fields_default_none(self): # Arrange # Act quota = ResourceQuota() # Assert - 所有字段默认 None assert quota.max_cpu is None assert quota.max_memory is None assert quota.max_connections is None assert quota.max_calls_per_sec is None def test_fields_settable(self): # Arrange # Act quota = ResourceQuota( max_cpu="500m", max_memory="512Mi", max_connections=100, max_calls_per_sec=50, ) # Assert assert quota.max_cpu == "500m" assert quota.max_memory == "512Mi" assert quota.max_connections == 100 assert quota.max_calls_per_sec == 50 def test_is_frozen_dataclass(self): # Arrange quota = ResourceQuota(max_cpu="500m") # Act & Assert with pytest.raises(Exception): quota.max_cpu = "1000m" # type: ignore[misc] def test_max_connections_immutable(self): # Arrange quota = ResourceQuota(max_connections=10) # Act & Assert with pytest.raises(Exception): quota.max_connections = 20 # type: ignore[misc] # ─── CredentialStrategy ───────────────────────────────────────────────── @pytest.mark.unit class TestCredentialStrategy: """CredentialStrategy 不可变值对象。""" def test_required_fields_set(self): # Arrange # Act strategy = CredentialStrategy( type=CredentialType.STATIC, acquire_method=AcquireMethod.MANUAL, required_fields=("app_id", "app_secret"), ) # Assert assert strategy.type is CredentialType.STATIC assert strategy.acquire_method is AcquireMethod.MANUAL assert strategy.required_fields == ("app_id", "app_secret") def test_optional_fields_defaults(self): # Arrange # Act strategy = CredentialStrategy( type=CredentialType.DYNAMIC, acquire_method=AcquireMethod.QR_LOGIN, required_fields=("token",), ) # Assert - optional_fields 默认空元组，supports_rotation/supports_revocation 默认 False assert strategy.optional_fields == () assert strategy.supports_rotation is False assert strategy.supports_revocation is False def test_optional_fields_settable(self): # Arrange # Act strategy = CredentialStrategy( type=CredentialType.HYBRID, acquire_method=AcquireMethod.OAUTH, required_fields=("app_id",), optional_fields=("user_token",), supports_rotation=True, supports_revocation=True, ) # Assert assert strategy.optional_fields == ("user_token",) assert strategy.supports_rotation is True assert strategy.supports_revocation is True def test_is_frozen_dataclass(self): # Arrange strategy = CredentialStrategy( type=CredentialType.STATIC, acquire_method=AcquireMethod.MANUAL, required_fields=("app_id",), ) # Act & Assert with pytest.raises(Exception): strategy.type = CredentialType.DYNAMIC # type: ignore[misc] def test_required_fields_immutable(self): # Arrange strategy = CredentialStrategy( type=CredentialType.STATIC, acquire_method=AcquireMethod.MANUAL, required_fields=("app_id",), ) # Act & Assert with pytest.raises(Exception): strategy.required_fields = ("other",) # type: ignore[misc] # ─── ChannelManifest ──────────────────────────────────────────────────── @pytest.mark.unit class TestChannelManifest: """ChannelManifest 不可变值对象。""" def test_required_fields_set(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.id == "com.yuxi.channels.feishu" assert manifest.name == "feishu" assert manifest.version == "1.0.0" assert manifest.channel_type == ChannelType("feishu") assert manifest.provides == ("inbound", "outbound") assert manifest.entry_module == "yuxi.channels.plugins.feishu.entry" assert isinstance(manifest.capabilities, ChannelCapabilities) assert isinstance(manifest.config_schema, tuple) assert isinstance(manifest.config_schema[0], ConfigField) def test_lifecycle_default(self): # Arrange # Act manifest = _build_channel_manifest() # Assert - 默认生命周期为 init/start/stop/unload assert manifest.lifecycle == ("init", "start", "stop", "unload") def test_failure_policy_default(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.failure_policy is FailurePolicy.DEGRADE def test_critical_default_false(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.critical is False def test_requires_dm_pairing_default_true(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.requires_dm_pairing is True def test_requires_outbound_delivery_default_true(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.requires_outbound_delivery is True def test_credential_strategy_default_none(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.credential_strategy is None def test_depends_default_empty_tuple(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.depends == () def test_compatibility_default_none(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.compatibility is None def test_resource_quota_default_none(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.resource_quota is None def test_accessible_ports_default_empty_tuple(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.accessible_ports == () def test_injectable_pipelines_default_empty_tuple(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.injectable_pipelines == () def test_skills_default_empty_tuple(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.skills == () def test_env_vars_default_empty_tuple(self): # Arrange # Act manifest = _build_channel_manifest() # Assert assert manifest.env_vars == () def test_capability_requirements_default_empty_tuple(self): # Arrange # Act manifest = _build_channel_manifest() # Assert - 能力运行时依赖声明默认空 assert manifest.capability_requirements == () def test_max_message_length_default_4096(self): # Arrange # Act manifest = _build_channel_manifest() # Assert - 单条消息最大文本长度默认 4096（FR19-P0-5） assert manifest.max_message_length == 4096 def test_supports_credential_cloning_default_false(self): # Arrange # Act manifest = _build_channel_manifest() # Assert - 默认不允许克隆时复制凭据（ACC-CLONE，fail-closed） assert manifest.supports_credential_cloning is False def test_icon_default_none(self): # Arrange # Act manifest = _build_channel_manifest() # Assert - 渠道展示图标名默认 None，前端回退默认图标 assert manifest.icon is None def test_transport_mode_default_both(self): # Arrange # Act manifest = _build_channel_manifest() # Assert - 传输模式默认 both（Task 11.3） assert manifest.transport_mode == "both" def test_max_restart_attempts_default_none(self): # Arrange # Act manifest = _build_channel_manifest() # Assert - None 表示无限重启 assert manifest.max_restart_attempts is None def test_transport_mode_settable(self): # Arrange # Act manifest = _build_channel_manifest(transport_mode="pull") # Assert assert manifest.transport_mode == "pull" def test_max_message_length_settable(self): # Arrange # Act manifest = _build_channel_manifest(max_message_length=8192) # Assert assert manifest.max_message_length == 8192 def test_credential_strategy_settable(self): # Arrange strategy = CredentialStrategy( type=CredentialType.STATIC, acquire_method=AcquireMethod.MANUAL, required_fields=("app_id",), ) # Act manifest = _build_channel_manifest(credential_strategy=strategy) # Assert assert manifest.credential_strategy is strategy def test_is_frozen_dataclass(self): # Arrange manifest = _build_channel_manifest() # Act & Assert with pytest.raises(Exception): manifest.id = "other" # type: ignore[misc] def test_critical_immutable(self): # Arrange manifest = _build_channel_manifest() # Act & Assert with pytest.raises(Exception): manifest.critical = True # type: ignore[misc] # ─── PluginManifest ───────────────────────────────────────────────────── @pytest.mark.unit class TestPluginManifest: """PluginManifest 不可变值对象。""" def test_required_field_manifest(self): # Arrange channel_manifest = _build_channel_manifest() # Act plugin_manifest = PluginManifest(manifest=channel_manifest) # Assert assert plugin_manifest.manifest is channel_manifest def test_adapters_default_empty_tuple(self): # Arrange # Act plugin_manifest = PluginManifest(manifest=_build_channel_manifest()) # Assert assert plugin_manifest.adapters == () def test_stages_default_empty_tuple(self): # Arrange # Act plugin_manifest = PluginManifest(manifest=_build_channel_manifest()) # Assert assert plugin_manifest.stages == () def test_event_subscriptions_default_empty_tuple(self): # Arrange # Act plugin_manifest = PluginManifest(manifest=_build_channel_manifest()) # Assert assert plugin_manifest.event_subscriptions == () def test_config_sources_default_empty_tuple(self): # Arrange # Act plugin_manifest = PluginManifest(manifest=_build_channel_manifest()) # Assert assert plugin_manifest.config_sources == () def test_installed_at_default_none(self): # Arrange # Act plugin_manifest = PluginManifest(manifest=_build_channel_manifest()) # Assert assert plugin_manifest.installed_at is None def test_install_source_default_none(self): # Arrange # Act plugin_manifest = PluginManifest(manifest=_build_channel_manifest()) # Assert assert plugin_manifest.install_source is None def test_install_version_default_none(self): # Arrange # Act plugin_manifest = PluginManifest(manifest=_build_channel_manifest()) # Assert assert plugin_manifest.install_version is None def test_optional_fields_settable(self): # Arrange now = datetime(2026, 1, 1, 0, 0, 0) # Act plugin_manifest = PluginManifest( manifest=_build_channel_manifest(), adapters=("InboundAdapter",), stages=("stage-1",), event_subscriptions=("PairingApproved",), config_sources=("source-1",), installed_at=now, install_source="/path/to/plugin", install_version="1.2.3", ) # Assert assert plugin_manifest.adapters == ("InboundAdapter",) assert plugin_manifest.stages == ("stage-1",) assert plugin_manifest.event_subscriptions == ("PairingApproved",) assert plugin_manifest.config_sources == ("source-1",) assert plugin_manifest.installed_at == now assert plugin_manifest.install_source == "/path/to/plugin" assert plugin_manifest.install_version == "1.2.3" def test_is_frozen_dataclass(self): # Arrange plugin_manifest = PluginManifest(manifest=_build_channel_manifest()) # Act & Assert with pytest.raises(Exception): plugin_manifest.manifest = _build_channel_manifest(id="other") # type: ignore[misc] # ─── lifecycle.py 枚举 ────────────────────────────────────────────────── @pytest.mark.unit class TestLifecycleEnums: """lifecycle.py 的 LifecycleState / LifecycleHook 枚举。""" def test_lifecycle_state_inherits_strenum(self): # Arrange # Act # Assert assert issubclass(LifecycleState, StrEnum) def test_lifecycle_state_member_count(self): # Arrange # Act # Assert - 11 个状态 assert len(list(LifecycleState)) == 11 def test_lifecycle_state_values(self): # Arrange # Act # Assert assert LifecycleState.DISCOVERED == "discovered" assert LifecycleState.RESOLVED == "resolved" assert LifecycleState.LOADED == "loaded" assert LifecycleState.INITIALIZED == "initialized" assert LifecycleState.STARTED == "started" assert LifecycleState.PAUSED == "paused" assert LifecycleState.STOPPED == "stopped" assert LifecycleState.UNLOADED == "unloaded" assert LifecycleState.FAILED == "failed" assert LifecycleState.NOT_INSTALLED == "not_installed" assert LifecycleState.INSTALLED == "installed" def test_lifecycle_hook_inherits_strenum(self): # Arrange # Act # Assert assert issubclass(LifecycleHook, StrEnum) def test_lifecycle_hook_member_count(self): # Arrange # Act # Assert - 8 个钩子 assert len(list(LifecycleHook)) == 8 def test_lifecycle_hook_values(self): # Arrange # Act # Assert assert LifecycleHook.INIT == "init" assert LifecycleHook.START == "start" assert LifecycleHook.STOP == "stop" assert LifecycleHook.PAUSE == "pause" assert LifecycleHook.RESUME == "resume" assert LifecycleHook.UNLOAD == "unload" assert LifecycleHook.RECONFIGURE == "reconfigure" assert LifecycleHook.FAIL == "fail" # ─── LifecycleHookHandler Protocol ────────────────────────────────────── @pytest.mark.unit class TestLifecycleHookHandlerProtocol: """LifecycleHookHandler 为 runtime_checkable Protocol。""" def test_is_protocol_subclass(self): # Arrange # Act # Assert assert issubclass(LifecycleHookHandler, Protocol) def test_runtime_checkable_isinstance_works(self): # Arrange - 使用最小实现实例替代 MagicMock，兼容 Python 3.12 # runtime_checkable Protocol 的 isinstance 行为 # Act # Assert assert isinstance(_make_instance(LifecycleHookHandler), LifecycleHookHandler) def test_protocol_has_expected_methods(self): # Arrange - LifecycleHookHandler 定义了 8 个异步钩子方法 expected_methods = [ "onInit", "onStart", "onStop", "onPause", "onResume", "onUnload", "onReconfigure", "onFail", ] # Act # Assert for name in expected_methods: assert hasattr(LifecycleHookHandler, name), f"缺少方法 {name}" def test_onReconfigure_signature_has_old_config_and_new_config(self): # Arrange - F-08 签名扩展：onReconfigure 接收关键字参数 old_config / new_config # Act sig = inspect.signature(LifecycleHookHandler.onReconfigure) params = sig.parameters # Assert - 包含 old_config 与 new_config 参数 assert "old_config" in params, "onReconfigure 缺少 old_config 参数" assert "new_config" in params, "onReconfigure 缺少 new_config 参数" # old_config 默认 None（向后兼容），new_config 必填 assert params["old_config"].default is None assert params["new_config"].default is inspect.Parameter.empty # 两者均为关键字参数（KEYWORD_ONLY），框架通过关键字传递 assert params["old_config"].kind is inspect.Parameter.KEYWORD_ONLY assert params["new_config"].kind is inspect.Parameter.KEYWORD_ONLY @pytest.mark.asyncio async def test_default_hook_methods_return_none(self): # Arrange - LifecycleHookHandler 方法体为 ``...``，默认返回 None # （与 capability.py 的 proveCapability 抛异常不同） instance = _make_instance(LifecycleHookHandler) # Act & Assert - 无参钩子默认返回 None assert await instance.onInit() is None assert await instance.onStart() is None assert await instance.onStop() is None assert await instance.onPause() is None assert await instance.onResume() is None assert await instance.onUnload() is None @pytest.mark.asyncio async def test_on_reconfigure_default_returns_none(self): # Arrange - onReconfigure 默认方法体 ``...`` 返回 None instance = _make_instance(LifecycleHookHandler) # Act result = await instance.onReconfigure(new_config={"k": "v"}) # Assert assert result is None @pytest.mark.asyncio async def test_on_fail_default_returns_none(self): # Arrange - onFail 接收 error 参数，默认方法体 ``...`` 返回 None instance = _make_instance(LifecycleHookHandler) # Act result = await instance.onFail("boom") # Assert assert result is None # ─── extension_point.py 枚举 ──────────────────────────────────────────── @pytest.mark.unit class TestExtensionPointEnums: """extension_point.py 的 ConflictStrategy / FailureStrategy 枚举。""" def test_conflict_strategy_inherits_strenum(self): # Arrange # Act # Assert assert issubclass(ConflictStrategy, StrEnum) def test_conflict_strategy_member_count(self): # Arrange # Act # Assert - 3 个冲突策略 assert len(list(ConflictStrategy)) == 3 def test_conflict_strategy_values(self): # Arrange # Act # Assert assert ConflictStrategy.CHAIN == "chain" assert ConflictStrategy.REJECT == "reject" assert ConflictStrategy.OVERRIDE == "override" def test_failure_strategy_inherits_strenum(self): # Arrange # Act # Assert assert issubclass(FailureStrategy, StrEnum) def test_failure_strategy_member_count(self): # Arrange # Act # Assert - 4 个阶段失败策略 assert len(list(FailureStrategy)) == 4 def test_failure_strategy_values(self): # Arrange # Act # Assert assert FailureStrategy.TERMINATE == "terminate" assert FailureStrategy.SKIP == "skip" assert FailureStrategy.COMPENSATE == "compensate" assert FailureStrategy.DEGRADE == "degrade" # ─── StageSlot ────────────────────────────────────────────────────────── @pytest.mark.unit class TestStageSlot: """StageSlot 不可变值对象。""" def test_required_fields_set(self): # Arrange - stage 字段为 Stage Protocol 实例，使用 MagicMock 模拟 stage = MagicMock() # Act slot = StageSlot(pipeline="inbound", anchor="before:route", stage=stage) # Assert assert slot.pipeline == "inbound" assert slot.anchor == "before:route" assert slot.stage is stage def test_priority_default_100(self): # Arrange # Act slot = StageSlot(pipeline="inbound", anchor="before:route", stage=MagicMock()) # Assert assert slot.priority == 100 def test_multi_default_false(self): # Arrange # Act slot = StageSlot(pipeline="inbound", anchor="before:route", stage=MagicMock()) # Assert assert slot.multi is False def test_conflict_strategy_default_chain(self): # Arrange # Act slot = StageSlot(pipeline="inbound", anchor="before:route", stage=MagicMock()) # Assert assert slot.conflict_strategy is ConflictStrategy.CHAIN def test_failure_policy_default_degrade(self): # Arrange # Act slot = StageSlot(pipeline="inbound", anchor="before:route", stage=MagicMock()) # Assert assert slot.failure_policy is FailurePolicy.DEGRADE def test_optional_fields_settable(self): # Arrange # Act slot = StageSlot( pipeline="outbound", anchor="after:format", stage=MagicMock(), priority=50, multi=True, conflict_strategy=ConflictStrategy.REJECT, failure_policy=FailurePolicy.ISOLATE, ) # Assert assert slot.priority == 50 assert slot.multi is True assert slot.conflict_strategy is ConflictStrategy.REJECT assert slot.failure_policy is FailurePolicy.ISOLATE def test_is_frozen_dataclass(self): # Arrange slot = StageSlot(pipeline="inbound", anchor="before:route", stage=MagicMock()) # Act & Assert with pytest.raises(Exception): slot.pipeline = "outbound" # type: ignore[misc] def test_priority_immutable(self): # Arrange slot = StageSlot(pipeline="inbound", anchor="before:route", stage=MagicMock()) # Act & Assert with pytest.raises(Exception): slot.priority = 200 # type: ignore[misc] # ─── EventSubscription ────────────────────────────────────────────────── @pytest.mark.unit class TestEventSubscription: """EventSubscription 不可变值对象。""" def test_required_fields_set(self): # Arrange - handler 字段为 EventHandler Protocol 实例，使用 MagicMock 模拟 handler = MagicMock() # Act sub = EventSubscription(event_type="PairingApproved", handler=handler) # Assert assert sub.event_type == "PairingApproved" assert sub.handler is handler def test_priority_default_100(self): # Arrange # Act sub = EventSubscription(event_type="PairingApproved", handler=MagicMock()) # Assert assert sub.priority == 100 def test_failure_policy_default_degrade(self): # Arrange # Act sub = EventSubscription(event_type="PairingApproved", handler=MagicMock()) # Assert assert sub.failure_policy is FailurePolicy.DEGRADE def test_optional_fields_settable(self): # Arrange # Act sub = EventSubscription( event_type="ConfigChanged", handler=MagicMock(), priority=50, failure_policy=FailurePolicy.CIRCUIT_BREAK, ) # Assert assert sub.priority == 50 assert sub.failure_policy is FailurePolicy.CIRCUIT_BREAK def test_is_frozen_dataclass(self): # Arrange sub = EventSubscription(event_type="PairingApproved", handler=MagicMock()) # Act & Assert with pytest.raises(Exception): sub.event_type = "other" # type: ignore[misc] # ─── ConfigSource ─────────────────────────────────────────────────────── @pytest.mark.unit class TestConfigSource: """ConfigSource 不可变值对象。""" def test_required_fields_set(self): # Arrange - loader 字段为 ConfigLoader Protocol 实例，使用 MagicMock 模拟 loader = MagicMock() # Act source = ConfigSource(source_id="src-1", loader=loader) # Assert assert source.source_id == "src-1" assert source.loader is loader def test_decryptor_default_none(self): # Arrange # Act source = ConfigSource(source_id="src-1", loader=MagicMock()) # Assert assert source.decryptor is None def test_failure_policy_default_degrade(self): # Arrange # Act source = ConfigSource(source_id="src-1", loader=MagicMock()) # Assert assert source.failure_policy is FailurePolicy.DEGRADE def test_optional_fields_settable(self): # Arrange # Act source = ConfigSource( source_id="src-1", loader=MagicMock(), decryptor=MagicMock(), failure_policy=FailurePolicy.ISOLATE, ) # Assert assert source.decryptor is not None assert source.failure_policy is FailurePolicy.ISOLATE def test_is_frozen_dataclass(self): # Arrange source = ConfigSource(source_id="src-1", loader=MagicMock()) # Act & Assert with pytest.raises(Exception): source.source_id = "other" # type: ignore[misc] # ─── Stage Protocol ───────────────────────────────────────────────────── @pytest.mark.unit class TestStageProtocol: """Stage 为 runtime_checkable Protocol。""" def test_is_protocol_subclass(self): # Arrange # Act # Assert assert issubclass(Stage, Protocol) def test_runtime_checkable_isinstance_works(self): # Arrange - 使用最小实现实例替代 MagicMock，兼容 Python 3.12 # runtime_checkable Protocol 的 isinstance 行为 # Act # Assert assert isinstance(_make_instance(Stage), Stage) def test_protocol_has_expected_attributes(self): # Arrange - Stage 定义了 id/reads/writes/idempotent/thread_safe/failure/compensate 属性 expected_attrs = [ "id", "reads", "writes", "idempotent", "thread_safe", "failure", "compensate", "process", ] # Act annotations = getattr(Stage, "__annotations__", {}) # Assert: 数据属性在 __annotations__ 中声明，process 为方法 for name in expected_attrs: assert name in annotations or hasattr(Stage, name), f"Stage 缺少属性 {name}" # ─── CapabilityDeclaration ────────────────────────────────────────────── @pytest.mark.unit class TestCapabilityDeclaration: """CapabilityDeclaration 不可变值对象。""" def test_capabilities_field_set(self): # Arrange caps = ChannelCapabilities(rich_message=True, streaming=True) # Act decl = CapabilityDeclaration(capabilities=caps) # Assert assert decl.capabilities is caps assert decl.capabilities.rich_message is True assert decl.capabilities.streaming is True def test_default_capabilities(self): # Arrange # Act decl = CapabilityDeclaration(capabilities=ChannelCapabilities()) # Assert assert isinstance(decl.capabilities, ChannelCapabilities) assert decl.capabilities.rich_message is False def test_is_frozen_dataclass(self): # Arrange decl = CapabilityDeclaration(capabilities=ChannelCapabilities()) # Act & Assert with pytest.raises(Exception): decl.capabilities = ChannelCapabilities() # type: ignore[misc] # ─── CapabilityProof ──────────────────────────────────────────────────── @pytest.mark.unit class TestCapabilityProof: """CapabilityProof 不可变值对象。""" def test_required_fields_set(self): # Arrange now = datetime(2026, 1, 1, 0, 0, 0) # Act proof = CapabilityProof( capability_name="streaming", proven=True, proof_timestamp=now, ) # Assert assert proof.capability_name == "streaming" assert proof.proven is True assert proof.proof_timestamp == now def test_cache_ttl_default_300(self): # Arrange # Act proof = CapabilityProof( capability_name="streaming", proven=False, proof_timestamp=datetime(2026, 1, 1, 0, 0, 0), ) # Assert assert proof.cache_ttl == 300 def test_cache_ttl_settable(self): # Arrange # Act proof = CapabilityProof( capability_name="streaming", proven=True, proof_timestamp=datetime(2026, 1, 1, 0, 0, 0), cache_ttl=600, ) # Assert assert proof.cache_ttl == 600 def test_is_frozen_dataclass(self): # Arrange proof = CapabilityProof( capability_name="streaming", proven=True, proof_timestamp=datetime(2026, 1, 1, 0, 0, 0), ) # Act & Assert with pytest.raises(Exception): proof.capability_name = "other" # type: ignore[misc] def test_proven_immutable(self): # Arrange proof = CapabilityProof( capability_name="streaming", proven=True, proof_timestamp=datetime(2026, 1, 1, 0, 0, 0), ) # Act & Assert with pytest.raises(Exception): proof.proven = False # type: ignore[misc] # ─── CapabilityProver Protocol ────────────────────────────────────────── @pytest.mark.unit class TestCapabilityProverProtocol: """CapabilityProver 为 runtime_checkable Protocol。""" def test_is_protocol_subclass(self): # Arrange # Act # Assert assert issubclass(CapabilityProver, Protocol) def test_runtime_checkable_isinstance_works(self): # Arrange - 使用最小实现实例替代 MagicMock，兼容 Python 3.12 # runtime_checkable Protocol 的 isinstance 行为 # Act # Assert assert isinstance(_make_instance(CapabilityProver), CapabilityProver) def test_protocol_has_prove_capability_method(self): # Arrange # Act # Assert assert hasattr(CapabilityProver, "proveCapability") @pytest.mark.asyncio async def test_prove_capability_default_raises_not_implemented(self): # Arrange - proveCapability 默认方法体 raise NotImplementedError instance = _make_instance(CapabilityProver) # Act & Assert with pytest.raises(NotImplementedError) as exc_info: await instance.proveCapability("streaming") assert exc_info.value.operation == "proveCapability" # ─── PluginHost Protocol ──────────────────────────────────────────────── @pytest.mark.unit class TestPluginHostProtocol: """PluginHost 为 runtime_checkable Protocol。""" def test_is_protocol_subclass(self): # Arrange # Act # Assert assert issubclass(PluginHost, Protocol) def test_runtime_checkable_isinstance_works(self): # Arrange - 使用最小实现实例替代 MagicMock，兼容 Python 3.12 # runtime_checkable Protocol 的 isinstance 行为 # Act # Assert assert isinstance(_make_instance(PluginHost), PluginHost) def test_protocol_has_port_getter_methods(self): # Arrange - PluginHost 定义了若干 getXxxPort 方法 expected_methods = [ "getConfigPort", "getLoggerPort", "getPersistencePort", "getCachePort", "getTracerPort", "getQueuePort", "getConversationPort", "getAgentRunPort", "getIdentityResolverPort", "getChannelAccountRepositoryPort", "getChannelSessionRepositoryPort", "getPairingRepositoryPort", "getAuditLogRepositoryPort", "getOutboxRepositoryPort", "getUserIdentityRepositoryPort", "getIdempotencyRepositoryPort", "getPersistenceHealthPort", ] # Act # Assert for name in expected_methods: assert hasattr(PluginHost, name), f"PluginHost 缺少方法 {name}" def test_protocol_has_register_methods(self): # Arrange - PluginHost 定义了若干 registerXxx 方法 expected_methods = [ "registerAdapter", "registerStageSlot", "registerEventSubscription", "registerConfigSource", "registerCapabilityProver", "registerMatchTier", "registerMatcher", ] # Act # Assert for name in expected_methods: assert hasattr(PluginHost, name), f"PluginHost 缺少方法 {name}" def test_protocol_no_declare_methods(self): # Arrange - PluginHost 不再定义 declareXxx 方法 # （能力边界声明已统一到 ChannelManifest 静态字段） removed_methods = [ "declareAccessiblePorts", "declareInjectablePipelines", "declareResourceQuota", ] # Act # Assert for name in removed_methods: assert not hasattr(PluginHost, name), f"PluginHost 不应再定义 {name}（已统一到 ChannelManifest）" def test_protocol_has_publish_event_method(self): # Arrange # Act # Assert assert hasattr(PluginHost, "publishEvent") # ─── CHANNEL_ENTRY 类型别名 ───────────────────────────────────────────── @pytest.mark.unit class TestChannelEntryAlias: """CHANNEL_ENTRY 为 Callable 类型别名。""" def test_is_callable_origin(self): # Arrange # Act origin = get_origin(CHANNEL_ENTRY) # Assert - CHANNEL_ENTRY 的 origin 为 collections.abc.Callable assert origin is Callable def test_args_contain_plugin_host_and_manifest(self): # Arrange - CHANNEL_ENTRY = Callable[[PluginHost, ChannelManifest], PluginManifest] # Act args = get_args(CHANNEL_ENTRY) # Assert - 参数列表含 PluginHost 与 ChannelManifest，返回类型为 PluginManifest assert PluginManifest in args # args 结构为 ([PluginHost, ChannelManifest], PluginManifest)，参数列表在 args[0] param_args = args[0] assert PluginHost in param_args assert ChannelManifest in param_args def test_callable_with_plugin_host_returns_manifest(self): # Arrange - 构造一个符合 CHANNEL_ENTRY 签名的函数（2 参数） def entry(host: PluginHost, manifest: ChannelManifest) -> PluginManifest: return PluginManifest(manifest=manifest) # Act result = entry(MagicMock(), _build_channel_manifest()) # Assert assert isinstance(result, PluginManifest) # ─── PluginAdapter Protocol ───────────────────────────────────────────── @pytest.mark.unit class TestPluginAdapterProtocol: """PluginAdapter 为插件适配器标记 Protocol。 标识 ``PluginHost.registerAdapter`` 接受的适配器实例类型，不含公共方法 声明，仅作为类型边界用于静态检查与文档化。 """ def test_is_protocol_subclass(self): # Arrange from yuxi.channels.contract.plugin.entry import PluginAdapter # Act # Assert assert issubclass(PluginAdapter, Protocol) def test_has_no_public_methods(self): # Arrange - PluginAdapter 为标记 Protocol，仅含 ``...`` 占位 from yuxi.channels.contract.plugin.entry import PluginAdapter # Act # Assert - 不含业务方法声明（23 类适配器各自定义独立方法，无公共基类） public_methods = [ name for name in dir(PluginAdapter) if not name.startswith("_") and callable(getattr(PluginAdapter, name, None)) ] # Protocol 自身引入的方法（如 ``__init__`` 等双下划线已排除）， # 标记 Protocol 不应声明任何业务方法 assert all(name.startswith("__") for name in public_methods) or not public_methods # ─── Matcher 类型别名 ─────────────────────────────────────────────────── @pytest.mark.unit class TestMatcherAlias: """Matcher 为路由匹配器类型别名（``Callable[..., Any]``）。""" def test_is_callable_origin(self): # Arrange from yuxi.channels.contract.plugin.entry import Matcher # Act origin = get_origin(Matcher) # Assert assert origin is Callable def test_accepts_any_callable(self): # Arrange - Matcher = Callable[..., Any]，任意可调用对象均符合 from yuxi.channels.contract.plugin.entry import Matcher def matcher(*args, **kwargs): return None # Act & Assert - 类型别名仅为静态检查用途，运行时不强制 assert callable(matcher) assert get_origin(Matcher) is Callable