# ============================================================================= # 三端认证重构 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"])