selfrelease
5278190750
- 后端: 凭证生成引擎、金蝶导出器、凭证模板服务 - 后端: 成本分析服务、AI问答服务 - 后端: 科目映射CRUD API、分析API、QA API - 后端: 集成测试(认证/任务/凭证) 49个测试全部通过 - 前端: 凭证管理、成本分析、导出中心、知识库、系统设置页面 - 前端: AuthGuard认证守卫、Dashboard AI聊天功能 - 前端: Playwright E2E测试 16 passed, 1 skipped - 基础设施: Docker Compose、Nginx反向代理、.env.example - 文档: 用户手册、管理员手册、发布检查清单
111 lines
4.1 KiB
Python
111 lines
4.1 KiB
Python
"""
|
|
对账任务 API 集成测试
|
|
|
|
测试任务列表、统计、详情等端点
|
|
使用同步 TestClient + Mock 数据库会话
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.models.reconciliation_task import ReconciliationTask
|
|
from tests.conftest import make_db_result
|
|
|
|
|
|
def _make_task(task_id=1, period="2026-07", status="COMPLETED",
|
|
total=100, matched=95, exceptions=5):
|
|
"""构造测试任务对象"""
|
|
return ReconciliationTask(
|
|
id=task_id,
|
|
company_id=1,
|
|
period=period,
|
|
status=status,
|
|
total_employees=total,
|
|
matched_count=matched,
|
|
exception_count=exceptions,
|
|
file_ids=[],
|
|
reconciliation_result={},
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow(),
|
|
)
|
|
|
|
|
|
class TestTasksAPI:
|
|
"""对账任务 API 集成测试"""
|
|
|
|
def test_list_tasks_empty(self, client: TestClient, auth_headers: dict, mock_db: AsyncMock):
|
|
"""测试空任务列表"""
|
|
mock_db.execute = AsyncMock(return_value=make_db_result(items=[], scalar=0))
|
|
|
|
response = client.get("/api/tasks/", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["items"] == []
|
|
assert data["total"] == 0
|
|
|
|
def test_list_tasks_with_data(self, client: TestClient, auth_headers: dict, mock_db: AsyncMock):
|
|
"""测试有数据的任务列表"""
|
|
task = _make_task()
|
|
# tasks.py 先执行 count 查询，再执行 list 查询
|
|
call_count = [0]
|
|
async def side_effect(*args, **kwargs):
|
|
call_count[0] += 1
|
|
if call_count[0] == 1:
|
|
return make_db_result(scalar=1) # count
|
|
return make_db_result(items=[task]) # list
|
|
mock_db.execute = AsyncMock(side_effect=side_effect)
|
|
|
|
response = client.get("/api/tasks/", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 1
|
|
assert data["items"][0]["period"] == "2026-07"
|
|
assert data["items"][0]["status"] == "COMPLETED"
|
|
|
|
def test_list_tasks_pagination(self, client: TestClient, auth_headers: dict, mock_db: AsyncMock):
|
|
"""测试分页"""
|
|
tasks = [_make_task(task_id=i, period=f"2025-{i:02d}") for i in range(1, 6)]
|
|
call_count = [0]
|
|
async def side_effect(*args, **kwargs):
|
|
call_count[0] += 1
|
|
if call_count[0] == 1:
|
|
return make_db_result(scalar=15)
|
|
return make_db_result(items=tasks)
|
|
mock_db.execute = AsyncMock(side_effect=side_effect)
|
|
|
|
response = client.get("/api/tasks/?page=1&page_size=5", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["items"]) == 5
|
|
assert data["total"] == 15
|
|
assert data["page"] == 1
|
|
|
|
def test_get_task_detail(self, client: TestClient, auth_headers: dict, mock_db: AsyncMock):
|
|
"""测试获取任务详情"""
|
|
task = _make_task(task_id=42)
|
|
mock_db.get = AsyncMock(return_value=task)
|
|
|
|
response = client.get("/api/tasks/42", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == 42
|
|
assert data["period"] == "2026-07"
|
|
|
|
def test_get_task_not_found(self, client: TestClient, auth_headers: dict, mock_db: AsyncMock):
|
|
"""测试获取不存在的任务"""
|
|
mock_db.get = AsyncMock(return_value=None)
|
|
|
|
response = client.get("/api/tasks/99999", headers=auth_headers)
|
|
assert response.status_code == 404
|
|
|
|
def test_list_tasks_no_company_header(self, client: TestClient, auth_headers: dict, mock_db: AsyncMock):
|
|
"""测试无企业 ID 头的任务列表（company_id=None 时查不到数据）"""
|
|
mock_db.execute = AsyncMock(return_value=make_db_result(items=[], scalar=0))
|
|
|
|
response = client.get("/api/tasks/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total"] == 0
|