path: root/tools/acceptance/e2e_acceptance.py

#!/usr/bin/env python3
"""Checked-in browser acceptance driver for the live pipeline viewport.

Starts a Release server on an isolated port with its own artifact root, drives
a real chromium browser through a live pipeline run (desktop and narrow
viewports), and asserts actual on-screen geometry/framing via the projected
bounding box of the loaded GLB - not just labels or whole-canvas hashes.

Playwright is provisioned into tools/acceptance/deps (repo-local, never
system/user site-packages). First run may download chromium into the default
user browser cache via `playwright install chromium`.

Usage:
  python3 tools/acceptance/e2e_acceptance.py [--port 18201]
      [--artifact-root DIR] [--evidence-dir DIR] [--stage-root DIR]
      [--server-dll PATH] [--bpython PATH] [--keep-server]

Preconditions:
  dotnet tool restore          (if tools are pinned in .config/dotnet-tools.json)
  dotnet fable src/SomhairlesDream.Frontend/SomhairlesDream.Frontend.fsproj --outDir public --noRestore --noCache
  dotnet build SomhairlesDream.sln -c Release
  real bpy interpreter available (default /tmp/bpyenv/bin/python)
"""

import argparse
import hashlib
import json
import os
import shutil
import signal
import socket
import subprocess
import sys
import time
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

TOOLS_DIR = Path(__file__).resolve().parent
REPO = TOOLS_DIR.parents[1]
DEPS = TOOLS_DIR / "deps"
DEFAULT_BPYTHON = Path("/tmp/bpyenv/bin/python")
DEFAULT_SERVER_DLL = (
    REPO / "src" / "SomhairlesDream.Server" / "bin" / "Release" / "net8.0" / "SomhairlesDream.Server.dll"
)

REFERENCE_SHA = {
    "steps/01-foundation.glb": "6082e735a65557665f68e673bd2899a38818d57a09e244a63fa3fb7f3cc687ef",
    "steps/03-cabin.glb": "4c54f96c66cdb6ad56e2d4118a90e2d6507a2dc061c941867bda86e4a177a9b2",
}

GEOMETRY_JS = """
() => {
  const v = window.SOMHAIRLES_VIEWER;
  if (!v || !v.camera || !v.group || !v.canvas) {
    return { ready: false, reason: 'viewer handle missing' };
  }
  const w = v.canvas.clientWidth, h = v.canvas.clientHeight;
  if (!w || !h) return { ready: false, reason: 'canvas has no size' };
  const box = new THREE.Box3().setFromObject(v.group);
  if (box.isEmpty()) return { ready: false, reason: 'empty bounding box' };
  const size = new THREE.Vector3(); box.getSize(size);
  const lo = box.min, hi = box.max;
  const corners = [
    [lo.x,lo.y,lo.z],[hi.x,lo.y,lo.z],[lo.x,hi.y,lo.z],[hi.x,hi.y,lo.z],
    [lo.x,lo.y,hi.z],[hi.x,lo.y,hi.z],[lo.x,hi.y,hi.z],[hi.x,hi.y,hi.z]
  ].map((p) => new THREE.Vector3(p[0], p[1], p[2]).project(v.camera));
  const px = corners.map((c) => [(c.x * 0.5 + 0.5) * w, (0.5 - c.y * 0.5) * h]);
  const xs = px.map((p) => p[0]);
  const ys = px.map((p) => p[1]);
  const minX = Math.min(...xs), maxX = Math.max(...xs);
  const minY = Math.min(...ys), maxY = Math.max(...ys);
  const minDim = Math.min(w, h);
  const extent = Math.max(maxX - minX, maxY - minY);
  const cx = (minX + maxX) / 2, cy = (minY + maxY) / 2;
  return {
    ready: true,
    canvas: [w, h],
    innerMargin: Math.min(minX, minY, w - maxX, h - maxY),
    coverage: extent / minDim,
    centerOffset: Math.hypot(cx - w / 2, cy - h / 2) / minDim,
    boxSize: [size.x, size.y, size.z],
    childCount: v.group.children.length
  };
}
"""


