shanxia/bx/bx-backend/run_e2e_tests.py

import sys
sys.path.insert(0, ".")
from datetime import datetime
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)
START = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
results = {}

def j(r):
    return r.json()

def tc(module, name, fn):
    try:
        ok, detail = fn()
        ds = str(detail)[:120]
        results.setdefault(module, []).append((name, ok, ds))
        print(f"  [{'PASS' if ok else 'FAIL'}] {module}/{name}: {ds[:100]}")
    except Exception as e:
        results.setdefault(module, []).append((name, False, str(e)[:120]))
        print(f"  [ERROR] {module}/{name}: {str(e)[:100]}")

def login(username, password):
    r = client.post("/api/auth/login", json={"username": username, "password": password})
    if r.status_code == 200:
        token = r.json()["data"]["access_token"]
        return {"Authorization": f"Bearer {token}"}
    return None

print("=" * 60)
print("报销系统 多角色端到端测试")
print("=" * 60)

# ===================== 1. 认证模块 =====================
print("\n=== 1. 认证模块 ===")

H_admin = login("admin", "admin123")
tc("认证", "管理员登录", lambda: (H_admin is not None, "admin"))
tc("认证", "获取当前用户信息", lambda: (
    client.get("/api/auth/me", headers=H_admin).status_code == 200,
    j(client.get("/api/auth/me", headers=H_admin)).get("data", {}).get("username", "")
))
tc("认证", "错误密码返回401", lambda: (
    client.post("/api/auth/login", json={"username": "admin", "password": "wrong"}).status_code == 401,
    "401"
))
tc("认证", "未登录访问返回401", lambda: (
    client.get("/api/users/?page=1&pageSize=10").status_code == 401,
    "401"
))

H_zhangsan = login("zhangsan", "test123")
tc("认证", "员工(zhangsan)登录", lambda: (H_zhangsan is not None, "zhangsan"))

H_lisi = login("lisi", "test123")
tc("认证", "总经理(lisi)登录", lambda: (H_lisi is not None, "lisi"))

H_wangwu = login("wangwu", "test123")
tc("认证", "财务(wangwu)登录", lambda: (H_wangwu is not None, "wangwu"))

# ===================== 2. 健康检查 =====================
print("\n=== 2. 健康检查 ===")
tc("系统", "健康检查", lambda: (client.get("/api/health").status_code == 200, "ok"))

# ===================== 3. 基础数据管理 (admin) =====================
print("\n=== 3. 基础数据管理 ===")

tc("用户管理", "用户列表", lambda: (
    client.get("/api/users/?page=1&pageSize=10", headers=H_admin).status_code == 200,
    f"total={j(client.get('/api/users/?page=1&pageSize=10', headers=H_admin))['data']['total']}"
))
tc("用户管理", "用户分页", lambda: (
    j(client.get("/api/users/?page=1&pageSize=2", headers=H_admin))["data"]["pageSize"] == 2,
    "pageSize=2"
))

tc("角色管理", "角色列表", lambda: (
    client.get("/api/roles/?page=1&pageSize=10", headers=H_admin).status_code == 200,
    f"total={j(client.get('/api/roles/?page=1&pageSize=10', headers=H_admin))['data']['total']}"
))
tc("角色管理", "角色详情", lambda: (
    client.get("/api/roles/1", headers=H_admin).status_code == 200,
    j(client.get("/api/roles/1", headers=H_admin))["data"]["roleName"]
))
tc("角色管理", "角色详情-不存在", lambda: (
    client.get("/api/roles/99999", headers=H_admin).status_code == 404,
    "404"
))

tc("菜单管理", "菜单树", lambda: (
    client.get("/api/menus/tree", headers=H_admin).status_code == 200,
    f"roots={len(j(client.get('/api/menus/tree', headers=H_admin))['data'])}"
))
tc("菜单管理", "按角色查菜单", lambda: (
    client.get("/api/menus/role/1", headers=H_admin).status_code == 200,
    "ok"
))

tc("费用项目", "费用项目树", lambda: (
    client.get("/api/expense-items/tree", headers=H_admin).status_code == 200,
    f"roots={len(j(client.get('/api/expense-items/tree', headers=H_admin))['data'])}"
))
tc("费用项目", "费用项目详情", lambda: (
    client.get("/api/expense-items/1", headers=H_admin).status_code == 200,
    j(client.get("/api/expense-items/1", headers=H_admin))["data"]["itemName"]
))

