# ============================================================================= # 企微IT智能服务台 — 测试配置与公共 fixtures # ============================================================================= # 说明：pytest 的全局 fixtures，包括： # 1. SQLite 内存数据库（替代 PostgreSQL） # 2. 模拟 Redis 客户端 # 3. FastAPI 测试客户端 # 4. 测试用数据库会话 # ============================================================================= import asyncio import uuid from datetime import datetime from typing import AsyncGenerator, Dict, Optional from unittest.mock import AsyncMock, MagicMock, patch import pytest import pytest_asyncio from httpx import ASGITransport, AsyncClient from sqlalchemy import event from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.pool import StaticPool from app.database import Base from app.models.agent import Agent from app.models.conversation import Conversation from app.models.message import Message from app.models.system_config import SystemConfig from app.models.funny_phrase import FunnyPhrase from app.models.approval_link import ApprovalLink from app.models.software_download import SoftwareDownload from app.models.quick_reply_template import QuickReplyTemplate from app.models.agent_note import AgentNote # ============================================================================= # 2026-06-15 修复: monkey-patch starlette.config.Config 强制 UTF-8 读 .env # 原因: Windows pytest 默认 GBK 读 .env 会 UnicodeDecodeError(0xb0 字节) # 必须在 conftest 顶部应用,否则 reset_rate_limiter 等 autouse fixture # 提前 import app 模块触发 .env 读取时会失败 # ============================================================================= import starlette.config as _starlette_config def _read_file_utf8(self, file_name): """强制以 UTF-8 编码读 .env,避免 Windows GBK 默认编码触发 UnicodeDecodeError。""" result = {} with open(file_name, encoding='utf-8') as f: for line in f: line = line.strip() if not line or line.startswith('#'): continue if '=' in line: k, v = line.split('=', 1) result[k.strip()] = v.strip().strip('"').strip("'") return result _starlette_config.Config._read_file = _read_file_utf8 # ============================================================================= # SQLite 内存数据库引擎 # ============================================================================= # 使用 aiosqlite 驱动的 SQLite 内存数据库替代 PostgreSQL # StaticPool 确保所有连接使用同一个内存数据库实例 # ============================================================================= TEST_DATABASE_URL = "sqlite+aiosqlite://" test_engine = create_async_engine( TEST_DATABASE_URL, connect_args={"check_same_thread": False}, poolclass=StaticPool, ) test_session_factory = async_sessionmaker( test_engine, class_=AsyncSession, expire_on_commit=False, ) # 为 SQLite 启用外键约束 @event.listens_for(test_engine.sync_engine, "connect") def _set_sqlite_pragma(dbapi_connection, connection_record): cursor = dbapi_connection.cursor() cursor.execute("PRAGMA foreign_keys=ON") cursor.close() # ============================================================================= # 模拟 Redis 客户端 # ============================================================================= class MockRedis: """模拟 Redis 客户端，使用内存字典存储数据。""" def __init__(self): self._data: Dict[str, str] = {} self._ttl: Dict[str, int] = {} async def get(self, key: str) -> Optional[bytes]: value = self._data.get(key) if value is not None: return value.encode("utf-8") if isinstance(value, str) else value return None async def setex(self, name: str, time: int, value: str) -> None: self._data[name] = value self._ttl[name] = time async def set(self, name: str, value: str, **kwargs) -> Optional[bool]: """模拟 Redis SET 命令，支持 nx 和 ex 参数。 Args: name: Redis key value: Redis value **kwargs: nx: SET IF NOT EXISTS — key 不存在时才设置，返回 True；已存在返回 None ex: 过期时间（秒） Returns: nx=True 时：True=设置成功，None=key 已存在未设置 其他情况：None（与真实 Redis SET 行为一致） """ nx = kwargs.get("nx", False) ex = kwargs.get("ex", None) if nx: if name in self._data: return None # key 已存在，SET NX 未设置 self._data[name] = value if ex is not None: self._ttl[name] = ex return True # 设置成功 self._data[name] = value if ex is not None: self._ttl[name] = ex return None async def delete(self, *names) -> int: count = 0 for name in names: if name in self._data: del self._data[name] count += 1 return count async def exists(self, *keys) -> int: return sum(1 for k in keys if k in self._data) async def expire(self, name: str, time: int) -> bool: if name in self._data: self._ttl[name] = time return True return False async def close(self) -> None: pass def reset(self) -> None: self._data.clear() self._ttl.clear() # ============================================================================= # Fixtures # ============================================================================= @pytest.fixture(scope="session") def event_loop(): """创建 session 级别的事件循环。""" loop = asyncio.new_event_loop() yield loop loop.close() @pytest_asyncio.fixture(scope="session", autouse=True) async def setup_database(): """创建所有数据库表（session 级别，只执行一次）。""" async with test_engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) yield async with test_engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) @pytest_asyncio.fixture async def db_session() -> AsyncGenerator[AsyncSession, None]: """提供干净的数据库会话，每个测试用例使用独立事务并在测试后回滚。""" async with test_session_factory() as session: # 开始一个嵌套事务 nested = await session.begin_nested() try: yield session finally: # 回滚嵌套事务，确保数据库干净 if nested.is_active: await nested.rollback() # 清理会话 await session.close() @pytest.fixture def mock_redis() -> MockRedis: """提供模拟 Redis 客户端。""" return MockRedis() # ============================================================================= # 模块级 Mock 外部服务（让子测试可覆盖其行为） # ============================================================================= # 2026-06-15 修复: 把 WecomService / AIService mock 提升到模块级 # 原因: client fixture 内的局部 mock 无法被测试内 `with patch.object(...)` 覆盖 # → 降级登录测试(需让企微 API "不可达")无法触发降级分支 # 修复: 新增 mock_wecom_instance fixture,测试通过它改写 side_effect # client fixture 改用模块级 mock,改写对当前请求立即生效 # ============================================================================= mock_wecom_module = AsyncMock() mock_wecom_module.send_message.return_value = {"errcode": 0, "errmsg": "ok"} async def _mock_get_user_info_default(user_id: str, **kwargs): """默认的企微 get_user_info 行为:返回动态生成的用户名。 测试可通过 mock_wecom_instance.get_user_info.side_effect = ... 改写。 """ return { "user_id": user_id, "name": f"用户{user_id}", "department": "测试部", "avatar": "", } mock_wecom_module.get_user_info.side_effect = _mock_get_user_info_default mock_wecom_module.get_department_users.return_value = [] mock_ai_module = AsyncMock() mock_ai_module.generate_response.return_value = "这是AI的模拟回复" @pytest.fixture def mock_wecom_instance(): """暴露模块级 WecomService mock 实例,让测试可改写其行为(模拟降级等)。 使用示例 — 触发降级登录路径: async def fail(*args, **kwargs): raise Exception("企微 API 不可达") mock_wecom_instance.get_user_info.side_effect = fail # ...发起请求后,用 try/finally 恢复原 side_effect """ return mock_wecom_module @pytest.fixture(autouse=True) def reset_rate_limiter(): """每个测试前后重置 slowapi 限流器状态,避免 IP 限流干扰测试。 背景: /agents/login 限流 10/min per IP,pytest 连续跑多个测试会撞 429。 """ from app.api.agents import limiter as agents_limiter try: agents_limiter._storage.reset() except Exception: pass yield try: agents_limiter._storage.reset() except Exception: pass @pytest_asyncio.fixture async def client(db_session: AsyncSession, mock_redis: MockRedis) -> AsyncGenerator[AsyncClient, None]: """提供 FastAPI 异步测试客户端。""" async def _override_get_db(): yield db_session async def _override_get_redis(): return mock_redis # 注: 2026-06-15 UTF-8 monkey-patch 已提升到 conftest 模块级,见文件顶部 # 原因: reset_rate_limiter 等 autouse fixture 提前 import 触发 .env 读取 from app.main import create_app from app.database import get_db app = create_app() # 覆盖数据库依赖 app.dependency_overrides[get_db] = _override_get_db # 模拟 Redis（同时 mock agents 和 h5 模块的 Redis 依赖） with patch("app.api.agents._get_redis", return_value=mock_redis): with patch("redis.asyncio.from_url", return_value=mock_redis): # ------------------------------------------------------------------ # Mock 外部服务：WecomService（企微API）和 AIService（AI大模型） # 为什么：测试中不应调用真实企微API/AI大模型 # 怎么做：patch 类构造函数，返回配置了默认返回值的 mock 对象 # ------------------------------------------------------------------ # 使用模块级 mock_wecom_module / mock_ai_module（2026-06-15 修复） # 原因: 模块级 mock 允许测试通过 mock_wecom_instance fixture 改写行为 # 例如降级登录测试改 side_effect = raise Exception("企微不可达") mock_wecom = mock_wecom_module mock_ai = mock_ai_module # Patch WecomService 类（端点函数中会新建实例） # 注意：只 patch 模块中实际引用的名字 # conversations.py 导入了 WecomService，但没有导入 AIService with patch("app.api.conversations.WecomService", return_value=mock_wecom): # h5.py 和 agents.py 也需要 patch with patch("app.api.h5.WecomService", return_value=mock_wecom): with patch("app.api.agents.WecomService", return_value=mock_wecom): with patch("app.api.agents._get_redis", return_value=mock_redis): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as ac: yield ac app.dependency_overrides.clear() @pytest_asyncio.fixture async def seeded_db(db_session: AsyncSession) -> AsyncSession: """插入测试基础数据并返回会话。""" # 系统配置 configs = [ SystemConfig(config_key="hand_raise_keywords", config_value='["转人工","人工","人工服务","真人","客服"]', description="举手关键词"), SystemConfig(config_key="emotion_keywords_angry", config_value='["崩溃","愤怒","投诉","差劲","垃圾"]', description="愤怒关键词"), SystemConfig(config_key="emotion_keywords_urgent", config_value='["急","紧急","马上","立刻","赶紧"]', description="紧急关键词"), SystemConfig(config_key="emotion_keywords_worried", config_value='["担心","害怕","出错","丢失","完蛋"]', description="担忧关键词"), SystemConfig(config_key="intervene_round_threshold", config_value="3", description="介入阈值"), SystemConfig(config_key="urgency_base_keyword_score", config_value="1", description="基础加分"), SystemConfig(config_key="urgency_emotion_bonus", config_value="1", description="情绪加成"), SystemConfig(config_key="urgency_vip_bonus", config_value="1", description="VIP加成"), SystemConfig(config_key="urgency_repeat_bonus", config_value="1", description="重复加成"), ] db_session.add_all(configs) # 趣味话术 phrases = [ FunnyPhrase(scene="shake", content="大哥，俺这就去摇人，稍等...", tone="亲切", sort_order=1), FunnyPhrase(scene="vip", content="这就帮您安排专家，请稍候", tone="正式", sort_order=1), ] db_session.add_all(phrases) # 审批链接 links = [ ApprovalLink(category="IT", title="软件安装申请", url="https://example.com/software", sort_order=1), ApprovalLink(category="HR", title="入职手续", url="https://example.com/onboarding", sort_order=2), ] db_session.add_all(links) # 软件下载 downloads = [ SoftwareDownload(category="办公", name="企业微信", version="最新版", platform="全平台", download_url="https://work.weixin.qq.com", sort_order=1), SoftwareDownload(category="开发", name="VS Code", version="1.90", platform="Windows/Mac/Linux", download_url="https://code.visualstudio.com", sort_order=2), ] db_session.add_all(downloads) await db_session.flush() return db_session # ============================================================================= # 辅助函数 # ============================================================================= def create_test_conversation( employee_id: str = "test_employee_001", employee_name: str = "测试员工", status: str = "queued", is_vip: bool = False, is_pinned: bool = False, is_todo: bool = False, urgency_score: int = 1, tags: Optional[Dict] = None, ) -> Conversation: """创建测试用的会话对象。""" return Conversation( employee_id=employee_id, employee_name=employee_name, department="技术部", position="工程师", level="", status=status, is_vip=is_vip, is_pinned=is_pinned, is_todo=is_todo, urgency_score=urgency_score, tags=tags or {}, last_message_at=datetime.now(), last_message_summary="测试消息", ) def create_test_agent( user_id: str = "test_agent_001", name: str = "测试坐席", status: str = "online", ) -> Agent: """创建测试用的坐席对象。""" return Agent( user_id=user_id, name=name, status=status, current_load=0, max_load=5, )