kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
80296b7db1
ForcePilot/backend/test/integration/api/channels/test_pairing_router.py
Kris df116a069c test: 补全全量接口参数校验与权限控制测试用例 
新增并完善多模块接口测试用例，覆盖：
1. 权限校验：认证缺失、普通用户越权访问场景
2. 参数合法性校验：空值、格式错误、边界值、枚举约束
3. 响应字段增强：补充业务字段断言与分页回传校验
4. 异常场景：不存在资源、超限参数、无效格式等400/404/422场景
2026-07-11 21:41:45 +08:00

583 lines
19 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 pairing_router endpoints."""
from __future__ import annotations
import httpx
import pytest
from .conftest import (
BASE_URL,
DEFAULT_CHANNEL_TYPE,
ISO_TIME_END,
ISO_TIME_START,
NON_EXISTENT_ACCOUNT_ID,
NON_EXISTENT_PAIRING_ID,
NON_EXISTENT_PEER_ID,
)
pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
PAIRINGS_URL = f"{BASE_URL}/pairings"
def _make_create_pairing_body() -> dict:
"""Build a minimal CreatePairingRequest body."""
return {
"channel_type": DEFAULT_CHANNEL_TYPE,
"account_id": NON_EXISTENT_ACCOUNT_ID,
"peer_id": NON_EXISTENT_PEER_ID,
"peer_name": "pytest_peer",
"expires_in_seconds": 604800,
}
# =============================================================================
# === Auth three-tier for list pairings ===
# =============================================================================
async def test_list_pairings_requires_auth(test_client: httpx.AsyncClient):
# Act
response = await test_client.get(PAIRINGS_URL)
# Assert
assert response.status_code == 401, response.text
async def test_list_pairings_forbids_standard_user(test_client: httpx.AsyncClient, standard_user):
# Act
response = await test_client.get(PAIRINGS_URL, headers=standard_user["headers"])
# Assert
assert response.status_code == 403, response.text
async def test_admin_can_list_pairings(test_client: httpx.AsyncClient, admin_headers):
# Act
response = await test_client.get(PAIRINGS_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)
# handler _pairingList 返回 {"pairings": [...]}（与前端 usePairingApproval 契约一致）
assert "pairings" in payload["data"]
# =============================================================================
# === Pagination + filter validation for list pairings ===
# =============================================================================
async def test_list_pairings_rejects_limit_zero(test_client: httpx.AsyncClient, admin_headers):
# Act — limit=0 violates ge=1
response = await test_client.get(
PAIRINGS_URL,
params={"limit": 0},
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_list_pairings_rejects_limit_over_max(test_client: httpx.AsyncClient, admin_headers):
# Act — limit=1001 violates le=1000
response = await test_client.get(
PAIRINGS_URL,
params={"limit": 1001},
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_list_pairings_rejects_invalid_status(test_client: httpx.AsyncClient, admin_headers):
# Act — status 不在枚举范围内
response = await test_client.get(
PAIRINGS_URL,
params={"status": "INVALID_STATUS"},
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
# =============================================================================
# === GET /pairings/count — FR-33 待办角标计数 ===
# =============================================================================
async def test_count_pairings_requires_auth(test_client: httpx.AsyncClient):
# Act
response = await test_client.get(f"{PAIRINGS_URL}/count")
# Assert
assert response.status_code == 401, response.text
async def test_count_pairings_forbids_standard_user(test_client: httpx.AsyncClient, standard_user):
# Act
response = await test_client.get(f"{PAIRINGS_URL}/count", headers=standard_user["headers"])
# Assert
assert response.status_code == 403, response.text
async def test_admin_can_count_pairings(test_client: httpx.AsyncClient, admin_headers):
# Act
response = await test_client.get(f"{PAIRINGS_URL}/count", 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_count_pairings_with_filters_returns_200(test_client: httpx.AsyncClient, admin_headers):
# Act — 支持按渠道/账户/对端/状态过滤
response = await test_client.get(
f"{PAIRINGS_URL}/count",
params={
"channel_type": DEFAULT_CHANNEL_TYPE,
"account_id": NON_EXISTENT_ACCOUNT_ID,
"peer_id": NON_EXISTENT_PEER_ID,
"status": "pending",
},
headers=admin_headers,
)
# Assert
assert response.status_code == 200, response.text
async def test_count_pairings_rejects_invalid_status(test_client: httpx.AsyncClient, admin_headers):
# Act — status 不在 PairingStatus 枚举范围内
response = await test_client.get(
f"{PAIRINGS_URL}/count",
params={"status": "INVALID_STATUS"},
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_count_static_path_not_captured_as_pairing_id(test_client: httpx.AsyncClient, admin_headers):
# Act — /pairings/count 是静态路径，必须先于 /pairings/{pairing_id} 声明，
# 避免 "count" 被捕获为 pairing_id。若路由顺序错误，此请求会走
# getPairing("count") 并返回 404，而非 count 端点的 200。
response = await test_client.get(f"{PAIRINGS_URL}/count", headers=admin_headers)
# Assert — 命中 count 端点返回 200，证明静态路径未被动态参数捕获
assert response.status_code == 200, response.text
# =============================================================================
# === create pairing ===
# =============================================================================
async def test_create_pairing_requires_auth(test_client: httpx.AsyncClient):
# Arrange
body = _make_create_pairing_body()
# Act
response = await test_client.post(PAIRINGS_URL, json=body)
# Assert
assert response.status_code == 401, response.text
async def test_create_pairing_forbids_standard_user(test_client: httpx.AsyncClient, standard_user):
# Arrange
body = _make_create_pairing_body()
# Act
response = await test_client.post(
PAIRINGS_URL,
json=body,
headers=standard_user["headers"],
)
# Assert
assert response.status_code == 403, response.text
async def test_create_pairing_non_existent_account_returns_404(test_client: httpx.AsyncClient, admin_headers):
# Arrange
body = _make_create_pairing_body()
# Act
response = await test_client.post(PAIRINGS_URL, json=body, headers=admin_headers)
# Assert — account_id 不存在：NOT_FOUND (404) 或适配器能力未实现 (501)
assert response.status_code in (404, 501), response.text
async def test_create_pairing_rejects_expires_zero(test_client: httpx.AsyncClient, admin_headers):
# Arrange — expires_in_seconds=0 低于下限 1
body = _make_create_pairing_body()
body["expires_in_seconds"] = 0
# Act
response = await test_client.post(PAIRINGS_URL, json=body, headers=admin_headers)
# Assert
assert response.status_code == 422, response.text
async def test_create_pairing_rejects_expires_over_max(test_client: httpx.AsyncClient, admin_headers):
# Arrange — expires_in_seconds > 2592000 超过上限
body = _make_create_pairing_body()
body["expires_in_seconds"] = 2592001
# Act
response = await test_client.post(PAIRINGS_URL, json=body, headers=admin_headers)
# Assert
assert response.status_code == 422, response.text
async def test_create_pairing_rejects_empty_account_id(test_client: httpx.AsyncClient, admin_headers):
# Arrange — account_id="" 违反 min_length=1
body = _make_create_pairing_body()
body["account_id"] = ""
# Act
response = await test_client.post(PAIRINGS_URL, json=body, headers=admin_headers)
# Assert
assert response.status_code == 422, response.text
async def test_create_pairing_rejects_empty_peer_id(test_client: httpx.AsyncClient, admin_headers):
# Arrange — peer_id="" 违反 min_length=1
body = _make_create_pairing_body()
body["peer_id"] = ""
# Act
response = await test_client.post(PAIRINGS_URL, json=body, headers=admin_headers)
# Assert
assert response.status_code == 422, response.text
# =============================================================================
# === batch approve / reject ===
# =============================================================================
async def test_batch_approve_pairings_requires_auth(test_client: httpx.AsyncClient):
# Arrange
body = {"pairing_ids": [NON_EXISTENT_PAIRING_ID]}
# Act
response = await test_client.post(f"{PAIRINGS_URL}/batch-approve", json=body)
# Assert
assert response.status_code == 401, response.text
async def test_batch_approve_pairings_forbids_standard_user(
test_client: httpx.AsyncClient, standard_user
):
# Arrange
body = {"pairing_ids": [NON_EXISTENT_PAIRING_ID]}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/batch-approve",
json=body,
headers=standard_user["headers"],
)
# Assert
assert response.status_code == 403, response.text
async def test_batch_approve_pairings_rejects_empty_ids(test_client: httpx.AsyncClient, admin_headers):
# Arrange — pairing_ids 为空违反 min_items=1
body = {"pairing_ids": []}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/batch-approve",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_batch_approve_pairings_rejects_too_many_ids(test_client: httpx.AsyncClient, admin_headers):
# Arrange — pairing_ids 超过 max_items=500
body = {"pairing_ids": [f"id_{i}" for i in range(501)]}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/batch-approve",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_batch_reject_pairings_requires_auth(test_client: httpx.AsyncClient):
# Arrange
body = {"pairing_ids": [NON_EXISTENT_PAIRING_ID]}
# Act
response = await test_client.post(f"{PAIRINGS_URL}/batch-reject", json=body)
# Assert
assert response.status_code == 401, response.text
async def test_batch_reject_pairings_forbids_standard_user(
test_client: httpx.AsyncClient, standard_user
):
# Arrange
body = {"pairing_ids": [NON_EXISTENT_PAIRING_ID]}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/batch-reject",
json=body,
headers=standard_user["headers"],
)
# Assert
assert response.status_code == 403, response.text
async def test_batch_reject_pairings_rejects_empty_ids(test_client: httpx.AsyncClient, admin_headers):
# Arrange
body = {"pairing_ids": []}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/batch-reject",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_batch_reject_pairings_rejects_too_many_ids(test_client: httpx.AsyncClient, admin_headers):
# Arrange — pairing_ids 超过 max_length=500
body = {"pairing_ids": [f"id_{i}" for i in range(501)]}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/batch-reject",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
# =============================================================================
# === clean expired ===
# =============================================================================
async def test_clean_expired_pairings_requires_auth(test_client: httpx.AsyncClient):
# Arrange
body = {"max_count": 100}
# Act
response = await test_client.post(f"{PAIRINGS_URL}/clean-expired", json=body)
# Assert
assert response.status_code == 401, response.text
async def test_clean_expired_pairings_forbids_standard_user(
test_client: httpx.AsyncClient, standard_user
):
# Arrange
body = {"max_count": 100}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/clean-expired",
json=body,
headers=standard_user["headers"],
)
# Assert
assert response.status_code == 403, response.text
async def test_clean_expired_pairings_rejects_max_count_zero(test_client: httpx.AsyncClient, admin_headers):
# Arrange — max_count=0 低于下限 1
body = {"max_count": 0}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/clean-expired",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
async def test_clean_expired_pairings_rejects_max_count_over_max(test_client: httpx.AsyncClient, admin_headers):
# Arrange — max_count=1001 超过上限 1000
body = {"max_count": 1001}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/clean-expired",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 422, response.text
# =============================================================================
# === pairings stats ===
# =============================================================================
async def test_get_pairings_stats_requires_auth(test_client: httpx.AsyncClient):
# Act
response = await test_client.get(f"{PAIRINGS_URL}/stats")
# Assert
assert response.status_code == 401, response.text
async def test_get_pairings_stats_forbids_standard_user(
test_client: httpx.AsyncClient, standard_user
):
# Act
response = await test_client.get(f"{PAIRINGS_URL}/stats", headers=standard_user["headers"])
# Assert
assert response.status_code == 403, response.text
async def test_get_pairings_stats_returns_200(test_client: httpx.AsyncClient, admin_headers):
# Act
response = await test_client.get(
f"{PAIRINGS_URL}/stats",
params={"start_time": ISO_TIME_START, "end_time": ISO_TIME_END},
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 pairing detail ===
# =============================================================================
async def test_get_pairing_requires_auth(test_client: httpx.AsyncClient):
# Act
response = await test_client.get(f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}")
# Assert
assert response.status_code == 401, response.text
async def test_get_pairing_forbids_standard_user(test_client: httpx.AsyncClient, standard_user):
# Act
response = await test_client.get(
f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}",
headers=standard_user["headers"],
)
# Assert
assert response.status_code == 403, response.text
async def test_get_pairing_non_existent_returns_404(test_client: httpx.AsyncClient, admin_headers):
# Act
response = await test_client.get(
f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}",
headers=admin_headers,
)
# Assert
assert response.status_code == 404, response.text
# =============================================================================
# === approve pairing ===
# =============================================================================
async def test_approve_pairing_requires_auth(test_client: httpx.AsyncClient):
# Act
response = await test_client.post(f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}/approve")
# Assert
assert response.status_code == 401, response.text
async def test_approve_pairing_forbids_standard_user(
test_client: httpx.AsyncClient, standard_user
):
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}/approve",
headers=standard_user["headers"],
)
# Assert
assert response.status_code == 403, response.text
async def test_approve_pairing_non_existent_returns_404(test_client: httpx.AsyncClient, admin_headers):
# Arrange
body = {"approver_id": "admin_xxx", "reason": "pytest"}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}/approve",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 404, response.text
# =============================================================================
# === reject pairing ===
# =============================================================================
async def test_reject_pairing_requires_auth(test_client: httpx.AsyncClient):
# Act
response = await test_client.post(f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}/reject")
# Assert
assert response.status_code == 401, response.text
async def test_reject_pairing_forbids_standard_user(
test_client: httpx.AsyncClient, standard_user
):
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}/reject",
headers=standard_user["headers"],
)
# Assert
assert response.status_code == 403, response.text
async def test_reject_pairing_non_existent_returns_404(test_client: httpx.AsyncClient, admin_headers):
# Arrange
body = {"approver_id": "admin_xxx", "reason": "pytest"}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}/reject",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 404, response.text
# =============================================================================
# === revoke pairing ===
# =============================================================================
async def test_revoke_pairing_requires_auth(test_client: httpx.AsyncClient):
# Act
response = await test_client.post(f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}/revoke")
# Assert
assert response.status_code == 401, response.text
async def test_revoke_pairing_forbids_standard_user(
test_client: httpx.AsyncClient, standard_user
):
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}/revoke",
headers=standard_user["headers"],
)
# Assert
assert response.status_code == 403, response.text
async def test_revoke_pairing_non_existent_returns_404(test_client: httpx.AsyncClient, admin_headers):
# Arrange
body = {"reason": "pytest"}
# Act
response = await test_client.post(
f"{PAIRINGS_URL}/{NON_EXISTENT_PAIRING_ID}/revoke",
json=body,
headers=admin_headers,
)
# Assert
assert response.status_code == 404, response.text
Reference in New Issue View Git Blame Copy Permalink