71 lines
2.6 KiB
Python
71 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
|