SCENE_JS = """
() => {
  const v = window.SOMHAIRLES_VIEWER;
  if (!v || !v.group) return { ready: false, reason: 'viewer handle missing' };
  const kids = Array.from(v.group.children).map((c) => ({
    type: c.type,
    isMesh: c.isMesh === true,
    color: c.material && c.material.color && c.material.color.getHexString
      ? c.material.color.getHexString() : null,
    childCount: c.children ? c.children.length : 0
  }));
  const box = new THREE.Box3().setFromObject(v.group);
  const size = new THREE.Vector3();
  if (!box.isEmpty()) box.getSize(size);
  return {
    ready: true,
    childCount: v.group.children.length,
    kids: kids,
    boxSize: box.isEmpty() ? null : [size.x, size.y, size.z],
    note: (document.querySelector('#viewport-note') || {}).textContent || '',
    meshLabel: (document.querySelector('#mesh-label') || {}).textContent || ''
  };
}
"""


def log(msg):
    print(f"[acceptance] {msg}", flush=True)


def sha256_bytes(data):
    return hashlib.sha256(data).hexdigest()


def ensure_playwright():
    if not (DEPS / "playwright").exists():
        log("provisioning playwright into tools/acceptance/deps (repo-local, not system-wide)")
        pinned = ""
        try:
            probe = subprocess.run(
                [sys.executable, "-c", "import playwright,sys; sys.stdout.write(playwright.__version__)"],
                capture_output=True, text=True, timeout=60,
            )
            if probe.returncode == 0 and probe.stdout.strip():
                pinned = f"=={probe.stdout.strip()}"
                log(f"pinning playwright{pinned} to reuse any existing chromium cache")
        except Exception:
            pass
        DEPS.mkdir(parents=True, exist_ok=True)
        subprocess.run(
            [sys.executable, "-m", "pip", "install", "--target", str(DEPS),
             "--break-system-packages", f"playwright{pinned}"],
            check=True,
        )
    env = os.environ.copy()
    env["PYTHONPATH"] = str(DEPS)
    log("ensuring chromium browser is present (no-op if already downloaded)")
    subprocess.run([sys.executable, "-m", "playwright", "install", "chromium"],
                   env=env, check=True)
    if str(DEPS) not in sys.path:
        sys.path.insert(0, str(DEPS))
    from playwright.sync_api import sync_playwright

    return sync_playwright


def wait_for_port(port, timeout=45.0):
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with socket.create_connection(("127.0.0.1", port), timeout=1.0):
                return
        except OSError:
            time.sleep(0.3)
    raise RuntimeError(f"server did not open port {port} within {timeout}s")


def fetch_bytes(url):
    with urllib.request.urlopen(url, timeout=30) as resp:
        return resp.status, resp.headers.get("Content-Type", ""), resp.read()


def wait_text(page, selector, needle, timeout_ms=90000):
    page.wait_for_function(
        "([sel, needle]) => (document.querySelector(sel)?.textContent ?? '').includes(needle)",
        arg=[selector, needle],
        timeout=timeout_ms,
    )


def assert_geometry_fit(geo, label, results):
    checks = []
    if not geo.get("ready"):
        results.append((False, f"{label}: geometry probe not ready ({geo.get('reason')})"))
        return checks
    w, h = geo["canvas"]
    min_dim = min(w, h)
    checks.append(("childCount", geo["childCount"] >= 1, geo))
    checks.append(("innerMargin>=2%minDim", geo["innerMargin"] >= 0.02 * min_dim, geo))
    checks.append(("coverage in [0.40,0.95]", 0.40 <= geo["coverage"] <= 0.95, geo))
    checks.append(("centerOffset<=10%minDim", geo["centerOffset"] <= 0.10, geo))
    sizes = geo["boxSize"]
    finite_pos = all(isinstance(s, (int, float)) and s > 0 and s == s and s != float("inf") for s in sizes)
    checks.append(("boxSize finite>0", finite_pos, geo))
    for name, ok, payload in checks:
        results.append((ok, f"{label}: {name} -> {json.dumps(payload) if not ok else 'ok'}"))
    return checks


