fix(report): 报表统计修复批次——对账口径、导出审计与月份校验
- 部门汇总全公司 5% 按每位有台账聚合行的营销经理各计一次， 团队应发小计取聚合额 + 调整净额；与月结 total_payable 对拍 （有调整单时 + adjust_net_total），分列舍入差单列 - 人员归属按行类型拆分（普通/经理个人户/仅调整单），调整单全量并入， 团队与未分配不重复；按经理筛选只汇总展示团队 - 导出改单 sheet 分块签字稿，审计覆盖取数与超限；筛选页眉 ID→名称 - 月份校验抽 core/month，报表/月结/调整单非法月份统一 400； 岗位只读选项供财务/领导筛选 - 前端修复：部门下钻带入筛选、回款多桶独立状态、打印截断判定、 summary 请求序号保护 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
liuwenxinandClaude Code committed
1 parent
294d99a0d3
commit
ebb3a92f25
21 files changed
+2377
-287
No files matched your search
@@ -147,9 +147,18 @@ class TestBuiltinProtection:
|
||||
|
||||
|
||||
class TestPermission:
|
||||
"""权限：普通用户全部 403（当前默认管理员专用）。"""
|
||||
"""权限：管理接口普通用户 403；只读选项登录即可（报表筛选依赖）。"""
|
||||
|
||||
def test_普通用户访问岗位列表_403(self, client: TestClient, make_user, auth_headers) -> None:
|
||||
user = make_user([ROLE_USER])
|
||||
headers = auth_headers(user)
|
||||
assert client.get(_ROLES, headers=headers).status_code == 403
|
||||
|
||||
def test_普通用户可读取岗位选项(self, client: TestClient, make_user, auth_headers) -> None:
|
||||
"""财务/领导做报表岗位筛选时不能因管理员接口 403 而失去筛选能力。"""
|
||||
user = make_user([ROLE_USER])
|
||||
headers = auth_headers(user)
|
||||
resp = client.get("/api/v1/base/business-roles/options", headers=headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
codes = {row["role_code"] for row in resp.json()["data"]}
|
||||
assert {"salesperson", "sales_manager", "info_officer"} <= codes
|
||||
@@ -306,3 +306,498 @@ 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"] == "无权访问报表统计"
|
||||
|
||||
|
||||
# ==== 审查回归：归属、调整单、经理个人户、导出审计 ====
|
||||
|
||||
|
||||
def _set_param(client: TestClient, headers: dict, code: str, value: str) -> None:
|
||||
"""立即生效的参数变更（用例不依赖种子默认值，结束显式恢复）。"""
|
||||
resp = client.post(
|
||||
"/api/v1/system/params/changes",
|
||||
json={
|
||||
"change_type": 1,
|
||||
"remark": f"用例设置 {code}={value}",
|
||||
"items": [{"param_code": code, "dimension_value": "", "value": value}],
|
||||
},
|
||||
headers=headers,
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
|
||||
def _insert_adjustment(person: dict, month: str, amount: str, direction: int = 1) -> None:
|
||||
"""直插生效调整单（status=2）：绕开审批链，聚焦报表归属口径。"""
|
||||
from app.core.db import get_session_factory
|
||||
from app.models import AdjustmentOrder
|
||||
|
||||
factory = get_session_factory()
|
||||
with factory() as session:
|
||||
session.add(
|
||||
AdjustmentOrder(
|
||||
order_no=f"TZ{uuid.uuid4().hex[:12]}",
|
||||
settlement_month=month,
|
||||
personnel_id=person["id"],
|
||||
personnel_name=person["personnel_name"],
|
||||
direction=direction,
|
||||
amount=Decimal(amount),
|
||||
reason="t_报表用例",
|
||||
status=2,
|
||||
submitter_name="t_admin",
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
|
||||
|
||||
def _insert_manager_personal_ledger(manager: dict, month: str, amount: str) -> None:
|
||||
"""直插一条经理个人户台账行（模拟单笔侧已计提的结果）。"""
|
||||
from datetime import date
|
||||
|
||||
from app.core.db import get_session_factory
|
||||
from app.models import PerformanceLedger
|
||||
|
||||
factory = get_session_factory()
|
||||
with factory() as session:
|
||||
session.add(
|
||||
PerformanceLedger(
|
||||
receipt_id=None,
|
||||
calc_version=1,
|
||||
calc_type=1,
|
||||
business_id=0,
|
||||
business_no="t_个人户",
|
||||
personnel_id=manager["id"],
|
||||
personnel_name=manager["personnel_name"],
|
||||
role_code="sales_manager",
|
||||
customer_type="manager_personal",
|
||||
product_category="mixed",
|
||||
receipt_date=date.today(),
|
||||
rate_text="0",
|
||||
performance_amount=Decimal(amount),
|
||||
barter_applied=0,
|
||||
payable_amount=Decimal(amount),
|
||||
settlement_month=month,
|
||||
is_probation=0,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
|
||||
|
||||
def _latest_export_log():
|
||||
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:
|
||||
return session.execute(
|
||||
select(OperationLog)
|
||||
.where(OperationLog.module == "报表统计", OperationLog.action == "导出")
|
||||
.order_by(OperationLog.id.desc())
|
||||
).scalars().first()
|
||||
|
||||
|
||||
def test_部门汇总_无台账经理挂上级不报500(admin, client: TestClient):
|
||||
"""经理 A 无台账且上级是经理 B：不能因 member_map 直接索引而 KeyError。"""
|
||||
manager_b = _person(client, admin["headers"], role="sales_manager")
|
||||
manager_a = _person(
|
||||
client, admin["headers"], role="sales_manager", manager_id=manager_b["id"]
|
||||
)
|
||||
|
||||
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_a = next(t for t in data["teams"] if t["manager_id"] == manager_a["id"])
|
||||
assert team_a["member_count"] == 0
|
||||
team_b = next(t for t in data["teams"] if t["manager_id"] == manager_b["id"])
|
||||
assert all(m["personnel_id"] != manager_a["id"] for m in team_b["members"])
|
||||
|
||||
|
||||
def test_部门汇总_仅有调整单人员并入对应团队(admin, client: TestClient):
|
||||
"""只有调整单、当月无台账行的人也要出现在团队明细中。"""
|
||||
manager = _person(client, admin["headers"], role="sales_manager")
|
||||
salesperson = _person(client, admin["headers"], manager_id=manager["id"])
|
||||
_insert_adjustment(salesperson, _month(), "500.00")
|
||||
|
||||
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"])
|
||||
member = next(m for m in team["members"] if m["personnel_id"] == salesperson["id"])
|
||||
assert member["member_kind"] == "adjustment"
|
||||
assert Decimal(str(member["performance_sum"])) == Decimal("0")
|
||||
assert Decimal(str(member["adjust_net"])) == Decimal("500.00")
|
||||
assert Decimal(str(team["adjust_net"])) == Decimal("500.00")
|
||||
assert Decimal(str(data["reconciliation"]["adjust_net_total"])) == Decimal("500.00")
|
||||
|
||||
|
||||
def test_部门汇总_仅调整单无经理人员进入未分配(admin, client: TestClient):
|
||||
"""无经理人员的调整单不能被静默丢弃。"""
|
||||
salesperson = _person(client, admin["headers"])
|
||||
_insert_adjustment(salesperson, _month(), "300.00")
|
||||
|
||||
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"]
|
||||
row = next(u for u in data["unassigned"] if u["personnel_id"] == salesperson["id"])
|
||||
assert row["member_kind"] == "adjustment"
|
||||
assert Decimal(str(row["adjust_net"])) == Decimal("300.00")
|
||||
|
||||
|
||||
def test_部门汇总_经理个人户计入时基数等于成员合计(admin, client: TestClient):
|
||||
"""开关 1：经理成员行只取个人户金额，团队基数 == 成员绩效合计。"""
|
||||
manager = _person(client, admin["headers"], role="sales_manager")
|
||||
_insert_manager_personal_ledger(manager, _month(), "100.00")
|
||||
_set_param(client, admin["headers"], "manager.personal.included", "1")
|
||||
try:
|
||||
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"]
|
||||
)
|
||||
assert Decimal(str(team["team_base"])) == Decimal("100.00")
|
||||
assert sum(
|
||||
(Decimal(str(m["performance_sum"])) for m in team["members"]), Decimal("0")
|
||||
) == Decimal("100.00")
|
||||
assert all(m["member_kind"] == "manager_personal" for m in team["members"])
|
||||
finally:
|
||||
_set_param(client, admin["headers"], "manager.personal.included", "0")
|
||||
|
||||
|
||||
def test_部门汇总_经理个人户关闭时不展示个人户金额(admin, client: TestClient):
|
||||
"""开关 0：个人户金额既不计入团队基数，也不进未分配（与全公司口径一致）。"""
|
||||
manager = _person(client, admin["headers"], role="sales_manager")
|
||||
_insert_manager_personal_ledger(manager, _month(), "100.00")
|
||||
_set_param(client, admin["headers"], "manager.personal.included", "0")
|
||||
|
||||
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"])
|
||||
assert team["member_count"] == 0
|
||||
assert all(u["personnel_id"] != manager["id"] for u in data["unassigned"])
|
||||
|
||||
|
||||
def test_部门汇总_按经理筛选不误判其他团队下属(admin, client: TestClient):
|
||||
"""筛选只影响展示的团队；其他经理的下属不能进入未分配。"""
|
||||
manager_1 = _person(client, admin["headers"], role="sales_manager")
|
||||
manager_2 = _person(client, admin["headers"], role="sales_manager")
|
||||
salesperson_1 = _person(client, admin["headers"], manager_id=manager_1["id"])
|
||||
salesperson_2 = _person(client, admin["headers"], manager_id=manager_2["id"])
|
||||
_full_chain(client, admin["headers"], owner_id=salesperson_1["id"])
|
||||
_full_chain(client, admin["headers"], owner_id=salesperson_2["id"])
|
||||
|
||||
resp = client.get(
|
||||
f"{_REPORTS}/department-summary",
|
||||
params={"settlement_month": _month(), "manager_id": manager_1["id"]},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()["data"]
|
||||
assert [t["manager_id"] for t in data["teams"]] == [manager_1["id"]]
|
||||
assert data["unassigned"] == []
|
||||
assert all(
|
||||
m["personnel_id"] != salesperson_2["id"]
|
||||
for team in data["teams"]
|
||||
for m in team["members"]
|
||||
)
|
||||
assert Decimal(str(data["reconciliation"]["other_team_base"])) > 0
|
||||
|
||||
|
||||
def test_部门汇总_对账恒等式(admin, client: TestClient):
|
||||
"""未筛选：团队基数 + 未分配绩效 = 全公司基数；应发公式成立。"""
|
||||
manager = _person(client, admin["headers"], role="sales_manager")
|
||||
salesperson = _person(client, admin["headers"], manager_id=manager["id"])
|
||||
unassigned = _person(client, admin["headers"])
|
||||
_full_chain(client, admin["headers"], owner_id=salesperson["id"])
|
||||
_full_chain(client, admin["headers"], owner_id=unassigned["id"])
|
||||
_insert_adjustment(salesperson, _month(), "200.00")
|
||||
|
||||
resp = client.get(
|
||||
f"{_REPORTS}/department-summary",
|
||||
params={"settlement_month": _month()},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
reconciliation = resp.json()["data"]["reconciliation"]
|
||||
assert Decimal(str(reconciliation["team_base_total"])) + Decimal(
|
||||
str(reconciliation["unassigned_performance_total"])
|
||||
) == Decimal(str(reconciliation["company_base"]))
|
||||
assert Decimal(str(reconciliation["other_team_base"])) == Decimal("0")
|
||||
assert Decimal(str(reconciliation["payable_total"])) == (
|
||||
Decimal(str(reconciliation["team_member_payable_total"]))
|
||||
+ Decimal(str(reconciliation["unassigned_payable_total"]))
|
||||
+ Decimal(str(reconciliation["aggregate_payable_total"]))
|
||||
+ Decimal(str(reconciliation["adjust_net_total"]))
|
||||
)
|
||||
# 分列合计 + 舍入差 = 聚合额（月结一次舍入口径）
|
||||
assert Decimal(str(reconciliation["aggregate_rounding_diff"])) == (
|
||||
Decimal(str(reconciliation["aggregate_payable_total"]))
|
||||
- Decimal(str(reconciliation["team_share_total"]))
|
||||
- Decimal(str(reconciliation["company_share_total"]))
|
||||
)
|
||||
|
||||
|
||||
def _two_manager_scene(
|
||||
client: TestClient, headers: dict
|
||||
) -> tuple[dict, dict, dict, dict, dict]:
|
||||
"""两位经理各有下属与业务 + 一位零成员经理（无聚合行）。"""
|
||||
manager_1 = _person(client, headers, role="sales_manager")
|
||||
manager_2 = _person(client, headers, role="sales_manager")
|
||||
empty_manager = _person(client, headers, role="sales_manager")
|
||||
salesperson_1 = _person(client, headers, manager_id=manager_1["id"])
|
||||
salesperson_2 = _person(client, headers, manager_id=manager_2["id"])
|
||||
_full_chain(client, headers, owner_id=salesperson_1["id"])
|
||||
_full_chain(client, headers, owner_id=salesperson_2["id"])
|
||||
return manager_1, manager_2, empty_manager, salesperson_1, salesperson_2
|
||||
|
||||
|
||||
def test_部门汇总_应发合计与月结对拍_两位经理(admin, client: TestClient):
|
||||
"""月结每位有聚合行的经理各计一次 5%；签字稿应发必须等于 total_payable。"""
|
||||
_two_manager_scene(client, admin["headers"])
|
||||
|
||||
resp = client.get(
|
||||
f"{_REPORTS}/department-summary",
|
||||
params={"settlement_month": _month()},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
before = resp.json()["data"]
|
||||
assert before["company"]["company_share_count"] == 2
|
||||
assert Decimal(str(before["company"]["company_share_total"])) == (
|
||||
Decimal(str(before["company"]["company_share_amount"])) * 2
|
||||
)
|
||||
|
||||
resp = client.post(
|
||||
f"{_SETTLEMENTS}/generate", json={"month": _month()}, headers=admin["headers"]
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
settlement = resp.json()["data"]
|
||||
|
||||
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"]
|
||||
teams = {team["manager_id"]: team for team in data["teams"]}
|
||||
frozen_total = sum(
|
||||
(
|
||||
Decimal(str(team["frozen_amount"]))
|
||||
for team in teams.values()
|
||||
if team["frozen_amount"] is not None
|
||||
),
|
||||
Decimal("0"),
|
||||
)
|
||||
reconciliation = data["reconciliation"]
|
||||
assert Decimal(str(reconciliation["aggregate_payable_total"])) == frozen_total
|
||||
assert Decimal(str(reconciliation["payable_total"])) == Decimal(
|
||||
str(settlement["total_payable"])
|
||||
)
|
||||
|
||||
|
||||
def test_部门汇总_有调整单时应发等于月结加调整(admin, client: TestClient):
|
||||
"""调整单不进月结 items；报表应发 = 月结总额 + 调整净额。"""
|
||||
_, _, _, salesperson, _ = _two_manager_scene(client, admin["headers"])
|
||||
_insert_adjustment(salesperson, _month(), "500.00")
|
||||
|
||||
resp = client.post(
|
||||
f"{_SETTLEMENTS}/generate", json={"month": _month()}, headers=admin["headers"]
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
settlement = resp.json()["data"]
|
||||
|
||||
resp = client.get(
|
||||
f"{_REPORTS}/department-summary",
|
||||
params={"settlement_month": _month()},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
reconciliation = resp.json()["data"]["reconciliation"]
|
||||
assert Decimal(str(reconciliation["adjust_net_total"])) == Decimal("500.00")
|
||||
assert Decimal(str(reconciliation["payable_total"])) == (
|
||||
Decimal(str(settlement["total_payable"])) + Decimal("500.00")
|
||||
)
|
||||
|
||||
|
||||
def test_部门汇总_按经理筛选时5百分比只计该团队(admin, client: TestClient):
|
||||
"""筛选态只汇总展示团队；隐藏团队的金额进 other_*，不参与应发。"""
|
||||
manager_1, _, _, _, _ = _two_manager_scene(client, admin["headers"])
|
||||
|
||||
resp = client.get(
|
||||
f"{_REPORTS}/department-summary",
|
||||
params={"settlement_month": _month(), "manager_id": manager_1["id"]},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()["data"]
|
||||
assert data["company"]["company_share_count"] == 1
|
||||
reconciliation = data["reconciliation"]
|
||||
assert Decimal(str(reconciliation["other_team_aggregate_total"])) > 0
|
||||
assert Decimal(str(reconciliation["other_team_company_share_total"])) > 0
|
||||
|
||||
|
||||
def test_导出_筛选页眉ID转名称与中文标签(admin, client: TestClient):
|
||||
"""页眉是签字稿追溯筛选范围的唯一凭据，ID 必须转名称。"""
|
||||
from io import BytesIO
|
||||
|
||||
from openpyxl import load_workbook
|
||||
|
||||
from app.core.db import get_session_factory
|
||||
from app.models import BusinessRole
|
||||
|
||||
person = _person(client, admin["headers"])
|
||||
_full_chain(client, admin["headers"], owner_id=person["id"])
|
||||
factory = get_session_factory()
|
||||
with factory() as session:
|
||||
role_name = (
|
||||
session.query(BusinessRole.role_name)
|
||||
.filter(BusinessRole.role_code == "salesperson")
|
||||
.scalar()
|
||||
)
|
||||
|
||||
resp = client.get(
|
||||
f"{_REPORTS}/personal-detail/export",
|
||||
params={
|
||||
"settlement_month": _month(),
|
||||
"personnel_id": person["id"],
|
||||
"role_code": "salesperson",
|
||||
"customer_type": "personal",
|
||||
"product_category": "municipal_water",
|
||||
},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
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 person["personnel_name"] in text
|
||||
assert f"人员 {person['personnel_name']}(#{person['id']})" in text
|
||||
assert "个人户" in text
|
||||
assert "市政水利" in text
|
||||
assert role_name in text
|
||||
|
||||
|
||||
def test_导出_回款维度sheet透传筛选条件(admin, client: TestClient):
|
||||
"""两个维度 sheet 的页眉不能固定写「全部」。"""
|
||||
from io import BytesIO
|
||||
|
||||
from openpyxl import load_workbook
|
||||
|
||||
resp = client.get(
|
||||
f"{_REPORTS}/payment-stats/export",
|
||||
params={"customer_type": "personal"},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
workbook = load_workbook(BytesIO(resp.content))
|
||||
for sheet_name in ("客户类型对比", "产品类别对比"):
|
||||
text = "\n".join(
|
||||
str(cell.value)
|
||||
for row in workbook[sheet_name].iter_rows()
|
||||
for cell in row
|
||||
if cell.value is not None
|
||||
)
|
||||
assert "客户类型 个人户" in text
|
||||
assert "筛选条件：全部" not in text
|
||||
|
||||
|
||||
def test_导出超限_失败审计留痕(admin, client: TestClient, monkeypatch):
|
||||
"""超限 400 发生在审计上下文内，必须留下失败记录。"""
|
||||
from app.services import report_service
|
||||
|
||||
person = _person(client, admin["headers"])
|
||||
_full_chain(client, admin["headers"], owner_id=person["id"])
|
||||
monkeypatch.setattr(report_service, "REPORT_EXPORT_MAX_ROWS", 0)
|
||||
|
||||
resp = client.get(
|
||||
f"{_REPORTS}/personal-detail/export",
|
||||
params={"settlement_month": _month()},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 400, resp.text
|
||||
log = _latest_export_log()
|
||||
assert log is not None
|
||||
assert log.success == 0
|
||||
assert "超过上限" in (log.error_message or "")
|
||||
|
||||
|
||||
def test_导出取数异常_失败审计留痕(admin, client: TestClient, monkeypatch):
|
||||
"""取数阶段的异常同样要留下失败记录。
|
||||
|
||||
TestClient 默认重抛服务端异常，这里直接调 service 层，聚焦审计落库。
|
||||
"""
|
||||
from app.core.db import get_session_factory
|
||||
from app.models import User
|
||||
from app.schemas.report import PersonalDetailQuery
|
||||
from app.services import report_service
|
||||
|
||||
def _boom(*args, **kwargs):
|
||||
raise RuntimeError("t_取数异常")
|
||||
|
||||
monkeypatch.setattr(report_service, "_export_payload", _boom)
|
||||
factory = get_session_factory()
|
||||
with factory() as session:
|
||||
operator = session.get(User, admin["id"])
|
||||
with pytest.raises(RuntimeError, match="t_取数异常"):
|
||||
report_service.export_report(
|
||||
session,
|
||||
"personal-detail",
|
||||
PersonalDetailQuery(settlement_month=_month()),
|
||||
operator=operator,
|
||||
)
|
||||
log = _latest_export_log()
|
||||
assert log is not None
|
||||
assert log.success == 0
|
||||
# 未预期异常按安全口径收敛为通用文案，摘要仍能定位到报表
|
||||
assert "导出个人绩效工资明细表" in (log.summary or "")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path", ["/personal-detail", "/department-summary", "/payment-stats"]
|
||||
)
|
||||
def test_非法月份_查询返回400(admin, client: TestClient, path: str):
|
||||
"""非法月份必须是参数错误 400，不能落 500。"""
|
||||
resp = client.get(
|
||||
f"{_REPORTS}{path}",
|
||||
params={"settlement_month": "2026-13"},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 400, resp.text
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"code", ["personal-detail", "department-summary", "payment-stats"]
|
||||
)
|
||||
def test_非法月份_导出返回400(admin, client: TestClient, code: str):
|
||||
"""导出手工构造查询模型同样要转 400。"""
|
||||
resp = client.get(
|
||||
f"{_REPORTS}/{code}/export",
|
||||
params={"settlement_month": "2026-13"},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 400, resp.text
|
||||
@@ -749,3 +749,24 @@ class TestLockInterception:
|
||||
resp = client.post(f"{_CALC}/receipts/{unsettled['id']}/calc", headers=admin["headers"])
|
||||
assert resp.status_code == 400, resp.text
|
||||
assert "锁定" in resp.json()["message"]
|
||||
|
||||
|
||||
def test_非法月份_月结生成与调整单均返回400(client: TestClient, admin: dict) -> None:
|
||||
"""月份范围必须在 schema 拦住：`2026-13` 不能传到 monthrange 变成 500。"""
|
||||
resp = client.post(
|
||||
f"{_SETTLEMENTS}/generate", json={"month": "2026-13"}, headers=admin["headers"]
|
||||
)
|
||||
assert resp.status_code == 400, resp.text
|
||||
|
||||
resp = client.post(
|
||||
f"{_SETTLEMENTS}/adjustments",
|
||||
json={
|
||||
"settlement_month": "2026-13",
|
||||
"personnel_id": 1,
|
||||
"direction": 1,
|
||||
"amount": "100.00",
|
||||
"reason": "t_非法月份",
|
||||
},
|
||||
headers=admin["headers"],
|
||||
)
|
||||
assert resp.status_code == 400, resp.text
|
||||
@@ -209,3 +209,214 @@ class TestExcel模板:
|
||||
]
|
||||
text = _sheet_text(_build("business-stats", rows))
|
||||
assert "已完成" in text
|
||||
|
||||
|
||||
def _member(
|
||||
personnel_id: int,
|
||||
name: str,
|
||||
*,
|
||||
kind: str = "regular",
|
||||
performance: str = "0.00",
|
||||
payable: str = "0.00",
|
||||
adjust: str = "0.00",
|
||||
reason: str | None = None,
|
||||
) -> dict:
|
||||
return {
|
||||
"personnel_id": personnel_id,
|
||||
"personnel_name": name,
|
||||
"role_code": "salesperson",
|
||||
"is_probation": 0,
|
||||
"member_kind": kind,
|
||||
"unassigned_reason": reason,
|
||||
"performance_sum": Decimal(performance),
|
||||
"payable_sum": Decimal(payable),
|
||||
"adjust_extra": Decimal(adjust),
|
||||
"adjust_deduct": Decimal("0"),
|
||||
"adjust_net": Decimal(adjust),
|
||||
"net_total": Decimal(payable) + Decimal(adjust),
|
||||
"ledger_ids": [],
|
||||
}
|
||||
|
||||
|
||||
def _department_data(
|
||||
team_text: str = "12.5",
|
||||
company_text: str = "5",
|
||||
*,
|
||||
second_has_aggregate: bool = False,
|
||||
) -> dict:
|
||||
"""一份含成员、零成员团队与未分配人员的部门汇总出参。
|
||||
|
||||
``second_has_aggregate`` 打开后两位经理都有月结聚合行，用于验证
|
||||
全公司 5% 按人数累计（零成员团队默认不发 5%）。
|
||||
"""
|
||||
second_aggregate = Decimal("600.00") if second_has_aggregate else Decimal("0")
|
||||
share_count = 2 if second_has_aggregate else 1
|
||||
aggregate_total = Decimal("1850.00") + second_aggregate
|
||||
payable_total = Decimal("10000.00") + Decimal("2000.00") + aggregate_total
|
||||
return {
|
||||
"settlement_month": "2026-10",
|
||||
"params": {
|
||||
"team_text": team_text,
|
||||
"company_text": company_text,
|
||||
"manager_personal_included": 0,
|
||||
"params_as_of": "2026-10-31",
|
||||
},
|
||||
"teams": [
|
||||
{
|
||||
"manager_id": 1,
|
||||
"manager_name": "张经理",
|
||||
"manager_status": 1,
|
||||
"member_count": 1,
|
||||
"team_base": Decimal("10000.00"),
|
||||
"team_share_amount": Decimal("1250.00"),
|
||||
"company_share_amount": Decimal("600.00"),
|
||||
"has_aggregate": True,
|
||||
"aggregate_amount": Decimal("1850.00"),
|
||||
"team_subtotal": Decimal("1850.00"),
|
||||
"company_base_ratio": Decimal("0.8333"),
|
||||
"adjust_extra": Decimal("0"),
|
||||
"adjust_deduct": Decimal("0"),
|
||||
"adjust_net": Decimal("0"),
|
||||
"net_total": Decimal("1850.00"),
|
||||
"frozen_amount": None,
|
||||
"aggregate_ledger_id": None,
|
||||
"members": [
|
||||
_member(
|
||||
2,
|
||||
"李业务",
|
||||
performance="10000.00",
|
||||
payable="10000.00",
|
||||
)
|
||||
],
|
||||
},
|
||||
{
|
||||
"manager_id": 3,
|
||||
"manager_name": "王经理",
|
||||
"manager_status": 1,
|
||||
"member_count": 0,
|
||||
"team_base": Decimal("0"),
|
||||
"team_share_amount": Decimal("0"),
|
||||
"company_share_amount": second_aggregate,
|
||||
"has_aggregate": second_has_aggregate,
|
||||
"aggregate_amount": second_aggregate,
|
||||
"team_subtotal": second_aggregate,
|
||||
"company_base_ratio": Decimal("0"),
|
||||
"adjust_extra": Decimal("0"),
|
||||
"adjust_deduct": Decimal("0"),
|
||||
"adjust_net": Decimal("0"),
|
||||
"net_total": second_aggregate,
|
||||
"frozen_amount": None,
|
||||
"aggregate_ledger_id": None,
|
||||
"members": [],
|
||||
},
|
||||
],
|
||||
"unassigned": [
|
||||
_member(
|
||||
4,
|
||||
"赵未分配",
|
||||
performance="2000.00",
|
||||
payable="2000.00",
|
||||
reason="无经理或上级非营销经理",
|
||||
)
|
||||
],
|
||||
"company": {
|
||||
"team_base_total": Decimal("10000.00"),
|
||||
"company_base": Decimal("12000.00"),
|
||||
"company_share_pct": company_text,
|
||||
"company_share_amount": Decimal("600.00"),
|
||||
"company_share_count": share_count,
|
||||
"company_share_total": Decimal("600.00") * share_count,
|
||||
},
|
||||
"reconciliation": {
|
||||
"team_member_payable_total": Decimal("10000.00"),
|
||||
"unassigned_payable_total": Decimal("2000.00"),
|
||||
"team_share_total": Decimal("1250.00"),
|
||||
"company_share_total": Decimal("600.00") * share_count,
|
||||
"aggregate_payable_total": aggregate_total,
|
||||
"aggregate_rounding_diff": Decimal("0"),
|
||||
"adjust_net_total": Decimal("0"),
|
||||
"payable_total": payable_total,
|
||||
"team_base_total": Decimal("10000.00"),
|
||||
"unassigned_performance_total": Decimal("2000.00"),
|
||||
"other_team_base": Decimal("0"),
|
||||
"other_team_member_payable_total": Decimal("0"),
|
||||
"other_team_share_total": Decimal("0"),
|
||||
"other_team_company_share_total": Decimal("0"),
|
||||
"other_team_aggregate_total": Decimal("0"),
|
||||
"other_adjust_net_total": Decimal("0"),
|
||||
"company_base": Decimal("12000.00"),
|
||||
},
|
||||
"totals": {},
|
||||
}
|
||||
|
||||
|
||||
def _build_department(data: dict) -> bytes:
|
||||
return report_export.build_workbook(
|
||||
code="department-summary",
|
||||
title="部门绩效工资汇总表",
|
||||
filters="统计月份 2026-10",
|
||||
rows=[],
|
||||
summary=None,
|
||||
buckets=None,
|
||||
department=data,
|
||||
template_version="V1.2",
|
||||
operator_name="admin",
|
||||
company_name="衍水集团",
|
||||
)
|
||||
|
||||
|
||||
class Test部门汇总签字稿:
|
||||
"""签字稿少了团队分成/5% 块/未分配/零成员团队，合计就会小于应发。"""
|
||||
|
||||
def test_五块顺序完整(self):
|
||||
text = _sheet_text(_build_department(_department_data()))
|
||||
titles = [
|
||||
"一、团队分成汇总",
|
||||
"二、全公司分成（每位有聚合行",
|
||||
"三、对账汇总",
|
||||
"四、团队成员明细",
|
||||
"五、未分配人员明细",
|
||||
]
|
||||
positions = [text.index(title) for title in titles]
|
||||
assert positions == sorted(positions), "签字稿块顺序错乱"
|
||||
assert "财务复核" in text and "领导签字" in text
|
||||
|
||||
def test_零成员团队与未分配行存在(self):
|
||||
text = _sheet_text(_build_department(_department_data()))
|
||||
assert "王经理" in text
|
||||
assert "赵未分配" in text
|
||||
assert "无经理或上级非营销经理" in text
|
||||
|
||||
def test_全公司分成按计入次数累计(self):
|
||||
"""两位有聚合行的经理各计一次 5%；零成员团队不产生分成。"""
|
||||
one = _department_data()
|
||||
text = _sheet_text(_build_department(one))
|
||||
assert text.count("全公司分成（每位有聚合行") == 1
|
||||
assert "共 1 次" in text
|
||||
assert "13850" in text
|
||||
|
||||
two = _department_data(second_has_aggregate=True)
|
||||
text = _sheet_text(_build_department(two))
|
||||
assert "共 2 次" in text
|
||||
# 团队成员应付 10000 + 未分配 2000 + 聚合 2450
|
||||
assert "14450" in text
|
||||
|
||||
def test_应发合计取对账口径(self):
|
||||
text = _sheet_text(_build_department(_department_data()))
|
||||
assert "应发合计" in text
|
||||
# Excel 数值单元格读回为 int/float，金额断言只看数值本身
|
||||
assert "13850" in text
|
||||
|
||||
def test_聚合行舍入差单列(self):
|
||||
data = _department_data()
|
||||
data["reconciliation"]["aggregate_rounding_diff"] = Decimal("-0.02")
|
||||
text = _sheet_text(_build_department(data))
|
||||
assert "聚合行舍入差" in text
|
||||
assert "-0.02" in text
|
||||
|
||||
def test_比例标题参数化(self):
|
||||
text = _sheet_text(_build_department(_department_data("20", "6")))
|
||||
assert "团队分成（20%）" in text
|
||||
assert "6%" in text
|
||||
assert "12.5%" not in text
|
||||
assert "5%" not in text.replace("6%", "")
|
||||
Reference in new issue
Block a user