04a1a0d307
- 个人绩效明细/部门绩效汇总/业务统计/回款统计四页，路由组「报表统计」order19-22（allowedRoles admin/finance/leader） - 明细分页页脚全筛选合计；部门汇总成员内联与跨页下钻；业务统计展开回款/台账；回款统计 8 档区间（档位由后端下发）与桶下钻 - 打印视图独立顶层路由（不挂 BasicLayout）：公司页眉走 public 端点、循环翻页取全量（2000 行截断提示）、@media print A4 横版每页表头、签字栏 - AppPagination 只派发 page-change/size-change，四页按事件绑定并补初始加载（@change/v-model 不生效的同类缺陷） - e2e report_flow.py：四页筛选与导出下载、打印视图 window.print 桩断言 Co-Authored-By: Claude Code <noreply@anthropic.com>
309 lines
11 KiB
Python
309 lines
11 KiB
Python
"""报表统计集成测试（M4 ⑤）。
|
||
|
||
守的口径（破坏了就是报表数字与台账对不上）：
|
||
- 个人明细与台账逐笔一致；summary 为**全筛选**合计而非当前页合计
|
||
- 部门汇总团队基数 = 成员绩效合计；12.5%/5% 分成按参数；无成员经理出 0 行
|
||
- 回款统计 8 个区间桶恒在（空桶为 0）；**二次分摊拆行时回款金额只计一次**
|
||
- 业务统计排除经理聚合行（calc_type=2），绩效不被整团队聚合额污染
|
||
- 导出 xlsx 可解析且含签字栏；导出行为入操作日志
|
||
- 权限：普通用户 403 且文案为「无权访问报表统计」
|
||
"""
|
||
|
||
import uuid
|
||
from decimal import Decimal
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
_REPORTS = "/api/v1/reports"
|
||
_PERSONNEL = "/api/v1/base/personnel"
|
||
_BUSINESSES = "/api/v1/business/businesses"
|
||
_SHIPMENTS = "/api/v1/business/shipments"
|
||
_RECEIPTS = "/api/v1/business/receipts"
|
||
_CALC = "/api/v1/calc"
|
||
_SETTLEMENTS = "/api/v1/settlements"
|
||
|
||
|
||
def _role_id(code: str) -> int:
|
||
"""取内置业务岗位 id（salesperson / sales_manager）。"""
|
||
from sqlalchemy import select
|
||
|
||
from app.core.db import get_session_factory
|
||
from app.models import BusinessRole
|
||
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
return int(
|
||
session.execute(
|
||
select(BusinessRole.id).where(
|
||
BusinessRole.role_code == code, BusinessRole.status == 1
|
||
)
|
||
).scalar_one()
|
||
)
|
||
|
||
|
||
def _person(client: TestClient, headers: dict, *, role: str = "salesperson", manager_id=None) -> dict:
|
||
body = {
|
||
"personnel_name": f"t_报表_{uuid.uuid4().hex[:8]}",
|
||
"business_role_id": _role_id(role),
|
||
"salary_standard": "3000.00",
|
||
}
|
||
if manager_id is not None:
|
||
body["manager_id"] = manager_id
|
||
resp = client.post(_PERSONNEL, json=body, headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _full_chain(
|
||
client: TestClient,
|
||
headers: dict,
|
||
*,
|
||
owner_id: int,
|
||
customer_id: int | None = None,
|
||
allocations: list[dict] | None = None,
|
||
) -> tuple[dict, dict]:
|
||
"""业务（单行 100×100，成本 3000）→ 发货（60 天前）→ 回款 → 核算。"""
|
||
from tests.integration.factories import business_item, business_payload, customer
|
||
|
||
cust = customer(client, headers) if customer_id is None else {"id": customer_id}
|
||
payload = business_payload(
|
||
cust["id"],
|
||
owner_personnel_id=owner_id,
|
||
items=[business_item(cost_amount="3000.00")],
|
||
)
|
||
if allocations:
|
||
payload["secondary_allocations"] = allocations
|
||
resp = client.post(_BUSINESSES, json=payload, headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
business = resp.json()["data"]
|
||
|
||
from datetime import date, timedelta
|
||
|
||
ship_date = (date.today() - timedelta(days=60)).isoformat()
|
||
resp = client.post(
|
||
_SHIPMENTS,
|
||
json={
|
||
"business_id": business["id"],
|
||
"ship_date": ship_date,
|
||
"items": [{"product_name": "t_发货物", "quantity": "1", "sale_price": "1.00"}],
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
shipment = resp.json()["data"]
|
||
|
||
resp = client.post(
|
||
_RECEIPTS,
|
||
json={
|
||
"business_id": business["id"],
|
||
"receipt_date": date.today().isoformat(),
|
||
"amount": "10000.00",
|
||
"shipment_id": shipment["id"],
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
receipt = resp.json()["data"]
|
||
|
||
resp = client.post(f"{_CALC}/receipts/{receipt['id']}/calc", headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
return business, receipt
|
||
|
||
|
||
def _month() -> str:
|
||
from datetime import date
|
||
|
||
return date.today().strftime("%Y-%m")
|
||
|
||
|
||
def _ledger_sum(headers: dict, month: str) -> Decimal:
|
||
"""台账（单笔生效行）同月绩效合计——报表 summary 的对照基准。"""
|
||
from sqlalchemy import func, select
|
||
|
||
from app.core.db import get_session_factory
|
||
from app.models import PerformanceLedger
|
||
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
return Decimal(
|
||
str(
|
||
session.execute(
|
||
select(func.coalesce(func.sum(PerformanceLedger.performance_amount), 0)).where(
|
||
PerformanceLedger.settlement_month == month,
|
||
PerformanceLedger.calc_type == 1,
|
||
PerformanceLedger.is_active == 1,
|
||
)
|
||
).scalar_one()
|
||
)
|
||
)
|
||
|
||
|
||
def test_个人明细_与台账一致且summary为全量合计(client: TestClient, admin: dict):
|
||
"""明细行金额必须等于台账；pageSize=1 时 summary 仍是全部筛选结果之和。"""
|
||
person = _person(client, admin["headers"])
|
||
_full_chain(client, admin["headers"], owner_id=person["id"])
|
||
month = _month()
|
||
|
||
resp = client.get(
|
||
f"{_REPORTS}/personal-detail",
|
||
params={"settlement_month": month, "pageSize": 1},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
data = resp.json()["data"]
|
||
assert data["total"] == 1
|
||
expected = _ledger_sum(admin["headers"], month)
|
||
assert Decimal(str(data["list"][0]["performance_amount"])) == expected
|
||
assert Decimal(str(data["summary"]["performance_total"])) == expected
|
||
|
||
|
||
def test_部门汇总_团队基数等于成员绩效且分成按参数(admin, client: TestClient):
|
||
"""团队基数=成员绩效合计；12.5% 团队分成与 5% 全公司分成分别可核。"""
|
||
manager = _person(client, admin["headers"], role="sales_manager")
|
||
salesperson = _person(client, admin["headers"], manager_id=manager["id"])
|
||
_full_chain(client, admin["headers"], owner_id=salesperson["id"])
|
||
month = _month()
|
||
|
||
resp = client.get(
|
||
f"{_REPORTS}/department-summary",
|
||
params={"settlement_month": month},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
data = resp.json()["data"]
|
||
|
||
team = next(t for t in data["teams"] if t["manager_id"] == manager["id"])
|
||
ledger_total = _ledger_sum(admin["headers"], month)
|
||
assert Decimal(str(team["team_base"])) == ledger_total
|
||
expected_share = (ledger_total * Decimal("0.125")).quantize(Decimal("0.01"))
|
||
assert Decimal(str(team["team_share_amount"])) == expected_share
|
||
assert team["member_count"] == 1
|
||
assert data["company"]["company_base"] == team["team_base"]
|
||
|
||
|
||
def test_部门汇总_无成员经理出零值行(admin, client: TestClient):
|
||
"""无团队成员 → 0 行展示（FR-030 异常处理）。"""
|
||
manager = _person(client, admin["headers"], role="sales_manager")
|
||
month = _month()
|
||
resp = client.get(
|
||
f"{_REPORTS}/department-summary",
|
||
params={"settlement_month": month},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
team = next(
|
||
(t for t in resp.json()["data"]["teams"] if t["manager_id"] == manager["id"]), None
|
||
)
|
||
assert team is not None
|
||
assert Decimal(str(team["team_base"])) == Decimal("0")
|
||
assert team["member_count"] == 0
|
||
|
||
|
||
def test_回款统计_空桶恒在且分摊拆行金额只计一次(admin, client: TestClient):
|
||
"""8 个区间桶恒在；二次分配拆行使台账多行但回款金额不得翻倍。"""
|
||
owner = _person(client, admin["headers"])
|
||
participant = _person(client, admin["headers"])
|
||
_full_chain(
|
||
client,
|
||
admin["headers"],
|
||
owner_id=owner["id"],
|
||
allocations=[{"personnel_id": participant["id"], "ratio": "30.00"}],
|
||
)
|
||
month = _month()
|
||
|
||
resp = client.get(
|
||
f"{_REPORTS}/payment-stats",
|
||
params={"settlement_month": month},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
data = resp.json()["data"]
|
||
|
||
assert len(data["buckets"]) == 8
|
||
assert [b["code"] for b in data["buckets"]] == [
|
||
"le0", "1-30", "31-60", "61-90", "91-180", "181-360", "361+", "none",
|
||
]
|
||
# 60 天前发货 → 落在 61-90 桶（发货日首日计第 1 天）
|
||
bucket = next(b for b in data["buckets"] if b["code"] == "61-90")
|
||
assert Decimal(str(bucket["receipt_amount"])) == Decimal("10000.00")
|
||
assert bucket["receipt_count"] == 1
|
||
# 拆行后台账两行，但绩效合计仍等于台账总和
|
||
assert bucket["ledger_count"] == 2
|
||
empty = next(b for b in data["buckets"] if b["code"] == "361+")
|
||
assert Decimal(str(empty["performance_amount"])) == Decimal("0")
|
||
|
||
|
||
def test_业务统计_绩效排除经理聚合行(admin, client: TestClient):
|
||
"""生成月结后出现 calc_type=2 聚合行，业务统计不得把聚合额算进单业务。"""
|
||
manager = _person(client, admin["headers"], role="sales_manager")
|
||
salesperson = _person(client, admin["headers"], manager_id=manager["id"])
|
||
business, _ = _full_chain(client, admin["headers"], owner_id=salesperson["id"])
|
||
|
||
resp = client.post(
|
||
f"{_SETTLEMENTS}/generate", json={"month": _month()}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
resp = client.get(
|
||
f"{_REPORTS}/business-stats",
|
||
params={"keyword": business["business_no"]},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
row = resp.json()["data"]["list"][0]
|
||
assert Decimal(str(row["performance_amount"])) == _ledger_sum(admin["headers"], _month())
|
||
|
||
|
||
def test_导出_excel含签字栏且入操作日志(admin, client: TestClient):
|
||
"""导出是文件流（非统一信封）；模板含复核/签字栏；行为入操作日志。"""
|
||
from io import BytesIO
|
||
|
||
from openpyxl import load_workbook
|
||
|
||
person = _person(client, admin["headers"])
|
||
_full_chain(client, admin["headers"], owner_id=person["id"])
|
||
|
||
resp = client.get(
|
||
f"{_REPORTS}/personal-detail/export",
|
||
params={"settlement_month": _month()},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
assert resp.headers["content-type"].startswith(
|
||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||
)
|
||
workbook = load_workbook(BytesIO(resp.content))
|
||
text = "\n".join(
|
||
str(cell.value)
|
||
for row in workbook.active.iter_rows()
|
||
for cell in row
|
||
if cell.value is not None
|
||
)
|
||
assert "个人绩效工资明细表" in text
|
||
assert "财务复核" in text
|
||
assert "领导签字" in text
|
||
|
||
from sqlalchemy import select
|
||
|
||
from app.core.db import get_session_factory
|
||
from app.models import OperationLog
|
||
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
log = session.execute(
|
||
select(OperationLog)
|
||
.where(OperationLog.module == "报表统计", OperationLog.action == "导出")
|
||
.order_by(OperationLog.id.desc())
|
||
).scalars().first()
|
||
assert log is not None
|
||
|
||
|
||
def test_权限_普通用户403文案(admin, normal, client: TestClient):
|
||
"""报表限管理员/财务/领导；普通用户被后端拦截且文案正确。"""
|
||
resp = client.get(f"{_REPORTS}/personal-detail", headers=normal["headers"])
|
||
assert resp.status_code == 403
|
||
assert resp.json()["message"] == "无权访问报表统计"
|