226 lines
7.4 KiB
Python
226 lines
7.4 KiB
Python
|
# =============================================================================
|
|||
|
|
# 三端认证重构 CTRT — 响应契约统一测试
|
|||
|
|
# =============================================================================
|
|||
|
|
# 验证后端响应格式符合统一信封规范：
|
|||
|
|
# - 成功响应: {code: 0, data: {...}, message: "success"}
|
|||
|
|
# - 失败响应: {code: 非0, data: null, message: "..."}
|
|||
|
|
#
|
|||
|
|
# 前端拦截器会根据 code 判断，成功时返回内层 data（CTRT-01）
|
|||
|
|
# 失败时抛出错对象 {code, message}（CTRT-03）
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
from tests.conftest import create_test_agent
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =============================================================================
|
|||
|
|
# 辅助函数
|
|||
|
|
# =============================================================================
|
|||
|
|
async def _login_and_get_token(client, user_id: str, name: str, otp_code: str = None) -> str:
|
|||
|
|
"""登录获取 token
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
client: 测试客户端
|
|||
|
|
user_id: 用户 ID
|
|||
|
|
name: 用户名
|
|||
|
|
otp_code: 可选的 OTP 验证码
|
|||
|
|
"""
|
|||
|
|
payload = {"user_id": user_id, "name": name}
|
|||
|
|
if otp_code:
|
|||
|
|
payload["otp_code"] = otp_code
|
|||
|
|
|
|||
|
|
response = await client.post(
|
|||
|
|
"/agents/login",
|
|||
|
|
json=payload,
|
|||
|
|
)
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
body = response.json()
|
|||
|
|
|
|||
|
|
# 如果返回 require_otp，说明需要 OTP 验证
|
|||
|
|
if body["data"].get("require_otp"):
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
return body["data"]["token"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _bearer(token: str) -> dict:
|
|||
|
|
return {"Authorization": f"Bearer {token}"}
|
|||
|
|
|
|||
|
|
|
|||
|
|
# =============================================================================
|
|||
|
|
# 响应格式测试
|
|||
|
|
# =============================================================================
|
|||
|
|
class TestResponseEnvelope:
|
|||
|
|
"""测试统一响应信封格式"""
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_success_response_format(self, client, db_session):
|
|||
|
|
"""成功响应应符合 {code: 0, data: {...}, message: "success"} 格式"""
|
|||
|
|
# 创建一个坐席
|
|||
|
|
agent = create_test_agent(user_id="test_agent_001", name="TestAgent")
|
|||
|
|
db_session.add(agent)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
# 登录
|
|||
|
|
token = await _login_and_get_token(client, "test_agent_001", "TestAgent")
|
|||
|
|
|
|||
|
|
# 调用需要认证的端点
|
|||
|
|
response = await client.get("/agents/me", headers=_bearer(token))
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
body = response.json()
|
|||
|
|
|
|||
|
|
# 验证响应格式
|
|||
|
|
assert "code" in body
|
|||
|
|
assert "data" in body
|
|||
|
|
assert "message" in body
|
|||
|
|
assert body["code"] == 0
|
|||
|
|
assert body["message"] == "success"
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_error_response_format(self, client):
|
|||
|
|
"""错误响应应符合 {code: 非0, data: null, message: "..."} 格式"""
|
|||
|
|
# 使用无效 token 调用
|
|||
|
|
response = await client.get(
|
|||
|
|
"/agents/me",
|
|||
|
|
headers=_bearer("invalid_token_12345")
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 401 会返回统一错误格式
|
|||
|
|
assert response.status_code in [200, 401]
|
|||
|
|
body = response.json()
|
|||
|
|
|
|||
|
|
# 验证错误响应格式
|
|||
|
|
assert "code" in body
|
|||
|
|
assert "message" in body
|
|||
|
|
# code 不为 0
|
|||
|
|
assert body["code"] != 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestOTPResponseContract:
|
|||
|
|
"""测试 OTP 端点响应契约"""
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_otp_status_response(self, client, db_session):
|
|||
|
|
"""OTP status 端点返回内层 data"""
|
|||
|
|
agent = create_test_agent(user_id="otp_test_001", name="OTPTest")
|
|||
|
|
db_session.add(agent)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
token = await _login_and_get_token(client, "otp_test_001", "OTPTest")
|
|||
|
|
response = await client.get("/auth/otp-status", headers=_bearer(token))
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
body = response.json()
|
|||
|
|
|
|||
|
|
# 验证统一格式
|
|||
|
|
assert body["code"] == 0
|
|||
|
|
assert body["data"]["bound"] is False
|
|||
|
|
assert body["data"]["enabled"] is False
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_otp_bind_response(self, client, db_session):
|
|||
|
|
"""OTP bind 端点返回内层 data"""
|
|||
|
|
agent = create_test_agent(user_id="otp_test_002", name="OTPTest2")
|
|||
|
|
db_session.add(agent)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
token = await _login_and_get_token(client, "otp_test_002", "OTPTest2")
|
|||
|
|
response = await client.post("/auth/otp-bind", headers=_bearer(token))
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
body = response.json()
|
|||
|
|
|
|||
|
|
# 验证统一格式
|
|||
|
|
assert body["code"] == 0
|
|||
|
|
assert "secret" in body["data"]
|
|||
|
|
assert "otpauth_url" in body["data"]
|
|||
|
|
assert "qr_code_base64" in body["data"]
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_otp_verify_response(self, client, db_session):
|
|||
|
|
"""OTP verify 端点返回内层 data"""
|
|||
|
|
from datetime import datetime
|
|||
|
|
import pyotp
|
|||
|
|
|
|||
|
|
agent = create_test_agent(user_id="otp_test_003", name="OTPTest3")
|
|||
|
|
secret = pyotp.random_base32()
|
|||
|
|
agent.mfa_secret = secret
|
|||
|
|
agent.mfa_enabled = True
|
|||
|
|
agent.mfa_bound_at = datetime.now()
|
|||
|
|
db_session.add(agent)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
# 已启用 MFA 的用户需要使用 OTP 验证码登录
|
|||
|
|
otp_code = pyotp.TOTP(secret).now()
|
|||
|
|
token = await _login_and_get_token(client, "otp_test_003", "OTPTest3", otp_code)
|
|||
|
|
response = await client.post(
|
|||
|
|
"/auth/otp-verify",
|
|||
|
|
headers=_bearer(token),
|
|||
|
|
json={"otp_code": otp_code},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
body = response.json()
|
|||
|
|
|
|||
|
|
# 验证统一格式
|
|||
|
|
assert body["code"] == 0
|
|||
|
|
assert body["data"]["verified"] is True
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestAgentsLoginResponse:
|
|||
|
|
"""测试登录响应契约"""
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_login_without_mfa_returns_token(self, client, db_session):
|
|||
|
|
"""无 MFA 时登录直接返回 token"""
|
|||
|
|
agent = create_test_agent(user_id="login_test_001", name="LoginTest")
|
|||
|
|
db_session.add(agent)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
response = await client.post(
|
|||
|
|
"/agents/login",
|
|||
|
|
json={"user_id": "login_test_001", "name": "LoginTest"}
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
body = response.json()
|
|||
|
|
|
|||
|
|
# 验证统一格式
|
|||
|
|
assert body["code"] == 0
|
|||
|
|
assert "token" in body["data"]
|
|||
|
|
assert "user_id" in body["data"]
|
|||
|
|
assert "name" in body["data"]
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_login_with_mfa_requires_otp(self, client, db_session):
|
|||
|
|
"""有 MFA 时登录返回 require_otp"""
|
|||
|
|
from datetime import datetime
|
|||
|
|
import pyotp
|
|||
|
|
|
|||
|
|
agent = create_test_agent(user_id="login_test_002", name="LoginTest2")
|
|||
|
|
secret = pyotp.random_base32()
|
|||
|
|
agent.mfa_secret = secret
|
|||
|
|
agent.mfa_enabled = True
|
|||
|
|
agent.mfa_bound_at = datetime.now()
|
|||
|
|
db_session.add(agent)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
# 登录时不带 otp_code
|
|||
|
|
response = await client.post(
|
|||
|
|
"/agents/login",
|
|||
|
|
json={"user_id": "login_test_002", "name": "LoginTest2"}
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert response.status_code == 200
|
|||
|
|
body = response.json()
|
|||
|
|
|
|||
|
|
# 应该返回 require_otp: true（不带 token）
|
|||
|
|
assert body["code"] == 0
|
|||
|
|
assert body["data"].get("require_otp") is True
|
|||
|
|
assert "token" not in body["data"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
pytest.main([__file__, "-v"])
|