From f93a24e283bb858c410c1dfe136e33b528cc4382 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 24 Oct 2026 22:37:49 +0200 Subject: [PATCH] flutter_web_drive: headless CDP driver + MCP server for Flutter web apps Drives a Flutter web build in a persistent headless chromium (systemd-run unit, own profile, CDP port) through the semantics tree: tree, tap by label, text, key, scroll, wait, shot, console. Never touches the live display; attaches only to HeadlessChrome. The MCP server wraps the same Driver. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01CZ4sULGfyav7EEwQ6vhbfg --- CLAUDE.md | 9 +- flutter_web_drive/README.md | 120 +++++++++ flutter_web_drive/__init__.py | 6 + flutter_web_drive/__main__.py | 5 + flutter_web_drive/actions.py | 239 ++++++++++++++++++ flutter_web_drive/api.py | 235 +++++++++++++++++ flutter_web_drive/browser.py | 81 ++++++ flutter_web_drive/cdp.py | 139 ++++++++++ flutter_web_drive/cli.py | 150 +++++++++++ flutter_web_drive/errors.py | 20 ++ flutter_web_drive/keys.py | 22 ++ flutter_web_drive/mcp_server.py | 173 +++++++++++++ flutter_web_drive/requirements.txt | 3 + flutter_web_drive/semantics.py | 228 +++++++++++++++++ flutter_web_drive/state.py | 59 +++++ flutter_web_drive/tests/__init__.py | 0 flutter_web_drive/tests/fakes.py | 179 +++++++++++++ flutter_web_drive/tests/test_actions.py | 158 ++++++++++++ .../tests/test_actions_output.py | 134 ++++++++++ flutter_web_drive/tests/test_api.py | 212 ++++++++++++++++ flutter_web_drive/tests/test_cdp.py | 157 ++++++++++++ flutter_web_drive/tests/test_cli.py | 158 ++++++++++++ flutter_web_drive/tests/test_mcp_server.py | 100 ++++++++ flutter_web_drive/tests/test_semantics.py | 105 ++++++++ flutter_web_drive/tests/test_state.py | 42 +++ 25 files changed, 2730 insertions(+), 4 deletions(-) create mode 100644 flutter_web_drive/README.md create mode 100644 flutter_web_drive/__init__.py create mode 100644 flutter_web_drive/__main__.py create mode 100644 flutter_web_drive/actions.py create mode 100644 flutter_web_drive/api.py create mode 100644 flutter_web_drive/browser.py create mode 100644 flutter_web_drive/cdp.py create mode 100644 flutter_web_drive/cli.py create mode 100644 flutter_web_drive/errors.py create mode 100644 flutter_web_drive/keys.py create mode 100644 flutter_web_drive/mcp_server.py create mode 100644 flutter_web_drive/requirements.txt create mode 100644 flutter_web_drive/semantics.py create mode 100644 flutter_web_drive/state.py create mode 100644 flutter_web_drive/tests/__init__.py create mode 100644 flutter_web_drive/tests/fakes.py create mode 100644 flutter_web_drive/tests/test_actions.py create mode 100644 flutter_web_drive/tests/test_actions_output.py create mode 100644 flutter_web_drive/tests/test_api.py create mode 100644 flutter_web_drive/tests/test_cdp.py create mode 100644 flutter_web_drive/tests/test_cli.py create mode 100644 flutter_web_drive/tests/test_mcp_server.py create mode 100644 flutter_web_drive/tests/test_semantics.py create mode 100644 flutter_web_drive/tests/test_state.py diff --git a/CLAUDE.md b/CLAUDE.md index 55aa400..495e19e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,7 +1,8 @@ # utils Monorepo of shared gates and libraries (dep_freshness, file_length, md_naming, -repo_contract, coverage_gaps, crdt-sync, gatelock, freedays, sickdays, earned_time, music_theory, ...). +repo_contract, coverage_gaps, crdt-sync, gatelock, freedays, sickdays, earned_time, music_theory, +flutter_web_drive (headless Flutter-web E2E driver + MCP), ...). Python packages with a root-level `/tests` run with `PYTHONPATH=.`; subprojects with their own `pyproject.toml` run via `scripts/run_subproject_tests.sh `. `scripts/check_repo_contract.sh` is the REAL shared gate (not a shim): never @@ -10,8 +11,8 @@ overwrite it with `repo_contract/templates/check_repo_contract.shim.sh`. ## Commands - run: n/a: collection of gates and libraries, no single entry point -- test: `PYTHONPATH=. python -m pytest -q agent_ready/tests coverage_gaps/tests dep_freshness/tests file_length/tests home_paths/tests md_naming/tests repo_contract/tests scripts/tests` +- test: `PYTHONPATH=. python -m pytest -q agent_ready/tests coverage_gaps/tests dep_freshness/tests file_length/tests flutter_web_drive/tests home_paths/tests md_naming/tests repo_contract/tests scripts/tests` - test-changed: `scripts/test_changed.sh` -- lint: `ruff check agent_ready coverage_gaps repo_contract` -- coverage: `PYTHONPATH=. python -m pytest -q agent_ready/tests repo_contract/tests --cov=agent_ready --cov=repo_contract --cov-branch --cov-report=xml --cov-fail-under=100` +- lint: `ruff check agent_ready coverage_gaps flutter_web_drive repo_contract` +- coverage: `PYTHONPATH=. python -m pytest -q agent_ready/tests flutter_web_drive/tests repo_contract/tests --cov=agent_ready --cov=flutter_web_drive --cov=repo_contract --cov-branch --cov-report=xml --cov-fail-under=100` - coverage-gaps: `coverage-gaps coverage.xml` diff --git a/flutter_web_drive/README.md b/flutter_web_drive/README.md new file mode 100644 index 0000000..37c6fee --- /dev/null +++ b/flutter_web_drive/README.md @@ -0,0 +1,120 @@ +# flutter_web_drive + +Headless, LLM-friendly driver for Flutter web apps. An agent can navigate and +test a desktop app (a Flutter web build in a Chrome `--app` window, the way +kartka and todo ship) **without touching the real screen, mouse or keyboard**. + +- A long-lived **headless** chromium runs as a `systemd-run --user` unit with a + persistent profile. It has to be a unit because `setsid` children die with + the calling Bash scope. +- Each CLI verb attaches over CDP, does one thing and exits. **The exit code + is the verdict.** On failure it prints one stderr line naming the step that + failed. +- Widgets are addressed through Flutter's **semantics tree**. Flutter paints + to a canvas, so the accessibility mirror (`[id^=flt-semantic-node-]`) is the + only addressable view. The driver switches it on by clicking + `flt-semantics-placeholder`. The app's `lib/` needs no changes. + +## Spike result (2026-10-24, chromium 155, Flutter 3.47.8 CanvasKit) + +`--headless=new` + CDP `Page.captureScreenshot` **renders CanvasKit frames +correctly**. (The old `--headless` `--screenshot` path was blank.) **No Xvfb +is needed**, and none is used. + +Other findings from the spike: +- Semantics enabled once stay enabled across later CDP connections. +- Console messages and exceptions are replayed on `Runtime.enable` / + `Log.enable`, so `console` works from a fresh attach. +- The viewport is smaller than `--window-size`: 1400x1000 gives 1400x879. + `start` reports the real viewport. + +## Safety + +- Every attach reads `/json/version`. It **refuses any browser that is not + `HeadlessChrome`** (exit 4). For example, Steam's CEF listens on 9222 on + this PC. Never point the driver at 9222. +- The unit unsets `DISPLAY`/`WAYLAND_DISPLAY`. `--password-store=basic` keeps + chromium away from the desktop keyring, which could prompt on :0. + `--disable-extensions` drops the Claude extension that chromium-flags.conf + loads. +- `start` refuses a CDP port that some other browser already answers on. +- Secrets go through `text --stdin` / `text --file PATH` (MCP: `from_file`), + never argv. `tree` shows password fields as `*`. + +## CLI + +```bash +export PYTHONPATH=~/src/utils +D="python3 -m flutter_web_drive --app myapp" # --app namespaces state/artefacts +$D start --url http://127.0.0.1:47392/ --profile ~/.local/share/myapp-agent/profile \ + --cdp-port 9342 [--unit fwd-myapp] [--size 1400x1000] +$D tree [--filter TEXT] [--row LABEL] [--all] +$D tap "Browse" [--contains] # exact label or one line of it +$D tap-xy 40 100 +$D text "hello" | $D text --stdin | $D text --file ~/.config/x/password +$D key Enter # Escape Tab Backspace Delete Space Arrow* Home End PageUp PageDown +$D scroll 0 600 [--at X Y] +$D wait "Sync now" [--timeout S<=50] [--contains] [--gone] [--flag current] [--row LABEL] +$D shot [OUT.png] [--min-colours 16] # prints "path WxH colours=N stddev=S" +$D eval "location.href" +$D console [--limit 20] [--warnings] [--fail-on-error] +$D stop +``` + +`tree` prints one line per node: + +``` +#5 button "Browse | Tab 2 of 4" [tap,current] @0,75 80x67 +#73 textbox "Sync account email" = "me@example.com" [focused] @16,203 1368x48 +``` + +- `[tags]`: `tap` (tappable), `current`, `selected`, `checked`, `mixed`, + `pressed`, `expanded`, `disabled`, `focused`. +- `= value` is the field value, `aria-valuenow` or `aria-description`. For + example, an ExpansionTile reads `"Collapsed"`. +- `--row LABEL` keeps the nodes inside the row of that label, e.g. a chapter + row's "Mark unread" button: `wait "Mark unread" --row "Chapter 5"`. +- `tap` prefers tappable matches and says when there were several. When + nothing matches, it lists what is tappable. + +Exit codes: `0` ok, `1` the step's check failed (not found, timeout, page +exception, blank shot), `2` usage, `3` no driver browser (not started or +crashed), `4` the CDP port belongs to a non-headless browser. + +Artefacts (screenshots, `state.json`) go to `~/data/claude-scratch/e2e//` +(override with `$FWD_ARTEFACTS`), never to /tmp. A `wait` timeout saves +`wait-timeout-*.png` and names it in the error. + +## MCP server + +`mcp_server.py` is a thin stdio server (mcp 2.x `MCPServer`). Its tools match +the verbs: `start stop shot tree tap tap_xy text key scroll wait eval_js +console`, each taking `app`. Every tool is one `Driver` call, so no logic is +duplicated. Register it (user scope): + +```bash +claude mcp add flutter-web-drive -s user -e PYTHONPATH=$HOME/src/utils \ + -- python3 -m flutter_web_drive.mcp_server +``` + +## Layout + +| file | role | +|---|---| +| `cdp.py` | websocket session, page discovery, headless check | +| `semantics.py` | the JS snippets, `Node`, matching, row scoping, tree format | +| `actions.py` / `keys.py` | verbs on one attached page | +| `browser.py` / `state.py` | chromium argv, `systemd-run` unit, per-app state | +| `api.py` | `Driver`: the verbs the CLI and the MCP server share | +| `cli.py` / `mcp_server.py` | the two front ends | + +Tests (`tests/`, 100 % branch, no network or browser) use a fake CDP page. +The real-browser check is a consumer flow, e.g. +`~/src/kartka/tools/agent/flows/f0_smoke.sh`. + +```bash +PYTHONPATH=. python -m pytest -q flutter_web_drive/tests \ + --cov=flutter_web_drive --cov-branch --cov-fail-under=100 +ruff check flutter_web_drive && ruff format --check flutter_web_drive +pip install --user -r flutter_web_drive/requirements.txt +``` diff --git a/flutter_web_drive/__init__.py b/flutter_web_drive/__init__.py new file mode 100644 index 0000000..5b6e1bb --- /dev/null +++ b/flutter_web_drive/__init__.py @@ -0,0 +1,6 @@ +"""Headless, LLM-friendly driver for Flutter web apps over the Chrome DevTools Protocol. + +A long-lived headless chromium runs as a ``systemd-run --user`` unit; every +verb attaches over CDP, does one thing through Flutter's semantics tree, and +exits with a code that is the verdict. See README.md. +""" diff --git a/flutter_web_drive/__main__.py b/flutter_web_drive/__main__.py new file mode 100644 index 0000000..597dd17 --- /dev/null +++ b/flutter_web_drive/__main__.py @@ -0,0 +1,5 @@ +"""Entry point: ``python -m flutter_web_drive``.""" + +from flutter_web_drive.cli import main + +raise SystemExit(main()) diff --git a/flutter_web_drive/actions.py b/flutter_web_drive/actions.py new file mode 100644 index 0000000..13540ec --- /dev/null +++ b/flutter_web_drive/actions.py @@ -0,0 +1,239 @@ +"""Verbs on one attached page: read the tree, send real input, capture output.""" + +from __future__ import annotations + +import base64 +import json +import time +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from flutter_web_drive import semantics +from flutter_web_drive.cdp import Session +from flutter_web_drive.errors import EXIT_USAGE, DriveError +from flutter_web_drive.keys import KEYS +from flutter_web_drive.semantics import Node + +POLL = 0.25 + + +@dataclass(frozen=True) +class Clock: + """Monotonic time + sleep, injectable so polling is testable.""" + + now: Callable[[], float] = time.monotonic + sleep: Callable[[float], None] = time.sleep + + +def poll_nodes(sess: Session) -> list[Node]: + """Ask for semantics (idempotent) and read whatever nodes exist right now.""" + sess.evaluate(semantics.ENABLE_JS) + return semantics.parse_nodes(sess.evaluate(semantics.COLLECT_JS)) + + +def ensure_semantics(sess: Session, clock: Clock, timeout: float = 5.0) -> list[Node]: + """Nodes once the semantics mirror is populated, or fail after ``timeout``.""" + deadline = clock.now() + timeout + while True: + nodes = poll_nodes(sess) + if nodes: + return nodes + if clock.now() >= deadline: + raise DriveError( + "semantics", + f"no Flutter semantics nodes after {timeout:g}s (not a Flutter page, or still loading?)", + ) + clock.sleep(POLL) + + +def viewport(sess: Session) -> tuple[int, int]: + """Layout viewport in CSS pixels (what screenshots and rects use).""" + width, height = sess.evaluate(semantics.VIEWPORT_JS) + return int(width), int(height) + + +def click_at(sess: Session, x: float, y: float) -> None: + """A real left click: move, press, release, so Flutter's pointer path sees it.""" + sess.call("Input.dispatchMouseEvent", type="mouseMoved", x=x, y=y) + for kind in ("mousePressed", "mouseReleased"): + sess.call( + "Input.dispatchMouseEvent", type=kind, x=x, y=y, button="left", clickCount=1 + ) + + +def tap(sess: Session, clock: Clock, target: str, *, contains: bool = False) -> str: + """Click the center of the best node labelled ``target``.""" + nodes = ensure_semantics(sess, clock) + hits = semantics.find(nodes, target, contains=contains) + if not hits: + raise DriveError( + "tap", f'no node labelled "{target}"; tappable: {semantics.hint(nodes)}' + ) + best = hits[0] + x, y = best.center + width, height = viewport(sess) + if not (0 <= x < width and 0 <= y < height): + raise DriveError( + "tap", f'"{best.one_line}" is off-screen at ({x:g}, {y:g}); scroll first' + ) + click_at(sess, x, y) + extra = f" ({len(hits)} matches; took the first)" if len(hits) > 1 else "" + return f"tapped {semantics.describe(best)}{extra}" + + +def tap_xy(sess: Session, x: float, y: float) -> str: + """Click at viewport coordinates.""" + click_at(sess, x, y) + return f"tapped ({x:g}, {y:g})" + + +def insert_text(sess: Session, text: str) -> str: + """Type into the focused field (never echoes the text back).""" + if not sess.evaluate(semantics.FOCUS_JS): + raise DriveError("text", "no focused text field; tap the field first") + sess.call("Input.insertText", text=text) + return f"inserted {len(text)} characters" + + +def press_key(sess: Session, name: str) -> str: + """Key down + up for one named key.""" + if name not in KEYS: + raise DriveError( + "key", f"unknown key {name!r}; known: {', '.join(KEYS)}", EXIT_USAGE + ) + key, code, vk, text = KEYS[name] + down: dict[str, Any] = { + "type": "keyDown", + "key": key, + "code": code, + "windowsVirtualKeyCode": vk, + } + if text: + down["text"] = text + sess.call("Input.dispatchKeyEvent", **down) + sess.call( + "Input.dispatchKeyEvent", + type="keyUp", + key=key, + code=code, + windowsVirtualKeyCode=vk, + ) + return f"pressed {name}" + + +def scroll( + sess: Session, dx: float, dy: float, at: tuple[float, float] | None = None +) -> str: + """Mouse-wheel by (dx, dy) at ``at`` (default: viewport center).""" + if at is None: + width, height = viewport(sess) + at = (width / 2, height / 2) + x, y = at + sess.call( + "Input.dispatchMouseEvent", type="mouseWheel", x=x, y=y, deltaX=dx, deltaY=dy + ) + return f"scrolled ({dx:g}, {dy:g}) at ({x:g}, {y:g})" + + +def wait_for( + sess: Session, + clock: Clock, + target: str, + *, + timeout: float, + on_timeout: Callable[[], str], + contains: bool = False, + gone: bool = False, + flag: str | None = None, + row: str | None = None, +) -> str: + """Poll until a node labelled ``target`` (carrying ``flag``) is present, or absent with ``gone``.""" + what = ( + f'"{target}"' + + (f' in row "{row}"' if row else "") + + (f" [{flag}]" if flag else "") + ) + deadline = clock.now() + timeout + while True: + nodes = poll_nodes(sess) + hits = semantics.find( + semantics.in_row(nodes, row) if row else nodes, target, contains=contains + ) + hits = [h for h in hits if flag is None or flag in h.flags] + if hits and not gone: + return f"found {semantics.describe(hits[0])}" + if gone and not hits: + return f"gone: {what}" + if clock.now() >= deadline: + state = "still present" if gone else "not present" + raise DriveError( + "wait", + f"{what} {state} after {timeout:g}s; screenshot {on_timeout()}", + ) + clock.sleep(POLL) + + +def screenshot(sess: Session, out: Path, min_colours: int = 0) -> str: + """Write a PNG of the viewport; fail if it has fewer than ``min_colours`` colours.""" + data = sess.call("Page.captureScreenshot", format="png")["data"] + out.parent.mkdir(parents=True, exist_ok=True) + out.write_bytes(base64.b64decode(data)) + stats = sess.evaluate(f"{semantics.IMAGE_STATS_FN}({json.dumps(data)})") + colours = int(stats["colours"]) + summary = f"{out} {stats['w']}x{stats['h']} colours={colours} stddev={float(stats['stddev']):.1f}" + if colours < min_colours: + raise DriveError( + "shot", + f"screenshot looks blank ({colours} colours < {min_colours}): {summary}", + ) + return summary + + +def evaluate(sess: Session, expr: str) -> str: + """Run JS in the page and return its value as JSON.""" + return json.dumps(sess.evaluate(expr)) + + +def _event_line(event: dict[str, Any]) -> tuple[str, str] | None: + """(level, text) for console/log events worth reporting, else None.""" + method, params = event.get("method"), event.get("params", {}) + if method == "Runtime.consoleAPICalled" and params.get("type") in { + "error", + "assert", + "warning", + }: + level = "warning" if params["type"] == "warning" else "error" + args = " ".join( + str(a.get("value", a.get("description", ""))) + for a in params.get("args", []) + ) + return level, args + if method == "Runtime.exceptionThrown": + det = params.get("exceptionDetails", {}) + desc = det.get("exception", {}).get("description", "") + return ( + "error", + f"{det.get('text', 'exception')} {desc.splitlines()[0] if desc else ''}".strip(), + ) + if method == "Log.entryAdded" and params.get("entry", {}).get("level") in { + "error", + "warning", + }: + return params["entry"]["level"], str(params["entry"].get("text", "")) + return None + + +def console( + sess: Session, limit: int = 20, *, warnings: bool = False +) -> tuple[str, int]: + """Errors (and optionally warnings) since page load; chromium replays them on enable.""" + sess.call("Runtime.enable") + sess.call("Log.enable") + lines = [line for e in sess.drain(0.5) if (line := _event_line(e))] + errors = sum(1 for level, _ in lines if level == "error") + shown = [f"{lv}: {text}" for lv, text in lines if warnings or lv == "error"][ + -limit: + ] + return ("\n".join(shown) if shown else "no console errors"), errors diff --git a/flutter_web_drive/api.py b/flutter_web_drive/api.py new file mode 100644 index 0000000..405c92e --- /dev/null +++ b/flutter_web_drive/api.py @@ -0,0 +1,235 @@ +"""Driver: the verbs the CLI and the MCP server share (no logic lives in either).""" + +from __future__ import annotations + +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from pathlib import Path + +from flutter_web_drive import actions, browser, cdp, state +from flutter_web_drive.errors import EXIT_BROWSER, EXIT_UNSAFE, EXIT_USAGE, DriveError + +CDP_UP_TIMEOUT = 15.0 +FLUTTER_BOOT_TIMEOUT = 30.0 +# One verb must finish inside the 60 s per-action cap, screenshot included. +MAX_WAIT = 50.0 + + +class Driver: + """Verbs for one named app; ``start`` records the CDP port, the rest attach to it.""" + + def __init__( + self, + app: str = "default", + *, + cdp_port: int | None = None, + root: Path | None = None, + fetch: cdp.Fetch = cdp.http_json, + create: Callable[[str], cdp.Socket] = cdp.create_socket, + runner: browser.Runner = browser.run, + clock: actions.Clock | None = None, + ) -> None: + """Bind to ``app``'s state under ``root`` (default ~/data/claude-scratch/e2e).""" + self.app = app + self.cdp_port = cdp_port + self.root = root or state.artefact_root() + self.fetch = fetch + self.create = create + self.runner = runner + self.clock = clock or actions.Clock() + + @property + def app_dir(self) -> Path: + """Artefacts for this app (screenshots, state).""" + return self.root / self.app + + def _connect(self, port: int) -> cdp.Session: + return cdp.connect(port, fetch=self.fetch, create=self.create) + + @contextmanager + def _session(self) -> Iterator[cdp.Session]: + port = self.cdp_port + if port is None: + saved = state.load(self.root, self.app) + if saved is None: + raise DriveError( + "connect", + f"no running browser for app {self.app!r}; run start first", + EXIT_BROWSER, + ) + port = saved.cdp_port + sess = self._connect(port) + try: + yield sess + finally: + sess.close() + + def _await_flutter(self, sess: cdp.Session) -> str: + deadline = self.clock.now() + FLUTTER_BOOT_TIMEOUT + while not sess.evaluate(actions.semantics.FLUTTER_READY_JS): + if self.clock.now() >= deadline: + raise DriveError( + "start", + f"Flutter never booted within {FLUTTER_BOOT_TIMEOUT:g}s; see `console`", + ) + self.clock.sleep(actions.POLL) + actions.ensure_semantics(sess, self.clock) + width, height = actions.viewport(sess) + return f"viewport {width}x{height}" + + def start( + self, + url: str, + profile: Path, + cdp_port: int, + *, + unit: str | None = None, + size: str = "1400x1000", + ) -> str: + """Launch (or reuse) the headless unit on ``url`` and enable semantics.""" + unit = unit or f"fwd-{self.app}" + dims = browser.parse_size(size) + saved = state.State(unit=unit, cdp_port=cdp_port, url=url, profile=str(profile)) + if browser.unit_active(unit, self.runner): + sess = self._connect(cdp_port) + state.save(self.root, self.app, saved) + try: + if not str(sess.evaluate("location.href")).startswith(url): + sess.call("Page.navigate", url=url) + return f"reused {unit}: {url} on CDP {cdp_port}, {self._await_flutter(sess)}" + finally: + sess.close() + if cdp.probe(cdp_port, fetch=self.fetch) is not None: + raise DriveError( + "start", + f"CDP port {cdp_port} is already in use by another browser", + EXIT_UNSAFE, + ) + profile.mkdir(parents=True, exist_ok=True) + self.runner(["systemctl", "--user", "reset-failed", unit]) + argv = browser.unit_argv( + unit, browser.chromium_argv(url, profile, cdp_port, dims) + ) + if (rc := self.runner(argv)) != 0: + raise DriveError("start", f"systemd-run exited {rc}", EXIT_BROWSER) + deadline = self.clock.now() + CDP_UP_TIMEOUT + while cdp.probe(cdp_port, fetch=self.fetch) is None: + if self.clock.now() >= deadline: + raise DriveError( + "start", + f"CDP port {cdp_port} never answered; journalctl --user -u {unit}", + EXIT_BROWSER, + ) + self.clock.sleep(actions.POLL) + state.save(self.root, self.app, saved) + sess = self._connect(cdp_port) + try: + return ( + f"started {unit}: {url} on CDP {cdp_port}, {self._await_flutter(sess)}" + ) + finally: + sess.close() + + def stop(self) -> str: + """Stop the unit (profile kept) and forget the state.""" + saved = state.load(self.root, self.app) + unit = saved.unit if saved else f"fwd-{self.app}" + self.runner(["systemctl", "--user", "stop", unit]) + state.clear(self.root, self.app) + return f"stopped {unit}" + + def _artefact(self, kind: str) -> Path: + return self.app_dir / f"{kind}-{state.stamp()}.png" + + def shot(self, out: Path | None = None, *, min_colours: int = 0) -> str: + """Screenshot to ``out`` (default: a timestamped file in the app dir).""" + with self._session() as sess: + return actions.screenshot(sess, out or self._artefact("shot"), min_colours) + + def tree( + self, *, filt: str | None = None, show_all: bool = False, row: str | None = None + ) -> str: + """Compact semantics listing.""" + with self._session() as sess: + nodes = actions.ensure_semantics(sess, self.clock) + return actions.semantics.format_tree( + nodes, filt=filt, show_all=show_all, row=row + ) + + def tap(self, label: str, *, contains: bool = False) -> str: + """Click a node by label.""" + with self._session() as sess: + return actions.tap(sess, self.clock, label, contains=contains) + + def tap_xy(self, x: float, y: float) -> str: + """Click at coordinates.""" + with self._session() as sess: + return actions.tap_xy(sess, x, y) + + def text(self, value: str) -> str: + """Insert text into the focused field.""" + with self._session() as sess: + return actions.insert_text(sess, value) + + def key(self, name: str) -> str: + """Press one named key.""" + with self._session() as sess: + return actions.press_key(sess, name) + + def scroll( + self, dx: float, dy: float, *, at: tuple[float, float] | None = None + ) -> str: + """Mouse-wheel scroll.""" + with self._session() as sess: + return actions.scroll(sess, dx, dy, at) + + def wait( + self, + label: str, + *, + timeout: float = 10.0, + contains: bool = False, + gone: bool = False, + flag: str | None = None, + row: str | None = None, + ) -> str: + """Poll for a label (with ``flag``, or its absence); on timeout save a screenshot and fail.""" + if timeout > MAX_WAIT: + raise DriveError( + "wait", + f"--timeout must be <= {MAX_WAIT:g} s (one call stays under the 60 s cap)", + EXIT_USAGE, + ) + with self._session() as sess: + + def snap() -> str: + out = self._artefact("wait-timeout") + actions.screenshot(sess, out) + return str(out) + + return actions.wait_for( + sess, + self.clock, + label, + timeout=timeout, + contains=contains, + gone=gone, + flag=flag, + row=row, + on_timeout=snap, + ) + + def eval(self, expr: str) -> str: + """Evaluate JS; JSON result.""" + with self._session() as sess: + return actions.evaluate(sess, expr) + + def console( + self, *, limit: int = 20, warnings: bool = False, fail_on_error: bool = False + ) -> str: + """Recent console errors; ``fail_on_error`` turns any error into a failure.""" + with self._session() as sess: + out, errors = actions.console(sess, limit, warnings=warnings) + if fail_on_error and errors: + raise DriveError("console", f"{errors} console error(s):\n{out}") + return out diff --git a/flutter_web_drive/browser.py b/flutter_web_drive/browser.py new file mode 100644 index 0000000..df18790 --- /dev/null +++ b/flutter_web_drive/browser.py @@ -0,0 +1,81 @@ +"""The headless chromium unit: argv, systemd-run wrapper, size parsing.""" + +from __future__ import annotations + +import re +import subprocess +from collections.abc import Callable +from pathlib import Path + +from flutter_web_drive.errors import EXIT_USAGE, DriveError + +# The real binary: /usr/local/bin/chromium is a hosts-installer wrapper. +CHROMIUM = "/usr/bin/chromium" + +Runner = Callable[[list[str]], int] + + +def run(cmd: list[str]) -> int: + """Run a short command quietly and return its exit code.""" + return subprocess.run(cmd, check=False, capture_output=True).returncode + + +def parse_size(size: str) -> tuple[int, int]: + """``1400x1000`` -> (1400, 1000).""" + m = re.fullmatch(r"(\d+)x(\d+)", size) + if not m: + raise DriveError( + "start", f"--size must look like 1400x1000, got {size!r}", EXIT_USAGE + ) + return int(m[1]), int(m[2]) + + +def chromium_argv( + url: str, profile: Path, cdp_port: int, size: tuple[int, int] +) -> list[str]: + """Headless chromium with CDP on loopback and a persistent profile. + + ``--disable-extensions`` keeps the Claude extension (loaded via + chromium-flags.conf) out of this CDP session; ``--password-store=basic`` + stops chromium asking the desktop keyring, which could prompt on :0. + """ + width, height = size + return [ + CHROMIUM, + "--headless=new", + f"--remote-debugging-port={cdp_port}", + "--remote-debugging-address=127.0.0.1", + f"--user-data-dir={profile}", + "--disable-extensions", + "--no-first-run", + "--no-default-browser-check", + "--password-store=basic", + "--mute-audio", + "--hide-scrollbars", + "--enable-unsafe-swiftshader", + f"--window-size={width},{height}", + url, + ] + + +def unit_argv(unit: str, argv: list[str]) -> list[str]: + """``systemd-run --user`` so the browser outlives the calling shell. + + DISPLAY/WAYLAND_DISPLAY are unset in the unit: even a mis-flagged + chromium then has no live display to open a window on. + """ + return [ + "systemd-run", + "--user", + f"--unit={unit}", + "--collect", + "--quiet", + "-p", + "UnsetEnvironment=DISPLAY WAYLAND_DISPLAY", + *argv, + ] + + +def unit_active(unit: str, runner: Runner) -> bool: + """Is the systemd user unit running?""" + return runner(["systemctl", "--user", "is-active", "--quiet", unit]) == 0 diff --git a/flutter_web_drive/cdp.py b/flutter_web_drive/cdp.py new file mode 100644 index 0000000..c70f48c --- /dev/null +++ b/flutter_web_drive/cdp.py @@ -0,0 +1,139 @@ +"""Minimal CDP client: find the headless page, call methods, collect events.""" + +from __future__ import annotations + +import json +import urllib.request +from collections.abc import Callable +from typing import Any, Protocol + +import websocket + +from flutter_web_drive.errors import EXIT_BROWSER, EXIT_UNSAFE, DriveError + +CALL_TIMEOUT = 30.0 + +Fetch = Callable[[int, str], Any] + + +class Socket(Protocol): + """The slice of websocket-client's WebSocket the session uses.""" + + def send(self, payload: str) -> Any: ... + def recv(self) -> Any: ... + def settimeout(self, timeout: float) -> None: ... + def close(self) -> None: ... + + +def http_json(port: int, path: str) -> Any: + """GET a DevTools HTTP endpoint on loopback and decode the JSON.""" + with urllib.request.urlopen(f"http://127.0.0.1:{port}{path}", timeout=2.0) as resp: + return json.load(resp) + + +def create_socket(url: str) -> Socket: + """Open the page websocket (origin suppressed: chromium rejects foreign ones).""" + return websocket.create_connection(url, timeout=CALL_TIMEOUT, suppress_origin=True) + + +class Session: + """One request/response CDP conversation over a page websocket.""" + + def __init__(self, ws: Socket) -> None: + """Wrap an open websocket.""" + self.ws = ws + self.next_id = 0 + self.events: list[dict[str, Any]] = [] + + def call(self, method: str, **params: Any) -> dict[str, Any]: + """Send one command and return its result; stash events seen meanwhile.""" + self.next_id += 1 + msg_id = self.next_id + try: + self.ws.send(json.dumps({"id": msg_id, "method": method, "params": params})) + while True: + msg = json.loads(self.ws.recv()) + if msg.get("id") == msg_id: + break + self.events.append(msg) + except (OSError, websocket.WebSocketException) as exc: + raise DriveError( + "cdp", f"{method}: connection lost ({exc})", EXIT_BROWSER + ) from exc + if "error" in msg: + raise DriveError( + "cdp", f"{method}: {msg['error'].get('message', msg['error'])}" + ) + return msg.get("result", {}) + + def evaluate(self, expr: str) -> Any: + """Evaluate JS in the page (promises awaited); page exceptions raise.""" + res = self.call( + "Runtime.evaluate", expression=expr, returnByValue=True, awaitPromise=True + ) + if "exceptionDetails" in res: + det = res["exceptionDetails"] + desc = det.get("exception", {}).get("description") or det.get( + "text", "exception" + ) + raise DriveError("eval", desc.splitlines()[0]) + return res.get("result", {}).get("value") + + def drain(self, seconds: float) -> list[dict[str, Any]]: + """Collect queued events until ``seconds`` pass with none; return all seen.""" + self.ws.settimeout(seconds) + try: + while True: + self.events.append(json.loads(self.ws.recv())) + except (OSError, websocket.WebSocketException): + pass # the timeout is how a drain ends + self.ws.settimeout(CALL_TIMEOUT) + return self.events + + def close(self) -> None: + """Detach; the browser and page keep running.""" + self.ws.close() + + +def pick_page(targets: list[dict[str, Any]]) -> str: + """Websocket URL of the app page: a ``page`` target, preferring non-chrome:// ones.""" + pages = [t for t in targets if t.get("type") == "page"] + if not pages: + raise DriveError("connect", "no page target in the browser", EXIT_BROWSER) + pages.sort(key=lambda t: str(t.get("url", "")).startswith("chrome")) + return str(pages[0]["webSocketDebuggerUrl"]) + + +def probe(port: int, *, fetch: Fetch = http_json) -> dict[str, Any] | None: + """``/json/version`` of whatever listens on ``port``, or None if nothing does.""" + try: + return dict(fetch(port, "/json/version")) + except OSError: + return None + + +def is_headless(version: dict[str, Any]) -> bool: + """True only for a headless chromium: the one browser we may send input to.""" + return "HeadlessChrome" in str(version.get("User-Agent", "")) + + +def connect( + port: int, + *, + fetch: Fetch = http_json, + create: Callable[[str], Socket] = create_socket, +) -> Session: + """Attach to the headless browser on ``port``; refuse anything with a window.""" + version = probe(port, fetch=fetch) + if version is None: + raise DriveError( + "connect", f"no browser on CDP port {port}; run start first", EXIT_BROWSER + ) + if not is_headless(version): + agent = version.get("User-Agent", "?") + raise DriveError( + "connect", + f"CDP port {port} is not headless ({agent}); refusing to drive it", + EXIT_UNSAFE, + ) + return Session(create(pick_page(fetch(port, "/json")))) diff --git a/flutter_web_drive/cli.py b/flutter_web_drive/cli.py new file mode 100644 index 0000000..e990a43 --- /dev/null +++ b/flutter_web_drive/cli.py @@ -0,0 +1,150 @@ +"""``python -m flutter_web_drive [--app A] ...``: one verb per call, exit code is the verdict.""" + +from __future__ import annotations + +import argparse +import os +import sys +from pathlib import Path +from typing import Any, TextIO + +from flutter_web_drive.api import Driver +from flutter_web_drive.errors import EXIT_OK, EXIT_USAGE, DriveError + + +class _Parser(argparse.ArgumentParser): + """argparse that exits with our usage code (2 already, but explicit).""" + + def error(self, message: str) -> Any: + self.print_usage(sys.stderr) + self.exit(EXIT_USAGE, f"flutter_web_drive: {message}\n") + + +def build_parser() -> argparse.ArgumentParser: + """All verbs and flags.""" + p = _Parser(prog="flutter_web_drive", description=__doc__) + p.add_argument( + "--app", + default=os.environ.get("FWD_APP", "default"), + help="state/artefact namespace", + ) + p.add_argument("--cdp-port", type=int, help="override the port saved by start") + sub = p.add_subparsers(dest="verb", required=True) + + s = sub.add_parser("start", help="launch the headless unit on a URL") + s.add_argument("--url", required=True) + s.add_argument("--profile", required=True, type=Path) + s.add_argument("--cdp-port", dest="start_port", required=True, type=int) + s.add_argument("--unit") + s.add_argument("--size", default="1400x1000") + sub.add_parser("stop", help="stop the unit (profile kept)") + s = sub.add_parser("shot", help="screenshot the viewport") + s.add_argument("out", nargs="?", type=Path) + s.add_argument( + "--min-colours", type=int, default=0, help="fail if fewer distinct colours" + ) + s = sub.add_parser("tree", help="list semantics nodes") + s.add_argument("--filter", dest="filt") + s.add_argument( + "--all", dest="show_all", action="store_true", help="include bare containers" + ) + s.add_argument("--row", help="only nodes in the row of the node with this label") + s = sub.add_parser("tap", help="click a node by label") + s.add_argument("label") + s.add_argument("--contains", action="store_true") + s = sub.add_parser("tap-xy", help="click at viewport coordinates") + s.add_argument("x", type=float) + s.add_argument("y", type=float) + s = sub.add_parser("text", help="insert text into the focused field") + s.add_argument("value", nargs="?") + s.add_argument( + "--stdin", action="store_true", help="read the text from stdin (secrets)" + ) + s.add_argument("--file", type=Path, help="read the text from a file (secrets)") + s = sub.add_parser("key", help="press Enter, Escape, Tab, Backspace, ...") + s.add_argument("name") + s = sub.add_parser("scroll", help="mouse-wheel scroll") + s.add_argument("dx", type=float) + s.add_argument("dy", type=float) + s.add_argument("--at", nargs=2, type=float, metavar=("X", "Y")) + s = sub.add_parser("wait", help="poll until a label is present (or --gone)") + s.add_argument("label") + s.add_argument("--timeout", type=float, default=10.0) + s.add_argument("--contains", action="store_true") + s.add_argument("--gone", action="store_true") + s.add_argument("--row", help="search only the row of the node with this label") + s.add_argument( + "--flag", + help="the node must carry this flag (current, selected, checked, focused, ...)", + ) + s = sub.add_parser("eval", help="evaluate JS, print JSON") + s.add_argument("expr") + s = sub.add_parser("console", help="console errors since page load") + s.add_argument("--limit", type=int, default=20) + s.add_argument("--warnings", action="store_true") + s.add_argument("--fail-on-error", action="store_true") + return p + + +def _text_value(ns: argparse.Namespace, stdin: TextIO) -> str: + if ns.file is not None: + try: + return ns.file.read_text().removesuffix("\n") + except OSError as exc: + raise DriveError( + "text", f"cannot read {ns.file}: {exc.strerror}", EXIT_USAGE + ) from exc + if ns.stdin: + return stdin.read().removesuffix("\n") + if ns.value is None: + raise DriveError("text", "text needs a value, --stdin or --file", EXIT_USAGE) + return str(ns.value) + + +def dispatch(drv: Driver, ns: argparse.Namespace, stdin: TextIO) -> str: + """Call the Driver method for the parsed verb.""" + verb = ns.verb + if verb == "start": + return drv.start(ns.url, ns.profile, ns.start_port, unit=ns.unit, size=ns.size) + if verb == "stop": + return drv.stop() + if verb == "shot": + return drv.shot(ns.out, min_colours=ns.min_colours) + if verb == "tree": + return drv.tree(filt=ns.filt, show_all=ns.show_all, row=ns.row) + if verb == "tap": + return drv.tap(ns.label, contains=ns.contains) + if verb == "tap-xy": + return drv.tap_xy(ns.x, ns.y) + if verb == "text": + return drv.text(_text_value(ns, stdin)) + if verb == "key": + return drv.key(ns.name) + if verb == "scroll": + return drv.scroll(ns.dx, ns.dy, at=tuple(ns.at) if ns.at else None) + if verb == "wait": + return drv.wait( + ns.label, + timeout=ns.timeout, + contains=ns.contains, + gone=ns.gone, + flag=ns.flag, + row=ns.row, + ) + if verb == "eval": + return drv.eval(ns.expr) + return drv.console( + limit=ns.limit, warnings=ns.warnings, fail_on_error=ns.fail_on_error + ) + + +def main(argv: list[str] | None = None, *, stdin: TextIO = sys.stdin) -> int: + """Parse, run one verb, print its result; failures print one line to stderr.""" + ns = build_parser().parse_args(argv) + drv = Driver(ns.app, cdp_port=ns.cdp_port) + try: + print(dispatch(drv, ns, stdin)) + except DriveError as err: + print(f"flutter_web_drive: {err.step} failed: {err.message}", file=sys.stderr) + return err.code + return EXIT_OK diff --git a/flutter_web_drive/errors.py b/flutter_web_drive/errors.py new file mode 100644 index 0000000..5727797 --- /dev/null +++ b/flutter_web_drive/errors.py @@ -0,0 +1,20 @@ +"""Exit codes and the one error type every verb raises.""" + +from __future__ import annotations + +EXIT_OK = 0 +EXIT_FAIL = 1 # the step ran and its check failed: not found, timeout, page exception +EXIT_USAGE = 2 # bad arguments +EXIT_BROWSER = 3 # no driver browser reachable (not started, crashed, CDP gone) +EXIT_UNSAFE = 4 # the CDP port belongs to a browser we must not drive + + +class DriveError(Exception): + """A verb failed at ``step``; ``code`` becomes the process exit code.""" + + def __init__(self, step: str, message: str, code: int = EXIT_FAIL) -> None: + """Record the failing step, a one-line reason and the exit code.""" + super().__init__(f"{step}: {message}") + self.step = step + self.message = message + self.code = code diff --git a/flutter_web_drive/keys.py b/flutter_web_drive/keys.py new file mode 100644 index 0000000..e91ca18 --- /dev/null +++ b/flutter_web_drive/keys.py @@ -0,0 +1,22 @@ +"""Named keys the ``key`` verb can press.""" + +from __future__ import annotations + +# name -> (key, code, windowsVirtualKeyCode, text). Text makes chromium emit +# the char event Flutter's text input listens for (Enter submits a field). +KEYS: dict[str, tuple[str, str, int, str]] = { + "Enter": ("Enter", "Enter", 13, "\r"), + "Escape": ("Escape", "Escape", 27, ""), + "Tab": ("Tab", "Tab", 9, ""), + "Backspace": ("Backspace", "Backspace", 8, ""), + "Delete": ("Delete", "Delete", 46, ""), + "Space": (" ", "Space", 32, " "), + "ArrowUp": ("ArrowUp", "ArrowUp", 38, ""), + "ArrowDown": ("ArrowDown", "ArrowDown", 40, ""), + "ArrowLeft": ("ArrowLeft", "ArrowLeft", 37, ""), + "ArrowRight": ("ArrowRight", "ArrowRight", 39, ""), + "Home": ("Home", "Home", 36, ""), + "End": ("End", "End", 35, ""), + "PageUp": ("PageUp", "PageUp", 33, ""), + "PageDown": ("PageDown", "PageDown", 34, ""), +} diff --git a/flutter_web_drive/mcp_server.py b/flutter_web_drive/mcp_server.py new file mode 100644 index 0000000..3d8bf83 --- /dev/null +++ b/flutter_web_drive/mcp_server.py @@ -0,0 +1,173 @@ +"""Stdio MCP server exposing the driver's verbs as tools (thin: every tool is one Driver call). + +Register (user scope) with: + claude mcp add flutter-web-drive -s user -e PYTHONPATH=$HOME/src/utils \ + -- python3 -m flutter_web_drive.mcp_server +""" + +from __future__ import annotations + +from collections.abc import Callable +from pathlib import Path + +from mcp.server.mcpserver import MCPServer +from mcp.server.mcpserver.exceptions import ToolError + +from flutter_web_drive.api import Driver +from flutter_web_drive.errors import DriveError + +server = MCPServer( + "flutter-web-drive", + instructions=( + "Drive a Flutter web app in a private headless chromium (never the user's screen). " + "start once, then tree -> tap/text/key/scroll -> wait -> shot. Widgets are addressed " + "by their semantics label as shown by `tree`. `app` namespaces state per app." + ), +) + + +def _run(app: str, call: Callable[[Driver], str]) -> str: + """Run one verb; a DriveError becomes a tool error naming the failing step.""" + try: + return call(Driver(app)) + except DriveError as err: + raise ToolError(f"{err.step} failed: {err.message}") from err + + +@server.tool() +def start( + url: str, + profile: str, + cdp_port: int, + app: str = "default", + unit: str | None = None, + size: str = "1400x1000", +) -> str: + """Launch (or reuse) a headless chromium unit on url with a persistent profile dir.""" + return _run( + app, lambda d: d.start(url, Path(profile), cdp_port, unit=unit, size=size) + ) + + +@server.tool() +def stop(app: str = "default") -> str: + """Stop the app's headless chromium unit (profile kept).""" + return _run(app, lambda d: d.stop()) + + +@server.tool() +def shot(out: str | None = None, min_colours: int = 0, app: str = "default") -> str: + """Screenshot the viewport to a PNG; returns 'path WxH colours=N stddev=S'.""" + return _run( + app, lambda d: d.shot(Path(out) if out else None, min_colours=min_colours) + ) + + +@server.tool() +def tree( + filt: str | None = None, + show_all: bool = False, + row: str | None = None, + app: str = "default", +) -> str: + """List semantics nodes: '#id role "label" = "value" [tap,current,...] @x,y wxh'. row: only the row of that label.""" + return _run(app, lambda d: d.tree(filt=filt, show_all=show_all, row=row)) + + +@server.tool() +def tap(label: str, contains: bool = False, app: str = "default") -> str: + """Click the node with this label (exact label or one of its lines; contains=substring).""" + return _run(app, lambda d: d.tap(label, contains=contains)) + + +@server.tool() +def tap_xy(x: float, y: float, app: str = "default") -> str: + """Click at viewport CSS-pixel coordinates.""" + return _run(app, lambda d: d.tap_xy(x, y)) + + +@server.tool() +def text( + value: str | None = None, from_file: str | None = None, app: str = "default" +) -> str: + """Insert text into the focused field. Use from_file for secrets so they never appear in the transcript.""" + if from_file: + try: + value = Path(from_file).read_text().removesuffix("\n") + except OSError as exc: + raise ToolError( + f"text failed: cannot read {from_file}: {exc.strerror}" + ) from exc + if value is None: + raise ToolError("text failed: pass value or from_file") + secret = value + return _run(app, lambda d: d.text(secret)) + + +@server.tool() +def key(name: str, app: str = "default") -> str: + """Press a key: Enter, Escape, Tab, Backspace, Delete, Space, Arrow*, Home, End, PageUp, PageDown.""" + return _run(app, lambda d: d.key(name)) + + +@server.tool() +def scroll( + dx: float, + dy: float, + at_x: float | None = None, + at_y: float | None = None, + app: str = "default", +) -> str: + """Mouse-wheel scroll by (dx, dy) at (at_x, at_y), default viewport center.""" + at = (at_x, at_y) if at_x is not None and at_y is not None else None + return _run(app, lambda d: d.scroll(dx, dy, at=at)) + + +@server.tool() +def wait( + label: str, + timeout: float = 10.0, + contains: bool = False, + gone: bool = False, + flag: str | None = None, + row: str | None = None, + app: str = "default", +) -> str: + """Poll until a label (carrying flag, e.g. current/checked) is present, or absent with gone; timeout <= 50 s, screenshot on failure.""" + return _run( + app, + lambda d: d.wait( + label, timeout=timeout, contains=contains, gone=gone, flag=flag + ), + ) + + +@server.tool() +def eval_js(expr: str, app: str = "default") -> str: + """Evaluate JavaScript in the page; returns the JSON value.""" + return _run(app, lambda d: d.eval(expr)) + + +@server.tool() +def console( + limit: int = 20, + warnings: bool = False, + fail_on_error: bool = False, + app: str = "default", +) -> str: + """Console errors (and optionally warnings) since page load.""" + return _run( + app, + lambda d: d.console( + limit=limit, warnings=warnings, fail_on_error=fail_on_error + ), + ) + + +def main() -> None: + """Serve over stdio.""" + server.run(transport="stdio") + + +if __name__ == "__main__": # pragma: no cover - exercised by the stdio smoke test + main() diff --git a/flutter_web_drive/requirements.txt b/flutter_web_drive/requirements.txt new file mode 100644 index 0000000..6de2ebf --- /dev/null +++ b/flutter_web_drive/requirements.txt @@ -0,0 +1,3 @@ +# Exact pins, newest stable (shared dependency-freshness gate). +mcp==2.3.0 +websocket-client==1.9.2 diff --git a/flutter_web_drive/semantics.py b/flutter_web_drive/semantics.py new file mode 100644 index 0000000..f9b12dd --- /dev/null +++ b/flutter_web_drive/semantics.py @@ -0,0 +1,228 @@ +"""Flutter web's semantics DOM: the only addressable view of a canvas-rendered app. + +Flutter paints to a canvas, so widgets are found through the accessibility +mirror it writes as ``[id^=flt-semantic-node-]`` elements (``flt-semantics``, +but also ``h1``-``h6`` for headings). The mirror is off until something asks +for it; clicking ``flt-semantics-placeholder`` is how a screen reader does. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from flutter_web_drive.errors import DriveError + +# Turns semantics on (idempotent). 'on' | 'clicked' | 'missing' (Flutter not booted). +ENABLE_JS = """(() => { + if (document.querySelector('[id^="flt-semantic-node-"]')) return 'on'; + const p = document.querySelector('flt-semantics-placeholder'); + if (!p) return 'missing'; + p.click(); + return 'clicked'; +})()""" + +FLUTTER_READY_JS = ( + "!!document.querySelector('flt-semantics-placeholder, flt-semantics-host')" +) + +FOCUS_JS = """(() => { + const e = document.activeElement; + return !!e && (e.tagName === 'INPUT' || e.tagName === 'TEXTAREA' || e.isContentEditable); +})()""" + +VIEWPORT_JS = "[innerWidth, innerHeight]" + +# One record per semantics node. A node's own label is its aria-label, else its +# direct text / children (not descendant nodes': those are listed on +# their own). value = field value, aria-valuenow or aria-description. +# Password fields report asterisks, never their value. +COLLECT_JS = """(() => { + const flagsOf = (e) => { + const a = (n) => e.getAttribute(n); + const f = []; + if (a('aria-current') && a('aria-current') !== 'false') f.push('current'); + if (a('aria-selected') === 'true') f.push('selected'); + if (a('aria-checked') === 'true') f.push('checked'); + if (a('aria-checked') === 'mixed') f.push('mixed'); + if (a('aria-pressed') === 'true') f.push('pressed'); + if (a('aria-expanded') === 'true') f.push('expanded'); + if (a('aria-disabled') === 'true' || e.disabled) f.push('disabled'); + if (document.activeElement === e) f.push('focused'); + return f; + }; + return [...document.querySelectorAll('[id^="flt-semantic-node-"]')].map((e) => { + const field = e.matches('input,textarea') + ? e : e.querySelector(':scope > input, :scope > textarea'); + let own = ''; + for (const c of e.childNodes) { + // Plain Text widgets render as a bare text node or a scaled . + if (c.nodeType === 3) own += c.textContent; + else if (c.tagName === 'SPAN' && !c.id.startsWith('flt-semantic-node-')) own += c.textContent; + } + const flags = flagsOf(e); + if (field && field !== e) for (const x of flagsOf(field)) if (!flags.includes(x)) flags.push(x); + const r = e.getBoundingClientRect(); + const fieldLabel = field && (field.getAttribute('aria-label') || field.placeholder); + let value = e.getAttribute('aria-valuenow') || e.getAttribute('aria-description') || ''; + if (field) value = field.type === 'password' ? '*'.repeat(field.value.length) : field.value; + return { + id: Number(e.id.slice(18)), tag: e.tagName.toLowerCase(), + role: e.getAttribute('role') || (field ? 'textbox' : ''), + label: (e.getAttribute('aria-label') || fieldLabel || own || '').trim(), value, + x: Math.round(r.x), y: Math.round(r.y), w: Math.round(r.width), h: Math.round(r.height), + tappable: e.hasAttribute('flt-tappable'), flags, + }; + }); +})()""" + +# Called as `${IMAGE_STATS_FN}("")`: decodes the screenshot in the +# page and returns size, distinct colours (capped) and luminance stddev. +IMAGE_STATS_FN = """(async (b64) => { + const bin = Uint8Array.from(atob(b64), (c) => c.charCodeAt(0)); + const bmp = await createImageBitmap(new Blob([bin], {type: 'image/png'})); + const g = new OffscreenCanvas(bmp.width, bmp.height).getContext('2d'); + g.drawImage(bmp, 0, 0); + const d = g.getImageData(0, 0, bmp.width, bmp.height).data; + const seen = new Set(); + let s = 0, s2 = 0; + for (let i = 0; i < d.length; i += 4) { + const l = 0.299 * d[i] + 0.587 * d[i + 1] + 0.114 * d[i + 2]; + s += l; s2 += l * l; + if (seen.size < 100000) seen.add((d[i] << 16) | (d[i + 1] << 8) | d[i + 2]); + } + const n = d.length / 4, mean = s / n; + return {w: bmp.width, h: bmp.height, colours: seen.size, + stddev: Math.sqrt(Math.max(0, s2 / n - mean * mean))}; +})""" + +HINT_LIMIT = 12 + + +@dataclass(frozen=True) +class Node: + """One semantics node with its on-screen rectangle (CSS pixels).""" + + id: int + tag: str + role: str + label: str + value: str + x: float + y: float + w: float + h: float + tappable: bool + flags: tuple[str, ...] + + @property + def lines(self) -> tuple[str, ...]: + """The label split on newlines (Flutter joins merged labels with them).""" + return tuple(part.strip() for part in self.label.split("\n")) + + @property + def center(self) -> tuple[float, float]: + """Where a pointer should land to hit this node.""" + return (self.x + self.w / 2, self.y + self.h / 2) + + @property + def one_line(self) -> str: + """Label with newlines shown as ' | '.""" + return " | ".join(self.lines) + + @property + def interesting(self) -> bool: + """Worth listing in the compact tree: not a bare layout container.""" + return bool(self.role or self.label or self.value or self.tappable) + + +def parse_nodes(raw: Any) -> list[Node]: + """Typed nodes from COLLECT_JS output.""" + if not isinstance(raw, list): + raise DriveError( + "semantics", f"unexpected semantics payload: {type(raw).__name__}" + ) + return [ + Node( + id=int(r["id"]), + tag=str(r["tag"]), + role=str(r["role"]), + label=str(r["label"]), + value=str(r["value"]), + x=float(r["x"]), + y=float(r["y"]), + w=float(r["w"]), + h=float(r["h"]), + tappable=bool(r["tappable"]), + flags=tuple(str(f) for f in r["flags"]), + ) + for r in raw + ] + + +def matches(node: Node, target: str, *, contains: bool = False) -> bool: + """Exact: the whole label, its `tree` form ("a | b") or one line. Contains: substring, any case.""" + if contains: + return target.lower() in node.label.lower() + return target in (node.label, node.one_line, *node.lines) + + +def find(nodes: list[Node], target: str, *, contains: bool = False) -> list[Node]: + """Matching nodes, tappable ones first, otherwise in document order.""" + hits = [n for n in nodes if matches(n, target, contains=contains)] + return sorted(hits, key=lambda n: not n.tappable) + + +def in_row(nodes: list[Node], anchor: str, *, contains: bool = False) -> list[Node]: + """Nodes whose vertical center lies inside the row of the node labelled ``anchor``. + + For "the Mark-read button of Chapter 5" style lookups. No anchor: []. + """ + hits = find(nodes, anchor, contains=contains) + if not hits: + return [] + top, bottom = hits[0].y, hits[0].y + hits[0].h + return [n for n in nodes if n.id != hits[0].id and top <= n.center[1] <= bottom] + + +def describe(node: Node) -> str: + """``#id role "label" = "value" [tags] @x,y wxh`` on one line.""" + out = f'#{node.id} {node.role or node.tag} "{node.one_line}"' + if node.value: + out += f' = "{node.value}"' + tags = (["tap"] if node.tappable else []) + list(node.flags) + if tags: + out += f" [{','.join(tags)}]" + return out + f" @{node.x:g},{node.y:g} {node.w:g}x{node.h:g}" + + +def format_tree( + nodes: list[Node], + *, + filt: str | None = None, + show_all: bool = False, + row: str | None = None, +) -> str: + """Compact listing, one node per line; ``filt`` matches label, role or value; ``row`` scopes to a row.""" + if not nodes: + return "(no semantics nodes)" + shown = [ + n for n in (in_row(nodes, row) if row else nodes) if show_all or n.interesting + ] + if filt: + needle = filt.lower() + shown = [ + n for n in shown if needle in f"{n.label}\n{n.role}\n{n.value}".lower() + ] + if not shown: + return "(no matching semantics nodes)" + return "\n".join(describe(n) for n in shown) + + +def hint(nodes: list[Node]) -> str: + """Tappable labels on screen, for 'not found' messages an agent can act on.""" + labels = [f'"{n.one_line}"' for n in nodes if n.tappable and n.label] + if not labels: + return "(no tappable nodes)" + more = ", ..." if len(labels) > HINT_LIMIT else "" + return ", ".join(labels[:HINT_LIMIT]) + more diff --git a/flutter_web_drive/state.py b/flutter_web_drive/state.py new file mode 100644 index 0000000..c50d3b6 --- /dev/null +++ b/flutter_web_drive/state.py @@ -0,0 +1,59 @@ +"""Per-app state file and artefact directory under ~/data/claude-scratch/e2e//.""" + +from __future__ import annotations + +import json +import os +import time +from dataclasses import asdict, dataclass +from pathlib import Path + + +def artefact_root() -> Path: + """``$FWD_ARTEFACTS`` or ~/data/claude-scratch/e2e (never /tmp: it is wiped).""" + env = os.environ.get("FWD_ARTEFACTS") + return Path(env) if env else Path.home() / "data" / "claude-scratch" / "e2e" + + +def stamp() -> str: + """Sortable timestamp for artefact names.""" + now = time.time() + return ( + time.strftime("%Y%m%d-%H%M%S", time.localtime(now)) + + f"-{int(now * 1000) % 1000:03d}" + ) + + +@dataclass(frozen=True) +class State: + """What ``start`` launched, so later verbs attach to the same browser.""" + + unit: str + cdp_port: int + url: str + profile: str + + +def state_path(root: Path, app: str) -> Path: + """Where ``app``'s state lives.""" + return root / app / "state.json" + + +def load(root: Path, app: str) -> State | None: + """The saved state, or None before the first start / after stop.""" + path = state_path(root, app) + if not path.is_file(): + return None + return State(**json.loads(path.read_text())) + + +def save(root: Path, app: str, state: State) -> None: + """Persist state (creates the app's artefact dir).""" + path = state_path(root, app) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(asdict(state), indent=2) + "\n") + + +def clear(root: Path, app: str) -> None: + """Forget the state after stop.""" + state_path(root, app).unlink(missing_ok=True) diff --git a/flutter_web_drive/tests/__init__.py b/flutter_web_drive/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/flutter_web_drive/tests/fakes.py b/flutter_web_drive/tests/fakes.py new file mode 100644 index 0000000..509fd5a --- /dev/null +++ b/flutter_web_drive/tests/fakes.py @@ -0,0 +1,179 @@ +"""In-memory stand-ins for a headless chromium page and its CDP websocket.""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any + +import websocket + +from flutter_web_drive import semantics + +Handler = Callable[[str, dict[str, Any]], dict[str, Any]] + + +def node( + nid: int, + label: str = "", + *, + role: str = "button", + value: str = "", + rect: tuple[int, int, int, int] = (0, 0, 80, 40), + tappable: bool = True, + flags: tuple[str, ...] = (), + tag: str = "flt-semantics", +) -> dict[str, Any]: + """Build one raw node record the way COLLECT_JS returns it.""" + x, y, w, h = rect + return { + "id": nid, + "tag": tag, + "role": role, + "label": label, + "value": value, + "x": x, + "y": y, + "w": w, + "h": h, + "tappable": tappable, + "flags": list(flags), + } + + +class FakeWs: + """A websocket whose replies come from a handler, with queued events.""" + + def __init__( + self, handler: Handler, events: list[dict[str, Any]] | None = None + ) -> None: + """Answer each request with ``handler``; replay ``events`` on enable.""" + self.handler = handler + self.events = list(events or []) + self.sent: list[dict[str, Any]] = [] + self.queue: list[dict[str, Any]] = [] + self.timeout: float | None = None + self.closed = False + + def send(self, raw: str) -> None: + """Record the request and queue its reply (events first on enable).""" + msg = json.loads(raw) + self.sent.append(msg) + if msg["method"] in {"Runtime.enable", "Log.enable"}: + self.queue.extend(self.events) + self.events = [] + self.queue.append( + {"id": msg["id"], **self.handler(msg["method"], msg["params"])} + ) + + def recv(self) -> str: + """Pop the next queued message, or time out like the real client.""" + if not self.queue: + raise websocket.WebSocketTimeoutException("timed out") + return json.dumps(self.queue.pop(0)) + + def settimeout(self, seconds: float) -> None: + """Remember the timeout so tests can assert drain restored it.""" + self.timeout = seconds + + def close(self) -> None: + """Mark closed.""" + self.closed = True + + +@dataclass +class FakePage: + """A Flutter web page as seen through Runtime.evaluate and Input.*.""" + + nodes: list[dict[str, Any]] = field(default_factory=list) + semantics_on: bool = True + flutter_ready: bool = True + focused: bool = True + href: str = "http://127.0.0.1:1/" + viewport: tuple[int, int] = (1400, 900) + stats: dict[str, Any] = field( + default_factory=lambda: {"w": 1400, "h": 900, "colours": 180, "stddev": 9.0}, + ) + evals: dict[str, Any] = field(default_factory=dict) + clicks: list[tuple[float, float]] = field(default_factory=list) + + def evaluate(self, expr: str) -> dict[str, Any]: + """Answer the driver's known JS snippets; fall back to ``evals``.""" + value: Any + if expr == semantics.ENABLE_JS: + value = "on" if self.semantics_on else "missing" + elif expr == semantics.COLLECT_JS: + value = self.nodes if self.semantics_on else [] + elif expr == semantics.FLUTTER_READY_JS: + value = self.flutter_ready + elif expr == semantics.FOCUS_JS: + value = self.focused + elif expr == semantics.VIEWPORT_JS: + value = list(self.viewport) + elif expr == "location.href": + value = self.href + elif expr.startswith(semantics.IMAGE_STATS_FN): + value = self.stats + elif expr in self.evals: + value = self.evals[expr] + if isinstance(value, Exception): + return { + "result": { + "result": {"type": "object"}, + "exceptionDetails": { + "text": "Uncaught", + "exception": {"description": f"Error: {value}\n at x"}, + }, + } + } + else: # pragma: no cover - a test sent JS the fake does not model + raise AssertionError(f"unexpected evaluate: {expr[:80]}") + return {"result": {"result": {"type": "object", "value": value}}} + + def handle(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + """CDP handler for FakeWs.""" + if method == "Runtime.evaluate": + return self.evaluate(params["expression"]) + if method == "Page.captureScreenshot": + return {"result": {"data": "iVBORw0KGgo="}} + if method == "Input.dispatchMouseEvent" and params["type"] == "mouseReleased": + self.clicks.append((params["x"], params["y"])) + return {"result": {}} + + def ws(self, events: list[dict[str, Any]] | None = None) -> FakeWs: + """A fresh websocket onto this page.""" + return FakeWs(self.handle, events) + + +class FakeClock: + """Deterministic monotonic clock; sleeping advances it.""" + + def __init__(self) -> None: + """Start at t=0.""" + self.t = 0.0 + self.slept: list[float] = [] + + def now(self) -> float: + """Current fake time.""" + return self.t + + def sleep(self, seconds: float) -> None: + """Advance fake time.""" + self.slept.append(seconds) + self.t += seconds + + +HEADLESS = { + "Browser": "Chrome/155", + "User-Agent": "Mozilla/5.0 HeadlessChrome/155.0.0.0", +} +HEADED = {"Browser": "Chrome/126", "User-Agent": "Mozilla/5.0 Valve Steam Client"} +PAGE_TARGETS = [ + { + "type": "browser_ui", + "url": "chrome://omnibox", + "webSocketDebuggerUrl": "ws://ui", + }, + {"type": "page", "url": "http://127.0.0.1:1/", "webSocketDebuggerUrl": "ws://page"}, +] diff --git a/flutter_web_drive/tests/test_actions.py b/flutter_web_drive/tests/test_actions.py new file mode 100644 index 0000000..d5c2f14 --- /dev/null +++ b/flutter_web_drive/tests/test_actions.py @@ -0,0 +1,158 @@ +"""Verbs that act on one attached page.""" + +from __future__ import annotations + +import pytest + +from flutter_web_drive import actions, cdp +from flutter_web_drive.errors import EXIT_USAGE, DriveError +from flutter_web_drive.tests.fakes import FakeClock, FakePage, node + +TABS = [ + node(4, "Library\nTab 1 of 4", rect=(0, 8, 80, 67), flags=("current",)), + node(5, "Browse\nTab 2 of 4", rect=(0, 75, 80, 67)), + node(12, "Library", role="", tag="h2", tappable=False), +] + + +def _sess(page: FakePage) -> tuple[cdp.Session, FakeClock]: + return cdp.Session(page.ws()), FakeClock() + + +def test_ensure_semantics_returns_nodes_once_they_appear() -> None: + page = FakePage(nodes=TABS, semantics_on=False) + sess, clock = _sess(page) + clock_sleep = clock.sleep + + def flip(seconds: float) -> None: + clock_sleep(seconds) + page.semantics_on = True + + clock.sleep = flip # type: ignore[method-assign] + assert [n.id for n in actions.ensure_semantics(sess, clock)] == [4, 5, 12] + + +def test_ensure_semantics_times_out() -> None: + sess, clock = _sess(FakePage(semantics_on=False)) + with pytest.raises(DriveError, match="no Flutter semantics"): + actions.ensure_semantics(sess, clock, timeout=1) + + +def test_tap_clicks_center_of_best_match_and_reports_ambiguity() -> None: + page = FakePage(nodes=TABS) + sess, clock = _sess(page) + out = actions.tap(sess, clock, "Library") + assert page.clicks == [(40.0, 41.5)] + assert out.startswith("tapped #4 button") + assert "(2 matches; took the first)" in out + assert "matches" not in actions.tap(sess, clock, "browse", contains=True) + + +def test_tap_unknown_label_lists_what_is_tappable() -> None: + sess, clock = _sess(FakePage(nodes=TABS)) + with pytest.raises( + DriveError, match=r'no node labelled "Sources".*"Browse \| Tab 2 of 4"' + ): + actions.tap(sess, clock, "Sources") + + +def test_tap_refuses_offscreen_nodes() -> None: + page = FakePage(nodes=[node(7, "Far", rect=(0, 2000, 10, 10))]) + sess, clock = _sess(page) + with pytest.raises(DriveError, match="off-screen"): + actions.tap(sess, clock, "Far") + assert page.clicks == [] + + +def test_tap_xy_sends_move_press_release() -> None: + page = FakePage() + sess = cdp.Session(ws := page.ws()) + assert actions.tap_xy(sess, 3, 4) == "tapped (3, 4)" + kinds = [ + m["params"]["type"] + for m in ws.sent + if m["method"] == "Input.dispatchMouseEvent" + ] + assert kinds == ["mouseMoved", "mousePressed", "mouseReleased"] + + +def test_insert_text_needs_a_focused_field() -> None: + page = FakePage(focused=False) + sess = cdp.Session(ws := page.ws()) + with pytest.raises(DriveError, match="tap the field first"): + actions.insert_text(sess, "x") + page.focused = True + assert actions.insert_text(sess, "secret") == "inserted 6 characters" + assert ws.sent[-1] == { + "id": 3, + "method": "Input.insertText", + "params": {"text": "secret"}, + } + + +def test_press_key_sends_down_and_up_with_text_for_enter() -> None: + page = FakePage() + sess = cdp.Session(ws := page.ws()) + assert actions.press_key(sess, "Enter") == "pressed Enter" + down, up = ws.sent + assert down["params"]["type"] == "keyDown" + assert down["params"]["text"] == "\r" + assert up["params"]["type"] == "keyUp" + actions.press_key(sess, "Escape") + assert "text" not in ws.sent[2]["params"] + + +def test_press_key_rejects_unknown_names() -> None: + sess = cdp.Session(FakePage().ws()) + with pytest.raises(DriveError) as err: + actions.press_key(sess, "Hyper") + assert err.value.code == EXIT_USAGE + + +def test_scroll_defaults_to_viewport_center() -> None: + page = FakePage(viewport=(1000, 800)) + sess = cdp.Session(ws := page.ws()) + assert actions.scroll(sess, 0, 300) == "scrolled (0, 300) at (500, 400)" + wheel = ws.sent[-1]["params"] + assert (wheel["type"], wheel["x"], wheel["deltaY"]) == ("mouseWheel", 500, 300) + assert actions.scroll(sess, 1, 2, at=(5, 6)) == "scrolled (1, 2) at (5, 6)" + + +def test_wait_for_present_and_gone() -> None: + page = FakePage(nodes=[]) + sess, clock = _sess(page) + sleep = clock.sleep + + def appear(seconds: float) -> None: + sleep(seconds) + page.nodes = TABS + + clock.sleep = appear # type: ignore[method-assign] + found = actions.wait_for(sess, clock, "Browse", timeout=5, on_timeout=lambda: "x") + assert found.startswith("found #5") + gone = actions.wait_for( + sess, clock, "Sources", timeout=5, gone=True, on_timeout=lambda: "x" + ) + assert gone == 'gone: "Sources"' + + +def test_wait_for_flag_requires_the_flag() -> None: + sess, clock = _sess(FakePage(nodes=TABS)) + found = actions.wait_for( + sess, clock, "Library", timeout=1, flag="current", on_timeout=lambda: "x" + ) + assert found.startswith("found #4") + with pytest.raises(DriveError, match=r'"Browse" \[current\] not present'): + actions.wait_for( + sess, clock, "Browse", timeout=1, flag="current", on_timeout=lambda: "x" + ) + gone = actions.wait_for( + sess, + clock, + "Browse", + timeout=1, + flag="current", + gone=True, + on_timeout=lambda: "x", + ) + assert gone == 'gone: "Browse" [current]' diff --git a/flutter_web_drive/tests/test_actions_output.py b/flutter_web_drive/tests/test_actions_output.py new file mode 100644 index 0000000..9dbe315 --- /dev/null +++ b/flutter_web_drive/tests/test_actions_output.py @@ -0,0 +1,134 @@ +"""Verbs that report from one attached page: shots, eval, console, rows.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from flutter_web_drive import actions, cdp +from flutter_web_drive.errors import DriveError +from flutter_web_drive.tests.fakes import FakeClock, FakePage, node + +TABS = [ + node(4, "Library\nTab 1 of 4", rect=(0, 8, 80, 67), flags=("current",)), + node(5, "Browse\nTab 2 of 4", rect=(0, 75, 80, 67)), + node(12, "Library", role="", tag="h2", tappable=False), +] + + +def _sess(page: FakePage) -> tuple[cdp.Session, FakeClock]: + return cdp.Session(page.ws()), FakeClock() + + +def test_wait_for_timeout_names_screenshot() -> None: + sess, clock = _sess(FakePage(nodes=TABS)) + with pytest.raises(DriveError, match=r"present after 1s; screenshot /s\.png"): + actions.wait_for( + sess, clock, "Library", timeout=1, gone=True, on_timeout=lambda: "/s.png" + ) + with pytest.raises(DriveError, match="not present"): + actions.wait_for(sess, clock, "Nope", timeout=1, on_timeout=lambda: "/s.png") + + +def test_screenshot_writes_png_and_reports_stats(tmp_path: Path) -> None: + sess = cdp.Session(FakePage().ws()) + out = tmp_path / "a" / "s.png" + msg = actions.screenshot(sess, out, min_colours=2) + assert out.read_bytes().startswith(b"\x89PNG") + assert msg == f"{out} 1400x900 colours=180 stddev=9.0" + + +def test_screenshot_flags_blank_frames(tmp_path: Path) -> None: + page = FakePage(stats={"w": 10, "h": 10, "colours": 1, "stddev": 0.0}) + with pytest.raises(DriveError, match="looks blank"): + actions.screenshot(cdp.Session(page.ws()), tmp_path / "b.png", min_colours=2) + + +def test_evaluate_serialises_value() -> None: + sess = cdp.Session(FakePage(evals={"x": {"a": [1]}}).ws()) + assert actions.evaluate(sess, "x") == '{"a": [1]}' + + +EVENTS = [ + { + "method": "Runtime.consoleAPICalled", + "params": {"type": "log", "args": [{"value": "hi"}]}, + }, + { + "method": "Runtime.consoleAPICalled", + "params": {"type": "error", "args": [{"value": "bad"}, {"description": "Obj"}]}, + }, + { + "method": "Runtime.consoleAPICalled", + "params": {"type": "warning", "args": [{"value": "meh"}]}, + }, + { + "method": "Runtime.exceptionThrown", + "params": { + "exceptionDetails": { + "text": "Uncaught", + "exception": {"description": "Error: x\n at y"}, + } + }, + }, + { + "method": "Runtime.exceptionThrown", + "params": {"exceptionDetails": {"text": "Uncaught"}}, + }, + { + "method": "Log.entryAdded", + "params": {"entry": {"level": "error", "text": "404"}}, + }, + { + "method": "Log.entryAdded", + "params": {"entry": {"level": "warning", "text": "webgl"}}, + }, + { + "method": "Log.entryAdded", + "params": {"entry": {"level": "info", "text": "fine"}}, + }, + {"method": "Page.loadEventFired", "params": {}}, +] + + +def test_console_lists_errors_and_optionally_warnings() -> None: + page = FakePage() + out, errors = actions.console(cdp.Session(page.ws(EVENTS))) + assert errors == 4 + assert out.splitlines() == [ + "error: bad Obj", + "error: Uncaught Error: x", + "error: Uncaught", + "error: 404", + ] + out, _ = actions.console(cdp.Session(page.ws(EVENTS)), limit=2, warnings=True) + assert out.splitlines() == ["error: 404", "warning: webgl"] + assert actions.console(cdp.Session(page.ws([]))) == ("no console errors", 0) + + +def test_wait_for_row_scopes_the_search() -> None: + page = FakePage( + nodes=[ + node(1, "Chapter 2", rect=(0, 100, 1000, 60)), + node(2, "Mark unread", rect=(900, 110, 40, 40)), + node(3, "Chapter 1", rect=(0, 160, 1000, 60)), + node(4, "Mark read", rect=(900, 170, 40, 40)), + ] + ) + sess, clock = _sess(page) + hit = actions.wait_for( + sess, clock, "Mark unread", row="Chapter 2", timeout=1, on_timeout=lambda: "x" + ) + assert hit.startswith("found #2") + with pytest.raises( + DriveError, match=r'"Mark unread" in row "Chapter 1" not present' + ): + actions.wait_for( + sess, + clock, + "Mark unread", + row="Chapter 1", + timeout=1, + on_timeout=lambda: "x", + ) diff --git a/flutter_web_drive/tests/test_api.py b/flutter_web_drive/tests/test_api.py new file mode 100644 index 0000000..c972b0b --- /dev/null +++ b/flutter_web_drive/tests/test_api.py @@ -0,0 +1,212 @@ +"""Driver: unit lifecycle, state file, and each verb's attach-act-detach.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +from flutter_web_drive import browser +from flutter_web_drive.api import Driver +from flutter_web_drive.errors import EXIT_BROWSER, EXIT_UNSAFE, EXIT_USAGE, DriveError +from flutter_web_drive.tests.fakes import ( + HEADLESS, + PAGE_TARGETS, + FakeClock, + FakePage, + node, +) + + +class Rig: + """A fake systemd + chromium: systemd-run makes the CDP port answer.""" + + def __init__( + self, tmp_path: Path, *, up: bool = False, active: bool = False + ) -> None: + self.page = FakePage(nodes=[node(5, "Browse")]) + self.up = up + self.active = active + self.cmds: list[list[str]] = [] + self.run_rc = 0 + self.comes_up = True + self.root = tmp_path / "e2e" + + def fetch(self, port: int, path: str) -> Any: + if not self.up: + raise OSError("refused") + return HEADLESS if path == "/json/version" else PAGE_TARGETS + + def runner(self, cmd: list[str]) -> int: + self.cmds.append(cmd) + if cmd[:3] == ["systemctl", "--user", "is-active"]: + return 0 if self.active else 3 + if cmd[0] == "systemd-run": + self.up = self.comes_up + return self.run_rc + return 0 + + def driver(self, **kw: Any) -> Driver: + return Driver( + "kartka", + root=self.root, + fetch=self.fetch, + create=lambda url: self.page.ws(), + runner=self.runner, + clock=FakeClock(), + **kw, + ) + + def state(self) -> dict[str, Any]: + return json.loads((self.root / "kartka" / "state.json").read_text()) + + +def _start(rig: Rig, tmp_path: Path) -> str: + return rig.driver().start( + "http://127.0.0.1:1/", tmp_path / "prof", 9341, size="800x600" + ) + + +def test_start_launches_unit_saves_state_and_enables_semantics(tmp_path: Path) -> None: + rig = Rig(tmp_path) + out = _start(rig, tmp_path) + assert ( + out == "started fwd-kartka: http://127.0.0.1:1/ on CDP 9341, viewport 1400x900" + ) + run = next(c for c in rig.cmds if c[0] == "systemd-run") + assert "--unit=fwd-kartka" in run + assert "UnsetEnvironment=DISPLAY WAYLAND_DISPLAY" in run + assert "--headless=new" in run + assert "--window-size=800,600" in run + assert ["systemctl", "--user", "reset-failed", "fwd-kartka"] in rig.cmds + assert (tmp_path / "prof").is_dir() + assert rig.state()["cdp_port"] == 9341 + + +def test_start_refuses_a_port_someone_else_owns(tmp_path: Path) -> None: + rig = Rig(tmp_path, up=True) + with pytest.raises(DriveError) as err: + _start(rig, tmp_path) + assert err.value.code == EXIT_UNSAFE + assert not any(c[0] == "systemd-run" for c in rig.cmds) + + +def test_start_reuses_a_running_unit_and_navigates_if_needed(tmp_path: Path) -> None: + rig = Rig(tmp_path, up=True, active=True) + rig.page.href = "about:blank" + rig.page.evals = {} + out = _start(rig, tmp_path) + assert out.startswith("reused fwd-kartka") + assert not any(c[0] == "systemd-run" for c in rig.cmds) + rig.page.href = "http://127.0.0.1:1/#/browse" + assert _start(rig, tmp_path).startswith("reused") + + +def test_start_fails_when_systemd_run_fails(tmp_path: Path) -> None: + rig = Rig(tmp_path) + rig.run_rc = 1 + with pytest.raises(DriveError, match="systemd-run exited 1") as err: + _start(rig, tmp_path) + assert err.value.code == EXIT_BROWSER + + +def test_start_fails_when_cdp_never_answers(tmp_path: Path) -> None: + rig = Rig(tmp_path) + rig.comes_up = False + with pytest.raises(DriveError, match="never answered"): + _start(rig, tmp_path) + + +def test_start_fails_when_flutter_never_boots(tmp_path: Path) -> None: + rig = Rig(tmp_path) + rig.page.flutter_ready = False + with pytest.raises(DriveError, match="Flutter never booted"): + _start(rig, tmp_path) + + +def test_start_rejects_bad_size(tmp_path: Path) -> None: + with pytest.raises(DriveError) as err: + Rig(tmp_path).driver().start("u", tmp_path, 1, size="big") + assert err.value.code == EXIT_USAGE + + +def test_stop_stops_unit_and_clears_state(tmp_path: Path) -> None: + rig = Rig(tmp_path) + _start(rig, tmp_path) + assert rig.driver().stop() == "stopped fwd-kartka" + assert ["systemctl", "--user", "stop", "fwd-kartka"] in rig.cmds + assert not (rig.root / "kartka" / "state.json").exists() + assert rig.driver().stop() == "stopped fwd-kartka" + + +def test_verbs_need_start_first_unless_port_given(tmp_path: Path) -> None: + rig = Rig(tmp_path, up=True) + with pytest.raises(DriveError, match="run start first") as err: + rig.driver().tree() + assert err.value.code == EXIT_BROWSER + assert "#5" in rig.driver(cdp_port=9341).tree() + + +def test_verbs_round_trip_through_the_page(tmp_path: Path) -> None: + rig = Rig(tmp_path) + _start(rig, tmp_path) + drv = rig.driver() + assert "#5" in drv.tree(filt="brow") + assert drv.tree(row="Nope") == "(no matching semantics nodes)" + assert drv.tap("Browse").startswith("tapped #5") + assert drv.tap_xy(1, 2) == "tapped (1, 2)" + assert drv.text("hi") == "inserted 2 characters" + assert drv.key("Tab") == "pressed Tab" + assert drv.scroll(0, 10, at=(1, 1)) == "scrolled (0, 10) at (1, 1)" + assert drv.wait("Browse").startswith("found #5") + rig.page.evals = {"1": 1} + assert drv.eval("1") == "1" + assert drv.console() == "no console errors" + shot = drv.shot() + assert shot.startswith(str(rig.root / "kartka" / "shot-")) + + +def test_wait_timeout_writes_screenshot_and_caps_timeout(tmp_path: Path) -> None: + rig = Rig(tmp_path) + _start(rig, tmp_path) + with pytest.raises(DriveError, match="wait-timeout-"): + rig.driver().wait("Nope", timeout=1) + with pytest.raises(DriveError, match="<= 50") as err: + rig.driver().wait("Nope", timeout=51) + assert err.value.code == EXIT_USAGE + + +def test_console_can_fail_on_errors(tmp_path: Path) -> None: + rig = Rig(tmp_path) + _start(rig, tmp_path) + events = [ + { + "method": "Log.entryAdded", + "params": {"entry": {"level": "error", "text": "x"}}, + } + ] + drv = Driver( + "kartka", + root=rig.root, + fetch=rig.fetch, + create=lambda u: rig.page.ws(events), + runner=rig.runner, + ) + with pytest.raises(DriveError, match="1 console error"): + drv.console(fail_on_error=True) + + +def test_browser_helpers() -> None: + argv = browser.chromium_argv("http://x/", Path("/p"), 9, (10, 20)) + assert argv[0] == browser.CHROMIUM + assert argv[-1] == "http://x/" + assert "--password-store=basic" in argv + assert browser.parse_size("10x20") == (10, 20) + assert browser.unit_argv("u", ["a"])[-1] == "a" + + +def test_real_runner_returns_exit_code() -> None: + assert browser.run(["true"]) == 0 + assert browser.run(["false"]) == 1 diff --git a/flutter_web_drive/tests/test_cdp.py b/flutter_web_drive/tests/test_cdp.py new file mode 100644 index 0000000..f8040a3 --- /dev/null +++ b/flutter_web_drive/tests/test_cdp.py @@ -0,0 +1,157 @@ +"""CDP session, target discovery and the headless safety check.""" + +from __future__ import annotations + +import io +import json +from typing import Any + +import pytest +import websocket + +from flutter_web_drive import cdp +from flutter_web_drive.errors import EXIT_BROWSER, EXIT_UNSAFE, DriveError +from flutter_web_drive.tests.fakes import ( + HEADED, + HEADLESS, + PAGE_TARGETS, + FakePage, + FakeWs, +) + + +def _fetch(version: dict[str, Any] | None, targets: list[dict[str, Any]]) -> cdp.Fetch: + def fetch(port: int, path: str) -> Any: + if version is None: + raise OSError("refused") + return version if path == "/json/version" else targets + + return fetch + + +def test_call_returns_result_and_collects_events() -> None: + ws = FakeWs( + lambda m, p: {"result": {"ok": m}}, events=[{"method": "Log.entryAdded"}] + ) + sess = cdp.Session(ws) + assert sess.call("Log.enable") == {"ok": "Log.enable"} + assert sess.events == [{"method": "Log.entryAdded"}] + assert sess.call("Page.enable", a=1) == {"ok": "Page.enable"} + assert ws.sent[1] == {"id": 2, "method": "Page.enable", "params": {"a": 1}} + + +def test_call_raises_on_protocol_error() -> None: + sess = cdp.Session(FakeWs(lambda m, p: {"error": {"message": "nope"}})) + with pytest.raises(DriveError, match="Bogus.x: nope"): + sess.call("Bogus.x") + + +def test_call_wraps_transport_failures_as_browser_errors() -> None: + class Dead(FakeWs): + def send(self, raw: str) -> None: + raise websocket.WebSocketConnectionClosedException("gone") + + with pytest.raises(DriveError) as err: + cdp.Session(Dead(lambda m, p: {})).call("Page.enable") + assert err.value.code == EXIT_BROWSER + + +def test_evaluate_returns_value_or_raises_page_exception() -> None: + page = FakePage(evals={"1+1": 2, "boom()": ValueError("bad")}) + sess = cdp.Session(page.ws()) + assert sess.evaluate("1+1") == 2 + with pytest.raises(DriveError, match="Error: bad"): + sess.evaluate("boom()") + assert ( + page.ws().handler("Runtime.evaluate", {"expression": "1+1"})["result"][ + "result" + ]["value"] + == 2 + ) + + +def test_evaluate_exception_without_description_uses_text() -> None: + ws = FakeWs( + lambda m, p: { + "result": {"result": {}, "exceptionDetails": {"text": "Uncaught"}} + } + ) + with pytest.raises(DriveError, match="Uncaught"): + cdp.Session(ws).evaluate("x") + + +def test_drain_collects_until_timeout_and_restores_timeout() -> None: + ws = FakeWs(lambda m, p: {"result": {}}) + ws.queue = [{"method": "A"}, {"method": "B"}] + sess = cdp.Session(ws) + assert [e["method"] for e in sess.drain(0.1)] == ["A", "B"] + assert ws.timeout == cdp.CALL_TIMEOUT + + +def test_close_closes_socket() -> None: + ws = FakeWs(lambda m, p: {}) + cdp.Session(ws).close() + assert ws.closed + + +def test_pick_page_skips_browser_ui_and_needs_a_page() -> None: + assert cdp.pick_page(PAGE_TARGETS) == "ws://page" + chrome_only = [ + {"type": "page", "url": "chrome://newtab", "webSocketDebuggerUrl": "ws://n"} + ] + assert cdp.pick_page(chrome_only) == "ws://n" + with pytest.raises(DriveError, match="no page target"): + cdp.pick_page([PAGE_TARGETS[0]]) + + +def test_probe_returns_none_when_nothing_listens() -> None: + assert cdp.probe(1, fetch=_fetch(None, [])) is None + assert cdp.probe(1, fetch=_fetch(HEADLESS, [])) == HEADLESS + + +def test_connect_refuses_headed_browsers() -> None: + with pytest.raises(DriveError) as err: + cdp.connect(9222, fetch=_fetch(HEADED, PAGE_TARGETS), create=lambda u: None) + assert err.value.code == EXIT_UNSAFE + assert "Valve Steam" in err.value.message + + +def test_connect_reports_missing_browser() -> None: + with pytest.raises(DriveError) as err: + cdp.connect(9, fetch=_fetch(None, []), create=lambda u: None) + assert err.value.code == EXIT_BROWSER + + +def test_connect_opens_the_page_socket() -> None: + seen: list[str] = [] + + def create(url: str) -> FakeWs: + seen.append(url) + return FakeWs(lambda m, p: {"result": {}}) + + sess = cdp.connect(9341, fetch=_fetch(HEADLESS, PAGE_TARGETS), create=create) + assert isinstance(sess, cdp.Session) + assert seen == ["ws://page"] + + +def test_http_json_reads_loopback(monkeypatch: pytest.MonkeyPatch) -> None: + urls: list[str] = [] + + def fake_urlopen(url: str, timeout: float) -> io.BytesIO: + urls.append(url) + return io.BytesIO(json.dumps({"ok": True}).encode()) + + monkeypatch.setattr(cdp.urllib.request, "urlopen", fake_urlopen) + assert cdp.http_json(9341, "/json/version") == {"ok": True} + assert urls == ["http://127.0.0.1:9341/json/version"] + + +def test_default_create_uses_websocket_client(monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[tuple[str, dict[str, Any]]] = [] + monkeypatch.setattr( + cdp.websocket, + "create_connection", + lambda url, **kw: calls.append((url, kw)) or "ws", + ) + assert cdp.create_socket("ws://x") == "ws" + assert calls == [("ws://x", {"timeout": cdp.CALL_TIMEOUT, "suppress_origin": True})] diff --git a/flutter_web_drive/tests/test_cli.py b/flutter_web_drive/tests/test_cli.py new file mode 100644 index 0000000..febc0ac --- /dev/null +++ b/flutter_web_drive/tests/test_cli.py @@ -0,0 +1,158 @@ +"""CLI: argument parsing, verb dispatch, exit codes and stderr messages.""" + +from __future__ import annotations + +import io +from pathlib import Path +from typing import Any, ClassVar + +import pytest + +from flutter_web_drive import cli +from flutter_web_drive.errors import EXIT_BROWSER, EXIT_USAGE, DriveError + + +class Recorder: + """Stands in for Driver: records each call and returns its name.""" + + instances: ClassVar[list[Recorder]] = [] + + def __init__(self, app: str, *, cdp_port: int | None = None) -> None: + self.app = app + self.cdp_port = cdp_port + self.calls: list[tuple[str, tuple[Any, ...], dict[str, Any]]] = [] + Recorder.instances.append(self) + + def __getattr__(self, name: str) -> Any: + def verb(*args: Any, **kw: Any) -> str: + self.calls.append((name, args, kw)) + if name == "stop": + raise DriveError("stop", "boom", EXIT_BROWSER) + return f"ok {name}" + + return verb + + +@pytest.fixture(autouse=True) +def _recorder(monkeypatch: pytest.MonkeyPatch) -> None: + Recorder.instances = [] + monkeypatch.setattr(cli, "Driver", Recorder) + + +def _run(argv: list[str], stdin: str = "") -> tuple[int, Any]: + code = cli.main(argv, stdin=io.StringIO(stdin)) + return code, Recorder.instances[-1].calls[-1] if Recorder.instances else None + + +@pytest.mark.parametrize( + ("argv", "call"), + [ + ( + ["start", "--url", "http://u/", "--profile", "/p", "--cdp-port", "9341"], + ( + "start", + ("http://u/", Path("/p"), 9341), + {"unit": None, "size": "1400x1000"}, + ), + ), + (["shot"], ("shot", (None,), {"min_colours": 0})), + ( + ["shot", "o.png", "--min-colours", "2"], + ("shot", (Path("o.png"),), {"min_colours": 2}), + ), + ( + ["tree", "--filter", "x", "--all", "--row", "R"], + ("tree", (), {"filt": "x", "show_all": True, "row": "R"}), + ), + (["tap", "Browse", "--contains"], ("tap", ("Browse",), {"contains": True})), + (["tap-xy", "1", "2.5"], ("tap_xy", (1.0, 2.5), {})), + (["text", "hello"], ("text", ("hello",), {})), + (["key", "Enter"], ("key", ("Enter",), {})), + (["scroll", "0", "300"], ("scroll", (0.0, 300.0), {"at": None})), + ( + ["scroll", "0", "3", "--at", "4", "5"], + ("scroll", (0.0, 3.0), {"at": (4.0, 5.0)}), + ), + ( + ["wait", "Library", "--timeout", "3", "--gone", "--flag", "current"], + ( + "wait", + ("Library",), + { + "timeout": 3.0, + "contains": False, + "gone": True, + "flag": "current", + "row": None, + }, + ), + ), + (["eval", "1+1"], ("eval", ("1+1",), {})), + ( + ["console", "--limit", "5", "--warnings", "--fail-on-error"], + ("console", (), {"limit": 5, "warnings": True, "fail_on_error": True}), + ), + ], +) +def test_verbs_dispatch( + argv: list[str], call: Any, capsys: pytest.CaptureFixture[str] +) -> None: + code, got = _run(argv) + assert code == 0 + assert got == call + assert capsys.readouterr().out.startswith("ok ") + + +def test_global_app_and_port() -> None: + cli.main(["--app", "kartka", "--cdp-port", "9", "tree"], stdin=io.StringIO()) + assert (Recorder.instances[-1].app, Recorder.instances[-1].cdp_port) == ( + "kartka", + 9, + ) + + +def test_app_defaults_from_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("FWD_APP", "envapp") + cli.main(["tree"], stdin=io.StringIO()) + assert Recorder.instances[-1].app == "envapp" + + +def test_text_stdin_strips_one_trailing_newline() -> None: + _, call = _run(["text", "--stdin"], stdin="p@ss\n") + assert call == ("text", ("p@ss",), {}) + + +def test_text_file_reads_secret_without_argv(tmp_path: Path) -> None: + secret = tmp_path / "pw" + secret.write_text("s3cret\n") + _, call = _run(["text", "--file", str(secret)]) + assert call == ("text", ("s3cret",), {}) + + +def test_text_file_missing_is_usage_error( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + assert ( + cli.main(["text", "--file", str(tmp_path / "nope")], stdin=io.StringIO()) + == EXIT_USAGE + ) + assert "cannot read" in capsys.readouterr().err + + +def test_text_needs_value_or_stdin(capsys: pytest.CaptureFixture[str]) -> None: + assert cli.main(["text"], stdin=io.StringIO()) == EXIT_USAGE + assert "text needs a value, --stdin or --file" in capsys.readouterr().err + + +def test_drive_error_becomes_exit_code_and_one_line( + capsys: pytest.CaptureFixture[str], +) -> None: + code, _ = _run(["stop"]) + assert code == EXIT_BROWSER + assert capsys.readouterr().err == "flutter_web_drive: stop failed: boom\n" + + +def test_usage_errors_exit_2() -> None: + with pytest.raises(SystemExit) as err: + cli.main(["bogus"], stdin=io.StringIO()) + assert err.value.code == EXIT_USAGE diff --git a/flutter_web_drive/tests/test_mcp_server.py b/flutter_web_drive/tests/test_mcp_server.py new file mode 100644 index 0000000..b253977 --- /dev/null +++ b/flutter_web_drive/tests/test_mcp_server.py @@ -0,0 +1,100 @@ +"""MCP server: same verbs as the CLI, delegating to Driver.""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +from typing import Any + +import pytest +from mcp.server.mcpserver.exceptions import ToolError + +from flutter_web_drive import mcp_server +from flutter_web_drive.errors import DriveError + +VERBS = { + "start", + "stop", + "shot", + "tree", + "tap", + "tap_xy", + "text", + "key", + "scroll", + "wait", + "eval_js", + "console", +} + + +class Fake: + """Driver stand-in: echoes the call, fails on 'stop'.""" + + def __init__(self, app: str, *, cdp_port: int | None = None) -> None: + self.app = app + + def __getattr__(self, name: str) -> Any: + def verb(*args: Any, **kw: Any) -> str: + if name == "stop": + raise DriveError("stop", "boom") + return f"{self.app}:{name}:{args}:{sorted(kw)}" + + return verb + + +@pytest.fixture(autouse=True) +def _fake(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(mcp_server, "Driver", Fake) + + +def _call(tool: str, **args: Any) -> str: + result = asyncio.run(mcp_server.server.call_tool(tool, args)) + return result.content[0].text # type: ignore[union-attr] + + +def test_tools_cover_every_verb() -> None: + tools = asyncio.run(mcp_server.server.list_tools()) + assert {t.name for t in tools} == VERBS + + +def test_tools_delegate_to_driver() -> None: + assert _call("tap", label="Browse", app="k") == "k:tap:('Browse',):['contains']" + assert _call("start", url="u", profile="/p", cdp_port=1).startswith( + "default:start:" + ) + assert _call("shot").startswith("default:shot:(None,)") + assert _call("shot", out="/x.png").startswith("default:shot:(PosixPath('/x.png'),)") + assert _call("tree").startswith("default:tree:") + assert _call("tap_xy", x=1, y=2).startswith("default:tap_xy:(1.0, 2.0)") + assert _call("text", value="v").startswith("default:text:('v',)") + assert _call("key", name="Enter").startswith("default:key:") + assert _call("scroll", dx=0, dy=1).startswith("default:scroll:") + assert ":scroll:" in _call("scroll", dx=0, dy=1, at_x=2, at_y=3) + assert _call("wait", label="L").startswith("default:wait:") + assert _call("eval_js", expr="1").startswith("default:eval:") + assert _call("console").startswith("default:console:") + + +def test_text_from_file_keeps_secret_out_of_arguments(tmp_path: Path) -> None: + secret = tmp_path / "pw" + secret.write_text("hunter2\n") + assert _call("text", from_file=str(secret)) == "default:text:('hunter2',):[]" + with pytest.raises(ToolError, match="cannot read"): + _call("text", from_file=str(tmp_path / "missing")) + with pytest.raises(ToolError, match="value or from_file"): + _call("text") + + +def test_drive_errors_surface_as_tool_errors() -> None: + with pytest.raises(ToolError, match="stop failed: boom"): + _call("stop") + + +def test_main_runs_stdio(monkeypatch: pytest.MonkeyPatch) -> None: + seen: list[str] = [] + monkeypatch.setattr( + mcp_server.server, "run", lambda transport: seen.append(transport) + ) + mcp_server.main() + assert seen == ["stdio"] diff --git a/flutter_web_drive/tests/test_semantics.py b/flutter_web_drive/tests/test_semantics.py new file mode 100644 index 0000000..b22027c --- /dev/null +++ b/flutter_web_drive/tests/test_semantics.py @@ -0,0 +1,105 @@ +"""Semantics records: parsing, label matching, ranking and the compact tree.""" + +from __future__ import annotations + +import pytest + +from flutter_web_drive import semantics +from flutter_web_drive.errors import DriveError +from flutter_web_drive.tests.fakes import node + +TAB = node(4, "Library\nTab 1 of 4", rect=(0, 8, 80, 67), flags=("current",)) +HEADING = node(12, "Library", role="", tag="h2", tappable=False, rect=(16, 13, 72, 30)) +CONTAINER = node(1, "", role="", tappable=False, rect=(0, 0, 1400, 900)) +FIELD = node( + 20, "Email", role="textbox", value="me@x", tappable=False, flags=("focused",) +) + + +def test_parse_nodes_builds_typed_records() -> None: + (parsed,) = semantics.parse_nodes([TAB]) + assert parsed.id == 4 + assert parsed.lines == ("Library", "Tab 1 of 4") + assert parsed.center == (40.0, 41.5) + assert parsed.flags == ("current",) + + +def test_parse_nodes_rejects_non_list() -> None: + with pytest.raises(DriveError, match="semantics"): + semantics.parse_nodes("nope") + + +def test_exact_match_hits_whole_label_or_any_line() -> None: + tab = semantics.parse_nodes([TAB])[0] + assert semantics.matches(tab, "Library") + assert semantics.matches(tab, "Library\nTab 1 of 4") + assert semantics.matches(tab, "Library | Tab 1 of 4") # as `tree` prints it + assert not semantics.matches(tab, "Lib") + assert semantics.matches(tab, "lib", contains=True) + assert not semantics.matches(tab, "Browse", contains=True) + + +def test_find_prefers_tappable_and_keeps_document_order() -> None: + nodes = semantics.parse_nodes( + [HEADING, TAB, node(9, "Library", rect=(0, 300, 10, 10))] + ) + hits = semantics.find(nodes, "Library") + assert [h.id for h in hits] == [4, 9, 12] + + +def test_describe_is_one_line_with_tags_and_value() -> None: + tab, field = semantics.parse_nodes([TAB, FIELD]) + assert ( + semantics.describe(tab) + == '#4 button "Library | Tab 1 of 4" [tap,current] @0,8 80x67' + ) + assert ( + semantics.describe(field) == '#20 textbox "Email" = "me@x" [focused] @0,0 80x40' + ) + heading = semantics.parse_nodes([HEADING])[0] + assert semantics.describe(heading) == '#12 h2 "Library" @16,13 72x30' + + +def test_format_tree_hides_empty_containers_unless_all() -> None: + nodes = semantics.parse_nodes([CONTAINER, TAB, FIELD]) + compact = semantics.format_tree(nodes) + assert "#1 " not in compact + assert "#4 " in compact + assert "#1 " in semantics.format_tree(nodes, show_all=True) + + +def test_format_tree_filter_matches_label_role_or_value() -> None: + nodes = semantics.parse_nodes([TAB, FIELD]) + assert semantics.format_tree(nodes, filt="textbox").count("\n") == 0 + assert "#20" in semantics.format_tree(nodes, filt="ME@X") + assert semantics.format_tree(nodes, filt="zzz") == "(no matching semantics nodes)" + assert semantics.format_tree([]) == "(no semantics nodes)" + + +def test_hint_lists_tappable_labels_capped() -> None: + many = [node(i, f"b{i}") for i in range(15)] + [HEADING, node(99, "")] + hint = semantics.hint(semantics.parse_nodes(many)) + assert hint.startswith('"b0", "b1"') + assert "b12" not in hint + assert hint.endswith("...") + assert semantics.hint(semantics.parse_nodes([HEADING])) == "(no tappable nodes)" + assert semantics.hint(semantics.parse_nodes([TAB])) == '"Library | Tab 1 of 4"' + + +def test_js_snippets_never_read_password_values() -> None: + assert "password" in semantics.COLLECT_JS + assert semantics.IMAGE_STATS_FN.startswith("(async") + + +def test_in_row_keeps_nodes_vertically_inside_the_anchor_row() -> None: + rows = semantics.parse_nodes( + [ + node(1, "Chapter 2\n2024-01-02", rect=(0, 100, 1000, 60)), + node(2, "Mark unread", rect=(900, 110, 40, 40)), + node(3, "Chapter 1\n2024-01-01", rect=(0, 160, 1000, 60)), + node(4, "Mark read", rect=(900, 170, 40, 40)), + ] + ) + assert [n.id for n in semantics.in_row(rows, "Chapter 2")] == [2] + assert [n.id for n in semantics.in_row(rows, "chapter 1", contains=True)] == [4] + assert semantics.in_row(rows, "Chapter 9") == [] diff --git a/flutter_web_drive/tests/test_state.py b/flutter_web_drive/tests/test_state.py new file mode 100644 index 0000000..1267b3f --- /dev/null +++ b/flutter_web_drive/tests/test_state.py @@ -0,0 +1,42 @@ +"""State file round-trip, artefact root and the module entry point.""" + +from __future__ import annotations + +import re +import runpy +import sys +from pathlib import Path + +import pytest + +from flutter_web_drive import state + + +def test_round_trip_and_clear(tmp_path: Path) -> None: + assert state.load(tmp_path, "a") is None + saved = state.State(unit="u", cdp_port=1, url="http://x/", profile="/p") + state.save(tmp_path, "a", saved) + assert state.load(tmp_path, "a") == saved + state.clear(tmp_path, "a") + state.clear(tmp_path, "a") + assert state.load(tmp_path, "a") is None + + +def test_artefact_root_defaults_under_home( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.delenv("FWD_ARTEFACTS", raising=False) + assert state.artefact_root() == Path.home() / "data" / "claude-scratch" / "e2e" + monkeypatch.setenv("FWD_ARTEFACTS", str(tmp_path)) + assert state.artefact_root() == tmp_path + + +def test_stamp_is_sortable() -> None: + assert re.fullmatch(r"\d{8}-\d{6}-\d{3}", state.stamp()) + + +def test_module_entry_point_runs_cli(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(sys, "argv", ["flutter_web_drive", "--help"]) + with pytest.raises(SystemExit) as err: + runpy.run_module("flutter_web_drive", run_name="__main__") + assert err.value.code == 0