selfrelease
5278190750
- 后端: 凭证生成引擎、金蝶导出器、凭证模板服务 - 后端: 成本分析服务、AI问答服务 - 后端: 科目映射CRUD API、分析API、QA API - 后端: 集成测试(认证/任务/凭证) 49个测试全部通过 - 前端: 凭证管理、成本分析、导出中心、知识库、系统设置页面 - 前端: AuthGuard认证守卫、Dashboard AI聊天功能 - 前端: Playwright E2E测试 16 passed, 1 skipped - 基础设施: Docker Compose、Nginx反向代理、.env.example - 文档: 用户手册、管理员手册、发布检查清单
98 lines
2.6 KiB
Python
98 lines
2.6 KiB
Python
"""
|
|
集成测试公共夹具
|
|
|
|
使用同步 TestClient + Mock AsyncSession 进行 API 集成测试
|
|
避免 AsyncClient + ASGITransport 在 pytest-asyncio 下的死锁问题
|
|
"""
|
|
|
|
from typing import AsyncGenerator
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.database import get_db
|
|
from app.main import app
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_db() -> AsyncMock:
|
|
"""
|
|
创建 Mock AsyncSession
|
|
|
|
返回一个 AsyncMock 对象，模拟异步数据库会话
|
|
"""
|
|
session = AsyncMock()
|
|
session.commit = AsyncMock()
|
|
session.refresh = AsyncMock()
|
|
session.rollback = AsyncMock()
|
|
session.close = AsyncMock()
|
|
session.add = MagicMock()
|
|
session.delete = AsyncMock()
|
|
session.get = AsyncMock(return_value=None)
|
|
result_mock = MagicMock()
|
|
result_mock.scalars.return_value.all.return_value = []
|
|
result_mock.scalars.return_value.first.return_value = None
|
|
result_mock.scalar_one_or_none.return_value = None
|
|
result_mock.scalar.return_value = 0
|
|
result_mock.one.return_value = MagicMock()
|
|
session.execute = AsyncMock(return_value=result_mock)
|
|
return session
|
|
|
|
|
|
@pytest.fixture
|
|
def client(mock_db: AsyncMock) -> TestClient:
|
|
"""
|
|
创建同步测试客户端，覆盖数据库依赖
|
|
"""
|
|
|
|
async def override_get_db() -> AsyncGenerator[AsyncMock, None]:
|
|
yield mock_db
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
with TestClient(app) as tc:
|
|
yield tc
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def make_db_result(items: list = None, scalar=None, one=None):
|
|
"""
|
|
构造 db.execute() 返回值的辅助函数
|
|
|
|
Args:
|
|
items: scalars().all() 返回的列表
|
|
scalar: scalar() 返回的标量值
|
|
one: one() 返回的行对象
|
|
"""
|
|
result = MagicMock()
|
|
result.scalars.return_value.all.return_value = items or []
|
|
result.scalars.return_value.first.return_value = items[0] if items else None
|
|
result.scalar_one_or_none.return_value = scalar
|
|
result.scalar.return_value = scalar if scalar is not None else 0
|
|
if one:
|
|
result.one.return_value = one
|
|
return result
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_token() -> str:
|
|
"""
|
|
生成测试用 JWT Token（不依赖数据库）
|
|
"""
|
|
from app.core.security import create_access_token
|
|
return create_access_token(data={"sub": "1", "email": "test@example.com"})
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_headers(auth_token: str) -> dict:
|
|
"""
|
|
返回认证请求头
|
|
|
|
Returns:
|
|
包含 Authorization 和 X-Company-ID 的请求头
|
|
"""
|
|
return {
|
|
"Authorization": f"Bearer {auth_token}",
|
|
"X-Company-ID": "1",
|
|
}
|