Kris
f419f57f63
1. 更新能力注册测试用例，修正空能力默认动作断言 2. 调整渠道API测试的错误状态码校验逻辑，新增停止不存在渠道的测试 3. 重构飞书动作测试的状态校验逻辑，使用枚举替代字符串判断 4. 新增渠道状态条目管理的集成测试用例
81 lines
3.3 KiB
Python
81 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestChannelsStatusAPI:
|
|
async def test_list_channels_status(self, test_client: AsyncClient, admin_headers: dict):
|
|
response = await test_client.get("/api/channels/status", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["code"] == 0
|
|
assert "channels" in data["data"]
|
|
|
|
async def test_get_single_channel_status_not_found(self, test_client: AsyncClient, admin_headers: dict):
|
|
response = await test_client.get("/api/channels/nonexistent/status", headers=admin_headers)
|
|
assert response.status_code == 404
|
|
|
|
async def test_list_channels_requires_admin(self, test_client: AsyncClient, standard_user: dict):
|
|
response = await test_client.get("/api/channels/status", headers=standard_user["headers"])
|
|
assert response.status_code in (401, 403)
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestChannelStartStop:
|
|
async def test_start_nonexistent_channel(self, test_client: AsyncClient, admin_headers: dict):
|
|
response = await test_client.post("/api/channels/nonexistent/start", headers=admin_headers)
|
|
assert response.status_code == 404
|
|
|
|
async def test_stop_not_running_channel(self, test_client: AsyncClient, admin_headers: dict):
|
|
response = await test_client.post("/api/channels/telegram/stop", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["code"] == 409
|
|
assert "not running" in data["message"]
|
|
|
|
async def test_stop_nonexistent_channel(self, test_client: AsyncClient, admin_headers: dict):
|
|
response = await test_client.post("/api/channels/nonexistent/stop", headers=admin_headers)
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "not registered" in data["detail"]
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestChannelConfig:
|
|
async def test_update_config_nonexistent_channel(self, test_client: AsyncClient, admin_headers: dict):
|
|
response = await test_client.put(
|
|
"/api/channels/nonexistent/config",
|
|
json={"enabled": True},
|
|
headers=admin_headers,
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestChannelStats:
|
|
async def test_get_channel_stats(self, test_client: AsyncClient, admin_headers: dict):
|
|
response = await test_client.get("/api/channels/telegram/stats", headers=admin_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["code"] == 0
|
|
assert "summary" in data["data"]
|
|
|
|
async def test_get_channel_stats_with_params(self, test_client: AsyncClient, admin_headers: dict):
|
|
response = await test_client.get(
|
|
"/api/channels/telegram/stats",
|
|
params={"days": 30, "granularity": "day"},
|
|
headers=admin_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["code"] == 0
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestChannelTestConnection:
|
|
async def test_test_channel_nonexistent(self, test_client: AsyncClient, admin_headers: dict):
|
|
response = await test_client.post("/api/channels/nonexistent/test", headers=admin_headers)
|
|
assert response.status_code == 404
|