kuhyx/utils
1
0
Fork 0
mirror of https://github.com/kuhyx/utils.git synced 2026-11-10 16:48:54 +01:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity

flutter_web_drive: headless CDP driver + MCP server for Flutter web apps

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CZ4sULGfyav7EEwQ6vhbfg
This commit is contained in:
Krzysztof Rudnicki 2026-10-24 22:37:49 +02:00
parent d0f928ae22
commit f93a24e283
25 changed files with 2730 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
CLAUDE.md
View File

@ -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 `<pkg>/tests` run with `PYTHONPATH=.`;
subprojects with their own `pyproject.toml` run via `scripts/run_subproject_tests.sh <dir>`.
`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`

120
flutter_web_drive/README.md Normal file
View File

@ -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/<app>/`
(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
```

6
flutter_web_drive/__init__.py Normal file
View File

@ -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.
"""

5
flutter_web_drive/__main__.py Normal file
View File

@ -0,0 +1,5 @@
"""Entry point: ``python -m flutter_web_drive``."""
from flutter_web_drive.cli import main
raise SystemExit(main())

239
flutter_web_drive/actions.py Normal file
View File

@ -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

235
flutter_web_drive/api.py Normal file
View File

@ -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

81
flutter_web_drive/browser.py Normal file
View File

@ -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

139
flutter_web_drive/cdp.py Normal file
View File

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

150
flutter_web_drive/cli.py Normal file
View File

@ -0,0 +1,150 @@
"""``python -m flutter_web_drive [--app A] <verb> ...``: 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

20
flutter_web_drive/errors.py Normal file
View File

@ -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

22
flutter_web_drive/keys.py Normal file
View File

@ -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, ""),
}

173
flutter_web_drive/mcp_server.py Normal file
View File

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

3
flutter_web_drive/requirements.txt Normal file
View File

@ -0,0 +1,3 @@
# Exact pins, newest stable (shared dependency-freshness gate).
mcp==2.3.0
websocket-client==1.9.2

228
flutter_web_drive/semantics.py Normal file
View File

@ -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 / <span> 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 <span>.
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}("<base64 png>")`: 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

59
flutter_web_drive/state.py Normal file
View File

@ -0,0 +1,59 @@
"""Per-app state file and artefact directory under ~/data/claude-scratch/e2e/<app>/."""
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)

0
flutter_web_drive/tests/__init__.py Normal file
View File

179
flutter_web_drive/tests/fakes.py Normal file
View File

@ -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"},
]

158
flutter_web_drive/tests/test_actions.py Normal file
View File

@ -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]'

134
flutter_web_drive/tests/test_actions_output.py Normal file
View File

@ -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",
)

212
flutter_web_drive/tests/test_api.py Normal file
View File

@ -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

157
flutter_web_drive/tests/test_cdp.py Normal file
View File

@ -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})]

158
flutter_web_drive/tests/test_cli.py Normal file
View File

@ -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

100
flutter_web_drive/tests/test_mcp_server.py Normal file
View File

@ -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"]

105
flutter_web_drive/tests/test_semantics.py Normal file
View File

@ -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") == []

42
flutter_web_drive/tests/test_state.py Normal file
View File

@ -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