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