def run_flow(page, base_url, run_id, viewport_label, results, evidence, shot_prefix):
    page.goto(f"{base_url}/?freeze=1", wait_until="domcontentloaded")
    wait_text(page, "#status-label", "待机")
    results.append((True, f"{viewport_label}: idle 待机 visible"))
    body = page.inner_text("body")
    results.append(("当前没有下一版正在开发" in body, f"{viewport_label}: empty-state text present"))

    page.fill("#run-id-input", run_id)
    render_box = page.query_selector("#run-render-input")
    if render_box and render_box.is_checked():
        render_box.set_checked(False)
    page.click("#run-start-button")

    wait_text(page, "#stream-state", "已连接")
    results.append((True, f"{viewport_label}: SSE 已连接"))

    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    wait_text(page, "#status-label", "运行中")
    results.append((True, f"{viewport_label}: stage1 live (运行中 + steps/01-foundation.glb)"))
    page.wait_for_timeout(600)
    geo1 = page.evaluate(GEOMETRY_JS)
    (evidence / f"{shot_prefix}-geo-stage1.json").write_text(json.dumps(geo1, indent=2))
    assert_geometry_fit(geo1, f"{viewport_label}/stage1", results)
    page.locator("#viewport-canvas").screenshot(path=str(evidence / f"{shot_prefix}-canvas-stage1.png"))
    page.screenshot(path=str(evidence / f"{shot_prefix}-page-stage1.png"), full_page=True)

    wait_text(page, "#viewport-note", "管线几何 · steps/03-cabin.glb")
    wait_text(page, "#status-label", "已完成")
    results.append((True, f"{viewport_label}: stage3 complete (已完成 + steps/03-cabin.glb)"))
    page.wait_for_timeout(600)
    geo3 = page.evaluate(GEOMETRY_JS)
    (evidence / f"{shot_prefix}-geo-stage3.json").write_text(json.dumps(geo3, indent=2))
    assert_geometry_fit(geo3, f"{viewport_label}/stage3", results)
    page.locator("#viewport-canvas").screenshot(path=str(evidence / f"{shot_prefix}-canvas-stage3.png"))

    if geo1.get("ready") and geo3.get("ready"):
        diffs = [abs(a - b) / max(abs(a), abs(b), 1e-9) for a, b in zip(geo1["boxSize"], geo3["boxSize"])]
        results.append((max(diffs) >= 0.05,
                        f"{viewport_label}: stage1 vs stage3 bbox differs (max rel diff {max(diffs):.3f})"))

    shot3 = evidence / f"{shot_prefix}-canvas-stage3.png"
    page.wait_for_timeout(800)
    repeat = evidence / f"{shot_prefix}-canvas-stage3-repeat.png"
    page.locator("#viewport-canvas").screenshot(path=str(repeat))
    results.append((sha256_bytes(shot3.read_bytes()) == sha256_bytes(repeat.read_bytes()),
                    f"{viewport_label}: frozen stage3 screenshot deterministic"))
    page.screenshot(path=str(evidence / f"{shot_prefix}-page-stage3.png"), full_page=True)

    href = page.get_attribute("#artifact-manifest-link", "href")
    results.append((bool(href), f"{viewport_label}: manifest link exposed ({href})"))


