terainia/test/run.py

"""
Run a declarative scenario YAML against the running launch.py browser.

A scenario is a YAML mapping with a `name`, `description`, and a list
of `steps`. Each step is one mapping like `{eval: "..."}`. See
test/README.md for the full grammar.

Usage:
    python3 run.py scenarios/lighting-times-of-day.yaml
    python3 run.py scenarios/sunset.yaml --halt-on-fail
"""

from __future__ import annotations

import argparse
import sys
import time
from pathlib import Path

import yaml
from playwright.sync_api import sync_playwright

HERE = Path(__file__).resolve().parent
SCREENSHOT_DIR = HERE / "screenshots"
CDP_URL = "http://localhost:9222"

# All keys we'll let scenarios send via page.keyboard.press. Mirrors
# cucucaracha's whitelist + the keys our game listens for.
KEY_ALLOWLIST = {
    "Enter", "Escape", "Tab", "Space",
    "ArrowUp", "ArrowDown", "ArrowLeft", "ArrowRight",
    "KeyW", "KeyA", "KeyS", "KeyD",
    "ShiftLeft", "ShiftRight", "ControlLeft", "ControlRight",
    "Digit0", "Digit1", "Digit2", "Digit3", "Digit4",
    "Digit5", "Digit6", "Digit7", "Digit8", "Digit9",
}


def attach_page():
    p = sync_playwright().start()
    try:
        browser = p.chromium.connect_over_cdp(CDP_URL)
    except Exception as e:  # noqa: BLE001
        p.stop()
        sys.exit(f"could not attach to launch.py via {CDP_URL}: {e}")
    pages = [pg for c in browser.contexts for pg in c.pages]
    if not pages:
        browser.close()
        p.stop()
        sys.exit("no open pages")
    return p, browser, pages[-1]


def step_wait_for(page, expr: str, timeout_ms: int = 30_000):
    """Block until `expr` evaluates truthy in the page."""
    start = time.time()
    while True:
        try:
            ok = page.evaluate(f"() => Boolean({expr})")
        except Exception:
            ok = False
        if ok:
            return
        if (time.time() - start) * 1000 > timeout_ms:
            raise TimeoutError(f"wait_for {expr!r} timed out after {timeout_ms} ms")
        time.sleep(0.1)


def run_step(page, step: dict, scenario_name: str, halt_on_fail: bool) -> bool:
    """Execute one step. Returns False on assertion failure."""
    if "wait_for" in step:
        print(f"  wait_for: {step['wait_for']}")
        step_wait_for(page, step["wait_for"], step.get("timeout_ms", 30_000))
        return True

    if "wait" in step:
        ms = int(step["wait"])
        print(f"  wait: {ms} ms")
        time.sleep(ms / 1000.0)
        return True

    if "eval" in step:
        print(f"  eval: {step['eval']}")
        # Wrap so a bare expression like `set_scene_time(0)` works too.
        js = step["eval"]
        page.evaluate(f"() => {{ {js}; return null; }}")
        return True

    if "key" in step:
        key = step["key"]
        if key not in KEY_ALLOWLIST:
            raise ValueError(f"key {key!r} not in allowlist {sorted(KEY_ALLOWLIST)}")
        hold = step.get("hold")
        if hold:
            print(f"  key: {key} hold={hold}ms")
            page.keyboard.down(key)
            time.sleep(int(hold) / 1000.0)
            page.keyboard.up(key)
        else:
            print(f"  key: {key}")
            page.keyboard.press(key)
        return True

    if "mouse_move" in step:
        dx, dy = step["mouse_move"]
        print(f"  mouse_move: ({dx}, {dy})")
        # Relative motion via DeviceEvent — Playwright doesn't expose
        # raw relative deltas, so we move from center by (dx, dy).
        # The game uses pointer lock which means we'd need a different
        # path for real-relative; this still works for many scenarios.
        page.mouse.move(640 + dx, 400 + dy)
        return True

    if "mouse" in step:
        action = step["mouse"]
        print(f"  mouse: {action}")
        if action == "down":
            page.mouse.down()
        elif action == "up":
            page.mouse.up()
        elif action == "click":
            page.mouse.click(640, 400)
        else:
            raise ValueError(f"unknown mouse action: {action}")
        return True

    if "screenshot" in step:
        SCREENSHOT_DIR.mkdir(exist_ok=True)
        name = step["screenshot"]
        out = SCREENSHOT_DIR / f"{scenario_name}_{name}"
        page.screenshot(path=str(out), full_page=False)
        print(f"  screenshot → {out}")
        return True

    if "assert" in step:
        expr = step["assert"]
        ok = page.evaluate(f"() => Boolean({expr})")
        symbol = "✓" if ok else "✗"
        print(f"  assert: {expr}  {symbol}")
        if not ok and halt_on_fail:
            return False
        return ok

    raise ValueError(f"unknown step type in {step!r}")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("scenario", help="path to scenario YAML")
    ap.add_argument(
        "--halt-on-fail",
        action="store_true",
        help="stop on first failed assert (default: continue)",
    )
    args = ap.parse_args()

    scenario_path = Path(args.scenario)
    if not scenario_path.exists():
        sys.exit(f"scenario not found: {scenario_path}")
    scenario = yaml.safe_load(scenario_path.read_text())
    name = scenario.get("name") or scenario_path.stem
    print(f"# scenario: {name}")
    if scenario.get("description"):
        print(f"# {scenario['description']}")
    print()

    p, browser, page = attach_page()
    failures = 0
    total_start = time.perf_counter()
    timings: list[tuple[str, float]] = []
    try:
        for i, step in enumerate(scenario.get("steps", [])):
            step_summary = next(iter(step.keys())) if step else f"step{i}"
            print(f"step {i}:")
            step_start = time.perf_counter()
            try:
                ok = run_step(page, step, name, args.halt_on_fail)
                step_ms = (time.perf_counter() - step_start) * 1000
                timings.append((step_summary, step_ms))
                print(f"  [{step_ms:6.1f}ms]")
                if not ok:
                    failures += 1
                    if args.halt_on_fail:
                        break
            except Exception as e:  # noqa: BLE001
                step_ms = (time.perf_counter() - step_start) * 1000
                timings.append((step_summary, step_ms))
                print(f"  [!] step error: {e}", file=sys.stderr)
                failures += 1
                if args.halt_on_fail:
                    break
        total_ms = (time.perf_counter() - total_start) * 1000
        print()
        print(f"# {name}: {len(scenario.get('steps', []))} steps in {total_ms / 1000:.2f}s "
              f"({failures} failure{'s' if failures != 1 else ''})")
        # Per-step breakdown so the slowest steps are obvious.
        if timings:
            timings_sorted = sorted(enumerate(timings), key=lambda x: -x[1][1])
            print("# slowest steps:")
            for i, (kind, ms) in timings_sorted[:5]:
                print(f"#   step {i:>2}  {kind:>12}  {ms:>7.1f}ms")
    finally:
        browser.close()
        p.stop()

    return 0 if failures == 0 else 1


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