# ============================================================================= # 企微IT智能服务台 — 统一认证 API 测试 # ============================================================================= # 测试覆盖: # 1. GET /auth/qrcode — 获取扫码登录二维码 # 2. GET /auth/scan/status — 轮询扫码状态 # 3. POST /auth/verify — 验证 Token # 4. POST /auth/logout — 登出 # 5. GET /auth/me — 获取当前用户信息 # 6. POST /auth/switch-role — 切换角色 # # 说明：这些端点复用现有的 auth_qrcode 服务，因此重点测试统一认证 API 层的封装 # ============================================================================= import json import secrets from datetime import datetime import pytest from tests.conftest import MockRedis # -------------------------------------------------------------------------- # 工具: 创建测试 Token 并写入 Redis # -------------------------------------------------------------------------- async def _create_test_token( mock_redis: MockRedis, employee_id: str = "test-user-001", name: str = "测试用户", roles: list = None, login_source: str = "h5", ) -> str: """在 mock_redis 里手动写一个用户 token,返回 token 字符串。 与 TokenService.create_token 一致: 写 user:token:{token} """ if roles is None: roles = ["user"] token = secrets.token_urlsafe(32) token_data = { "employee_id": employee_id, "name": name, "department": "测试部", "avatar": "", "roles": roles, "current_role": roles[0] if roles else "user", "login_source": login_source, "created_at": datetime.now().isoformat(), "last_active": datetime.now().isoformat(), } await mock_redis.setex( f"user:token:{token}", 8 * 60 * 60, json.dumps(token_data, ensure_ascii=False), ) return token # -------------------------------------------------------------------------- # 1. GET /auth/qrcode — 获取扫码登录二维码 # -------------------------------------------------------------------------- class TestQrcodeEndpoint: """测试统一认证的二维码获取端点。""" @pytest.mark.asyncio async def test_get_qrcode_returns_ticket_and_url(self, client, mock_redis): """验证获取二维码返回 ticket + qrcode_url + qrcode_png_base64 + expires_in。""" response = await client.get("/auth/qrcode") assert response.status_code == 200 body = response.json() assert body["code"] == 0 assert "data" in body assert body["data"] is not None data = body["data"] assert "ticket" in data assert len(data["ticket"]) >= 16 assert "qrcode_url" in data assert "qrcode_png_base64" in data assert "expires_in" in data assert "expires_at" in data # 有效期应该是 120 秒 (从 QrcodeService 来的) assert data["expires_in"] == 120 # -------------------------------------------------------------------------- # 2. POST /auth/verify — 验证 Token # -------------------------------------------------------------------------- class TestVerifyTokenEndpoint: """测试统一认证的 Token 验证端点。""" @pytest.mark.asyncio async def test_verify_invalid_token_returns_false(self, client, mock_redis): """验证无效 Token 返回 valid=False。""" response = await client.post( "/auth/verify", json={"token": "invalid-token-xyz"}, ) assert response.status_code == 200 body = response.json() assert body["code"] == 0 data = body["data"] assert data["valid"] is False assert data["employee_id"] is None assert data["name"] is None @pytest.mark.asyncio async def test_verify_blacklisted_token_returns_false(self, client, mock_redis): """验证已加入黑名单的 Token 返回 valid=False。""" # 创建 token token = await _create_test_token( mock_redis, employee_id="test-user-001", name="测试用户", ) # 将 token 加入黑名单 import hashlib token_hash = hashlib.sha256(token.encode()).hexdigest() await mock_redis.setex(f"token:blacklist:{token_hash}", 8 * 60 * 60, "1") # 验证黑名单中的 token response = await client.post( "/auth/verify", json={"token": token}, ) assert response.status_code == 200 body = response.json() assert body["code"] == 0 data = body["data"] assert data["valid"] is False # -------------------------------------------------------------------------- # 3. POST /auth/logout — 登出 # -------------------------------------------------------------------------- class TestLogoutEndpoint: """测试统一认证的登出端点。""" @pytest.mark.asyncio async def test_logout_without_auth_returns_error(self, client, mock_redis): """验证未登录登出返回错误。""" response = await client.post("/auth/logout") # 未鉴权应该返回 401 或 403 assert response.status_code in (401, 403) # -------------------------------------------------------------------------- # 4. GET /auth/me — 获取当前用户信息 # -------------------------------------------------------------------------- class TestCurrentUserEndpoint: """测试统一认证的获取当前用户信息端点。""" @pytest.mark.asyncio async def test_get_current_user_returns_info(self, client, mock_redis): """验证获取当前用户信息成功。""" # 创建有效 token token = await _create_test_token( mock_redis, employee_id="test-user-001", name="测试用户", roles=["agent"], login_source="agent", ) # 获取当前用户 response = await client.get( "/auth/me", headers={"Authorization": f"Bearer {token}"}, ) assert response.status_code == 200 body = response.json() assert body["code"] == 0 data = body["data"] assert data["employee_id"] == "test-user-001" assert data["name"] == "测试用户" assert data["roles"] == ["agent"] assert data["current_role"] == "agent" assert data["login_source"] == "agent" @pytest.mark.asyncio async def test_get_current_user_without_auth_returns_error(self, client, mock_redis): """验证未登录获取用户信息返回错误。""" response = await client.get("/auth/me") # 未鉴权应该返回 401 或 403 assert response.status_code in (401, 403) # -------------------------------------------------------------------------- # 5. POST /auth/switch-role — 切换角色 # -------------------------------------------------------------------------- class TestSwitchRoleEndpoint: """测试统一认证的角色切换端点。""" @pytest.mark.asyncio async def test_switch_role_without_auth_returns_error(self, client, mock_redis): """验证未登录切换角色返回错误。""" response = await client.post( "/auth/switch-role", json={"role": "admin"}, ) # 未鉴权应该返回 401 或 403 assert response.status_code in (401, 403) # -------------------------------------------------------------------------- # 6. 废弃 API 测试 # -------------------------------------------------------------------------- class TestDeprecatedAPIs: """测试废弃的登录 API 是否正确禁用或返回提示。""" @pytest.mark.asyncio async def test_agents_login_still_works(self, client, mock_redis, db_session): """验证坐席登录端点仍然可用。""" # agents/login 是坐席登录，仍应可用 response = await client.post( "/agents/login", json={"user_id": "test-agent-001", "name": "测试坐席"}, ) # 应该返回 200 assert response.status_code == 200 body = response.json() assert body["code"] == 0 assert "token" in body["data"]