Kris
4c2c6b956d
1. 格式化代码与参数换行，提升可读性 2. 修正测试断言与测试逻辑，覆盖更多边界场景 3. 新增外部系统告警仓库、工具服务、系统服务等测试用例 4. 完善插件注册表的重启状态检测逻辑测试 5. 更新测试断言计数与预期结果，匹配代码变更 6. 修复Instagram适配器的签名校验测试逻辑 7. 优化会话合并、上下文测试的类型检查与断言
129 lines
5.0 KiB
Python
129 lines
5.0 KiB
Python
"""Integration tests for channels capability_router endpoints.
|
||
|
||
覆盖能力查询域（CAP-01 列出全部渠道能力 / CAP-02 查询单渠道能力 /
|
||
CAP-03 能力画像矩阵）的鉴权矩阵与渠道类型校验。``ChannelType`` 为路径
|
||
参数（str 子类，无白名单），未注册插件的渠道类型返回 404。``/capabilities
|
||
/matrix`` 静态路径先于 ``/capabilities/{channel_type}`` 声明，避免 "matrix"
|
||
被捕获为 channel_type（规范 §6.5）。
|
||
"""
|
||
|
||
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/matrix — CAP-03 capability matrix ===
|
||
# =============================================================================
|
||
|
||
|
||
async def test_list_capability_matrix_requires_auth(test_client: httpx.AsyncClient):
|
||
# Act
|
||
response = await test_client.get(f"{CAPABILITIES_URL}/matrix")
|
||
# Assert
|
||
assert response.status_code == 401
|
||
|
||
|
||
async def test_list_capability_matrix_requires_admin(test_client: httpx.AsyncClient, standard_user):
|
||
# Act
|
||
response = await test_client.get(f"{CAPABILITIES_URL}/matrix", headers=standard_user["headers"])
|
||
# Assert
|
||
assert response.status_code == 403
|
||
|
||
|
||
async def test_admin_can_list_capability_matrix(test_client: httpx.AsyncClient, admin_headers):
|
||
# Act
|
||
response = await test_client.get(f"{CAPABILITIES_URL}/matrix", headers=admin_headers)
|
||
# Assert
|
||
assert response.status_code == 200, response.text
|
||
|
||
payload = response.json()
|
||
assert payload["success"] is True
|
||
assert isinstance(payload["data"], dict)
|
||
|
||
|
||
async def test_capability_matrix_accepts_filter_query_params(test_client: httpx.AsyncClient, admin_headers):
|
||
# Act
|
||
response = await test_client.get(
|
||
f"{CAPABILITIES_URL}/matrix",
|
||
params={"status": "available", "capability": "streaming"},
|
||
headers=admin_headers,
|
||
)
|
||
# Assert
|
||
assert response.status_code == 200, response.text
|
||
|
||
payload = response.json()
|
||
assert payload["success"] is True
|
||
assert isinstance(payload["data"], dict)
|
||
assert "capabilities" in payload["data"]
|
||
|
||
|
||
async def test_matrix_static_path_not_captured_as_channel_type(test_client: httpx.AsyncClient, admin_headers):
|
||
# Act — /capabilities/matrix 是静态路径，必须先于 /capabilities/{channel_type}
|
||
# 声明，避免 "matrix" 被捕获为 channel_type。若路由顺序错误，此请求会走
|
||
# getCapability("matrix") 并返回 404（未注册插件），而非 matrix 端点的 200。
|
||
response = await test_client.get(f"{CAPABILITIES_URL}/matrix", headers=admin_headers)
|
||
# Assert — 命中 matrix 端点返回 200，证明静态路径未被动态参数捕获
|
||
assert response.status_code == 200, response.text
|
||
assert response.json()["success"] is True
|
||
|
||
|
||
# =============================================================================
|
||
# === 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
|