39a3fa7796
- 新增 router/access.ts 统一准入与落地页（工作台→业务数据→回款）， 守卫/登录回跳/菜单/面包屑复用；普通用户不再被无限重定向 - 待办按角色映射在途审批节点（财务/领导/admin），admin 不代办但可查全部 - 预警排除已终止业务；WarningBlock.total 为全量命中数；回款趋势改日期范围 - 工作台加载请求序号防竞态；KPI 口径说明补经理分成/调整单 - 新增 role_home_flow e2e，更新 system_management 普通用户断言 Co-Authored-By: Claude Code <noreply@anthropic.com>
454 lines
16 KiB
Python
454 lines
16 KiB
Python
"""工作台集成测试（M4 ⑧）。
|
||
|
||
守的口径：
|
||
- 权限：普通用户 403 且文案「无权访问工作台」
|
||
- 总览：绩效合计/人数与台账一致；月份参数非法 400
|
||
- 待办：月结提交后出现待审批；回款未指定批次可命中；同步未建时显式降级
|
||
- 风险预警：审批超时命中、参数关闭后不产生条目、阈值参数化生效
|
||
- 最近动态：来源操作日志（写操作可见）
|
||
|
||
待办与预警为实时派生（不落表），测试只验证查询结果，不清理派生数据。
|
||
"""
|
||
|
||
import uuid
|
||
from datetime import date, datetime, timedelta
|
||
from decimal import Decimal
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
_DASHBOARD = "/api/v1/dashboard"
|
||
_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"
|
||
_PARAMS = "/api/v1/system/params"
|
||
|
||
|
||
def _role_id(code: str) -> int:
|
||
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) -> dict:
|
||
resp = client.post(
|
||
_PERSONNEL,
|
||
json={
|
||
"personnel_name": f"t_工作台_{uuid.uuid4().hex[:8]}",
|
||
"business_role_id": _role_id("salesperson"),
|
||
"salary_standard": "3000.00",
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _chain(client: TestClient, headers: dict, *, assign_shipment: bool = True) -> dict:
|
||
"""客户/业务（100×100，成本 3000）→ 发货（60 天前）→ 回款 → 核算。"""
|
||
from tests.integration.factories import business_item, business_payload, customer
|
||
|
||
cust = customer(client, headers)
|
||
person = _person(client, headers)
|
||
resp = client.post(
|
||
_BUSINESSES,
|
||
json=business_payload(
|
||
cust["id"],
|
||
owner_personnel_id=person["id"],
|
||
items=[business_item(cost_amount="3000.00")],
|
||
),
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
business = resp.json()["data"]
|
||
|
||
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"]
|
||
|
||
payload = {
|
||
"business_id": business["id"],
|
||
"receipt_date": date.today().isoformat(),
|
||
"amount": "10000.00",
|
||
}
|
||
if assign_shipment:
|
||
payload["shipment_id"] = shipment["id"]
|
||
resp = client.post(_RECEIPTS, json=payload, headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
receipt = resp.json()["data"]
|
||
if assign_shipment:
|
||
resp = client.post(f"{_CALC}/receipts/{receipt['id']}/calc", headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
return {"business": business, "person": person, "receipt": receipt}
|
||
|
||
|
||
def _month() -> str:
|
||
return date.today().strftime("%Y-%m")
|
||
|
||
|
||
def _dashboard(client: TestClient, headers: dict, month: str | None = None) -> dict:
|
||
params = {"month": month} if month else {}
|
||
resp = client.get(_DASHBOARD, params=params, headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]
|
||
|
||
|
||
def test_权限_普通用户403文案(client: TestClient, normal: dict):
|
||
resp = client.get(_DASHBOARD, headers=normal["headers"])
|
||
assert resp.status_code == 403
|
||
assert resp.json()["message"] == "无权访问工作台"
|
||
|
||
|
||
def test_总览_绩效合计与台账一致且非法月份400(client: TestClient, admin: dict):
|
||
_chain(client, admin["headers"])
|
||
data = _dashboard(client, admin["headers"], _month())
|
||
|
||
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:
|
||
expected = 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()
|
||
)
|
||
)
|
||
assert Decimal(str(data["overview"]["performance_total"])) == expected
|
||
assert data["overview"]["person_count"] >= 1
|
||
assert len(data["overview"]["trend"]) == 6
|
||
|
||
resp = client.get(_DASHBOARD, params={"month": "2026-13"}, headers=admin["headers"])
|
||
assert resp.status_code == 400
|
||
|
||
|
||
def test_待办_待审批与未指定批次与同步降级(client: TestClient, admin: dict):
|
||
# 先造一笔已核算业务（月结生成要求有可结算记录），再补一笔未指定批次的回款
|
||
_chain(client, admin["headers"], assign_shipment=True)
|
||
ctx = _chain(client, admin["headers"], assign_shipment=False)
|
||
# 月结提交 → 产生待审批单
|
||
resp = client.post(
|
||
f"{_SETTLEMENTS}/generate", json={"month": _month()}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
settlement_id = resp.json()["data"]["id"]
|
||
resp = client.post(
|
||
f"{_SETTLEMENTS}/{settlement_id}/submit", headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
data = _dashboard(client, admin["headers"])
|
||
assert data["todos"]["pending_approval_total"] >= 1
|
||
assert any(
|
||
item["order_type"] == 2 for item in data["todos"]["pending_approvals"]
|
||
)
|
||
assert data["todos"]["unassigned_receipt_total"] >= 1
|
||
assert data["todos"]["unassigned_receipts"][0]["receipt_id"] == ctx["receipt"]["id"]
|
||
# ⑥ 未建：显式降级
|
||
assert data["todos"]["sync_available"] is False
|
||
|
||
|
||
def test_预警_审批超时命中且参数关闭后不产生(client: TestClient, admin: dict):
|
||
_chain(client, admin["headers"])
|
||
resp = client.post(
|
||
f"{_SETTLEMENTS}/generate", json={"month": _month()}, headers=admin["headers"]
|
||
)
|
||
settlement_id = resp.json()["data"]["id"]
|
||
client.post(f"{_SETTLEMENTS}/{settlement_id}/submit", headers=admin["headers"])
|
||
|
||
# 把待审批单提交时间改到 4 天前（阈值默认 3 天）
|
||
from sqlalchemy import update
|
||
|
||
from app.core.db import get_session_factory
|
||
from app.models import ApprovalOrder
|
||
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
session.execute(
|
||
update(ApprovalOrder)
|
||
.where(ApprovalOrder.status.in_((1, 2)))
|
||
.values(create_time=datetime.now() - timedelta(days=4))
|
||
)
|
||
session.commit()
|
||
|
||
data = _dashboard(client, admin["headers"])
|
||
assert data["warnings"]["enabled"] is True
|
||
assert any(item["code"] == "approval_timeout" for item in data["warnings"]["items"])
|
||
|
||
# 关闭预警开关 → 不产生条目
|
||
resp = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 1,
|
||
"remark": "e2e 工作台测试：关闭风险预警",
|
||
"items": [{"param_code": "warning.enabled", "dimension_value": "", "value": "0"}],
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
data = _dashboard(client, admin["headers"])
|
||
assert data["warnings"]["enabled"] is False
|
||
assert data["warnings"]["items"] == []
|
||
|
||
|
||
def test_动态_来源操作日志(client: TestClient, admin: dict):
|
||
_chain(client, admin["headers"])
|
||
data = _dashboard(client, admin["headers"])
|
||
assert data["activities"], "工作台动态为空：写操作应产生操作日志"
|
||
assert {"module", "action", "summary", "create_time"} <= set(data["activities"][0])
|
||
|
||
|
||
# ==== 角色待办 / 预警阈值 / 趋势（2026-10-19 code-review 修复回归）====
|
||
|
||
|
||
@pytest.fixture
|
||
def finance(make_user, auth_headers) -> dict:
|
||
user = make_user(["finance"])
|
||
user["headers"] = auth_headers(user)
|
||
return user
|
||
|
||
|
||
@pytest.fixture
|
||
def leader(make_user, auth_headers) -> dict:
|
||
user = make_user(["leader"])
|
||
user["headers"] = auth_headers(user)
|
||
return user
|
||
|
||
|
||
def _insert_approval(status: int) -> None:
|
||
"""直插在途审批单（避免依赖两次流转，专注工作台过滤口径）。"""
|
||
from app.core.db import get_session_factory
|
||
from app.models import ApprovalOrder
|
||
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
session.add(
|
||
ApprovalOrder(
|
||
order_no=f"AP{uuid.uuid4().hex[:12]}",
|
||
order_type=2,
|
||
ref_id=0,
|
||
refer_no="t_工作台",
|
||
status=status,
|
||
submitter_name="t_admin",
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
|
||
def _set_param(client: TestClient, headers: dict, code: str, value: str) -> None:
|
||
resp = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 1,
|
||
"remark": f"t_工作台：{code}",
|
||
"items": [{"param_code": code, "dimension_value": "", "value": value}],
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
|
||
def _shipment_scene(client: TestClient, headers: dict, *, days_ago: int) -> tuple[dict, dict]:
|
||
"""造一笔无回款的发货（用于回款逾期预警）。"""
|
||
from tests.integration.factories import business_item, business_payload, customer
|
||
|
||
person = _person(client, headers)
|
||
cust = customer(client, headers)
|
||
resp = client.post(
|
||
_BUSINESSES,
|
||
json=business_payload(
|
||
cust["id"],
|
||
owner_personnel_id=person["id"],
|
||
items=[business_item(cost_amount="3000.00")],
|
||
),
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
business = resp.json()["data"]
|
||
ship_date = (date.today() - timedelta(days=days_ago)).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
|
||
return business, resp.json()["data"]
|
||
|
||
|
||
def test_待办_按角色仅显示可操作节点(
|
||
client: TestClient, admin: dict, finance: dict, leader: dict, make_user, auth_headers
|
||
):
|
||
"""财务只看到待财务复核、领导只看到待领导审核；admin 与多角色看并集。"""
|
||
_insert_approval(1)
|
||
_insert_approval(2)
|
||
|
||
assert _dashboard(client, finance["headers"])["todos"]["pending_approval_total"] == 1
|
||
assert _dashboard(client, leader["headers"])["todos"]["pending_approval_total"] == 1
|
||
assert _dashboard(client, admin["headers"])["todos"]["pending_approval_total"] == 2
|
||
|
||
both = make_user(["finance", "leader"])
|
||
both["headers"] = auth_headers(both)
|
||
assert _dashboard(client, both["headers"])["todos"]["pending_approval_total"] == 2
|
||
|
||
finance_orders = {
|
||
item["order_no"] for item in _dashboard(client, finance["headers"])["todos"]["pending_approvals"]
|
||
}
|
||
leader_orders = {
|
||
item["order_no"] for item in _dashboard(client, leader["headers"])["todos"]["pending_approvals"]
|
||
}
|
||
assert finance_orders.isdisjoint(leader_orders)
|
||
|
||
|
||
def test_待办_财务通过后领导出现待办(
|
||
client: TestClient, admin: dict, finance: dict, leader: dict
|
||
):
|
||
"""真实流转：财务节点通过后，条目从财务待办消失并进入领导待办。"""
|
||
_chain(client, admin["headers"])
|
||
resp = client.post(
|
||
f"{_SETTLEMENTS}/generate", json={"month": _month()}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
settlement_id = resp.json()["data"]["id"]
|
||
resp = client.post(
|
||
f"{_SETTLEMENTS}/{settlement_id}/submit", headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
order = _dashboard(client, admin["headers"])["todos"]["pending_approvals"][0]
|
||
assert _dashboard(client, finance["headers"])["todos"]["pending_approval_total"] == 1
|
||
assert _dashboard(client, leader["headers"])["todos"]["pending_approval_total"] == 0
|
||
|
||
resp = client.post(
|
||
f"/api/v1/system/approvals/{order['order_id']}/action",
|
||
json={"action": 1},
|
||
headers=finance["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
assert _dashboard(client, finance["headers"])["todos"]["pending_approval_total"] == 0
|
||
assert _dashboard(client, leader["headers"])["todos"]["pending_approval_total"] == 1
|
||
|
||
|
||
def test_预警_回款逾期命中且阈值参数生效(client: TestClient, admin: dict):
|
||
"""发货超阈值且无任何回款 → 命中；调大阈值后不再命中。"""
|
||
_, shipment = _shipment_scene(client, admin["headers"], days_ago=120)
|
||
|
||
data = _dashboard(client, admin["headers"])
|
||
assert any(
|
||
item["code"] == "receipt_overdue" and item["object_id"] == shipment["id"]
|
||
for item in data["warnings"]["items"]
|
||
), "120 天未回款的发货批次未命中预警"
|
||
|
||
_set_param(client, admin["headers"], "warning.receipt.overdue.days", "200")
|
||
data = _dashboard(client, admin["headers"])
|
||
assert not any(
|
||
item["code"] == "receipt_overdue" and item["object_id"] == shipment["id"]
|
||
for item in data["warnings"]["items"]
|
||
), "阈值调大后仍命中，说明参数未生效"
|
||
|
||
|
||
def test_预警_终止业务不再产生回款逾期(client: TestClient, admin: dict):
|
||
"""已终止业务不再催收，其未回款发货单不得留在预警列表。"""
|
||
business, shipment = _shipment_scene(client, admin["headers"], days_ago=120)
|
||
resp = client.put(
|
||
f"{_BUSINESSES}/{business['id']}/status",
|
||
json={"status": 3, "reason": "t_工作台终止"},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
data = _dashboard(client, admin["headers"])
|
||
assert not any(item["object_id"] == shipment["id"] for item in data["warnings"]["items"])
|
||
assert data["warnings"]["receipt_overdue_total"] == 0
|
||
|
||
|
||
def test_预警_总数不受展示上限截断(client: TestClient, admin: dict, monkeypatch):
|
||
"""items 只展示前 N 条，徽标 total 必须仍是全量命中数。"""
|
||
from app.services import dashboard_service
|
||
|
||
_shipment_scene(client, admin["headers"], days_ago=120)
|
||
_shipment_scene(client, admin["headers"], days_ago=120)
|
||
monkeypatch.setattr(dashboard_service, "_WARNING_LIMIT", 1)
|
||
|
||
data = _dashboard(client, admin["headers"])
|
||
assert data["warnings"]["total"] == 2
|
||
assert len(data["warnings"]["items"]) == 1
|
||
assert data["warnings"]["total"] == data["warnings"]["receipt_overdue_total"]
|
||
|
||
|
||
def test_总览_回款趋势按到账月跨月归属(client: TestClient, admin: dict):
|
||
"""趋势窗口按 receipt_date 范围过滤，月初/月末不得串月。"""
|
||
from tests.integration.factories import business_item, business_payload, customer
|
||
|
||
person = _person(client, admin["headers"])
|
||
cust = customer(client, admin["headers"])
|
||
resp = client.post(
|
||
_BUSINESSES,
|
||
json=business_payload(
|
||
cust["id"],
|
||
owner_personnel_id=person["id"],
|
||
items=[business_item(cost_amount="3000.00")],
|
||
),
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
business = resp.json()["data"]
|
||
|
||
first_of_month = date.today().replace(day=1)
|
||
previous_last = first_of_month - timedelta(days=1)
|
||
for receipt_date in (previous_last, first_of_month):
|
||
resp = client.post(
|
||
_RECEIPTS,
|
||
json={
|
||
"business_id": business["id"],
|
||
"receipt_date": receipt_date.isoformat(),
|
||
"amount": "1000.00",
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
trend = {
|
||
item["month"]: Decimal(str(item["receipt_total"] or 0))
|
||
for item in _dashboard(client, admin["headers"])["overview"]["trend"]
|
||
}
|
||
assert trend[previous_last.strftime("%Y-%m")] == Decimal("1000.00")
|
||
assert trend[first_of_month.strftime("%Y-%m")] == Decimal("1000.00")
|