398 lines
12 KiB
Python
398 lines
12 KiB
Python
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_statistics_all_users_success(
|
|
client: AsyncClient, admin_token: str, db_session, admin_user
|
|
):
|
|
"""测试所有用户获取基础统计成功"""
|
|
from src.models.project import Project
|
|
from datetime import date
|
|
|
|
# 创建测试项目
|
|
project1 = Project(
|
|
project_no="PRJ001",
|
|
name="项目1",
|
|
engineering_type="基建",
|
|
contract_amount=100.0,
|
|
actual_receipt_amount=50.0,
|
|
actual_payment_amount=40.0,
|
|
cumulative_progress=50.0,
|
|
signing_date=date(2026, 1, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
project2 = Project(
|
|
project_no="PRJ002",
|
|
name="项目2",
|
|
engineering_type="业扩",
|
|
contract_amount=200.0,
|
|
actual_receipt_amount=100.0,
|
|
actual_payment_amount=80.0,
|
|
cumulative_progress=50.0,
|
|
signing_date=date(2026, 1, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
db_session.add(project1)
|
|
db_session.add(project2)
|
|
await db_session.commit()
|
|
|
|
response = await client.get(
|
|
"/api/v1/projects/statistics",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert data["data"]["total_count"] == 2
|
|
assert data["data"]["total_contract_amount"] == 300.0
|
|
assert data["data"]["total_receipt_amount"] == 150.0
|
|
assert data["data"]["total_payment_amount"] == 120.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_statistics_with_filters(
|
|
client: AsyncClient, admin_token: str, db_session, admin_user
|
|
):
|
|
"""测试带筛选条件的统计"""
|
|
from src.models.project import Project
|
|
from datetime import date
|
|
|
|
# 创建不同类型的项目
|
|
project1 = Project(
|
|
project_no="PRJ001",
|
|
name="基建项目",
|
|
engineering_type="基建",
|
|
contract_amount=100.0,
|
|
signing_date=date(2026, 1, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
project2 = Project(
|
|
project_no="PRJ002",
|
|
name="业扩项目",
|
|
engineering_type="业扩",
|
|
contract_amount=200.0,
|
|
signing_date=date(2026, 2, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
db_session.add(project1)
|
|
db_session.add(project2)
|
|
await db_session.commit()
|
|
|
|
# 筛选：工程类别=基建
|
|
response = await client.get(
|
|
"/api/v1/projects/statistics?engineering_type=基建",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert data["data"]["total_count"] == 1
|
|
assert data["data"]["total_contract_amount"] == 100.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_statistics_empty_data(
|
|
client: AsyncClient, admin_token: str
|
|
):
|
|
"""测试无数据时的统计"""
|
|
response = await client.get(
|
|
"/api/v1/projects/statistics",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert data["data"]["total_count"] == 0
|
|
assert data["data"]["total_contract_amount"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_group_statistics_by_engineering_type(
|
|
client: AsyncClient, admin_token: str, db_session, admin_user
|
|
):
|
|
"""测试按工程类别分组统计"""
|
|
from src.models.project import Project
|
|
from datetime import date
|
|
|
|
# 创建不同类型的项目
|
|
for i in range(3):
|
|
project = Project(
|
|
project_no=f"PRJ{i:03d}",
|
|
name=f"基建项目{i}",
|
|
engineering_type="基建",
|
|
contract_amount=100.0,
|
|
actual_receipt_amount=80.0,
|
|
actual_payment_amount=70.0,
|
|
signing_date=date(2026, 1, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
db_session.add(project)
|
|
|
|
for i in range(2):
|
|
project = Project(
|
|
project_no=f"PRJ{i+3:03d}",
|
|
name=f"业扩项目{i}",
|
|
engineering_type="业扩",
|
|
contract_amount=200.0,
|
|
actual_receipt_amount=150.0,
|
|
actual_payment_amount=130.0,
|
|
signing_date=date(2026, 1, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
db_session.add(project)
|
|
|
|
await db_session.commit()
|
|
|
|
response = await client.get(
|
|
"/api/v1/projects/statistics/group?group_by=engineering_type",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert len(data["data"]) == 2
|
|
|
|
# 验证基建工程统计
|
|
infra = next(item for item in data["data"] if item["engineering_type"] == "基建")
|
|
assert infra["count"] == 3
|
|
assert infra["total_contract_amount"] == 300.0
|
|
|
|
# 验证业扩工程统计
|
|
market = next(item for item in data["data"] if item["engineering_type"] == "业扩")
|
|
assert market["count"] == 2
|
|
assert market["total_contract_amount"] == 400.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_group_statistics_by_department(
|
|
client: AsyncClient, admin_token: str, db_session, admin_user
|
|
):
|
|
"""测试按项目部分组统计"""
|
|
from src.models.project import Project
|
|
from datetime import date
|
|
|
|
# 创建不同项目部的项目
|
|
project1 = Project(
|
|
project_no="PRJ001",
|
|
name="项目部一项目",
|
|
engineering_type="基建",
|
|
contract_amount=100.0,
|
|
project_department="项目部一",
|
|
signing_date=date(2026, 1, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
project2 = Project(
|
|
project_no="PRJ002",
|
|
name="项目部二项目",
|
|
engineering_type="业扩",
|
|
contract_amount=200.0,
|
|
project_department="项目部二",
|
|
signing_date=date(2026, 1, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
db_session.add(project1)
|
|
db_session.add(project2)
|
|
await db_session.commit()
|
|
|
|
response = await client.get(
|
|
"/api/v1/projects/statistics/group?group_by=project_department",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert len(data["data"]) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_group_statistics_with_date_filter(
|
|
client: AsyncClient, admin_token: str, db_session, admin_user
|
|
):
|
|
"""测试带日期筛选的分组统计"""
|
|
from src.models.project import Project
|
|
from datetime import date
|
|
|
|
# 创建不同日期的项目
|
|
project1 = Project(
|
|
project_no="PRJ001",
|
|
name="一月项目",
|
|
engineering_type="基建",
|
|
contract_amount=100.0,
|
|
signing_date=date(2026, 1, 15),
|
|
created_by=admin_user.id,
|
|
)
|
|
project2 = Project(
|
|
project_no="PRJ002",
|
|
name="二月项目",
|
|
engineering_type="业扩",
|
|
contract_amount=200.0,
|
|
signing_date=date(2026, 2, 15),
|
|
created_by=admin_user.id,
|
|
)
|
|
db_session.add(project1)
|
|
db_session.add(project2)
|
|
await db_session.commit()
|
|
|
|
# 只统计1月的项目
|
|
response = await client.get(
|
|
"/api/v1/projects/statistics/group?group_by=engineering_type&signing_date_start=2026-01-01&signing_date_end=2026-01-31",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert len(data["data"]) == 1
|
|
assert data["data"][0]["count"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_timeline_statistics_by_month(
|
|
client: AsyncClient, admin_token: str, db_session, admin_user
|
|
):
|
|
"""测试按月统计"""
|
|
from src.models.project import Project
|
|
from datetime import date
|
|
|
|
# 创建不同月份的项目
|
|
for month in [1, 1, 2, 2, 2]:
|
|
project = Project(
|
|
project_no=f"PRJ{month:02d}",
|
|
name=f"{month}月项目",
|
|
engineering_type="基建",
|
|
contract_amount=100.0,
|
|
signing_date=date(2026, month, 15),
|
|
created_by=admin_user.id,
|
|
)
|
|
db_session.add(project)
|
|
|
|
await db_session.commit()
|
|
|
|
response = await client.get(
|
|
"/api/v1/projects/statistics/timeline?time_field=signing_date&group_by=month",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert len(data["data"]) == 2
|
|
|
|
# 验证1月统计
|
|
jan = next(item for item in data["data"] if item["month"] == "2026-01")
|
|
assert jan["count"] == 2
|
|
|
|
# 验证2月统计
|
|
feb = next(item for item in data["data"] if item["month"] == "2026-02")
|
|
assert feb["count"] == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_timeline_statistics_by_year(
|
|
client: AsyncClient, admin_token: str, db_session, admin_user
|
|
):
|
|
"""测试按年统计"""
|
|
from src.models.project import Project
|
|
from datetime import date
|
|
|
|
# 创建不同年份的项目
|
|
for year in [2025, 2026, 2026]:
|
|
project = Project(
|
|
project_no=f"PRJ{year}",
|
|
name=f"{year}年项目",
|
|
engineering_type="基建",
|
|
contract_amount=100.0,
|
|
signing_date=date(year, 1, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
db_session.add(project)
|
|
|
|
await db_session.commit()
|
|
|
|
response = await client.get(
|
|
"/api/v1/projects/statistics/timeline?time_field=signing_date&group_by=year",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert len(data["data"]) == 2
|
|
|
|
# 验证2025年统计
|
|
y2025 = next(item for item in data["data"] if item["year"] == "2025")
|
|
assert y2025["count"] == 1
|
|
|
|
# 验证2026年统计
|
|
y2026 = next(item for item in data["data"] if item["year"] == "2026")
|
|
assert y2026["count"] == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_timeline_statistics_with_filter(
|
|
client: AsyncClient, admin_token: str, db_session, admin_user
|
|
):
|
|
"""测试带筛选条件的时间统计"""
|
|
from src.models.project import Project
|
|
from datetime import date
|
|
|
|
# 创建不同类型和日期的项目
|
|
project1 = Project(
|
|
project_no="PRJ001",
|
|
name="基建一月",
|
|
engineering_type="基建",
|
|
contract_amount=100.0,
|
|
signing_date=date(2026, 1, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
project2 = Project(
|
|
project_no="PRJ002",
|
|
name="业扩一月",
|
|
engineering_type="业扩",
|
|
contract_amount=200.0,
|
|
signing_date=date(2026, 1, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
project3 = Project(
|
|
project_no="PRJ003",
|
|
name="业扩二月",
|
|
engineering_type="业扩",
|
|
contract_amount=300.0,
|
|
signing_date=date(2026, 2, 1),
|
|
created_by=admin_user.id,
|
|
)
|
|
db_session.add(project1)
|
|
db_session.add(project2)
|
|
db_session.add(project3)
|
|
await db_session.commit()
|
|
|
|
# 只统计业扩项目
|
|
response = await client.get(
|
|
"/api/v1/projects/statistics/timeline?time_field=signing_date&group_by=month&engineering_type=业扩",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["success"] is True
|
|
assert len(data["data"]) == 2
|
|
|
|
# 验证1月业扩项目统计
|
|
jan = next(item for item in data["data"] if item["month"] == "2026-01")
|
|
assert jan["count"] == 1
|