"""报表统计端到端验证（M4 ⑤：四张报表 + 打印视图）。 覆盖真实浏览器链路（接口测试守不住的部分）： - 「报表统计」组四页菜单可见、可进入 - 个人明细：API 预置的已核算业务出现在列表；导出触发真实下载 - 部门汇总：经理团队行出现且团队基数非 0；全公司 5% 卡片可见 - 业务统计：关键字筛选命中；三列统计非空 - 回款统计：8 档区间表；目标桶（61-90 天）有金额；空桶显示 0 - 打印视图：独立路由带出报表与签字栏；window.print 被调用 前置（缺一不可）： docker compose up -d cd backend && uv run uvicorn app.main:app --port 8000 cd frontend && pnpm dev cd frontend && E2E_ADMIN_PASSWORD=<开发库admin密码> python e2e/report_flow.py 数据说明：核算后数据不可删（业务规则），按 `e2e_` 前缀留档开发库； 每次运行用随机后缀隔离，断言以本次业务编号定位。 """ import json import os import time import urllib.error import urllib.request from decimal import Decimal from playwright.sync_api import Page, sync_playwright from common import ( BASE_URL, E2EFailure, collect_console_errors, current_path, fail, login, ) ADMIN = "admin" SUFFIX = f"{int(time.time())}" console_errors: list[str] = [] MONTH = time.strftime("%Y-%m") def _api(method: str, path: str, token: str, body: dict | None = None) -> dict: """带令牌的轻量 API 调用（urllib，避免额外依赖）。""" request = urllib.request.Request( f"{BASE_URL}{path}", method=method, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", }, data=json.dumps(body).encode() if body is not None else None, ) try: with urllib.request.urlopen(request, timeout=20) as response: payload = json.loads(response.read().decode()) except urllib.error.HTTPError as exc: detail = exc.read().decode()[:200] raise E2EFailure(f"API {method} {path} HTTP {exc.code}：{detail}") from exc if payload.get("code") != 200: raise E2EFailure(f"API {method} {path} 业务失败：{payload.get('message')}") return payload["data"] def step_login(page: Page, admin_password: str) -> None: login(page, ADMIN, admin_password) print("✓ 1. 登录") def step_setup_via_api(page: Page) -> dict: """预置：经理 + 业务员 → 客户/业务 → 发货（60 天前）→ 回款 → 核算。""" token = page.evaluate("localStorage.getItem('ys_token')") if not token: fail("API 预置", "localStorage 中未取到 ys_token") # 回款统计断言取增量：脚本可重复运行（上次失败/历史 e2e 数据会累计到桶） baseline = _api( "GET", f"/api/v1/reports/payment-stats?settlement_month={MONTH}", token ) baseline_bucket = next(b for b in baseline["buckets"] if b["code"] == "61-90") expected_receipt_amount = ( Decimal(str(baseline_bucket["receipt_amount"])) + Decimal("10000.00") ) roles = _api("GET", "/api/v1/system/business-roles?pageNum=1&pageSize=50", token)["list"] salesperson_role = next((r["id"] for r in roles if r["role_code"] == "salesperson"), None) manager_role = next((r["id"] for r in roles if r["role_code"] == "sales_manager"), None) if salesperson_role is None or manager_role is None: fail("API 预置", "未找到 salesperson / sales_manager 岗位种子") manager = _api( "POST", "/api/v1/base/personnel", token, { "personnel_name": f"e2e_报表现经理_{SUFFIX}", "business_role_id": manager_role, "salary_standard": "3000.00", }, ) person = _api( "POST", "/api/v1/base/personnel", token, { "personnel_name": f"e2e_报表现业务员_{SUFFIX}", "business_role_id": salesperson_role, "salary_standard": "3000.00", "manager_id": manager["id"], }, ) customer = _api( "POST", "/api/v1/base/customers", token, { "company_name": f"e2e_报表客户_{SUFFIX}", "owner_name": "张老板", "customer_type": "personal", }, ) business = _api( "POST", "/api/v1/business/businesses", token, { "customer_id": customer["id"], "business_date": f"{MONTH}-01", "owner_personnel_id": person["id"], "items": [ { "product_name": f"e2e_报表产品_{SUFFIX}", "product_category": "municipal_water", "quantity": "100", "sale_price": "100.00", "cost_amount": "3000.00", } ], }, ) from datetime import date, timedelta ship_date = (date.today() - timedelta(days=60)).isoformat() shipment = _api( "POST", "/api/v1/business/shipments", token, { "business_id": business["id"], "ship_date": ship_date, "items": [{"product_name": "e2e_报表发货", "quantity": "1", "sale_price": "1.00"}], }, ) receipt = _api( "POST", "/api/v1/business/receipts", token, { "business_id": business["id"], "receipt_date": date.today().isoformat(), "amount": "10000.00", "shipment_id": shipment["id"], }, ) _api("POST", f"/api/v1/calc/receipts/{receipt['id']}/calc", token) print(f"✓ 2. API 预置 业务 {business['business_no']} 已核算") return { "month": MONTH, "business_no": business["business_no"], "personnel_id": person["id"], "personnel_name": person["personnel_name"], "manager_name": manager["personnel_name"], "expected_receipt_amount": f"{expected_receipt_amount:.2f}", } def step_personal_detail(page: Page, ctx: dict) -> None: page.goto(f"{BASE_URL}/reports/personal-detail", wait_until="networkidle") page.wait_for_timeout(800) if current_path(page) != "/reports/personal-detail": fail("导航", f"未进入个人绩效明细页，当前 {page.url}") # 月份筛选：Element Plus 月份选择器可直接键入 YYYY-MM month_input = page.locator(".filters input").first month_input.fill(ctx["month"]) month_input.press("Enter") page.locator(".filters button:has-text('查询')").first.click() page.wait_for_timeout(900) row = page.locator(".el-table__row", has_text=ctx["business_no"]).first if row.count() == 0: fail("个人明细", f"未找到业务 {ctx['business_no']} 的明细行") print("✓ 3. 个人明细 列表命中本次核算业务") with page.expect_download(timeout=20000) as download_info: page.locator("button:has-text('导出 Excel')").first.click() download = download_info.value if not download.suggested_filename.endswith(".xlsx"): fail("导出", f"下载文件名非 xlsx：{download.suggested_filename}") print(f"✓ 4. 导出下载 {download.suggested_filename}") def step_department_summary(page: Page, ctx: dict) -> None: page.goto(f"{BASE_URL}/reports/department-summary", wait_until="networkidle") page.wait_for_timeout(900) page.locator(".filters button:has-text('查询')").first.click() page.wait_for_timeout(900) row = page.locator(".el-table__row", has_text=ctx["manager_name"]).first if row.count() == 0: fail("部门汇总", f"未找到经理 {ctx['manager_name']} 的团队行") if page.locator(".company-card").count() == 0: fail("部门汇总", "全公司 5% 卡片未渲染") print("✓ 5. 部门汇总 团队行与全公司卡片可见") def step_business_stats(page: Page, ctx: dict) -> None: page.goto(f"{BASE_URL}/reports/business-stats", wait_until="networkidle") page.wait_for_timeout(800) keyword = page.locator(".filters input").first keyword.fill(ctx["business_no"]) keyword.press("Enter") page.locator(".filters button:has-text('查询')").first.click() page.wait_for_timeout(800) row = page.locator(".el-table__row", has_text=ctx["business_no"]).first if row.count() == 0: fail("业务统计", f"未找到业务 {ctx['business_no']}") print("✓ 7. 业务统计 关键字筛选命中") def step_payment_stats(page: Page, ctx: dict) -> None: page.goto(f"{BASE_URL}/reports/payment-stats", wait_until="networkidle") page.wait_for_timeout(900) page.locator(".filters button:has-text('查询')").first.click() page.wait_for_timeout(900) # 页面同时有区间表与两个维度对比表，必须限定主区间表 rows = page.locator('[data-testid="payment-bucket-table"] .el-table__row') if rows.count() != 8: fail("回款统计", f"区间表应为 8 行，实际 {rows.count()}") bucket = page.locator( '[data-testid="payment-bucket-table"] .el-table__row', has_text="61-90 天" ).first if bucket.count() == 0: fail("回款统计", "未找到 61-90 天区间行") bucket_text = bucket.inner_text() expected = ctx["expected_receipt_amount"] if expected not in bucket_text: fail("回款统计", f"61-90 天桶回款金额不符（期望含 {expected}）：{bucket_text[:120]}") print("✓ 8. 回款统计 8 档区间与目标桶金额正确") def step_payment_multi_expand(page: Page, ctx: dict) -> None: """多个区间同时展开时，各自的下钻面板不能串行显示同一批明细。""" page.goto(f"{BASE_URL}/reports/payment-stats", wait_until="networkidle") page.wait_for_timeout(900) page.locator(".filters button:has-text('查询')").first.click() page.wait_for_timeout(900) for label in ("61-90 天", "1-30 天"): row = page.locator( '[data-testid="payment-bucket-table"] .el-table__row', has_text=label ).first row.locator(".el-table__expand-icon").click() page.wait_for_timeout(700) drills = page.locator(".drill-wrap") if drills.count() != 2: fail("回款多桶展开", f"应展开 2 个下钻面板，实际 {drills.count()}") texts = [drills.nth(index).inner_text() for index in range(2)] # 面板是明细行（每笔 10000），不是桶合计；串行时两个面板文本会完全相同 if texts[0] == texts[1] or "10000.00" not in "".join(texts): fail("回款多桶展开", "两个桶面板显示了同一批明细（drill 状态串行）") print("✓ 9. 回款多桶展开 两个下钻面板互不串行") def step_department_drill(page: Page, ctx: dict) -> None: """部门汇总「查看明细」必须把人员与月份带到个人明细页。""" page.goto(f"{BASE_URL}/reports/department-summary", wait_until="networkidle") page.wait_for_timeout(900) page.locator(".filters button:has-text('查询')").first.click() page.wait_for_timeout(900) row = page.locator(".el-table__row", has_text=ctx["manager_name"]).first row.locator(".el-table__expand-icon").click() page.wait_for_timeout(500) page.locator("button:has-text('查看明细')").first.click() page.wait_for_timeout(900) if f"personnel_id={ctx['personnel_id']}" not in page.url: fail("部门下钻", f"URL 未携带人员筛选：{page.url}") if ctx["personnel_name"] not in page.locator(".filters").inner_text(): fail("部门下钻", "个人明细页未应用人员筛选") print("✓ 6. 部门下钻 人员与月份筛选已生效") def step_print_truncated(page: Page, ctx: dict) -> None: """超过打印上限时必须给出截断提示（修复前提示永不出现）。""" def handler(route) -> None: rows = [ { "ledger_id": index, "business_no": f"E2E-{index}", "ship_date": None, "receipt_date": f"{ctx['month']}-01", "payment_days": 10, "product_category": "municipal_water", "customer_type": "personal", "role_name": "业务员", "rate_text": "1", "weight_r": 1, "performance_amount": "1.00", "payable_amount": "1.00", } for index in range(200) ] route.fulfill( json={ "code": 200, "message": "ok", "data": { "list": rows, "total": 2500, "pageNum": 1, "pageSize": 200, "summary": {"row_count": 2500}, }, } ) page.route("**/api/v1/reports/personal-detail*", handler) try: page.goto( f"{BASE_URL}/reports/print/personal-detail?settlement_month={ctx['month']}", wait_until="networkidle", ) page.wait_for_timeout(1200) if page.locator(".truncated").count() == 0: fail("打印截断", "超过 2000 行未显示截断提示") if "2000" not in page.locator(".truncated").inner_text(): fail("打印截断", "截断提示未说明上限行数") finally: page.unroute("**/api/v1/reports/personal-detail*", handler) print("✓ 11. 打印截断 超限提示可见") def step_print_department(page: Page, ctx: dict) -> None: """部门汇总打印稿必须含团队分成、5% 块、对账、成员与未分配五块。""" page.goto( f"{BASE_URL}/reports/print/department-summary?settlement_month={ctx['month']}", wait_until="networkidle", ) page.wait_for_timeout(1200) for testid in ( "print-department-teams", "print-company-block", "print-reconciliation", "print-department-members", "print-unassigned", ): if page.locator(f'[data-testid="{testid}"]').count() == 0: fail("部门打印", f"缺少区块 {testid}") if "财务复核" not in page.locator(".print-page").inner_text(): fail("部门打印", "签字栏缺失") print("✓ 12. 部门打印 五块结构与签字栏完整") def step_print_view(page: Page, ctx: dict) -> None: page.add_init_script("window.print = () => { window.__printed = true }") page.goto( f"{BASE_URL}/reports/print/personal-detail?settlement_month={ctx['month']}", wait_until="networkidle", ) page.wait_for_timeout(1200) if page.locator(".sheet-table").count() == 0: fail("打印视图", "报表表格未渲染") text = page.locator(".print-page").inner_text() if "财务复核" not in text or "领导签字" not in text: fail("打印视图", "签字栏缺失") page.locator("button:has-text('打印')").first.click() page.wait_for_timeout(400) if page.evaluate("window.__printed") is not True: fail("打印视图", "window.print 未被调用") print("✓ 10. 打印视图 表格/签字栏/打印调用均正常") def run(page: Page) -> None: collect_console_errors(page, console_errors) admin_password = os.environ.get("E2E_ADMIN_PASSWORD", "") if not admin_password: fail("环境变量", "未设置 E2E_ADMIN_PASSWORD（开发库 admin 密码）") step_login(page, admin_password) ctx = step_setup_via_api(page) step_personal_detail(page, ctx) step_department_summary(page, ctx) step_department_drill(page, ctx) step_business_stats(page, ctx) step_payment_stats(page, ctx) step_payment_multi_expand(page, ctx) step_print_view(page, ctx) step_print_truncated(page, ctx) step_print_department(page, ctx) if console_errors: print(f" （控制台错误 {len(console_errors)} 条：{console_errors[:2]}）") def main() -> int: with sync_playwright() as playwright: browser = playwright.chromium.launch(headless=True) page = browser.new_page(viewport={"width": 1600, "height": 1000}) page.set_default_timeout(15000) try: run(page) except E2EFailure as exc: page.screenshot(path="e2e/failure.png", full_page=True) print(f"\n ✗ {exc}") print(" 已保存截图 e2e/failure.png") return 1 except Exception as exc: # noqa: BLE001 — 测试脚本需打印完整失败原因 page.screenshot(path="e2e/failure.png", full_page=True) print(f"\n ✗ 执行中断：{type(exc).__name__}: {exc}") return 1 finally: browser.close() print("\n全部通过。") return 0 if __name__ == "__main__": raise SystemExit(main())