Kris
361dbe2149
本次提交修复了多个测试文件中的问题： 1. 为各管道测试类添加tracer_port属性初始化 2. 修复QQBot客户端关闭异常的错误类型匹配 3. 修正配对接口查询的状态值小写格式 4. 移除ChannelType枚举值的显式.value调用 5. 新增QQ常量的导出项与Instagram测试桩函数 6. 修复API接口返回值解析，正确访问data字段 7. 调整企业微信富文本消息的断言结构 8. 修正微信iLink的消息类型测试用例 9. 替换废弃的datetime.utc相关导入为UTC常量 10. 修复QQBot白名单适配器的返回值结构 11. 调整outbox工具的channel_msg_id处理逻辑 12. 新增多个适配器与服务的测试用例，覆盖异常降级、缓存处理等场景 13. 重构部分QQBot适配器的辅助函数测试，清理冗余代码 14. 修复微信iLink客户端的上传接口调用参数与缓存清理逻辑 15. 修正配置导出接口的敏感字段过滤与返回值结构
224 lines
9.5 KiB
Python
224 lines
9.5 KiB
Python
"""wecom 插件 ``_constants.py`` 模块测试。
|
||
|
||
验证 wecom 插件共享常量模块的关键不变量：
|
||
- 模块导出完整（``__all__`` 与实际定义一致，共 16 项常量）。
|
||
- 常量值与 ``manifest.json`` 中对应字段对齐（F-03 单真相源）：
|
||
- ``CHANNEL_TARGET`` 与 manifest ``channel_type`` / ``provides`` 一致。
|
||
- ``HTTP_TIMEOUT_SECONDS`` 与 manifest ``request_timeout_ms.default`` 对齐。
|
||
- ``HTTP_MAX_CONNECTIONS`` 与 manifest ``resource_quota.max_connections`` 对齐。
|
||
- ``TOKEN_CACHE_TTL_SEC`` 与 manifest ``token_cache_ttl.default`` 对齐。
|
||
- ``USER_CACHE_TTL_SEC`` 与 manifest ``user_cache_ttl.default`` 对齐。
|
||
- ``DUPLICATE_CHECK_INTERVAL_SEC`` 与 manifest ``duplicate_check_interval.default`` 对齐。
|
||
- 常量值合理性（取值范围、相互关系）校验。
|
||
|
||
``_constants.py`` 作为 ConfigPort 未命中时的本地回退默认值，其值必须与
|
||
manifest 声明保持同步，避免双源不一致。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
|
||
import pytest
|
||
from yuxi.channels.application.lifecycle.manifest_loader import load_manifest_from_dir
|
||
from yuxi.channels.plugins.wecom import _constants
|
||
|
||
pytestmark = pytest.mark.unit
|
||
|
||
|
||
@pytest.fixture
|
||
def wecom_manifest():
|
||
"""加载 wecom 插件真实 manifest.json。"""
|
||
plugin_dir = os.path.join(
|
||
os.path.dirname(__file__),
|
||
"..",
|
||
"..",
|
||
"..",
|
||
"..",
|
||
"..",
|
||
"package",
|
||
"yuxi",
|
||
"channels",
|
||
"plugins",
|
||
"wecom",
|
||
)
|
||
return load_manifest_from_dir(os.path.abspath(plugin_dir))
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# 模块导出完整性
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestConstantsExports:
|
||
"""验证 ``_constants.py`` 模块导出完整。"""
|
||
|
||
def test_all_contains_channel_target(self) -> None:
|
||
assert "CHANNEL_TARGET" in _constants.__all__
|
||
|
||
def test_all_contains_wecom_api_dep(self) -> None:
|
||
assert "WECOM_API_DEP" in _constants.__all__
|
||
|
||
def test_all_contains_http_constants(self) -> None:
|
||
assert "HTTP_TIMEOUT_SECONDS" in _constants.__all__
|
||
assert "HTTP_MAX_CONNECTIONS" in _constants.__all__
|
||
assert "HTTP_MAX_KEEPALIVE_CONNECTIONS" in _constants.__all__
|
||
|
||
def test_all_contains_token_cache_constants(self) -> None:
|
||
assert "TOKEN_REFRESH_LEAD_TIME_SEC" in _constants.__all__
|
||
assert "TOKEN_CACHE_TTL_SEC" in _constants.__all__
|
||
|
||
def test_all_contains_streaming_constants(self) -> None:
|
||
assert "STREAMING_CACHE_KEY_PREFIX" in _constants.__all__
|
||
assert "STREAMING_BUF_CACHE_KEY_PREFIX" in _constants.__all__
|
||
|
||
def test_all_contains_other_ttl_constants(self) -> None:
|
||
assert "IDENTITY_CACHE_TTL_SEC" in _constants.__all__
|
||
assert "USER_CACHE_TTL_SEC" in _constants.__all__
|
||
assert "RESPONSE_CODE_CACHE_TTL_SEC" in _constants.__all__
|
||
assert "MSG_CACHE_TTL_SEC" in _constants.__all__
|
||
assert "KF_OPEN_KFID_CACHE_TTL_SEC" in _constants.__all__
|
||
|
||
def test_all_contains_api_constants(self) -> None:
|
||
assert "WEBHOOK_TIMEOUT_SEC" in _constants.__all__
|
||
assert "DUPLICATE_CHECK_INTERVAL_SEC" in _constants.__all__
|
||
assert "RESPONSE_CODE_VALID_HOURS" in _constants.__all__
|
||
|
||
def test_all_matches_defined_names(self) -> None:
|
||
"""``__all__`` 应恰好包含模块中定义的全部公开常量（17 项）。"""
|
||
expected = {
|
||
"CHANNEL_TARGET",
|
||
"WECOM_API_DEP",
|
||
"HTTP_TIMEOUT_SECONDS",
|
||
"HTTP_MAX_CONNECTIONS",
|
||
"HTTP_MAX_KEEPALIVE_CONNECTIONS",
|
||
"TOKEN_REFRESH_LEAD_TIME_SEC",
|
||
"TOKEN_CACHE_TTL_SEC",
|
||
"STREAMING_CACHE_KEY_PREFIX",
|
||
"STREAMING_BUF_CACHE_KEY_PREFIX",
|
||
"IDENTITY_CACHE_TTL_SEC",
|
||
"USER_CACHE_TTL_SEC",
|
||
"RESPONSE_CODE_CACHE_TTL_SEC",
|
||
"MSG_CACHE_TTL_SEC",
|
||
"KF_OPEN_KFID_CACHE_TTL_SEC",
|
||
"WEBHOOK_TIMEOUT_SEC",
|
||
"DUPLICATE_CHECK_INTERVAL_SEC",
|
||
"RESPONSE_CODE_VALID_HOURS",
|
||
}
|
||
assert set(_constants.__all__) == expected
|
||
|
||
def test_all_has_17_entries(self) -> None:
|
||
assert len(_constants.__all__) == 17
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# 常量值与 manifest 对齐（F-03 单真相源）
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestConstantsAlignWithManifest:
|
||
"""验证 ``_constants.py`` 常量值与 ``manifest.json`` 声明对齐。"""
|
||
|
||
def test_channel_target_matches_manifest_channel_type(self, wecom_manifest) -> None:
|
||
assert _constants.CHANNEL_TARGET == wecom_manifest.channel_type
|
||
|
||
def test_channel_target_matches_manifest_provides(self, wecom_manifest) -> None:
|
||
"""CHANNEL_TARGET 也应出现在 manifest provides 中。"""
|
||
assert _constants.CHANNEL_TARGET in wecom_manifest.provides
|
||
|
||
def test_http_timeout_aligns_with_request_timeout_ms(self, wecom_manifest) -> None:
|
||
"""HTTP_TIMEOUT_SECONDS（秒）与 manifest request_timeout_ms.default（毫秒）对齐。"""
|
||
field = next(f for f in wecom_manifest.config_schema if f.key == "request_timeout_ms")
|
||
assert _constants.HTTP_TIMEOUT_SECONDS == field.default / 1000.0
|
||
|
||
def test_http_max_connections_aligns_with_resource_quota(self, wecom_manifest) -> None:
|
||
"""HTTP_MAX_CONNECTIONS 与 manifest resource_quota.max_connections 对齐。"""
|
||
assert wecom_manifest.resource_quota is not None
|
||
assert _constants.HTTP_MAX_CONNECTIONS == wecom_manifest.resource_quota.max_connections
|
||
|
||
def test_token_cache_ttl_aligns_with_token_cache_ttl_default(self, wecom_manifest) -> None:
|
||
"""TOKEN_CACHE_TTL_SEC 与 manifest token_cache_ttl.default 对齐。"""
|
||
field = next(f for f in wecom_manifest.config_schema if f.key == "token_cache_ttl")
|
||
assert _constants.TOKEN_CACHE_TTL_SEC == field.default
|
||
|
||
def test_user_cache_ttl_aligns_with_user_cache_ttl_default(self, wecom_manifest) -> None:
|
||
"""USER_CACHE_TTL_SEC 与 manifest user_cache_ttl.default 对齐。"""
|
||
field = next(f for f in wecom_manifest.config_schema if f.key == "user_cache_ttl")
|
||
assert _constants.USER_CACHE_TTL_SEC == field.default
|
||
|
||
def test_duplicate_check_interval_aligns_with_default(self, wecom_manifest) -> None:
|
||
"""DUPLICATE_CHECK_INTERVAL_SEC 与 manifest duplicate_check_interval.default 对齐。"""
|
||
field = next(f for f in wecom_manifest.config_schema if f.key == "duplicate_check_interval")
|
||
assert _constants.DUPLICATE_CHECK_INTERVAL_SEC == field.default
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# 常量值合理性
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestConstantsSanity:
|
||
"""验证 ``_constants.py`` 常量值的基本合理性。"""
|
||
|
||
def test_channel_target_is_wecom(self) -> None:
|
||
assert _constants.CHANNEL_TARGET == "wecom"
|
||
|
||
def test_wecom_api_dep_is_string(self) -> None:
|
||
assert isinstance(_constants.WECOM_API_DEP, str)
|
||
assert _constants.WECOM_API_DEP == "wecom_api"
|
||
|
||
def test_http_timeout_positive(self) -> None:
|
||
assert _constants.HTTP_TIMEOUT_SECONDS > 0
|
||
|
||
def test_http_max_connections_positive(self) -> None:
|
||
assert _constants.HTTP_MAX_CONNECTIONS > 0
|
||
|
||
def test_http_keepalive_le_max_connections(self) -> None:
|
||
"""keepalive 连接数上限不应超过最大连接数。"""
|
||
assert _constants.HTTP_MAX_KEEPALIVE_CONNECTIONS <= _constants.HTTP_MAX_CONNECTIONS
|
||
|
||
def test_token_refresh_lead_time_positive(self) -> None:
|
||
assert _constants.TOKEN_REFRESH_LEAD_TIME_SEC > 0
|
||
|
||
def test_token_refresh_lead_time_lt_token_ttl(self) -> None:
|
||
"""提前刷新时间应小于 token 缓存 TTL，否则 token 永远不会被缓存复用。"""
|
||
assert _constants.TOKEN_REFRESH_LEAD_TIME_SEC < _constants.TOKEN_CACHE_TTL_SEC
|
||
|
||
def test_token_cache_ttl_positive(self) -> None:
|
||
assert _constants.TOKEN_CACHE_TTL_SEC > 0
|
||
|
||
def test_identity_cache_ttl_positive(self) -> None:
|
||
assert _constants.IDENTITY_CACHE_TTL_SEC > 0
|
||
|
||
def test_user_cache_ttl_positive(self) -> None:
|
||
assert _constants.USER_CACHE_TTL_SEC > 0
|
||
|
||
def test_user_cache_ttl_ge_identity_cache_ttl(self) -> None:
|
||
"""用户信息缓存 TTL 应不小于身份解析缓存 TTL（用户信息变化频率低于身份解析）。"""
|
||
assert _constants.USER_CACHE_TTL_SEC >= _constants.IDENTITY_CACHE_TTL_SEC
|
||
|
||
def test_response_code_cache_ttl_positive(self) -> None:
|
||
assert _constants.RESPONSE_CODE_CACHE_TTL_SEC > 0
|
||
|
||
def test_response_code_cache_ttl_equals_valid_hours_times_3600(self) -> None:
|
||
"""RESPONSE_CODE_CACHE_TTL_SEC 应与 RESPONSE_CODE_VALID_HOURS 保持一致（秒 = 小时 × 3600）。"""
|
||
assert _constants.RESPONSE_CODE_CACHE_TTL_SEC == _constants.RESPONSE_CODE_VALID_HOURS * 3600
|
||
|
||
def test_msg_cache_ttl_positive(self) -> None:
|
||
assert _constants.MSG_CACHE_TTL_SEC > 0
|
||
|
||
def test_kf_open_kfid_cache_ttl_positive(self) -> None:
|
||
assert _constants.KF_OPEN_KFID_CACHE_TTL_SEC > 0
|
||
|
||
def test_webhook_timeout_positive(self) -> None:
|
||
assert _constants.WEBHOOK_TIMEOUT_SEC > 0
|
||
|
||
def test_duplicate_check_interval_positive(self) -> None:
|
||
assert _constants.DUPLICATE_CHECK_INTERVAL_SEC > 0
|
||
|
||
def test_response_code_valid_hours_positive(self) -> None:
|
||
assert _constants.RESPONSE_CODE_VALID_HOURS > 0
|