Kris
532e7c59ec
1. 新增QQBot、Wecom、Dingtalk、Instagram、WechatMp等渠道的单元测试夹具与测试用例 2. 修复WechatILink的缓存键顺序与测试用例 3. 补全Feishu适配器的日志断言与测试逻辑 4. 清理WechatILink冗余的clear_context_token测试代码
171 lines
6.8 KiB
Python
171 lines
6.8 KiB
Python
"""dingtalk 插件 _constants.py 模块测试。
|
||
|
||
验证共享常量模块的导出完整性与值正确性：
|
||
- ``__all__`` 与实际定义一致（模块导出完整）。
|
||
- 常量值与 ``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`` 对齐。
|
||
- ``DEFAULT_STREAM_ENDPOINT_PATH`` 与 manifest ``stream_endpoint_path.default`` 对齐。
|
||
- token 缓存与签名相关常量值合理（设计方案 §5.4 / §3.3）。
|
||
|
||
_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.dingtalk import _constants
|
||
|
||
pytestmark = pytest.mark.unit
|
||
|
||
|
||
@pytest.fixture
|
||
def dingtalk_manifest():
|
||
plugin_dir = os.path.join(
|
||
os.path.dirname(__file__),
|
||
"..",
|
||
"..",
|
||
"..",
|
||
"..",
|
||
"..",
|
||
"package",
|
||
"yuxi",
|
||
"channels",
|
||
"plugins",
|
||
"dingtalk",
|
||
)
|
||
return load_manifest_from_dir(os.path.abspath(plugin_dir))
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# 模块导出完整性
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestConstantsExports:
|
||
"""验证 _constants.py 模块导出完整。"""
|
||
|
||
def test_all_contains_channel_target(self):
|
||
assert "CHANNEL_TARGET" in _constants.__all__
|
||
|
||
def test_all_contains_dingtalk_api_dep(self):
|
||
assert "DINGTALK_API_DEP" in _constants.__all__
|
||
|
||
def test_all_contains_http_constants(self):
|
||
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_stream_and_token_constants(self):
|
||
assert "DEFAULT_STREAM_ENDPOINT_PATH" in _constants.__all__
|
||
assert "TOKEN_CACHE_KEY_PREFIX" in _constants.__all__
|
||
assert "TOKEN_REFRESH_LEAD_TIME_SEC" in _constants.__all__
|
||
assert "HTTP_SIGN_TIMESTAMP_TOLERANCE_SEC" in _constants.__all__
|
||
assert "DEFAULT_CARD_TEMPLATE_ID" in _constants.__all__
|
||
|
||
def test_all_matches_defined_names(self):
|
||
"""``__all__`` 应恰好包含模块中定义的全部公开常量。"""
|
||
expected = {
|
||
"CHANNEL_TARGET",
|
||
"DINGTALK_API_DEP",
|
||
"HTTP_TIMEOUT_SECONDS",
|
||
"HTTP_MAX_CONNECTIONS",
|
||
"HTTP_MAX_KEEPALIVE_CONNECTIONS",
|
||
"DEFAULT_STREAM_ENDPOINT_PATH",
|
||
"TOKEN_CACHE_KEY_PREFIX",
|
||
"TOKEN_REFRESH_LEAD_TIME_SEC",
|
||
"HTTP_SIGN_TIMESTAMP_TOLERANCE_SEC",
|
||
"DEFAULT_CARD_TEMPLATE_ID",
|
||
}
|
||
assert set(_constants.__all__) == expected
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# 常量值与 manifest 对齐（F-03 单真相源）
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestConstantsAlignWithManifest:
|
||
"""验证 _constants.py 常量值与 manifest.json 声明对齐。"""
|
||
|
||
def test_channel_target_matches_manifest_channel_type(self, dingtalk_manifest):
|
||
assert _constants.CHANNEL_TARGET == dingtalk_manifest.channel_type
|
||
|
||
def test_channel_target_matches_manifest_provides(self, dingtalk_manifest):
|
||
assert _constants.CHANNEL_TARGET in dingtalk_manifest.provides
|
||
|
||
def test_http_timeout_aligns_with_request_timeout_ms(self, dingtalk_manifest):
|
||
"""HTTP_TIMEOUT_SECONDS（秒）与 manifest request_timeout_ms.default（毫秒）对齐。"""
|
||
field = next(f for f in dingtalk_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, dingtalk_manifest):
|
||
"""HTTP_MAX_CONNECTIONS 与 manifest resource_quota.max_connections 对齐。"""
|
||
assert dingtalk_manifest.resource_quota is not None
|
||
assert _constants.HTTP_MAX_CONNECTIONS == dingtalk_manifest.resource_quota.max_connections
|
||
|
||
def test_default_stream_endpoint_path_aligns_with_schema(self, dingtalk_manifest):
|
||
"""DEFAULT_STREAM_ENDPOINT_PATH 与 manifest stream_endpoint_path.default 对齐。"""
|
||
field = next(f for f in dingtalk_manifest.config_schema if f.key == "stream_endpoint_path")
|
||
assert _constants.DEFAULT_STREAM_ENDPOINT_PATH == field.default
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# 常量值合理性
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestConstantsSanity:
|
||
"""验证 _constants.py 常量值的基本合理性。"""
|
||
|
||
def test_channel_target_is_dingtalk(self):
|
||
assert _constants.CHANNEL_TARGET == "dingtalk"
|
||
|
||
def test_dingtalk_api_dep_is_string(self):
|
||
assert isinstance(_constants.DINGTALK_API_DEP, str)
|
||
assert _constants.DINGTALK_API_DEP == "dingtalk_api"
|
||
|
||
def test_http_timeout_positive(self):
|
||
assert _constants.HTTP_TIMEOUT_SECONDS > 0
|
||
|
||
def test_http_timeout_is_30_seconds(self):
|
||
"""HTTP_TIMEOUT_SECONDS 对齐 manifest request_timeout_ms.default=30000。"""
|
||
assert _constants.HTTP_TIMEOUT_SECONDS == 30.0
|
||
|
||
def test_http_max_connections_positive(self):
|
||
assert _constants.HTTP_MAX_CONNECTIONS > 0
|
||
|
||
def test_http_max_connections_is_50(self):
|
||
"""HTTP_MAX_CONNECTIONS 对齐 manifest resource_quota.max_connections=50。"""
|
||
assert _constants.HTTP_MAX_CONNECTIONS == 50
|
||
|
||
def test_http_keepalive_le_max_connections(self):
|
||
"""keepalive 连接数上限不应超过最大连接数。"""
|
||
assert _constants.HTTP_MAX_KEEPALIVE_CONNECTIONS <= _constants.HTTP_MAX_CONNECTIONS
|
||
|
||
def test_default_stream_endpoint_path_value(self):
|
||
assert _constants.DEFAULT_STREAM_ENDPOINT_PATH == "/v1.0/gateway/connections/open"
|
||
|
||
def test_token_cache_key_prefix_value(self):
|
||
assert _constants.TOKEN_CACHE_KEY_PREFIX == "dingtalk:token:"
|
||
|
||
def test_token_refresh_lead_time_sec_value(self):
|
||
"""token 提前刷新量 300 秒（设计方案 §5.4）。"""
|
||
assert _constants.TOKEN_REFRESH_LEAD_TIME_SEC == 300
|
||
|
||
def test_http_sign_timestamp_tolerance_sec_value(self):
|
||
"""HTTP 模式签名时效窗口 3600 秒（60 分钟，设计方案 §3.3）。"""
|
||
assert _constants.HTTP_SIGN_TIMESTAMP_TOLERANCE_SEC == 3600
|
||
|
||
def test_default_card_template_id_is_string(self):
|
||
assert isinstance(_constants.DEFAULT_CARD_TEMPLATE_ID, str)
|
||
assert _constants.DEFAULT_CARD_TEMPLATE_ID == "DEFAULT_CARD_TEMPLATE"
|