kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
00a5577aac
ForcePilot/backend/test/integration/api/channels/test_wizard_router.py
Kris 361dbe2149 test: 批量修复单元测试与集成测试用例的各类问题 
本次提交修复了多个测试文件中的问题：
1.  为各管道测试类添加tracer_port属性初始化
2.  修复QQBot客户端关闭异常的错误类型匹配
3.  修正配对接口查询的状态值小写格式
4.  移除ChannelType枚举值的显式.value调用
5.  新增QQ常量的导出项与Instagram测试桩函数
6.  修复API接口返回值解析，正确访问data字段
7.  调整企业微信富文本消息的断言结构
8.  修正微信iLink的消息类型测试用例
9.  替换废弃的datetime.utc相关导入为UTC常量
10. 修复QQBot白名单适配器的返回值结构
11. 调整outbox工具的channel_msg_id处理逻辑
12. 新增多个适配器与服务的测试用例，覆盖异常降级、缓存处理等场景
13. 重构部分QQBot适配器的辅助函数测试，清理冗余代码
14. 修复微信iLink客户端的上传接口调用参数与缓存清理逻辑
15. 修正配置导出接口的敏感字段过滤与返回值结构
2026-07-09 04:23:07 +08:00

349 lines
12 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Integration tests for channels wizard_router endpoints."""
from __future__ import annotations
import httpx
import pytest
from .conftest import BASE_URL, DEFAULT_CHANNEL_TYPE
pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
def _make_validate_payload() -> dict:
"""Build a valid ValidateWizardRequest body (step_id 在路径参数中)."""
return {
"values": {"app_id": "pytest_app", "app_secret": "pytest_secret"},
}
def _make_finalize_payload() -> dict:
"""Build a valid FinalizeWizardRequest body."""
return {
"patches": [
{
"step_id": "account_info",
"values": {"app_id": "pytest_app", "app_secret": "pytest_secret"},
}
]
}
# =============================================================================
# === Auth three-tier for GET /wizard/steps ===
# =============================================================================
async def test_list_wizard_steps_requires_auth(test_client: httpx.AsyncClient):
# Act
response = await test_client.get(f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/steps")
# Assert
assert response.status_code == 401
async def test_list_wizard_steps_requires_admin(test_client: httpx.AsyncClient, standard_user):
# Act
response = await test_client.get(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/steps",
headers=standard_user["headers"],
)
# Assert
assert response.status_code == 403
async def test_admin_can_list_wizard_steps_returns_200_or_404_or_501(test_client: httpx.AsyncClient, admin_headers):
# Act
response = await test_client.get(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/steps",
headers=admin_headers,
)
# Assert — adapter not registered → 404 or 501, or 200 with steps
assert response.status_code in (200, 404, 501), response.text
# =============================================================================
# === POST /wizard/steps/{step_id}/validate ===
# =============================================================================
async def test_wizard_validate_returns_200_or_404_or_501(test_client: httpx.AsyncClient, admin_headers):
# Arrange
step_id = "account_info"
body = _make_validate_payload()
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/steps/{step_id}/validate",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code in (200, 404, 501), response.text
async def test_wizard_validate_rejects_empty_step_id(test_client: httpx.AsyncClient, admin_headers):
# Arrange — step_id 路径参数为空字符串，FastAPI 路由匹配规则下
# 可能不匹配（404）或匹配到空 step_id 后由下游校验拒绝（400/422）
body = _make_validate_payload()
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/steps//validate",
json=body,
headers=admin_headers,
)
# Assert — 空路径段不被接受，返回 4xx 错误
assert response.status_code in (400, 404, 422), response.text
# =============================================================================
# === POST /wizard/steps/{step_id}/apply ===
# =============================================================================
async def test_wizard_apply_returns_200_or_404_or_501(test_client: httpx.AsyncClient, admin_headers):
# Arrange
step_id = "account_info"
body = {"values": {"app_id": "pytest_app", "app_secret": "pytest_secret"}}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/steps/{step_id}/apply",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code in (200, 404, 501), response.text
# =============================================================================
# === POST /wizard/finalize ===
# =============================================================================
async def test_wizard_finalize_returns_200_or_404_or_501(test_client: httpx.AsyncClient, admin_headers):
# Arrange
body = _make_finalize_payload()
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/finalize",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code in (200, 404, 501), response.text
async def test_wizard_finalize_rejects_empty_patches(test_client: httpx.AsyncClient, admin_headers):
# Arrange — patches min_length=1
body = {"patches": []}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/finalize",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
# =============================================================================
# === POST /wizard/oauth/initiate ===
# =============================================================================
async def test_oauth_initiate_rejects_invalid_redirect_uri_scheme(test_client: httpx.AsyncClient, admin_headers):
# Arrange — redirect_uri 必须为 http/https，ftp 等非法 scheme 拒绝
body = {"redirect_uri": "ftp://example.com/callback"}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/oauth/initiate",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_oauth_initiate_with_valid_params_returns_404_or_501_or_400(
test_client: httpx.AsyncClient, admin_headers
):
# Arrange — valid redirect_uri but channel/adapter 未注册
body = {"redirect_uri": "https://example.com/callback"}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/oauth/initiate",
json=body,
headers=admin_headers,
)
# Assert — non-existent channel / adapter → 404, 501, or 400
assert response.status_code in (400, 404, 501), response.text
# =============================================================================
# === POST /wizard/oauth-callback ===
# =============================================================================
async def test_oauth_callback_requires_code(test_client: httpx.AsyncClient, admin_headers):
# Arrange — code required (Pydantic min_length=1)
body = {"state": "pytest_state", "redirect_uri": "https://example.com/callback"}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/oauth-callback",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_oauth_callback_requires_state(test_client: httpx.AsyncClient, admin_headers):
# Arrange — state required (Pydantic min_length=1)
body = {"code": "pytest_code", "redirect_uri": "https://example.com/callback"}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/oauth-callback",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_oauth_callback_requires_redirect_uri(test_client: httpx.AsyncClient, admin_headers):
# Arrange — redirect_uri required
body = {"code": "pytest_code", "state": "pytest_state"}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/oauth-callback",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_oauth_callback_rejects_invalid_redirect_uri_scheme(test_client: httpx.AsyncClient, admin_headers):
# Arrange — redirect_uri 必须为 http/https，ftp 等非法 scheme 拒绝
body = {
"code": "pytest_code",
"state": "pytest_state",
"redirect_uri": "ftp://example.com/callback",
}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/oauth-callback",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_oauth_callback_with_all_params_returns_404_or_501_or_400(test_client: httpx.AsyncClient, admin_headers):
# Arrange — all params present but channel not registered
body = {
"code": "pytest_code",
"state": "pytest_state",
"redirect_uri": "https://example.com/callback",
}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/oauth-callback",
json=body,
headers=admin_headers,
)
# Assert — non-existent channel / adapter → 404, 501, or 400
assert response.status_code in (400, 404, 501), response.text
# =============================================================================
# === POST /wizard/qr/initiate (WIZ-07) ===
# =============================================================================
async def test_wizard_qr_initiate_requires_auth(test_client: httpx.AsyncClient):
# Act
response = await test_client.post(f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/qr/initiate")
# Assert
assert response.status_code == 401
async def test_wizard_qr_initiate_requires_admin(test_client: httpx.AsyncClient, standard_user):
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/qr/initiate",
headers=standard_user["headers"],
)
# Assert
assert response.status_code == 403
async def test_wizard_qr_initiate_returns_200_or_404_or_501(test_client: httpx.AsyncClient, admin_headers):
# Act — 账户尚未创建时复用 LoginAdapter 获取二维码，适配器未注册 → 404/501
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/qr/initiate",
headers=admin_headers,
)
# Assert
assert response.status_code in (200, 404, 501), response.text
# =============================================================================
# === POST /wizard/qr/wait (WIZ-08) ===
# =============================================================================
async def test_wizard_qr_wait_requires_session_id(test_client: httpx.AsyncClient, admin_headers):
# Arrange — session_id required (Pydantic min_length=1)
body = {"current_qr_data_url": "https://example.com/qr"}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/qr/wait",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_wizard_qr_wait_rejects_empty_session_id(test_client: httpx.AsyncClient, admin_headers):
# Arrange — session_id min_length=1，空字符串拒绝
body = {"session_id": ""}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/qr/wait",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_wizard_qr_wait_returns_200_or_404_or_501(test_client: httpx.AsyncClient, admin_headers):
# Arrange — 合法 session_id 但扫码会话不存在 / 适配器未注册
body = {"session_id": "nonexistent_qr_session"}
# Act
response = await test_client.post(
f"{BASE_URL}/{DEFAULT_CHANNEL_TYPE}/wizard/qr/wait",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code in (200, 404, 501), response.text
# =============================================================================
# === Invalid channel_type validation ===
# =============================================================================
async def test_wizard_unregistered_channel_type(test_client: httpx.AsyncClient, admin_headers):
# Arrange — channel_type 为 str 子类无白名单,未注册插件返回 404
unregistered_channel = "invalid_channel_type_xyz"
# Act
response = await test_client.get(
f"{BASE_URL}/{unregistered_channel}/wizard/steps",
headers=admin_headers,
)
# Assert
assert response.status_code == 404, response.text
Reference in New Issue View Git Blame Copy Permalink