Hermes CI Fix
b796897f8c
require_auth now uses HTTPBearer(auto_error=False) and raises 401 explicitly when the Authorization header is missing or non-Bearer, matching require_auth_401. Unauthenticated requests are 401 for every user endpoint; authenticated-but-forbidden (cross-entity / role) stays 403. - app/auth_middleware.py: single optional bearer, 401 on missing token (require_auth_401 kept as a compatibility alias) - tests: update 4 pre-existing cases that asserted 403 without a token (test_auth, test_reports, test_kpi_causality, test_expenses, test_api_auth_surface_20260914) - intentional behavior change - add tests/test_auth_401_unification_20260915.py: sample + full sweep (all AUTHED GET routes -> 401), invalid token -> 401, cross-entity and role-denied still 403, bot channel (X-BOT-KEY) and public paths unchanged
487 lines
23 KiB
Python
487 lines
23 KiB
Python
"""费用审核智能体模块测试 — 规则CRUD + 报销单自动校验 + 审批流 + 看板统计
|
||
|
||
覆盖 expenses.py 全部13个端点：
|
||
rules(CRUD+seed) / reimbursements(submit/list/detail/approve/reject/return/resubmit) / stats
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
||
from app.models import ExpenseRule, ExpenseReimbursement, User
|
||
import hashlib
|
||
|
||
|
||
def mk_rule(db: Session, **kwargs) -> ExpenseRule:
|
||
defaults = {
|
||
"rule_name": "测试规则", "dimension": "expense_type", "dimension_value": "",
|
||
"expense_type": "entertainment", "limit_type": "single", "limit_amount": 2000.0,
|
||
"cycle": "single", "status": "active", "remark": "",
|
||
}
|
||
defaults.update(kwargs)
|
||
r = ExpenseRule(**defaults)
|
||
db.add(r)
|
||
db.commit()
|
||
db.refresh(r)
|
||
return r
|
||
|
||
|
||
def mk_reimb(db: Session, **kwargs) -> ExpenseReimbursement:
|
||
defaults = {
|
||
"reimb_no": "BX_TEST001", "applicant": "张三", "department": "销售部",
|
||
"expense_type": "entertainment", "title": "客户招待", "amount": 500.0,
|
||
"status": "pending", "check_result": "pass", "created_by": "test",
|
||
}
|
||
defaults.update(kwargs)
|
||
r = ExpenseReimbursement(**defaults)
|
||
db.add(r)
|
||
db.commit()
|
||
db.refresh(r)
|
||
return r
|
||
|
||
|
||
class TestRules:
|
||
"""费用规则 CRUD"""
|
||
|
||
BASE = "/api/cma/expenses/rules"
|
||
|
||
def test_list_empty(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(self.BASE, headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"] == []
|
||
assert resp.json()["total"] == 0
|
||
|
||
def test_list_filters(self, client: TestClient, db: Session):
|
||
"""expense_type / status 过滤"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
mk_rule(db, rule_name="招待限额", expense_type="entertainment")
|
||
mk_rule(db, rule_name="差旅限额", expense_type="travel")
|
||
mk_rule(db, rule_name="停用规则", expense_type="office", status="inactive")
|
||
|
||
r = client.get(f"{self.BASE}?expense_type=travel", headers=auth_header(token))
|
||
assert [x["rule_name"] for x in r.json()["data"]] == ["差旅限额"]
|
||
r = client.get(f"{self.BASE}?status=inactive", headers=auth_header(token))
|
||
assert [x["rule_name"] for x in r.json()["data"]] == ["停用规则"]
|
||
# 标签映射
|
||
r = client.get(f"{self.BASE}?expense_type=entertainment", headers=auth_header(token))
|
||
assert r.json()["data"][0]["expense_type_label"] == "招待费"
|
||
|
||
def test_create_rule(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(self.BASE, headers=auth_header(token),
|
||
json={"rule_name": "招待费单笔限额", "expense_type": "entertainment",
|
||
"limit_type": "single", "limit_amount": 2000.0})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "规则已创建"
|
||
assert resp.json()["id"] > 0
|
||
|
||
def test_create_rule_missing_params(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(self.BASE, headers=auth_header(token), json={"rule_name": "缺参数"})
|
||
assert resp.status_code == 400
|
||
assert "缺少必要参数" in resp.json()["detail"]
|
||
|
||
def test_create_rule_invalid_expense_type(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(self.BASE, headers=auth_header(token),
|
||
json={"rule_name": "非法类型", "expense_type": "gambling",
|
||
"limit_amount": 100})
|
||
assert resp.status_code == 400
|
||
assert "无效费用类型" in resp.json()["detail"]
|
||
|
||
def test_create_rule_invalid_limit_type(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(self.BASE, headers=auth_header(token),
|
||
json={"rule_name": "非法限额", "expense_type": "office",
|
||
"limit_type": "per_month", "limit_amount": 100})
|
||
assert resp.status_code == 400
|
||
assert "无效限额类型" in resp.json()["detail"]
|
||
|
||
def test_update_rule(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
rule = mk_rule(db)
|
||
resp = client.put(f"{self.BASE}/{rule.id}", headers=auth_header(token),
|
||
json={"rule_name": "新规则名", "limit_amount": 5000.0, "status": "inactive"})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "规则已更新"
|
||
r = client.get(self.BASE, headers=auth_header(token))
|
||
updated = r.json()["data"][0]
|
||
assert updated["rule_name"] == "新规则名"
|
||
assert updated["limit_amount"] == 5000.0
|
||
assert updated["status"] == "inactive"
|
||
|
||
def test_update_rule_not_found(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.put(f"{self.BASE}/99999", headers=auth_header(token), json={"rule_name": "x"})
|
||
assert resp.status_code == 404
|
||
|
||
def test_update_rule_invalid_type(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
rule = mk_rule(db)
|
||
resp = client.put(f"{self.BASE}/{rule.id}", headers=auth_header(token),
|
||
json={"expense_type": "bogus"})
|
||
assert resp.status_code == 400
|
||
|
||
def test_delete_rule(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
rule = mk_rule(db)
|
||
resp = client.delete(f"{self.BASE}/{rule.id}", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "规则已删除"
|
||
r = client.get(self.BASE, headers=auth_header(token))
|
||
assert r.json()["data"] == []
|
||
|
||
def test_delete_rule_not_found(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.delete(f"{self.BASE}/99999", headers=auth_header(token))
|
||
assert resp.status_code == 404
|
||
|
||
def test_seed_rules_idempotent(self, client: TestClient, db: Session):
|
||
"""预置规则幂等：首次7条，再次0条"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
r1 = client.post(f"{self.BASE}/seed", headers=auth_header(token))
|
||
assert r1.status_code == 200
|
||
assert "新增 7 条" in r1.json()["message"]
|
||
r2 = client.post(f"{self.BASE}/seed", headers=auth_header(token))
|
||
assert "新增 0 条" in r2.json()["message"]
|
||
r = client.get(self.BASE, headers=auth_header(token))
|
||
assert r.json()["total"] == 7
|
||
|
||
def test_rule_permission_business_denied(self, client: TestClient, db: Session):
|
||
"""business角色不能创建规则 → 403"""
|
||
u = User(username="biz_expense", password_hash=hashlib.sha256("pass123".encode()).hexdigest(),
|
||
name="业务员", role="business")
|
||
db.add(u)
|
||
db.commit()
|
||
token = get_token_for_user(client, username="biz_expense", password="pass123")
|
||
resp = client.post(self.BASE, headers=auth_header(token),
|
||
json={"rule_name": "越权规则", "expense_type": "office", "limit_amount": 10})
|
||
assert resp.status_code == 403
|
||
|
||
|
||
class TestReimbursements:
|
||
"""报销单提交/校验/审批"""
|
||
|
||
BASE = "/api/cma/expenses/reimbursements"
|
||
|
||
def test_submit_pass_no_rules(self, client: TestClient, db: Session):
|
||
"""无规则 → 提交通过，状态 pending"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(self.BASE, headers=auth_header(token),
|
||
json={"expense_type": "entertainment", "title": "客户餐费",
|
||
"amount": 500.0, "expense_date": "2026-06-10"})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["message"] == "报销单已提交"
|
||
assert data["data"]["auto_check_passed"] is True
|
||
d = data["data"]
|
||
assert d["status"] == "pending"
|
||
assert d["reimb_no"].startswith("BX")
|
||
assert d["expense_type_label"] == "招待费"
|
||
assert d["applicant"] == "测试管理员" # 默认取当前用户name
|
||
|
||
def test_submit_over_single_limit_returned(self, client: TestClient, db: Session):
|
||
"""单笔超限 → 自动打回 returned + check_result fail"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
mk_rule(db, rule_name="招待单笔≤100", expense_type="entertainment",
|
||
limit_type="single", limit_amount=100.0)
|
||
resp = client.post(self.BASE, headers=auth_header(token),
|
||
json={"expense_type": "entertainment", "title": "大额招待",
|
||
"amount": 500.0})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["message"] == "报销单超限，已自动打回"
|
||
assert data["data"]["auto_check_passed"] is False
|
||
d = data["data"]
|
||
assert d["status"] == "returned"
|
||
assert d["check_result"] == "fail"
|
||
assert "超限" in d["check_reason"]
|
||
assert d["check_detail"][0]["passed"] is False
|
||
|
||
def test_submit_within_single_limit(self, client: TestClient, db: Session):
|
||
"""单笔限额内 → 通过，check_detail 有记录"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
mk_rule(db, rule_name="招待单笔≤2000", expense_type="entertainment",
|
||
limit_type="single", limit_amount=2000.0)
|
||
resp = client.post(self.BASE, headers=auth_header(token),
|
||
json={"expense_type": "entertainment", "title": "正常招待",
|
||
"amount": 500.0})
|
||
assert resp.status_code == 200
|
||
d = resp.json()["data"]
|
||
assert d["status"] == "pending"
|
||
assert d["check_result"] == "pass"
|
||
assert d["check_detail"][0]["passed"] is True
|
||
|
||
def test_submit_monthly_limit(self, client: TestClient, db: Session):
|
||
"""月度累计超限 → 第二次自动打回"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
mk_rule(db, rule_name="招待月限额1000", expense_type="entertainment",
|
||
limit_type="monthly", limit_amount=1000.0, cycle="monthly")
|
||
r1 = client.post(self.BASE, headers=auth_header(token),
|
||
json={"expense_type": "entertainment", "title": "第一批",
|
||
"amount": 500.0, "expense_date": "2026-06-10"})
|
||
assert r1.json()["data"]["auto_check_passed"] is True
|
||
r2 = client.post(self.BASE, headers=auth_header(token),
|
||
json={"expense_type": "entertainment", "title": "第二批",
|
||
"amount": 800.0, "expense_date": "2026-06-20"})
|
||
assert r2.status_code == 200
|
||
assert r2.json()["data"]["auto_check_passed"] is False
|
||
assert r2.json()["data"]["status"] == "returned"
|
||
# 累计明细：500 + 800 = 1300
|
||
assert r2.json()["data"]["check_detail"][0]["actual"] == 1300.0
|
||
|
||
def test_submit_missing_params(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(self.BASE, headers=auth_header(token),
|
||
json={"title": "没类型没金额"})
|
||
assert resp.status_code == 400
|
||
assert "缺少必要参数" in resp.json()["detail"]
|
||
|
||
def test_submit_invalid_type(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(self.BASE, headers=auth_header(token),
|
||
json={"expense_type": "bogus", "title": "x", "amount": 100})
|
||
assert resp.status_code == 400
|
||
|
||
def test_submit_zero_amount(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(self.BASE, headers=auth_header(token),
|
||
json={"expense_type": "office", "title": "x", "amount": 0})
|
||
assert resp.status_code == 400
|
||
assert "必须大于0" in resp.json()["detail"]
|
||
|
||
def test_submit_bad_date(self, client: TestClient, db: Session):
|
||
"""非法日期 → expense_date 为 None 不报错"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(self.BASE, headers=auth_header(token),
|
||
json={"expense_type": "office", "title": "x", "amount": 100,
|
||
"expense_date": "not-a-date"})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["expense_date"] is None
|
||
|
||
def test_list_filters(self, client: TestClient, db: Session):
|
||
"""状态/类型/申请人/关键字 过滤"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
mk_reimb(db, reimb_no="BX0001", applicant="张三", title="客户招待", status="pending")
|
||
mk_reimb(db, reimb_no="BX0002", applicant="李四", title="差旅住宿", status="approved")
|
||
mk_reimb(db, reimb_no="BX0003", applicant="张三", title="办公采购", status="rejected",
|
||
expense_type="office")
|
||
|
||
r = client.get(f"{self.BASE}?status=approved", headers=auth_header(token))
|
||
assert [x["reimb_no"] for x in r.json()["data"]] == ["BX0002"]
|
||
r = client.get(f"{self.BASE}?expense_type=office", headers=auth_header(token))
|
||
assert [x["reimb_no"] for x in r.json()["data"]] == ["BX0003"]
|
||
r = client.get(f"{self.BASE}?applicant=张三", headers=auth_header(token))
|
||
assert {x["reimb_no"] for x in r.json()["data"]} == {"BX0001", "BX0003"}
|
||
r = client.get(f"{self.BASE}?keyword=差旅", headers=auth_header(token))
|
||
assert [x["reimb_no"] for x in r.json()["data"]] == ["BX0002"]
|
||
r = client.get(f"{self.BASE}?keyword=BX0003", headers=auth_header(token))
|
||
assert [x["reimb_no"] for x in r.json()["data"]] == ["BX0003"]
|
||
|
||
def test_get_detail(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
reimb = mk_reimb(db, reimb_no="BX0099")
|
||
resp = client.get(f"{self.BASE}/{reimb.id}", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["reimb_no"] == "BX0099"
|
||
|
||
def test_get_detail_not_found(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(f"{self.BASE}/99999", headers=auth_header(token))
|
||
assert resp.status_code == 404
|
||
|
||
def test_approve(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
reimb = mk_reimb(db, reimb_no="BX0100")
|
||
resp = client.post(f"{self.BASE}/{reimb.id}/approve", headers=auth_header(token),
|
||
json={"comment": "同意"})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "已审批通过"
|
||
d = resp.json()["data"]
|
||
assert d["status"] == "approved"
|
||
assert d["approver"] == "测试管理员"
|
||
assert d["approve_comment"] == "同意"
|
||
|
||
def test_approve_wrong_state(self, client: TestClient, db: Session):
|
||
"""已审批的单不能再审 → 400"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
reimb = mk_reimb(db, reimb_no="BX0101", status="approved")
|
||
resp = client.post(f"{self.BASE}/{reimb.id}/approve", headers=auth_header(token))
|
||
assert resp.status_code == 400
|
||
assert "不可审批" in resp.json()["detail"]
|
||
|
||
def test_approve_not_found(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{self.BASE}/99999/approve", headers=auth_header(token))
|
||
assert resp.status_code == 404
|
||
|
||
def test_reject_requires_comment(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
reimb = mk_reimb(db, reimb_no="BX0102")
|
||
resp = client.post(f"{self.BASE}/{reimb.id}/reject", headers=auth_header(token))
|
||
assert resp.status_code == 400
|
||
assert "必须填写审批意见" in resp.json()["detail"]
|
||
|
||
def test_reject_with_comment(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
reimb = mk_reimb(db, reimb_no="BX0103")
|
||
resp = client.post(f"{self.BASE}/{reimb.id}/reject", headers=auth_header(token),
|
||
json={"comment": "发票不合规"})
|
||
assert resp.status_code == 200
|
||
d = resp.json()["data"]
|
||
assert d["status"] == "rejected"
|
||
assert d["approve_comment"] == "发票不合规"
|
||
|
||
def test_return(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
reimb = mk_reimb(db, reimb_no="BX0104")
|
||
resp = client.post(f"{self.BASE}/{reimb.id}/return", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
d = resp.json()["data"]
|
||
assert d["status"] == "returned"
|
||
assert d["approve_comment"] == "人工打回"
|
||
|
||
def test_resubmit_after_return(self, client: TestClient, db: Session):
|
||
"""打回后改金额重新提交 → 重新校验通过"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
mk_rule(db, rule_name="招待单笔≤100", expense_type="entertainment",
|
||
limit_type="single", limit_amount=100.0)
|
||
# 超限被自动打回
|
||
r = client.post(self.BASE, headers=auth_header(token),
|
||
json={"expense_type": "entertainment", "title": "超标单",
|
||
"amount": 500.0})
|
||
rid = r.json()["data"]["id"]
|
||
assert r.json()["data"]["status"] == "returned"
|
||
# 改金额到限额内 → 重新提交通过
|
||
resp = client.post(f"{self.BASE}/{rid}/resubmit", headers=auth_header(token),
|
||
json={"amount": 80.0})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "已重新提交"
|
||
d = resp.json()["data"]
|
||
assert d["status"] == "pending"
|
||
assert d["amount"] == 80.0
|
||
assert d["approver"] is None
|
||
|
||
def test_resubmit_still_over(self, client: TestClient, db: Session):
|
||
"""重提仍超限 → 再次打回"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
mk_rule(db, rule_name="招待单笔≤100", expense_type="entertainment",
|
||
limit_type="single", limit_amount=100.0)
|
||
r = client.post(self.BASE, headers=auth_header(token),
|
||
json={"expense_type": "entertainment", "title": "超标单",
|
||
"amount": 500.0})
|
||
rid = r.json()["data"]["id"]
|
||
resp = client.post(f"{self.BASE}/{rid}/resubmit", headers=auth_header(token),
|
||
json={"amount": 300.0})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "仍超限，已再次打回"
|
||
assert resp.json()["data"]["status"] == "returned"
|
||
|
||
def test_resubmit_wrong_state(self, client: TestClient, db: Session):
|
||
"""pending 状态不能重提 → 400"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
reimb = mk_reimb(db, reimb_no="BX0105")
|
||
resp = client.post(f"{self.BASE}/{reimb.id}/resubmit", headers=auth_header(token))
|
||
assert resp.status_code == 400
|
||
assert "不可重新提交" in resp.json()["detail"]
|
||
|
||
def test_approve_permission_business_denied(self, client: TestClient, db: Session):
|
||
"""business不能审批 → 403"""
|
||
reimb = mk_reimb(db, reimb_no="BX0106")
|
||
u = User(username="biz_approve", password_hash=hashlib.sha256("pass123".encode()).hexdigest(),
|
||
name="业务员", role="business")
|
||
db.add(u)
|
||
db.commit()
|
||
token = get_token_for_user(client, username="biz_approve", password="pass123")
|
||
resp = client.post(f"{self.BASE}/{reimb.id}/approve", headers=auth_header(token))
|
||
assert resp.status_code == 403
|
||
|
||
|
||
class TestStats:
|
||
"""审核看板统计"""
|
||
|
||
BASE = "/api/cma/expenses/stats"
|
||
|
||
def test_stats_empty(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(self.BASE, headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["counts"]["total"] == 0
|
||
assert data["monthly_total"] == 0
|
||
assert len(data["budget_usage"]) == 2
|
||
|
||
def test_stats_with_data(self, client: TestClient, db: Session):
|
||
"""有报销数据 → 状态计数/类型统计/超限预警/月度总额"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
from datetime import datetime
|
||
mk_reimb(db, reimb_no="BX0201", applicant="张三", amount=500.0,
|
||
expense_date=datetime(2026, 6, 10), status="pending")
|
||
mk_reimb(db, reimb_no="BX0202", applicant="李四", amount=300.0,
|
||
expense_date=datetime(2026, 6, 11), status="approved")
|
||
mk_reimb(db, reimb_no="BX0203", applicant="王五", amount=800.0,
|
||
expense_date=datetime(2026, 5, 20), status="approved") # 上期不计
|
||
# 超限打回
|
||
mk_reimb(db, reimb_no="BX0204", applicant="赵六", amount=99999.0,
|
||
expense_date=datetime(2026, 6, 12), status="returned", check_result="fail",
|
||
check_reason="招待费单笔限额: 超限")
|
||
resp = client.get(f"{self.BASE}?period=2026-06", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["period"] == "2026-06"
|
||
assert data["counts"]["pending"] == 1
|
||
assert data["counts"]["approved"] == 2
|
||
assert data["counts"]["returned"] == 1
|
||
assert data["counts"]["total"] == 4
|
||
# 月度总额 = 500 + 300 = 800（上期不计，returned不计）
|
||
assert data["monthly_total"] == 800.0
|
||
assert data["amounts_by_type"][0]["amount"] == 800.0
|
||
# 超限预警
|
||
assert data["over_limit_count"] == 1
|
||
assert data["over_limit"][0]["reimb_no"] == "BX0204"
|
||
# 预算使用率
|
||
ent_usage = next(u for u in data["budget_usage"] if u["expense_type"] == "entertainment")
|
||
assert ent_usage["used"] == 800.0
|
||
assert ent_usage["usage_rate"] == round(800 / 150000 * 100, 1)
|
||
# 最近列表
|
||
assert len(data["recent"]) == 4
|
||
|
||
def test_stats_no_token(self, client: TestClient, db: Session):
|
||
resp = client.get(self.BASE)
|
||
# 语义统一(2026-09-15)：未认证=401（原 HTTPBearer auto_error 给 403）
|
||
assert resp.status_code == 401
|