Kris
d58331e0c0
本次提交修复了多个测试文件中的问题： 1. 将 ChannelType 枚举调用改为字符串实例化方式 2. 修正了日志断言、异步mock使用、配置参数等多处测试细节 3. 新增了会话聚合根、跨渠道关联策略等单元测试用例 4. 修复了路由测试中的路径方法错误与断言逻辑 5. 调整了依赖导入与测试夹具的兼容性 6. 统一了重试回退调度的列表/元组使用规范
79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
"""Integration tests for channels capability_router endpoints.
|
||
|
||
覆盖能力查询域（CAP-01 列出全部渠道能力 / CAP-02 查询单渠道能力）的
|
||
鉴权矩阵与渠道类型校验。``ChannelType`` 为路径参数（str 子类，无白名单），
|
||
未注册插件的渠道类型返回 404。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from .conftest import BASE_URL, DEFAULT_CHANNEL_TYPE
|
||
|
||
pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
|
||
|
||
CAPABILITIES_URL = f"{BASE_URL}/capabilities"
|
||
|
||
|
||
# =============================================================================
|
||
# === Auth three-tier for GET /capabilities ===
|
||
# =============================================================================
|
||
|
||
|
||
async def test_list_capabilities_requires_auth(test_client: httpx.AsyncClient):
|
||
# Act
|
||
response = await test_client.get(CAPABILITIES_URL)
|
||
# Assert
|
||
assert response.status_code == 401
|
||
|
||
|
||
async def test_list_capabilities_requires_admin(
|
||
test_client: httpx.AsyncClient, standard_user
|
||
):
|
||
# Act
|
||
response = await test_client.get(CAPABILITIES_URL, headers=standard_user["headers"])
|
||
# Assert
|
||
assert response.status_code == 403
|
||
|
||
|
||
async def test_admin_can_list_capabilities(test_client: httpx.AsyncClient, admin_headers):
|
||
# Act
|
||
response = await test_client.get(CAPABILITIES_URL, headers=admin_headers)
|
||
# Assert
|
||
assert response.status_code == 200, response.text
|
||
|
||
payload = response.json()
|
||
assert payload["success"] is True
|
||
assert isinstance(payload["data"], dict)
|
||
|
||
|
||
# =============================================================================
|
||
# === GET /capabilities/{channel_type} ===
|
||
# =============================================================================
|
||
|
||
|
||
async def test_admin_can_get_capability_for_channel_type(
|
||
test_client: httpx.AsyncClient, admin_headers
|
||
):
|
||
# Act — DEFAULT_CHANNEL_TYPE 已注册则 200，未注册则 404
|
||
response = await test_client.get(
|
||
f"{CAPABILITIES_URL}/{DEFAULT_CHANNEL_TYPE}", headers=admin_headers
|
||
)
|
||
# Assert
|
||
assert response.status_code in (200, 404), response.text
|
||
|
||
if response.status_code == 200:
|
||
payload = response.json()
|
||
assert payload["success"] is True
|
||
|
||
|
||
async def test_get_capability_unregistered_channel_type(
|
||
test_client: httpx.AsyncClient, admin_headers
|
||
):
|
||
# Act — "foo" 是合法字符串但未注册插件
|
||
response = await test_client.get(f"{CAPABILITIES_URL}/foo", headers=admin_headers)
|
||
# Assert - channel_type 无白名单,任何字符串都合法,但未注册插件返回 404
|
||
assert response.status_code == 404, response.text
|