tc("工作流配置", "工作流列表", lambda: (
    client.get("/api/workflows/?page=1&pageSize=10", headers=H_admin).status_code == 200,
    f"total={j(client.get('/api/workflows/?page=1&pageSize=10', headers=H_admin))['data']['total']}"
))
tc("工作流配置", "按biz_type查询-备用金", lambda: (
    client.get("/api/workflows/biz/advance", headers=H_admin).status_code == 200,
    j(client.get("/api/workflows/biz/advance", headers=H_admin))["data"]["wfName"]
))
tc("工作流配置", "按biz_type查询-报销", lambda: (
    client.get("/api/workflows/biz/reimburse", headers=H_admin).status_code == 200,
    j(client.get("/api/workflows/biz/reimburse", headers=H_admin))["data"]["wfName"]
))

wf_res = j(client.get("/api/workflows/1", headers=H_admin))
if wf_res.get("code") == 200:
    nodes = wf_res["data"].get("nodes", [])
    has_approvers = all(n.get("approver_id") for n in nodes)
    tc("工作流配置", "备用金工作流节点有审批人", lambda: (has_approvers, f"nodes={len(nodes)}, all_have_approver={has_approvers}"))

# ===================== 4. 备用金完整流程 =====================
print("\n=== 4. 备用金完整流程 (员工申请 → 总经理审批 → 财务复核 → 总经理审批 → 发放) ===")

r_adv = client.post("/api/advances/", json={"reason": "出差备用金-E2E测试", "purpose": "北京出差", "amount": 8000}, headers=H_zhangsan)
adv_id = r_adv.json()["data"]["id"] if r_adv.status_code == 200 else 0
tc("备用金流程", "员工创建备用金申请", lambda: (r_adv.status_code == 200, f"id={adv_id}"))
tc("备用金流程", "备用金状态为draft", lambda: (
    j(client.get(f"/api/advances/{adv_id}", headers=H_zhangsan))["data"]["status"] == "draft",
    f"status={j(client.get(f'/api/advances/{adv_id}', headers=H_zhangsan))['data']['status']}"
))

r_sub = client.post(f"/api/advances/{adv_id}/submit", headers=H_zhangsan)
tc("备用金流程", "员工提交备用金", lambda: (r_sub.status_code == 200, f"status={r_sub.json()['data']['status']}"))

r_appr_list = client.get(f"/api/advances/{adv_id}/approvals", headers=H_zhangsan)
tc("备用金流程", "获取审批记录", lambda: (r_appr_list.status_code == 200, f"count={len(r_appr_list.json()['data'])}"))

tc("备用金流程", "我的备用金列表包含新申请", lambda: (
    any(r["id"] == adv_id for r in j(client.get("/api/advances/my?page=1&pageSize=50", headers=H_zhangsan))["data"]["rows"]),
    f"found id={adv_id}"
))

r_approve_1 = client.post(f"/api/advances/{adv_id}/approve", json={"approvalStatus": "approved", "approval_opinion": "同意-总经理"}, headers=H_lisi)
tc("备用金流程", "总经理通过(节点1)", lambda: (r_approve_1.status_code == 200, f"status after={r_approve_1.json()['data']['status']}"))

r_approve_2 = client.post(f"/api/advances/{adv_id}/approve", json={"approvalStatus": "approved", "approval_opinion": "复核通过-财务"}, headers=H_wangwu)
tc("备用金流程", "财务复核通过(节点2)", lambda: (r_approve_2.status_code == 200, f"status after={r_approve_2.json()['data']['status']}"))

detail_after = j(client.get(f"/api/advances/{adv_id}", headers=H_zhangsan))
current_status = detail_after["data"]["status"]
current_node = detail_after["data"].get("currentNode", "")
tc("备用金流程", "审批中状态为pending且有当前节点", lambda: (
    current_status == "pending" and bool(current_node),
    f"status={current_status}, node={current_node}"
))

r_approve_3 = client.post(f"/api/advances/{adv_id}/approve", json={"approvalStatus": "approved", "approval_opinion": "最终批准-总经理"}, headers=H_lisi)
tc("备用金流程", "总经理最终审批通过(节点3)", lambda: (r_approve_3.status_code == 200, f"status={r_approve_3.json()['data']['status']}"))

detail_final = j(client.get(f"/api/advances/{adv_id}", headers=H_zhangsan))
final_status = detail_final["data"]["status"]
tc("备用金流程", "备用金最终状态为approved", lambda: (final_status == "approved", f"status={final_status}"))

# ===================== 5. 备用金发放 =====================
print("\n=== 5. 备用金发放 (财务人员) ===")

r_issues = client.get("/api/payments/advance-issues?page=1&pageSize=10", headers=H_wangwu)
tc("备用金发放", "查看待发放列表(财务角色)", lambda: (r_issues.status_code == 200, f"total={j(r_issues)['data']['total']}"))

