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
  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
  };
}
"""


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 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):
                page.on("console", lambda m: console_errors.append(m.text) if m.type == "error" else None)
                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]}…)"))

            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())