feat: 凭证生成、成本分析、AI问答、前端页面、集成测试与E2E测试
- 后端: 凭证生成引擎、金蝶导出器、凭证模板服务 - 后端: 成本分析服务、AI问答服务 - 后端: 科目映射CRUD API、分析API、QA API - 后端: 集成测试(认证/任务/凭证) 49个测试全部通过 - 前端: 凭证管理、成本分析、导出中心、知识库、系统设置页面 - 前端: AuthGuard认证守卫、Dashboard AI聊天功能 - 前端: Playwright E2E测试 16 passed, 1 skipped - 基础设施: Docker Compose、Nginx反向代理、.env.example - 文档: 用户手册、管理员手册、发布检查清单
This commit is contained in:
@@ -0,0 +1,97 @@
|
||||
"""
|
||||
集成测试公共夹具
|
||||
|
||||
使用同步 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",
|
||||
}
|
||||
Reference in New Issue
Block a user