"""entry 模块单元测试。 覆盖 ``channel_entry(host)`` 入口函数（端口声明、适配器注册、生命周期钩子 注册、PluginManifest 返回）、``_build_config_schema()``（18 个 ConfigField） 与 ``_build_env_vars()``（7 个 EnvVar）。使用 fake host 记录所有调用， 验证 18 个适配器均被注册且 manifest 字段正确。 """ from __future__ import annotations from unittest.mock import AsyncMock, MagicMock import pytest from yuxi.channels.contract.dtos.channel import ChannelType from yuxi.channels.contract.dtos.config import ConfigField from yuxi.channels.contract.dtos.plugin import EnvVar from yuxi.channels.contract.plugin.manifest import ( FailurePolicy, PluginManifest, ) from yuxi.channels.plugins.feishu.entry import ( _ACCESSIBLE_PORTS, _ADAPTER_TYPES, _INJECTABLE_PIPELINES, _build_config_schema, _build_env_vars, channel_entry, ) from yuxi.channels.plugins.feishu.lifecycle import FeishuLifecycleHandler pytestmark = pytest.mark.unit # ------------------------------------------------------------------ # 桩辅助 # ------------------------------------------------------------------ class _FakeHost: """PluginHost 替身，记录所有注册调用。""" def __init__(self) -> None: self.config_port: AsyncMock = AsyncMock() self.logger_port: AsyncMock = AsyncMock() self.cache_port: AsyncMock = AsyncMock() self.persistence_port: AsyncMock = AsyncMock() self.registered_adapters: dict[str, object] = {} self.lifecycle_handler: object | None = None def getConfigPort(self) -> AsyncMock: return self.config_port def getLoggerPort(self) -> AsyncMock: return self.logger_port def getCachePort(self) -> AsyncMock: return self.cache_port def getPersistencePort(self) -> AsyncMock: return self.persistence_port def registerAdapter(self, adapter_type: str, adapter: object) -> None: self.registered_adapters[adapter_type] = adapter def registerLifecycleHandler(self, handler: object) -> None: self.lifecycle_handler = handler # ------------------------------------------------------------------ # channel_entry # ------------------------------------------------------------------ @pytest.mark.unit class TestChannelEntry: """channel_entry 入口函数测试。""" def test_manifest_declares_accessible_ports(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert - 能力边界由 manifest 静态字段声明（不再运行时 declareXxx） assert manifest.manifest.accessible_ports == _ACCESSIBLE_PORTS assert len(manifest.manifest.accessible_ports) == 7 def test_manifest_declares_injectable_pipelines(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert assert manifest.manifest.injectable_pipelines == _INJECTABLE_PIPELINES assert "inbound" in manifest.manifest.injectable_pipelines assert "outbound" in manifest.manifest.injectable_pipelines def test_manifest_declares_resource_quota(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert quota = manifest.manifest.resource_quota assert quota is not None assert quota.max_cpu == "20.0" assert quota.max_memory == "256MB" assert quota.max_connections == 50 assert quota.max_calls_per_sec == 50 def test_registers_all_18_adapters(self) -> None: # Arrange host = _FakeHost() # Act channel_entry(host) # Assert assert len(host.registered_adapters) == 18 for adapter_type in _ADAPTER_TYPES: assert adapter_type in host.registered_adapters, f"Missing adapter: {adapter_type}" def test_registers_lifecycle_handler(self) -> None: # Arrange host = _FakeHost() # Act channel_entry(host) # Assert assert host.lifecycle_handler is not None assert isinstance(host.lifecycle_handler, FeishuLifecycleHandler) def test_returns_plugin_manifest(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert assert isinstance(manifest, PluginManifest) def test_manifest_has_correct_id_and_name(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert assert manifest.manifest.id == "com.yuxi.channels.feishu" assert manifest.manifest.name == "飞书" assert manifest.manifest.version == "1.0.0" def test_manifest_has_correct_channel_type(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert assert manifest.manifest.channel_type == ChannelType("feishu") def test_manifest_has_correct_failure_policy(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert assert manifest.manifest.failure_policy == FailurePolicy.DEGRADE def test_manifest_adapters_match_registered_types(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert assert manifest.adapters == _ADAPTER_TYPES assert len(manifest.adapters) == 18 def test_manifest_critical_flag_is_false(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert assert manifest.manifest.critical is False def test_manifest_requires_dm_pairing_is_true(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert assert manifest.manifest.requires_dm_pairing is True def test_manifest_capabilities_flags(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert caps = manifest.manifest.capabilities assert caps.rich_message is True assert caps.streaming is True assert caps.typing_indicator is True assert caps.message_edit is True assert caps.message_recall is True assert caps.mention is True assert caps.command is True assert caps.directory is True assert caps.doctor is True assert caps.whitelist is True assert caps.supports_reaction is True assert caps.supports_pin is True assert caps.supports_card_update is True assert caps.supports_card_update_streaming is True assert caps.supports_image_inbound is True assert caps.supports_video_inbound is False assert caps.supports_image_outbound is True assert caps.supports_video_outbound is True assert caps.lifecycle is True assert caps.supports_qr_login is True assert caps.identity_resolver is True def test_manifest_accessible_ports_and_pipelines(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert assert manifest.manifest.accessible_ports == _ACCESSIBLE_PORTS assert manifest.manifest.injectable_pipelines == _INJECTABLE_PIPELINES def test_manifest_lifecycle_hooks(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert assert manifest.manifest.lifecycle == ("init", "start", "stop", "unload") def test_manifest_compatibility_info(self) -> None: # Arrange host = _FakeHost() # Act manifest = channel_entry(host) # Assert assert manifest.manifest.compatibility is not None assert manifest.manifest.compatibility["min_host_version"] == "1.0.0" assert manifest.manifest.compatibility["feishu_api_version"] == "v2" # ------------------------------------------------------------------ # _build_config_schema # ------------------------------------------------------------------ @pytest.mark.unit class TestBuildConfigSchema: """_build_config_schema 配置 schema 构建测试。""" def test_returns_18_config_fields(self) -> None: # Arrange / Act schema = _build_config_schema() # Assert assert len(schema) == 18 assert all(isinstance(f, ConfigField) for f in schema) def test_app_id_not_hot_reloadable_with_pattern(self) -> None: # Arrange / Act schema = _build_config_schema() field = next(f for f in schema if f.key == "app_id") # Assert assert field.required is True assert field.hot_reloadable is False assert field.constraints is not None assert "pattern" in field.constraints def test_app_secret_not_hot_reloadable_and_sensitive(self) -> None: # Arrange / Act schema = _build_config_schema() field = next(f for f in schema if f.key == "app_secret") # Assert assert field.required is True assert field.hot_reloadable is False assert field.constraints is not None assert field.constraints.get("sensitive") is True def test_event_mode_not_hot_reloadable_enum(self) -> None: # Arrange / Act schema = _build_config_schema() field = next(f for f in schema if f.key == "event_mode") # Assert assert field.type == "enum" assert field.hot_reloadable is False assert field.default == "websocket" assert "websocket" in field.constraints["values"] assert "webhook" in field.constraints["values"] def test_webhook_url_not_hot_reloadable(self) -> None: # Arrange / Act schema = _build_config_schema() field = next(f for f in schema if f.key == "webhook_url") # Assert assert field.hot_reloadable is False def test_bot_name_hot_reloadable(self) -> None: # Arrange / Act schema = _build_config_schema() field = next(f for f in schema if f.key == "bot_name") # Assert assert field.required is True assert field.hot_reloadable is True def test_bot_open_id_hot_reloadable_with_pattern(self) -> None: # Arrange / Act schema = _build_config_schema() field = next(f for f in schema if f.key == "bot_open_id") # Assert assert field.hot_reloadable is True assert "pattern" in field.constraints def test_enable_qr_login_boolean_default_false(self) -> None: # Arrange / Act schema = _build_config_schema() field = next(f for f in schema if f.key == "enable_qr_login") # Assert assert field.type == "boolean" assert field.default is False assert field.hot_reloadable is True def test_qr_timeout_ms_has_min_max_constraints(self) -> None: # Arrange / Act schema = _build_config_schema() field = next(f for f in schema if f.key == "qr_timeout_ms") # Assert assert field.type == "integer" assert field.default == 120000 assert field.constraints["min"] == 30000 assert field.constraints["max"] == 300000 def test_api_base_url_has_default(self) -> None: # Arrange / Act schema = _build_config_schema() field = next(f for f in schema if f.key == "api_base_url") # Assert assert field.default == "https://open.feishu.cn" def test_all_keys_unique(self) -> None: # Arrange / Act schema = _build_config_schema() keys = [f.key for f in schema] # Assert assert len(keys) == len(set(keys)) # ------------------------------------------------------------------ # _build_env_vars # ------------------------------------------------------------------ @pytest.mark.unit class TestBuildEnvVars: """_build_env_vars 环境变量声明构建测试。""" def test_returns_7_env_vars(self) -> None: # Arrange / Act env_vars = _build_env_vars() # Assert assert len(env_vars) == 7 assert all(isinstance(v, EnvVar) for v in env_vars) def test_feishu_app_id_required(self) -> None: # Arrange / Act env_vars = _build_env_vars() var = next(v for v in env_vars if v.name == "FEISHU_APP_ID") # Assert assert var.required is True assert var.sensitive is False def test_feishu_app_secret_required_and_sensitive(self) -> None: # Arrange / Act env_vars = _build_env_vars() var = next(v for v in env_vars if v.name == "FEISHU_APP_SECRET") # Assert assert var.required is True assert var.sensitive is True def test_feishu_encrypt_key_optional_and_sensitive(self) -> None: # Arrange / Act env_vars = _build_env_vars() var = next(v for v in env_vars if v.name == "FEISHU_ENCRYPT_KEY") # Assert assert var.required is False assert var.sensitive is True def test_feishu_verification_token_optional_and_sensitive(self) -> None: # Arrange / Act env_vars = _build_env_vars() var = next(v for v in env_vars if v.name == "FEISHU_VERIFICATION_TOKEN") # Assert assert var.required is False assert var.sensitive is True def test_feishu_bot_name_required(self) -> None: # Arrange / Act env_vars = _build_env_vars() var = next(v for v in env_vars if v.name == "FEISHU_BOT_NAME") # Assert assert var.required is True def test_feishu_qr_redirect_uri_optional(self) -> None: # Arrange / Act env_vars = _build_env_vars() var = next(v for v in env_vars if v.name == "FEISHU_QR_REDIRECT_URI") # Assert assert var.required is False def test_feishu_qr_timeout_ms_has_default(self) -> None: # Arrange / Act env_vars = _build_env_vars() var = next(v for v in env_vars if v.name == "FEISHU_QR_TIMEOUT_MS") # Assert assert var.required is False assert var.default == "120000" def test_all_env_var_names_unique(self) -> None: # Arrange / Act env_vars = _build_env_vars() names = [v.name for v in env_vars] # Assert assert len(names) == len(set(names))