crm-backend-matt/.scratch/opportunity-verify/webapp_test.py

"""
商机模块 webapp-testing：用 Playwright 跑前端验收页 + e2e API 测试，截图存档。
"""
import sys
import io
import json
import time
from playwright.sync_api import sync_playwright

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')

FRONTEND_DIR = "D:/code/crm-backend-matt/.scratch/opportunity-verify/frontend"
BASE_URL = "http://localhost:8080"
TOKEN_URL = f"{BASE_URL}/api/auth/debug/token?userId=739564171091247104"
SCREENSHOT_DIR = "D:/code/crm-backend-matt/.scratch/opportunity-verify"

results = []

def log(status, name, detail=""):
    icon = "[OK]  " if status else "[FAIL]"
    results.append((status, name, detail))
    print(f"{icon} {name}" + (f" | {detail}" if detail else ""))


with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    ctx = browser.new_context()
    page = ctx.new_page()

    console_errors = []
    page.on("console", lambda msg: console_errors.append(msg.text) if msg.type == "error" else None)
    page.on("pageerror", lambda err: console_errors.append(str(err)))

    # ── 1. 骨架自检页 index.html ──────────────────────────────────
    page.goto(f"file:///{FRONTEND_DIR}/index.html")
    page.wait_for_load_state("networkidle")
    page.screenshot(path=f"{SCREENSHOT_DIR}/webapp-01-index.png", full_page=True)
    log(True, "index.html 加载无崩溃")

    # 点击自检按钮
    page.click("button")
    page.wait_for_timeout(4000)
    page.screenshot(path=f"{SCREENSHOT_DIR}/webapp-02-index-check.png", full_page=True)

    assertions = page.locator("#assertions li").all_text_contents()
    for a in assertions:
        ok = "PASS" in a or "✅" in a or "pass" in a.lower()
        # 兼容各种格式，只要不含 FAIL/fail 就算通过
        ok = "FAIL" not in a and "fail" not in a.lower() and "❌" not in a
        log(ok, f"index.html 自检: {a[:80]}")

    result_text = page.locator("#result").text_content()
    log(bool(result_text and "等待" not in result_text), "index.html 结果框已渲染", result_text[:60])

    # ── 2. 列表族页 03-list.html ──────────────────────────────────
    console_errors.clear()
    page.goto(f"file:///{FRONTEND_DIR}/03-list.html")
    page.wait_for_load_state("networkidle")
    page.screenshot(path=f"{SCREENSHOT_DIR}/webapp-03-list.png", full_page=True)
    log(True, "03-list.html 加载无崩溃")

    # 找并点击运行按钮
    btns = page.locator("button").all()
    if btns:
        btns[0].click()
        page.wait_for_timeout(4000)
        page.screenshot(path=f"{SCREENSHOT_DIR}/webapp-04-list-result.png", full_page=True)
        log(True, "03-list.html 运行按钮已点击")

    list_errors = [e for e in console_errors if "error" in e.lower() or "failed" in e.lower()]
    log(len(list_errors) == 0, "03-list.html 无 console 错误",
        f"{len(list_errors)} 个错误" if list_errors else "")

    # ── 3. 直接调 API 验证关键接口返回结构 ──────────────────────────
    import urllib.request
    import urllib.parse

    def api_get(path, params=None):
        url = BASE_URL + path
        if params:
            url += "?" + urllib.parse.urlencode(params)
        req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
        with urllib.request.urlopen(req, timeout=10) as r:
            return json.loads(r.read())

    def api_post(path, data):
        url = BASE_URL + path
        body = urllib.parse.urlencode(data).encode()
        req = urllib.request.Request(url, data=body, headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/x-www-form-urlencoded"
        })
        with urllib.request.urlopen(req, timeout=10) as r:
            return json.loads(r.read())

    # 先拿 token
    with urllib.request.urlopen(TOKEN_URL, timeout=5) as r:
        token = json.loads(r.read())["data"]
    log(bool(token), "获取 debug token", token[:20] + "...")

    # 核心接口结构检查
    OPP_ID = 999999999001

    d = api_get("/api/opportunity/detail", {"id": OPP_ID})
    log(d.get("code") == 0, "GET /detail code=0")
    log("oppName" in (d.get("data") or {}), "GET /detail data.oppName 存在")

    d = api_post("/api/opportunity/page", {"pageNum": 1, "pageSize": 5})
    log(d.get("code") == 0, "POST /page code=0")
    data_page = d.get("data") or {}
    log("content" in data_page, "POST /page data.content 存在（PageResult）",
        f"keys={list(data_page.keys())[:6]}")
    log("total" in data_page, "POST /page data.total 存在")

    d = api_post("/api/opportunity/follow/page", {"oppId": OPP_ID, "pageNum": 1, "pageSize": 5})
    log(d.get("code") == 0, "POST /follow/page code=0")
    fp = d.get("data") or {}
    log("content" in fp and "total" in fp, "POST /follow/page PageResult 结构",
        f"keys={list(fp.keys())[:6]}")

    d = api_post("/api/opportunity/oplog/page", {"oppId": OPP_ID, "pageNum": 1, "pageSize": 5})
    log(d.get("code") == 0, "POST /oplog/page code=0")
    op = d.get("data") or {}
    log("content" in op and "total" in op, "POST /oplog/page PageResult 结构")

    d = api_post("/api/rule/opp-stage-template/page", {"pageNum": 1, "pageSize": 5})
    log(d.get("code") == 0, "POST /rule/opp-stage-template/page code=0")

    browser.close()

# ── 汇总 ──────────────────────────────────────────────────────────
passed = sum(1 for ok, _, _ in results if ok)
failed = sum(1 for ok, _, _ in results if not ok)
print(f"\n{'='*60}")
print(f"webapp-testing 结果：OK={passed}  FAIL={failed}  共{len(results)}条")
print(f"{'='*60}")
if failed:
    print("\n失败项：")
    for ok, name, detail in results:
        if not ok:
            print(f"  [FAIL] {name}" + (f" | {detail}" if detail else ""))