Kris
04a2472014
1. 新增channels模块集成测试目录与基础fixture 2. 为capability、doctor、reports、dashboard、directory、webhook、wizard、health、analytics、allowlist、config等路由编写完整的鉴权、参数校验与异常场景测试 3. 修复基础集成测试setup，新增external/scheduler/channel数据库schema初始化步骤
79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
"""Integration tests for channels capability_router endpoints.
|
||
|
||
覆盖能力查询域（CAP-01 列出全部渠道能力 / CAP-02 查询单渠道能力）的
|
||
鉴权矩阵与渠道类型枚举校验。``ChannelType`` 为路径枚举参数，非法值返回
|
||
422；单渠道能力查询在渠道未绑定插件时返回 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_rejects_invalid_channel_type(
|
||
test_client: httpx.AsyncClient, admin_headers
|
||
):
|
||
# Act — "foo" 不在 ChannelType 枚举中
|
||
response = await test_client.get(f"{CAPABILITIES_URL}/foo", headers=admin_headers)
|
||
# Assert
|
||
assert response.status_code == 422, response.text
|