fix(cma): entity scope for expenses resubmit + cumulative limit check

F1 resubmit_reimbursement now requires session entity_id (cross-entity -> 404).

F2 validate_reimbursement monthly/yearly cumulative sum is entity-scoped,

so other entities' in-flight amounts no longer pollute the limit verdict.

Tests: backend/tests/test_cross_entity_isolation_expenses_residual_20260921.py
This commit is contained in:
Hermes CI Fix
2026-09-21 14:56:43 +08:00
parent adfc3ac845
commit b4f94a12b3
2 changed files with 233 additions and 2 deletions
+8 -2
View File
@@ -290,7 +290,9 @@ def validate_reimbursement(db: Session, reimb: ExpenseReimbursement, rules: list
})
else:
# 月度/年度累计: 统计同维度+同费用类型在周期内的已提交金额(含本次)
# 账套隔离 2026-09-21:累计口径限本账套(他账套在途不得污染本账套额度判定)
q = db.query(func.coalesce(func.sum(ExpenseReimbursement.amount), 0)).filter(
ExpenseReimbursement.entity_id == reimb.entity_id,
ExpenseReimbursement.expense_type == reimb.expense_type,
ExpenseReimbursement.id != reimb.id,
ExpenseReimbursement.status.in_(["pending", "approved"]),
@@ -519,9 +521,13 @@ def resubmit_reimbursement(
data: dict = None,
db: Session = Depends(get_db),
_=Depends(require_auth),
entity_id: int = Depends(get_entity_id),
):
"""被驳回/打回后重新提交 → 重新自动校验"""
r = db.query(ExpenseReimbursement).filter(ExpenseReimbursement.id == reimb_id).first()
"""被驳回/打回后重新提交 → 重新自动校验(账套隔离 2026-09-21:跨账套 → 404)"""
r = db.query(ExpenseReimbursement).filter(
ExpenseReimbursement.id == reimb_id,
ExpenseReimbursement.entity_id == entity_id,
).first()
if not r:
raise HTTPException(404, "报销单不存在")
if r.status not in ("returned", "rejected"):
@@ -0,0 +1,225 @@
"""CMA expenses 残余跨账套面收口回归测试 — 2026-09-21
任务:fullstack-20260921-expenses-residual-entity-scope(batch5 派生单)
本单范围(仅此两处,均在 backend/app/api/expenses.py):
F1 (P1 跨租户写) resubmit_reimbursement 仅按 id 取行、无账套归属校验
→ 本账套会话持他企报销单 id 即可改写金额并触发重新校验
F2 (P2 跨租户读污染判定) validate_reimbursement 月度/年度累计分支的累计查询未按
entity_id 过滤 → 他账套同费用类型在途金额计入本账套「月度/年度累计是否超限」判定
每条均「跨账套负例 + 本账套正对照」。
只跑内存 SQLite(tests/conftest.py 机制),零生产库 / 零生产 Redis 连接。
"""
from datetime import datetime
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models import Entity, ExpenseReimbursement, ExpenseRule
from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user
EXPENSES = "/api/cma/expenses"
# ──────────────────────────────────────────────
# 夹具
# ──────────────────────────────────────────────
def _ensure_entity(db: Session, eid: int, name: str) -> Entity:
ent = db.query(Entity).filter(Entity.id == eid).first()
if not ent:
ent = Entity(id=eid, name=name, short_name=name, status="active")
db.add(ent)
db.commit()
db.refresh(ent)
return ent
def _mk_rule(db: Session, **kw) -> ExpenseRule:
defaults = {
"rule_name": "ZZEXP_RESIDUAL_RULE", "dimension": "expense_type",
"dimension_value": "", "expense_type": "office",
"limit_type": "monthly", "limit_amount": 1000.0,
"cycle": "monthly", "status": "active", "remark": "",
}
defaults.update(kw)
r = ExpenseRule(**defaults)
db.add(r)
db.commit()
db.refresh(r)
return r
def _mk_reimb(db: Session, eid: int, no: str, amount: float = 100.0,
status: str = "pending", expense_type: str = "office",
expense_date: datetime = datetime(2026, 9, 10)) -> ExpenseReimbursement:
r = ExpenseReimbursement(entity_id=eid, reimb_no=no, applicant="张三",
department="财务部", expense_type=expense_type,
title=f"{no}-报销", amount=amount,
expense_date=expense_date, status=status,
check_result="pass")
db.add(r)
db.commit()
db.refresh(r)
return r
def _login(client: TestClient, entity_id: int) -> str:
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin",
"password": TEST_USER_PASSWORD,
"entity_id": entity_id,
})
assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}"
return resp.json()["token"]
def _seed_two_entities(db: Session):
_ensure_entity(db, 1, "测试企业1")
_ensure_entity(db, 2, "测试企业2")
create_test_user(db)
def _check_entry(detail, rule_type: str) -> dict:
for c in (detail or []):
if c.get("rule_type") == rule_type:
return c
raise AssertionError(f"check_detail 缺少 {rule_type} 条目: {detail}")
# ──────────────────────────────────────────────
# F1. resubmit_reimbursement 账套归属(跨账套 → 404)
# ──────────────────────────────────────────────
def test_resubmit_cross_entity_404(client: TestClient, db: Session):
"""e1 会话持 e2 报销单 id 重新提交(含改金额)→ 404,且 e2 行零变更"""
_seed_two_entities(db)
other = _mk_reimb(db, 2, "ZZEXP_E2_RESUBMIT_1", amount=100.0, status="returned")
token1 = _login(client, 1)
r = client.post(f"{EXPENSES}/reimbursements/{other.id}/resubmit",
headers=auth_header(token1), json={"amount": 9999.0})
assert r.status_code == 404, (
f"跨账套 resubmit 应 404,实际 {r.status_code} {r.text[:200]}")
assert "报销单不存在" in r.json()["detail"]
# 跨租户写必须零生效:金额/状态/校验结果均未改
db.expire_all()
row = db.query(ExpenseReimbursement).filter(
ExpenseReimbursement.id == other.id).first()
assert row is not None
assert row.amount == 100.0, f"跨账套 resubmit 改写了金额: {row.amount}"
assert row.status == "returned", f"跨账套 resubmit 改写了状态: {row.status}"
def test_resubmit_own_entity_ok(client: TestClient, db: Session):
"""本账套报销单重新提交仍可用 → 200(正对照,未被收口误杀)"""
_seed_two_entities(db)
mine = _mk_reimb(db, 1, "ZZEXP_E1_RESUBMIT_1", amount=100.0, status="returned")
token1 = _login(client, 1)
r = client.post(f"{EXPENSES}/reimbursements/{mine.id}/resubmit",
headers=auth_header(token1), json={"amount": 80.0})
assert r.status_code == 200, r.text[:300]
assert r.json()["message"] == "已重新提交"
d = r.json()["data"]
assert d["amount"] == 80.0
assert d["status"] == "pending"
db.expire_all()
row = db.query(ExpenseReimbursement).filter(
ExpenseReimbursement.id == mine.id).first()
assert row.amount == 80.0 and row.status == "pending"
# ──────────────────────────────────────────────
# F2. validate_reimbursement 累计判定按账套(月度/年度)
# ──────────────────────────────────────────────
def test_monthly_cumulative_cross_entity_excluded(client: TestClient, db: Session):
"""e2 在途金额不得计入 e1 月度累计:e1 单笔 200 / 月限额 1000 应通过"""
_seed_two_entities(db)
_mk_rule(db, rule_name="ZZEXP月限额1000", expense_type="office",
limit_type="monthly", limit_amount=1000.0, cycle="monthly")
_mk_reimb(db, 2, "ZZEXP_E2_MONTH_900", amount=900.0, status="pending",
expense_date=datetime(2026, 9, 5))
token1 = _login(client, 1)
r = client.post(EXPENSES + "/reimbursements", headers=auth_header(token1),
json={"expense_type": "office", "title": "e1月度单",
"amount": 200.0, "expense_date": "2026-09-15"})
assert r.status_code == 200, r.text[:300]
d = r.json()["data"]
assert d["auto_check_passed"] is True, (
f"他账套在途金额污染了本账套月度判定: {d['check_detail']}")
assert d["status"] == "pending"
entry = _check_entry(d["check_detail"], "月度累计")
assert entry["used"] == 0.0, f"月度累计 used 含他账套金额: {entry}"
assert entry["actual"] == 200.0
assert entry["passed"] is True
def test_monthly_cumulative_own_entity_still_counted(client: TestClient, db: Session):
"""本账套月度累计仍照常计入:e1 已有 500,再报 600 → 超 1000 自动打回"""
_seed_two_entities(db)
_mk_rule(db, rule_name="ZZEXP月限额1000b", expense_type="office",
limit_type="monthly", limit_amount=1000.0, cycle="monthly")
_mk_reimb(db, 1, "ZZEXP_E1_MONTH_500", amount=500.0, status="pending",
expense_date=datetime(2026, 9, 5))
token1 = _login(client, 1)
r = client.post(EXPENSES + "/reimbursements", headers=auth_header(token1),
json={"expense_type": "office", "title": "e1月度超限",
"amount": 600.0, "expense_date": "2026-09-16"})
assert r.status_code == 200, r.text[:300]
d = r.json()["data"]
assert d["auto_check_passed"] is False, (
f"本账套累计未生效(收口过度): {d['check_detail']}")
assert d["status"] == "returned"
entry = _check_entry(d["check_detail"], "月度累计")
assert entry["used"] == 500.0, f"本账套月度累计 used 应为 500: {entry}"
assert entry["actual"] == 1100.0
def test_yearly_cumulative_cross_entity_excluded(client: TestClient, db: Session):
"""e2 在途金额不得计入 e1 年度累计(跨月同年度):e1 报 200 / 年限额 1000 应通过"""
_seed_two_entities(db)
_mk_rule(db, rule_name="ZZEXP年限额1000", expense_type="office",
limit_type="yearly", limit_amount=1000.0, cycle="yearly")
_mk_reimb(db, 2, "ZZEXP_E2_YEAR_900", amount=900.0, status="approved",
expense_date=datetime(2026, 3, 5))
token1 = _login(client, 1)
r = client.post(EXPENSES + "/reimbursements", headers=auth_header(token1),
json={"expense_type": "office", "title": "e1年度单",
"amount": 200.0, "expense_date": "2026-09-15"})
assert r.status_code == 200, r.text[:300]
d = r.json()["data"]
assert d["auto_check_passed"] is True, (
f"他账套在途金额污染了本账套年度判定: {d['check_detail']}")
entry = _check_entry(d["check_detail"], "年度累计")
assert entry["used"] == 0.0, f"年度累计 used 含他账套金额: {entry}"
assert entry["actual"] == 200.0
def test_yearly_cumulative_own_entity_still_counted(client: TestClient, db: Session):
"""本账套年度累计仍照常计入:e1 已有 900(3月),再报 200 → 超 1000 自动打回"""
_seed_two_entities(db)
_mk_rule(db, rule_name="ZZEXP年限额1000b", expense_type="office",
limit_type="yearly", limit_amount=1000.0, cycle="yearly")
_mk_reimb(db, 1, "ZZEXP_E1_YEAR_900", amount=900.0, status="approved",
expense_date=datetime(2026, 3, 5))
token1 = _login(client, 1)
r = client.post(EXPENSES + "/reimbursements", headers=auth_header(token1),
json={"expense_type": "office", "title": "e1年度超限",
"amount": 200.0, "expense_date": "2026-09-16"})
assert r.status_code == 200, r.text[:300]
d = r.json()["data"]
assert d["auto_check_passed"] is False, (
f"本账套年度累计未生效(收口过度): {d['check_detail']}")
entry = _check_entry(d["check_detail"], "年度累计")
assert entry["used"] == 900.0, f"本账套年度累计 used 应为 900: {entry}"
assert entry["actual"] == 1100.0