def stale_load_regression(browser, base_url, track, results, evidence):
    """Regression for async step-load ownership across runs.

    Intercepts steps GLB requests and fulfills them late, out of order, or
    with errors to prove stale responses from superseded runs never repaint
    geometry or notes, while the current owner's loads keep working.
    """
    page = browser.new_page(viewport={"width": 1280, "height": 900})

    def ignore_intentional_500(m):
        return "status of 500" in m.text \
            and "/api/artifacts/steps" in (m.location or {}).get("url", "")

    track(page, ignore=ignore_intentional_500)
    stamp = int(time.time() * 1000)
    held = []

    def hold_route(route):
        q = urllib.parse.parse_qs(urllib.parse.urlparse(route.request.url).query)
        held.append({
            "route": route,
            "project": (q.get("projectId") or [""])[0],
            "run": (q.get("runId") or [""])[0],
            "path": (q.get("path") or [""])[0],
            "done": False,
        })

    page.route("**/api/artifacts/steps*", hold_route)
    page.goto(f"{base_url}/?freeze=1", wait_until="domcontentloaded")
    wait_text(page, "#status-label", "待机")

    def start_run(run_id):
        page.fill("#run-id-input", run_id)
        render_box = page.query_selector("#run-render-input")
        if render_box and render_box.is_checked():
            render_box.set_checked(False)
        page.click("#run-start-button")
        wait_text(page, "#stream-state", "已连接")

    def held_for(run_id, path, timeout_s=120):
        deadline = time.time() + timeout_s
        while time.time() < deadline:
            for entry in held:
                if not entry["done"] and entry["run"] == run_id and entry["path"] == path:
                    return entry
            page.wait_for_timeout(300)
        raise AssertionError(
            f"no held steps request for run={run_id} path={path}; "
            f"held={[(e['run'], e['path'], e['done']) for e in held]}")

    def release(entry, status=200):
        if status == 200:
            q = urllib.parse.urlencode({
                "projectId": entry["project"], "runId": entry["run"], "path": entry["path"],
            })
            resp_status, ctype, body = fetch_bytes(f"{base_url}/api/artifacts/steps?{q}")
            if resp_status != 200:
                raise AssertionError(f"artifact fetch for fulfill failed ({resp_status})")
        else:
            ctype, body = "text/plain", b"boom"
        entry["route"].fulfill(status=status, content_type=ctype, body=body)
        entry["done"] = True

    def is_frame(g):
        return bool(g.get("kids")) and g["kids"][0]["type"] == "Mesh" \
            and g["kids"][0]["color"] == "d7a948"

    def is_glb(g):
        return bool(g.get("kids")) and g["kids"][0]["type"] == "Group"

    def box_ok(g):
        return g.get("boxSize") and all(
            isinstance(s, (int, float)) and s > 0 for s in g["boxSize"])

    # Scenario A: same path across two distinct runs must trigger a reload.
    run_a = f"run-stale-{stamp}-a"
    start_run(run_a)
    release(held_for(run_a, "steps/01-foundation.glb"))
    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    page.wait_for_timeout(600)
    ga = page.evaluate(SCENE_JS)
    results.append((ga["ready"] and is_glb(ga) and box_ok(ga),
                    f"stale-load: run A displays GLB scene (kids={ga.get('kids')})"))
    (evidence / "stale-scene-a.json").write_text(json.dumps(ga, indent=2, ensure_ascii=False))

    run_b = f"run-stale-{stamp}-b"
    start_run(run_b)
    eb = held_for(run_b, "steps/01-foundation.glb")
    results.append((True, f"stale-load: same path across runs reloads for run B ({eb['run']})"))
    release(eb)
    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    page.wait_for_timeout(600)
    gb = page.evaluate(SCENE_JS)
    results.append((gb["ready"] and is_glb(gb) and box_ok(gb),
                    f"stale-load: run B repaints GLB after same-path switch (kids={gb.get('kids')})"))
    page.screenshot(path=str(evidence / "stale-page-b.png"), full_page=True)

    # Scenario B: late success from a superseded run must be ignored.
    run_c = f"run-stale-{stamp}-c"
    start_run(run_c)
    held_for(run_c, "steps/01-foundation.glb")

    run_d = f"run-stale-{stamp}-d"
    start_run(run_d)
    frame_geo = None
    deadline = time.time() + 30
    while time.time() < deadline:
        g = page.evaluate(SCENE_JS)
        if g["ready"] and is_frame(g):
            frame_geo = g
            break
        page.wait_for_timeout(300)
    results.append((frame_geo is not None,
                    "stale-load: run switch with empty steps repaints replay frame mesh "
                    f"({(frame_geo or {}).get('kids')})"))
    page.screenshot(path=str(evidence / "stale-page-d-frame.png"), full_page=True)

    release(held_for(run_c, "steps/01-foundation.glb"))
    page.wait_for_timeout(600)
    g2 = page.evaluate(SCENE_JS)
    results.append((g2["ready"] and is_frame(g2)
                    and g2["note"] != "管线几何 · steps/01-foundation.glb",
                    f"stale-load: late old success ignored (note='{g2['note'][:40]}', "
                    f"kids={g2['kids']})"))

    release(held_for(run_d, "steps/01-foundation.glb"))
    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    page.wait_for_timeout(600)
    g3 = page.evaluate(SCENE_JS)
    results.append((g3["ready"] and is_glb(g3) and box_ok(g3),
                    "stale-load: run D current load displays after late-old-success"))

    # Scenario C: late error from a superseded run must not fail the current load.
    run_e = f"run-stale-{stamp}-e"
    start_run(run_e)
    ee = held_for(run_e, "steps/01-foundation.glb")

    run_f = f"run-stale-{stamp}-f"
    start_run(run_f)
    ef = held_for(run_f, "steps/01-foundation.glb")

    release(ee, status=500)
    page.wait_for_timeout(600)
    g4 = page.evaluate(SCENE_JS)
    results.append(("加载失败" not in g4["note"],
                    f"stale-load: late old error ignored (note='{g4['note'][:40]}')"))

    release(ef)
    wait_text(page, "#viewport-note", "管线几何 · steps/01-foundation.glb")
    page.wait_for_timeout(600)
    g5 = page.evaluate(SCENE_JS)
    results.append((g5["ready"] and is_glb(g5) and box_ok(g5) and "加载失败" not in g5["note"],
                    "stale-load: current owner load unaffected by stale error"))
    (evidence / "stale-scene-final.json").write_text(json.dumps(g5, indent=2, ensure_ascii=False))
    page.screenshot(path=str(evidence / "stale-page-final.png"), full_page=True)
    page.close()


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--port", type=int, default=18201)
    parser.add_argument("--artifact-root", default=str(REPO / "scratch" / "acceptance" / "artifacts"))
    parser.add_argument("--evidence-dir", default=None)
    parser.add_argument("--stage-root", default=str(REPO / "stage"))
    parser.add_argument("--server-dll", default=str(DEFAULT_SERVER_DLL))
    parser.add_argument("--bpython", default=str(DEFAULT_BPYTHON))
    parser.add_argument("--keep-server", action="store_true")
    args = parser.parse_args()

    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    evidence = Path(args.evidence_dir or (REPO / "scratch" / "acceptance" / f"evidence-{stamp}"))
    evidence.mkdir(parents=True, exist_ok=True)
    artifact_root = Path(args.artifact_root)
    artifact_root.mkdir(parents=True, exist_ok=True)
    server_log = evidence / "server.log"

    if not Path(args.server_dll).exists():
        log(f"server dll missing: {args.server_dll} - run 'dotnet build SomhairlesDream.sln -c Release' first")
        return 2
    if not Path(args.bpython).exists():
        log(f"bpy interpreter missing: {args.bpython} (pass --bpython)")
        return 2
    try:
        with socket.create_connection(("127.0.0.1", args.port), timeout=0.5):
            log(f"port {args.port} already in use - choose another --port")
            return 2
    except OSError:
        pass

    env = os.environ.copy()
    env["SOMHAIRLES_ARTIFACT_ROOT"] = str(artifact_root)
    env["SOMHAIRLES_STAGE_ROOT"] = str(args.stage_root)
    env["SOMHAIRLES_BPYTHON"] = str(args.bpython)
    server = subprocess.Popen(
        ["dotnet", args.server_dll, "--urls", f"http://127.0.0.1:{args.port}"],
        env=env, stdout=open(server_log, "wb"), stderr=subprocess.STDOUT,
        start_new_session=True, cwd=str(REPO),
    )
    base_url = f"http://127.0.0.1:{args.port}"
    results = []
    console_errors = []

    try:
        wait_for_port(args.port)
        with urllib.request.urlopen(f"{base_url}/", timeout=15) as resp:
            results.append((resp.status == 200, f"GET / -> {resp.status}"))
        sync_playwright = ensure_playwright()
        with sync_playwright() as pw:
            browser = pw.chromium.launch(args=["--enable-unsafe-swiftshader"])

            def track(page, ignore=None):
                def on_console(m):
                    if m.type == "error" and not (ignore and ignore(m)):
                        console_errors.append(m.text)
                page.on("console", on_console)
                page.on("pageerror", lambda e: console_errors.append(str(e)))

            desktop_ms = int(time.time() * 1000)
            page = browser.new_page(viewport={"width": 1440, "height": 1000})
            track(page)
            run_id = f"run-acc-{desktop_ms}-desktop"
            run_flow(page, base_url, run_id, "desktop", results, evidence, "desktop")
            project_id = page.input_value("#project-id-input")

            narrow_ms = int(time.time() * 1000) + 1
            page2 = browser.new_page(viewport={"width": 390, "height": 844})
            track(page2)
            run_id2 = f"run-acc-{narrow_ms}-narrow"
            run_flow(page2, base_url, run_id2, "narrow", results, evidence, "narrow")

            for step_path in ("steps/01-foundation.glb", "steps/03-cabin.glb"):
                q = urllib.parse.urlencode({
                    "projectId": project_id, "runId": run_id, "path": step_path,
                })
                status, ctype, body = fetch_bytes(f"{base_url}/api/artifacts/steps?{q}")
                digest = sha256_bytes(body)
                results.append((status == 200, f"GET steps {step_path} -> {status}"))
                results.append(("model/gltf-binary" in ctype, f"content-type {step_path} -> {ctype}"))
                expected = REFERENCE_SHA[step_path]
                results.append((digest == expected, f"sha256 {step_path} -> {digest[:12]}… (expect {expected[:12]}…)"))

            stale_load_regression(browser, base_url, track, results, evidence)

            fail_ms = int(time.time() * 1000) + 2
            fail_run_id = f"run-acc-{fail_ms}-failure"
            (artifact_root / project_id / fail_run_id).mkdir(parents=True, exist_ok=True)
            page.fill("#run-id-input", fail_run_id)
            page.click("#run-start-button")
            wait_text(page, "#status-label", "失败")
            msg = page.inner_text("#run-message")
            results.append(("run directory already exists" in msg,
                            f"failure run surfaces pipeline error (message: {msg.strip()[:80]}…)"))

            browser.close()
    finally:
        if not args.keep_server:
            try:
                os.killpg(os.getpgid(server.pid), signal.SIGTERM)
            except ProcessLookupError:
                pass
            try:
                server.wait(timeout=10)
            except subprocess.TimeoutExpired:
                os.killpg(os.getpgid(server.pid), signal.SIGKILL)
        else:
            log(f"server kept alive on port {args.port} (pid {server.pid})")

    if console_errors:
        results.append((False, f"console/page errors: {console_errors[:5]}"))
    else:
        results.append((True, "zero browser console/page errors"))

    passed = sum(1 for ok, _ in results if ok)
    failed = len(results) - passed
    report = {
        "timestamp": stamp,
        "base_url": base_url,
        "evidence_dir": str(evidence),
        "console_errors": console_errors,
        "checks": [{"ok": ok, "detail": detail} for ok, detail in results],
    }
    (evidence / "acceptance-report.json").write_text(json.dumps(report, indent=2, ensure_ascii=False))
    for ok, detail in results:
        log(f"{'PASS' if ok else 'FAIL'}: {detail}")
    log(f"RESULT: {passed}/{len(results)} checks passed ({failed} failed); evidence in {evidence}")
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main())