test(e2e): 绩效核算与月度结算端到端用例——台账重算与月结审批链路
覆盖台账页按回款 ID 重算、版本筛选三态、月结生成/提交审批/明细回显， 用随机远未来月份造数，重复运行互不碰撞。 Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
liuwenxin
parent
c5f219423b
commit
fb398c3835
353
frontend/e2e/calc_settlement_flow.py
Normal file
353
frontend/e2e/calc_settlement_flow.py
Normal file
@ -0,0 +1,353 @@
|
||||
"""绩效核算与月度结算端到端验证（M3 ④：台账/月结两页）。
|
||||
|
||||
覆盖真实浏览器链路（重点守 code-review 修复过的两处前端缺陷）：
|
||||
- 台账页「重算」对话框按**回款 ID** 发起重算（修复前传台账行主键，会
|
||||
作废并重算无关回款的台账）；重算后新版本生效、旧版本翻「已替代」
|
||||
- 版本筛选三态：仅当前生效（1）/ 仅历史版本（0）/ 清空=全部（不过滤）
|
||||
（修复前「含历史版本」传 0 只回历史行）
|
||||
- 月结页：生成快照（草稿）→ 提交审批（自动通过并锁定）→ 详情抽屉展示
|
||||
按人明细（净工资/试用期），锁定人回显
|
||||
|
||||
前置（API 预置；前端无「核算」入口——核算为批量动作走接口）：
|
||||
客户/人员/业务/发货/回款经 HTTP API 直建（与集成测试同形入参），核算经
|
||||
API 触发；浏览器只负责台账/月结两页的真实交互。
|
||||
|
||||
数据说明：核算后的回款与已生成月结按设计不可删（变更只走调整单），
|
||||
本流程按 `e2e_` 前缀 + **随机远未来月份（209x）**造数——每次运行独占
|
||||
一个月份，重复运行互不碰撞；行数据留档开发库（与 business_data.py 同）。
|
||||
|
||||
运行前置（缺一不可）：
|
||||
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/calc_settlement_flow.py
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
from playwright.sync_api import Page, sync_playwright
|
||||
|
||||
from common import (
|
||||
BASE_URL,
|
||||
E2EFailure,
|
||||
collect_console_errors,
|
||||
current_path,
|
||||
expect_toast,
|
||||
fail,
|
||||
login,
|
||||
)
|
||||
|
||||
ADMIN = "admin"
|
||||
API_BASE = "http://127.0.0.1:8000"
|
||||
SUFFIX = f"{int(time.time())}"
|
||||
console_errors: list[str] = []
|
||||
|
||||
_SETTLEMENT_SWITCH = "approval.monthly_settlement.enabled"
|
||||
|
||||
|
||||
def _api(method: str, path: str, token: str, body: dict | None = None) -> dict:
|
||||
"""直连后端接口（仅用于 API 预置；浏览器交互一律走页面）。"""
|
||||
request = urllib.request.Request(
|
||||
f"{API_BASE}{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 _set_settlement_switch(token: str, value: str) -> None:
|
||||
"""显式控制月结审批开关（用例不依赖种子默认值；结束还原默认 1）。"""
|
||||
_api(
|
||||
"POST",
|
||||
"/api/v1/system/params/changes",
|
||||
token,
|
||||
{
|
||||
"change_type": 1,
|
||||
"remark": f"e2e calc_settlement_flow 设置月结审批开关={value}",
|
||||
"items": [
|
||||
{"param_code": _SETTLEMENT_SWITCH, "dimension_value": "", "value": value}
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _filter_item(page: Page, label: str):
|
||||
"""可见筛选表单中指定标签的表单项（select 带 :visible 避开隐藏实例）。"""
|
||||
return page.locator(
|
||||
f'.filters:visible .el-form-item:has(.el-form-item__label:text-is("{label}"))'
|
||||
).first
|
||||
|
||||
|
||||
def _query(page: Page) -> None:
|
||||
page.locator(".filters:visible button:has-text('查询')").first.click()
|
||||
page.wait_for_timeout(800)
|
||||
|
||||
|
||||
def _select_filter(page: Page, label: str, option: str) -> None:
|
||||
item = _filter_item(page, label)
|
||||
item.locator(".el-select").click()
|
||||
page.wait_for_timeout(300)
|
||||
page.locator(
|
||||
".el-select-dropdown:visible .el-select-dropdown__item", has_text=option
|
||||
).first.click()
|
||||
page.wait_for_timeout(200)
|
||||
|
||||
|
||||
def step_login_and_open_ledger(page: Page, admin_password: str) -> None:
|
||||
login(page, ADMIN, admin_password)
|
||||
page.goto(f"{BASE_URL}/calc/ledger", wait_until="networkidle")
|
||||
page.wait_for_timeout(600)
|
||||
if current_path(page) != "/calc/ledger":
|
||||
fail("导航", f"未进入绩效台账页，当前 {page.url}")
|
||||
print("✓ 1. 登录与导航 进入绩效台账页")
|
||||
|
||||
|
||||
def step_setup_via_api(page: Page, month: str) -> dict:
|
||||
"""API 预置完整链路：角色 → 客户/人员 → 业务 → 发货 → 回款 → 核算。"""
|
||||
token = page.evaluate("localStorage.getItem('ys_token')")
|
||||
if not token:
|
||||
fail("API 预置", "localStorage 中未取到 ys_token")
|
||||
|
||||
roles = _api("GET", "/api/v1/system/business-roles?pageNum=1&pageSize=50", token)["list"]
|
||||
role_id = next((r["id"] for r in roles if r["role_code"] == "salesperson"), None)
|
||||
if role_id is None:
|
||||
fail("API 预置", "未找到 salesperson 岗位种子")
|
||||
|
||||
customer = _api(
|
||||
"POST",
|
||||
"/api/v1/base/customers",
|
||||
token,
|
||||
{
|
||||
"company_name": f"e2e_月结客户_{SUFFIX}",
|
||||
"owner_name": "张老板",
|
||||
"customer_type": "personal",
|
||||
},
|
||||
)
|
||||
person = _api(
|
||||
"POST",
|
||||
"/api/v1/base/personnel",
|
||||
token,
|
||||
{
|
||||
"personnel_name": f"e2e_月结人员_{SUFFIX}",
|
||||
"business_role_id": role_id,
|
||||
"salary_standard": "3000.00",
|
||||
},
|
||||
)
|
||||
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",
|
||||
}
|
||||
],
|
||||
},
|
||||
)
|
||||
shipment = _api(
|
||||
"POST",
|
||||
"/api/v1/business/shipments",
|
||||
token,
|
||||
{
|
||||
"business_id": business["id"],
|
||||
"ship_date": f"{month}-02",
|
||||
"items": [{"product_name": "e2e_月结发货", "quantity": "1", "sale_price": "1.00"}],
|
||||
},
|
||||
)
|
||||
receipt = _api(
|
||||
"POST",
|
||||
"/api/v1/business/receipts",
|
||||
token,
|
||||
{
|
||||
"business_id": business["id"],
|
||||
"receipt_date": f"{month}-15",
|
||||
"amount": "20000.00",
|
||||
"shipment_id": shipment["id"],
|
||||
},
|
||||
)
|
||||
calc = _api("POST", f"/api/v1/calc/receipts/{receipt['id']}/calc", token)
|
||||
if calc.get("repeated") is not False:
|
||||
fail("API 预置", f"核算应为首算，实际 repeated={calc.get('repeated')}")
|
||||
print(f"✓ 2. API 预置 业务 {business['business_no']} 已核算（{month}）")
|
||||
return {
|
||||
"token": token,
|
||||
"month": month,
|
||||
"business_no": business["business_no"],
|
||||
"personnel_name": person["personnel_name"],
|
||||
}
|
||||
|
||||
|
||||
def step_ledger_row_visible(page: Page, ctx: dict) -> None:
|
||||
_filter_item(page, "归属月").locator("input").first.fill(ctx["month"])
|
||||
_query(page)
|
||||
row = page.locator(".el-table__row", has_text=ctx["business_no"]).first
|
||||
if row.count() == 0:
|
||||
fail("台账列表", f"筛选 {ctx['month']} 未见业务 {ctx['business_no']} 的台账行")
|
||||
text = row.text_content()
|
||||
if "v1" not in text or "已替代" in text:
|
||||
fail("台账列表", f"首算行应为当前生效 v1，实际：「{text.strip()[:80]}」")
|
||||
print("✓ 3. 台账筛选 首算 v1 行可见且为当前生效")
|
||||
|
||||
|
||||
def step_recalc_via_dialog(page: Page, ctx: dict) -> None:
|
||||
"""重算对话框：修复前按台账行主键发起，会作废无关回款的台账。"""
|
||||
row = page.locator(".el-table__row", has_text=ctx["business_no"]).first
|
||||
row.locator("button:has-text('重算')").click()
|
||||
dialog = page.locator(".el-dialog:visible").last
|
||||
dialog.wait_for(state="visible")
|
||||
dialog.locator("textarea").fill(f"e2e 重算验证 {SUFFIX}")
|
||||
dialog.locator("button:has-text('确认重算')").click()
|
||||
expect_toast(page, "重算完成", "重算")
|
||||
page.wait_for_timeout(600)
|
||||
|
||||
row = page.locator(".el-table__row", has_text=ctx["business_no"]).first
|
||||
text = row.text_content()
|
||||
if "v2" not in text:
|
||||
fail("重算", f"重算后应显示 v2，实际：「{text.strip()[:80]}」")
|
||||
print("✓ 4. 重算对话框 按回款 ID 重算成功，v2 生效")
|
||||
|
||||
|
||||
def step_version_filter(page: Page, ctx: dict) -> None:
|
||||
# 仅历史版本：0 必须精确命中历史行（修复前「含历史版本」即传 0，语义相反）
|
||||
_select_filter(page, "版本", "仅历史版本")
|
||||
_query(page)
|
||||
rows = page.locator(".el-table__row", has_text=ctx["business_no"])
|
||||
if rows.count() != 1 or "已替代" not in rows.first.text_content():
|
||||
fail("版本筛选", "「仅历史版本」应只剩一条带「已替代」的 v1 行")
|
||||
|
||||
# 清空 = 不过滤（全部版本）：v1+v2 双版本可见
|
||||
item = _filter_item(page, "版本")
|
||||
item.hover()
|
||||
clear = item.locator(".el-select__clear").first
|
||||
if clear.count() == 0:
|
||||
fail("版本筛选", "未找到下拉清空按钮")
|
||||
clear.click()
|
||||
page.wait_for_timeout(200)
|
||||
_query(page)
|
||||
rows = page.locator(".el-table__row", has_text=ctx["business_no"])
|
||||
texts = [rows.nth(i).text_content() for i in range(rows.count())]
|
||||
if rows.count() != 2 or not any("已替代" in t for t in texts):
|
||||
fail("版本筛选", f"清空后应见 v1+v2 两行，实际 {rows.count()} 行：{texts}")
|
||||
print("✓ 5. 版本筛选 仅历史/全部三态正确（v1 已替代 + v2 生效）")
|
||||
|
||||
|
||||
def step_generate_settlement(page: Page, ctx: dict) -> None:
|
||||
page.goto(f"{BASE_URL}/settlements", wait_until="networkidle")
|
||||
page.wait_for_timeout(600)
|
||||
page.locator('input[placeholder="结算月份 YYYY-MM"]').fill(ctx["month"])
|
||||
page.click("button:has-text('生成月结快照')")
|
||||
expect_toast(page, "月结已生成", "生成月结")
|
||||
|
||||
row = page.locator(".el-table__row", has_text=ctx["month"]).first
|
||||
row.wait_for(state="visible")
|
||||
text = row.text_content()
|
||||
if "草稿" not in text:
|
||||
fail("月结生成", f"生成后应为草稿状态，实际：「{text.strip()[:80]}」")
|
||||
print("✓ 6. 月结生成 快照生成，草稿状态")
|
||||
|
||||
|
||||
def step_submit_and_locked(page: Page, ctx: dict) -> None:
|
||||
row = page.locator(".el-table__row", has_text=ctx["month"]).first
|
||||
row.locator("button:has-text('提交审批')").click()
|
||||
# 开关已显式关闭（见 run 的 finally 还原）→ 自动通过并锁定
|
||||
expect_toast(page, "月结已锁定", "提交审批")
|
||||
page.wait_for_timeout(600)
|
||||
row = page.locator(".el-table__row", has_text=ctx["month"]).first
|
||||
if "已审定" not in row.text_content():
|
||||
fail("提交审批", "提交后应显示已审定（锁定）")
|
||||
print("✓ 7. 提交审批 自动通过并锁定（已审定）")
|
||||
|
||||
|
||||
def step_detail_drawer(page: Page, ctx: dict) -> None:
|
||||
row = page.locator(".el-table__row", has_text=ctx["month"]).first
|
||||
row.locator("button:has-text('详情')").click()
|
||||
drawer = page.locator(".el-drawer:visible").last
|
||||
drawer.wait_for(state="visible")
|
||||
page.wait_for_timeout(500)
|
||||
text = drawer.text_content()
|
||||
if ctx["month"] not in text or "已审定" not in text:
|
||||
fail("月结详情", f"抽屉应含月份与已审定状态，实际：「{text.strip()[:120]}」")
|
||||
if ctx["personnel_name"] not in text:
|
||||
fail("月结详情", f"明细应含本次人员 {ctx['personnel_name']}")
|
||||
if "净工资" not in text:
|
||||
fail("月结详情", "明细应含净工资列")
|
||||
print("✓ 8. 月结详情 抽屉明细含人员/净工资/锁定状态")
|
||||
|
||||
|
||||
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 密码）")
|
||||
|
||||
# 随机远未来月份：每次运行独占一月，重复运行互不碰撞（数据留档）
|
||||
month = f"20{random.randint(90, 99)}-{random.randint(1, 12):02d}"
|
||||
|
||||
step_login_and_open_ledger(page, admin_password)
|
||||
ctx = step_setup_via_api(page, month)
|
||||
_set_settlement_switch(ctx["token"], "0")
|
||||
try:
|
||||
step_ledger_row_visible(page, ctx)
|
||||
step_recalc_via_dialog(page, ctx)
|
||||
step_version_filter(page, ctx)
|
||||
step_generate_settlement(page, ctx)
|
||||
step_submit_and_locked(page, ctx)
|
||||
step_detail_drawer(page, ctx)
|
||||
finally:
|
||||
# 还原审批开关默认（V8 起默认开启），中断也不遗留
|
||||
_set_settlement_switch(ctx["token"], "1")
|
||||
|
||||
print(f" （本次数据留档：{month} 已锁月结 + 客户/人员/业务 e2e_ 前缀，不可删）")
|
||||
if console_errors:
|
||||
fail("控制台", f"存在 {len(console_errors)} 条 JS 错误：{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())
|
||||
Loading…
Reference in New Issue
Block a user