r_issue_proof = client.post(
    "/api/attachments/upload",
    headers=H_wangwu,
    files={"file": ("issue_proof.txt", b"voucher", "text/plain")},
    params={"biz_type": "advance_issue", "biz_id": adv_id},
)
tc("备用金发放", "上传发放附件(发放前必传)", lambda: (r_issue_proof.status_code == 200, f"status={r_issue_proof.status_code}"))

r_issue = client.post(f"/api/payments/advance/{adv_id}/issue", json={"issue_remark": "已发放到银行卡"}, headers=H_wangwu)
tc("备用金发放", "发放备用金", lambda: (r_issue.status_code == 200, f"ok"))

# ===================== 6. 报销完整流程 =====================
print("\n=== 6. 报销完整流程 (员工申请 → 总经理审批 → 财务复核 → 总经理审批) ===")

r_reim = client.post("/api/reimburses/", json={"title": "差旅报销-E2E测试", "expenseItemId": 9, "amount": 3500}, headers=H_zhangsan)
reim_id = r_reim.json()["data"]["id"] if r_reim.status_code == 200 else 0
tc("报销流程", "员工创建报销申请", lambda: (r_reim.status_code == 200, f"id={reim_id}"))

r_reim_sub = client.post(f"/api/reimburses/{reim_id}/submit", headers=H_zhangsan)
tc("报销流程", "员工提交报销", lambda: (r_reim_sub.status_code == 200, f"status={r_reim_sub.json()['data']['status']}"))

r_reim_appr_1 = client.post(f"/api/reimburses/{reim_id}/approve", json={"approvalStatus": "approved", "approval_opinion": "同意-总经理"}, headers=H_lisi)
tc("报销流程", "总经理通过(节点1)", lambda: (r_reim_appr_1.status_code == 200, f"ok"))

r_reim_appr_2 = client.post(f"/api/reimburses/{reim_id}/approve", json={"approvalStatus": "approved", "approval_opinion": "复核通过-财务"}, headers=H_wangwu)
tc("报销流程", "财务复核通过(节点2)", lambda: (r_reim_appr_2.status_code == 200, f"ok"))

r_reim_appr_3 = client.post(f"/api/reimburses/{reim_id}/approve", json={"approvalStatus": "approved", "approval_opinion": "最终批准-总经理"}, headers=H_lisi)
tc("报销流程", "总经理最终审批通过(节点3)", lambda: (r_reim_appr_3.status_code == 200, f"ok"))

reim_final = j(client.get(f"/api/reimburses/{reim_id}", headers=H_zhangsan))
tc("报销流程", "报销最终状态为approved", lambda: (reim_final["data"]["status"] == "approved", f"status={reim_final['data']['status']}"))

# ===================== 7. 驳回流程 =====================
print("\n=== 7. 驳回流程测试 ===")

r_adv2 = client.post("/api/advances/", json={"reason": "测试驳回", "amount": 1000}, headers=H_zhangsan)
adv2_id = r_adv2.json()["data"]["id"] if r_adv2.status_code == 200 else 0
tc("驳回流程", "员工创建备用金", lambda: (r_adv2.status_code == 200, f"id={adv2_id}"))

if adv2_id:
    client.post(f"/api/advances/{adv2_id}/submit", headers=H_zhangsan)
    r_rej = client.post(f"/api/advances/{adv2_id}/reject", json={"approvalStatus": "rejected", "approval_opinion": "金额不合理"}, headers=H_lisi)
    tc("驳回流程", "总经理驳回", lambda: (r_rej.status_code == 200, f"status={r_rej.json()['data']['status']}"))

    rej_detail = j(client.get(f"/api/advances/{adv2_id}", headers=H_zhangsan))
    tc("驳回流程", "状态为rejected", lambda: (rej_detail["data"]["status"] == "rejected", f"status={rej_detail['data']['status']}"))

# ===================== 8. 各角色权限 =====================
print("\n=== 8. 角色权限测试 ===")

tc("权限", "员工能访问我的申请", lambda: (
    client.get("/api/advances/my?page=1&pageSize=10", headers=H_zhangsan).status_code == 200,
    "ok"
))
tc("权限", "员工能创建备用金", lambda: (
    client.post("/api/advances/", json={"reason": "权限测试", "amount": 500}, headers=H_zhangsan).status_code == 200,
    "ok"
))

tc("权限", "总经理能查看所有备用金", lambda: (
    client.get("/api/advances/?page=1&pageSize=10", headers=H_lisi).status_code == 200,
    "ok"
))

tc("权限", "财务能查看待支付", lambda: (
    client.get("/api/payments/pending?page=1&pageSize=10", headers=H_wangwu).status_code == 200,
    "ok"
))
tc("权限", "财务能查看备用金发放列表", lambda: (
    client.get("/api/payments/advance-issues?page=1&pageSize=10", headers=H_wangwu).status_code == 200,
    "ok"
))

