"""D2 业务员移动开单 H5 全流程验证（Chrome 移动模拟）。 跑法: python scripts/verify_salesman_h5.py 前置: 后端 dev 在 127.0.0.1:18099, vite(verify配置) 在 127.0.0.1:15173 流程: 登录(alice/default) → 开单tab(客户/仓库/商品/单位/取价/数量/抹零/保存过账) → 欠款tab(未结单+余额) → 收款tab(登记收款单) → 断言后端落库 """ import asyncio import json import sys import urllib.request BASE = "http://127.0.0.1:18099" PAGE = "http://127.0.0.1:15173/static/#/salesman" # D2收尾：残留隔离 —— 每次运行用时间戳唯一收款单号 + 前后计数断言， # 不再依赖固定 SMH50002（重复跑会撞唯一约束），且不删单（过账有副作用）。 import time as _time RUN_TAG = _time.strftime("%m%d%H%M%S") RECEIPT_NO = f"SMH{ RUN_TAG }" def api(method, path, token=None, tenant="default", body=None): req = urllib.request.Request( BASE + path, data=json.dumps(body).encode() if body is not None else None, method=method, headers={"Content-Type": "application/json"}, ) if token: req.add_header("Authorization", f"Bearer {token}") req.add_header("X-Tenant-Id", tenant) try: with urllib.request.urlopen(req, timeout=20) as r: return r.status, json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read() or b"{}") except Exception: return e.code, {} async def main(): from playwright.async_api import async_playwright errors = [] # --- 后端预检: 登录拿 token --- import urllib.error # noqa s, login = api("POST", "/api/v1/auth/token/", body={"username": "alice", "password": "alice12345"}) assert s == 200, f"login failed: {s} {login}" token = login["access"] s, custs = api("GET", "/api/v1/partner/customers/?page_size=200", token=token) assert s == 200 and custs["count"] >= 1, f"customers: {s}" # 前后计数基线：验证前后差值断言（残留隔离，不删单） s, _b0 = api("GET", "/api/v1/sales/bills/?page_size=1", token=token) assert s == 200, f"bills baseline: {s}" bills_before = _b0["count"] s, _r0 = api("GET", "/api/v1/finance/receipts/?page_size=1", token=token) assert s == 200, f"receipts baseline: {s}" receipts_before = _r0["count"] print(f"[pre] baseline bills={bills_before} receipts={receipts_before} " f"run_tag={RUN_TAG} receipt_no={RECEIPT_NO}") s, whs = api("GET", "/api/v1/inventory/warehouses/?page_size=200", token=token) assert s == 200 and whs["count"] >= 1, f"warehouses: {s}" s, prods = api("GET", "/api/v1/catalog/products/?page_size=200", token=token) assert s == 200 and prods["count"] >= 1, f"products: {s}" customer = custs["results"][0] warehouse = whs["results"][0] product = prods["results"][0] print(f"[pre] customer={customer['code']} warehouse={warehouse['code']} " f"product={product['code']} token=ok") async with async_playwright() as pw: browser = await pw.chromium.launch() # Chrome 移动模拟: Pixel 5 视口 + 触摸 + 移动 UA ctx = await browser.new_context( viewport={"width": 393, "height": 851}, has_touch=True, is_mobile=True, user_agent=("Mozilla/5.0 (Linux; Android 13; Pixel 5) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0 Mobile Safari/537.36"), ) page = await ctx.new_page() page.on("console", lambda m: errors.append(m.text) if m.type == "error" else None) page.on("pageerror", lambda e: errors.append(str(e))) await page.goto(PAGE, wait_until="networkidle") # --- 登录 ---（密码框无 placeholder，用 input 序号定位） inputs = page.locator(".sm-login input") await inputs.nth(0).fill("alice") await inputs.nth(1).fill("alice12345") await page.get_by_role("button", name="登 录").click() await page.wait_for_timeout(3000) body = await page.content() assert "外勤开单" in body or "开单" in body, "login did not land" print("[1] login ok (mobile viewport 393x851)") # --- 开单: 选客户/仓库 --- # 移动视口下 el-option 常判定不可见，用 JS 直点可见 option async def pick(select_idx, opt_idx=0): sels = page.locator(".el-select") await sels.nth(select_idx).click() await page.wait_for_timeout(900) txt = await page.evaluate("""(idx) => { const opts = document.querySelectorAll('li[role=option]'); const vis = [...opts].filter(o => o.offsetParent !== null); if (!vis.length || idx >= vis.length) return 'NONE total='+opts.length; vis[idx].click(); return 'OK:' + vis[idx].textContent.trim().slice(0, 30); }""", opt_idx) await page.wait_for_timeout(1200) return txt print("[2] customer:", await pick(0, 0)) print("[2] warehouse:", await pick(1, 0)) # 搜商品并加一行: 取第一行的数量框填 2, 点"加" await page.wait_for_timeout(2000) add_btns = page.get_by_role("button", name="加", exact=True) n_add = await add_btns.count() assert n_add >= 1, "no product rows rendered" qty_boxes = page.locator(".sm-item .el-input-number input") await qty_boxes.first.fill("2") await add_btns.first.click() await page.wait_for_timeout(800) order_title = await page.locator(".sm-order-title").count() assert order_title == 1, "order section not shown after add" print("[3] add-to-order ok (qty=2)") # 抹零选抹角 + 保存并过账 await page.locator(".sm-totals .el-select").click() await page.wait_for_timeout(700) print("[4] round_to:", await page.evaluate("""() => { const opts = document.querySelectorAll('li[role=option]'); const vis = [...opts].filter(o => o.offsetParent !== null); if (vis.length < 3) return 'NONE total='+opts.length; vis[2].click(); return 'OK:' + vis[2].textContent.trim().slice(0, 20); }""")) await page.wait_for_timeout(500) await page.get_by_role("button", name="保存并过账").click() await page.wait_for_timeout(4000) print("[4] submit+confirm clicked") # --- 欠款 tab ---（tab 项是 div#tab-xxx，直接点） await page.locator("#tab-debt").click() await page.wait_for_timeout(2500) debt_body = await page.locator("#pane-debt").text_content() print(f"[5] debt tab rendered: {(debt_body or '').strip()[:100]}") assert "未结" in (debt_body or "") or "暂无欠款" in (debt_body or ""), \ f"debt pane unexpected: {(debt_body or '')[:100]}" # --- 收款 tab: 填单登记 --- await page.locator("#tab-receipt").click() await page.wait_for_timeout(1000) # 收款面板内第一个 select = 客户 cust_sel = page.locator("#pane-receipt .el-select").first await cust_sel.scroll_into_view_if_needed() await cust_sel.click() await page.wait_for_timeout(900) print("[6] receipt customer:", await page.evaluate("""() => { const opts = document.querySelectorAll('li[role=option]'); const vis = [...opts].filter(o => o.offsetParent !== null); if (!vis.length) return 'NONE'; vis[0].click(); return 'OK:' + vis[0].textContent.trim().slice(0, 30); }""")) await page.wait_for_timeout(500) # 关掉可能残留的下拉浮层后再填单号（Esc 关闭 popper） await page.keyboard.press("Escape") await page.wait_for_timeout(400) bill_no = RECEIPT_NO await page.locator("#pane-receipt input[placeholder]").nth(0).scroll_into_view_if_needed() await page.locator("#pane-receipt input[placeholder]").nth(0).fill(bill_no) amt_input = page.locator("#pane-receipt .el-input-number input") await amt_input.fill("50") await page.get_by_role("button", name="登记收款").click() await page.wait_for_timeout(3000) print("[6] receipt submit clicked") errs = [e for e in errors if "favicon" not in e.lower()] print(f"[7] console errors: {len(errs)}") for e in errs[:10]: print(" ERR:", e[:200]) await browser.close() # --- 后端断言落库 --- s, bills = api("GET", "/api/v1/sales/bills/?page_size=5", token=token) assert s == 200, f"bills: {s}" latest = bills["results"][0] if bills["results"] else None print(f"[8] latest bill: {latest['bill_no'] if latest else None} " f"state={latest['state'] if latest else None} " f"total={latest['total_amount'] if latest else None}") assert latest and latest["state"] == "confirmed", \ f"expected confirmed bill, got {latest}" s, found = api("GET", f"/api/v1/finance/receipts/?page_size=50&search={RECEIPT_NO}", token=token) assert s == 200, f"receipts: {s}" hit = [r for r in found["results"] if r["bill_no"] == RECEIPT_NO] assert hit, f"receipt {RECEIPT_NO} not found: {[r['bill_no'] for r in found['results'][:5]]}" print(f"[9] receipt {RECEIPT_NO} ok amount={hit[0]['amount']}") # 残留隔离断言：本次恰好新增 1 单 + 1 收款（可重复跑，不删单） s, _b1 = api("GET", "/api/v1/sales/bills/?page_size=1", token=token) assert s == 200 and _b1["count"] == bills_before + 1, \ f"bills delta: before={bills_before} after={_b1['count']}" s, _r1 = api("GET", "/api/v1/finance/receipts/?page_size=1", token=token) assert s == 200 and _r1["count"] == receipts_before + 1, \ f"receipts delta: before={receipts_before} after={_r1['count']}" print(f"[9b] delta ok: bills {bills_before}->{_b1['count']}, " f"receipts {receipts_before}->{_r1['count']}") s, debts = api("GET", "/api/v1/finance/receivables/?page_size=50", token=token) assert s == 200 open_debts = [d for d in debts["results"] if d["status"] in ("open", "partial")] assert open_debts, "expected open receivables" print(f"[10] open receivables: {len(open_debts)}, " f"first balance={open_debts[0]['balance']}") real_errors = [e for e in errors if "favicon" not in e.lower()] assert not real_errors, f"console errors: {real_errors[:5]}" print("ALL SALESMAN-H5 CHECKS PASSED") if __name__ == "__main__": sys.exit(asyncio.run(main()) or 0)