tc("权限", "员工不能查看待支付(无FINANCE角色)", lambda: (
    client.get("/api/payments/pending?page=1&pageSize=10", headers=H_zhangsan).status_code == 403,
    "403"
))

# ===================== 9. 列表/筛选/分页 =====================
print("\n=== 9. 列表筛选测试 ===")

tc("列表筛选", "备用金列表按状态筛选(pending)", lambda: (
    client.get("/api/advances/?page=1&pageSize=10&status=pending", headers=H_admin).status_code == 200,
    "ok"
))
tc("列表筛选", "备用金列表按状态筛选(approved)", lambda: (
    client.get("/api/advances/?page=1&pageSize=10&status=approved", headers=H_admin).status_code == 200,
    "ok"
))
tc("列表筛选", "报销列表分页", lambda: (
    j(client.get("/api/reimburses/?page=1&pageSize=2", headers=H_admin))["data"]["pageSize"] == 2,
    "pageSize=2"
))

# ===================== 10. 移动端API兼容性 =====================
print("\n=== 10. 移动端API兼容性 ===")

tc("移动端", "登录接口返回access_token", lambda: (
    "access_token" in j(client.post("/api/auth/login", json={"username": "zhangsan", "password": "test123"})).get("data", {}),
    "has_token"
))
tc("移动端", "me接口返回角色信息", lambda: (
    isinstance(j(client.get("/api/auth/me", headers=H_zhangsan))["data"].get("roleList"), list),
    "has_roleList"
))
tc("移动端", "备用金创建返回id", lambda: (
    "id" in j(client.post("/api/advances/", json={"reason": "移动端测试", "amount": 200}, headers=H_zhangsan)).get("data", {}),
    "has_id"
))
tc("移动端", "备用金详情包含全部字段", lambda: (
    all(k in j(client.get(f"/api/advances/{adv_id}", headers=H_admin))["data"] for k in ["id", "reason", "amount", "status", "applyCode", "currentNode"]),
    "all_fields_present"
))

# ===================== SUMMARY =====================
print("\n" + "=" * 60)
print("测试汇总")
print("=" * 60)

total_pass = total_fail = 0
for mod, cases in results.items():
    p = sum(1 for _, ok, _ in cases if ok)
    f = len(cases) - p
    total_pass += p
    total_fail += f
    print(f"  {mod}: {p}/{len(cases)} passed")
rate = total_pass / (total_pass + total_fail) * 100 if (total_pass + total_fail) > 0 else 0
print(f"\n  TOTAL: {total_pass}/{total_pass + total_fail} passed ({rate:.1f}%)")

END = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
report = f"# 多角色端到端测试报告\n\n> 执行时间: {START} ~ {END}\n\n"
report += f"## 汇总\n\n总用例: {total_pass + total_fail}, 通过: {total_pass}, 失败: {total_fail}, 通过率: {rate:.1f}%\n\n"
report += "| 模块 | 通过/总数 | 通过率 |\n|------|----------|-------|\n"
for mod, cases in results.items():
    p = sum(1 for _, ok, _ in cases if ok)
    r = p / len(cases) * 100 if cases else 0
    report += f"| {mod} | {p}/{len(cases)} | {r:.0f}% |\n"
report += "\n## 详细结果\n\n"
for mod, cases in results.items():
    report += f"\n### {mod}\n\n| 用例 | 状态 | 结果 |\n|------|------|------|\n"
    for name, ok, detail in cases:
        s = "✅" if ok else "❌"
        report += f"| {name} | {s} | {detail[:100]} |\n"
report += "\n## 测试用户\n\n| 用户名 | 密码 | 角色 |\n|--------|------|------|\n"
report += "| zhangsan | test123 | 员工(EMPLOYEE) |\n"
report += "| lisi | test123 | 总经理(GENERAL_MANAGER) |\n"
report += "| wangwu | test123 | 财务(FINANCE) |\n"
report += "| admin | admin123 | 管理员(ADMIN) |\n"
report += "\n## 测试覆盖范围\n\n"
report += "- 认证登录（多角色）\n"
report += "- 基础数据管理（用户/角色/菜单/费用项目/工作流）\n"
report += "- 备用金完整流程：创建→提交→审批(3节点)→发放\n"
report += "- 报销完整流程：创建→提交→审批(3节点)\n"
report += "- 驳回流程\n"
report += "- 角色权限（员工/总经理/财务）\n"
report += "- 列表筛选分页\n"
report += "- 移动端API兼容性\n"

with open("/Users/wuzilong/work/00code/shanxia/bx/报销业务/多角色端到端测试报告.md", "w") as f:
    f.write(report)
print(f"\n报告已写入: 报销业务/多角色端到端测试报告.md")