bfe-core1106-sdk/qemu/tests/qmp.py

#!/usr/bin/env python3
"""Tiny QMP client for the qemu/ device-sim tests.

    qmp.py <socket> screendump <out.ppm>
    qmp.py <socket> tap <x> <y>          # absolute 0..32767 (virtio-tablet)
    qmp.py <socket> quit
    qmp.py <socket> drive <script> <outdir> [--size N]

`drive` runs a whole interaction on ONE connection, so the VM boots once and
the scenario costs a couple of seconds per step instead of a full boot. Its
script is one command per line, '#' comments and blank lines ignored, and
coordinates are PANEL PIXELS (0..size-1, default 720) rather than the
0..32767 absolute axis the raw `tap` takes -- the point of the mode is to write
"tap the gear at 47,676" straight off a screenshot.

    shot NAME              screendump to <outdir>/NAME.ppm
    tap X Y                press, hold past several LVGL poll periods, release
    swipe X1 Y1 X2 Y2 [MS] drag, interpolated so LVGL sees real motion
    fling X1 Y1 X2 Y2      swipe fast enough to leave momentum behind
    sleep SECONDS          let animations settle or timers fire
    echo TEXT              progress marker in the driver's own output

With `--ctl SOCK` (the unix socket run.sh --ctl creates; ui-drive.sh passes
it) the script can also ask the UI itself, through the debug channel in
warden_debug.c. The same channel is what tools/warden-ctl reaches over SSH on
a real panel, so these verbs mean the same thing on the rig and on hardware:

    nav MENU[/TAB]         jump to a page; fails if the page is unknown
    scroll X Y DY          scroll the scrollable under pixel X,Y by DY
                           pixels (positive reveals content below), no
                           animation: the deterministic stand-in for a
                           swipe that only brings a control into view
    home                   close every popup and go to Dashboard/Dashboard,
                           the state a fresh boot starts in
    wake                   wake the UI as if touched, minus the swallow
                           (sleep.c dims and then blanks an idle panel and
                           eats the touch that ends either); every compiled
                           flow script opens with it
    page | stats | hit X Y print the reply, judge nothing
    ctl WORDS...           raw passthrough for anything the channel grows
    assert_page MENU/TAB   the active page is exactly this
    assert_hit X Y CLASS [TEXT...]
                           a real tap at X,Y lands on an object of CLASS whose
                           caption contains TEXT: the pre-tap self-check that
                           reports a MOVED button as such, not as broken
                           behaviour

Two more channels reach behavioural state instead of the widget tree.
`--refs FILE` (default <outdir>/refs.json) is the reference store the pixel
ones read and write; imgtools.py in this directory does the actual image
math, this file only drives it.

    assert_json PATH OP VALUE
    wait_json PATH OP VALUE TIMEOUT_S
                           PATH is dotted with optional [N] indexes (e.g.
                           network.scan.ranges[0]) into the guest's
                           /tmp/warden-web-status.json, fetched fresh on
                           every check with `@cat` over the SAME control
                           channel as nav/hit (see the bridge added to
                           rootfs/sbin/init). OP is one of eq, ne, contains,
                           len_eq, len_ge, gt, lt. `wait_json` polls every
                           0.5s until OP holds or TIMEOUT_S elapses and
                           `assert_json` reads once. A PATH the document
                           doesn't have is a `fail`, not a crash -- that
                           document is a live snapshot and the key might
                           simply not be built yet.
    assert_stat FIELD OP VALUE
                           FIELD is cpu, fps, render or idle (0 awake,
                           1 dimmed, 2 asleep), read off a fresh `stats`
                           reply (warden_debug.c). Same OP vocabulary as
                           assert_json.
    capture_region NAME X Y W H TOLERANCE
                           screendump now, crop to X,Y,WxH, and (over)write
                           NAME in the refs file with both a phash and a
                           structural hash plus TOLERANCE (exact, loose or
                           structural). Run this once, by hand, to author a
                           reference -- the same way targets are authored,
                           see flows/README.md.
    assert_region NAME [TOLERANCE]
                           fresh screendump, crop to NAME's stored box,
                           compare against its stored hash at its stored
                           tolerance. TOLERANCE, when given, must be the one
                           NAME was captured under (flowc.py always emits it).
    assert_ocr NAME REGEX  same crop, run through `tesseract`, REGEX searched
                           against the extracted text (spaces in REGEX are
                           fine, same as assert_hit's TEXT).

assert_region/assert_ocr against a NAME with no captured reference, and
assert_ocr with no `tesseract` on PATH, are FATAL for that one step: a
silently-skipped check reads as coverage that was never actually there. So
are a TOLERANCE that differs from the captured one and a stored box that does
not fit the screendump: both are the reference's fault, not the UI's, and one
bad reference must not take the rest of the script with it.

Every step is recorded to <outdir>/results.jsonl as ok / fail / fatal. An
assertion mismatch is a `fail` and the script continues, so one run reports
every broken expectation. With `--console LOG` the console is checked after
EVERY step for the stage-2 init's "warden-ui EXITED" line, so a crash is a
`fatal` pinned to the step that caused it rather than a stale picture found at
the end. The exit status is non-zero if anything failed.

A step that names an unknown command is a FATAL error rather than a skip: a
silently-ignored line in a scenario is a test that proves nothing.
"""
import json
import math
import os
import re
import shutil
import socket
import subprocess
import sys
import time

# imgtools.py lives beside this file and does the actual pixel math (phash,
# structural hash, crop, compare) for the region/ocr verbs. Imported at
# module load but never let a missing/broken imgtools take down the verbs
# that don't need it: nav/tap/page/... must keep working while it is being
# written, so the failure is deferred to need_imgtools() at first use.
try:
    import imgtools
except ImportError:
    imgtools = None

# Pillow, likewise, is only needed to hand tesseract an image file (it has no
# raw-RGB stdin mode); a host without it still runs every other verb.
try:
    from PIL import Image as _PILImage
except ImportError:
    _PILImage = None

AXIS_MAX = 32767

# Where webstatus.c (ui-src/src/warden/webstatus.c) publishes its atomic
# snapshot inside the guest. assert_json/wait_json `@cat` this path over the
# control channel; see the bridge in rootfs/sbin/init.
STATUS_JSON_PATH = "/tmp/warden-web-status.json"


def rpc(sock, sock_file, obj):
    sock.sendall((json.dumps(obj) + "\n").encode())
    while True:
        line = sock_file.readline()
        if not line:
            raise RuntimeError("QMP connection closed")
        msg = json.loads(line)
        if "return" in msg:
            return msg["return"]
        if "error" in msg:
            raise RuntimeError(f"QMP error: {msg['error']}")
        # asynchronous events are interleaved; skip them


def send_events(s, f, events):
    rpc(s, f, {"execute": "input-send-event", "arguments": {"events": events}})


def abs_ev(axis, value):
    return {"type": "abs", "data": {"axis": axis, "value": value}}


def btn_ev(down):
    return {"type": "btn", "data": {"down": down, "button": "left"}}


def to_axis(px, size):
    """Panel pixels -> the tablet's absolute axis, clamped to the panel.

    LVGL's evdev driver maps the axis back with integer truncation
    (lv_evdev.c _evdev_calibrate: px = axis * (width - 1) / AXIS_MAX), so the
    axis value must be the SMALLEST one that truncates to px, i.e. rounded up.
    Truncating here as well composed two floors and landed one pixel short for
    many values (130 -> 5924 -> 129), which is how a tap could miss the control
    that `hit` at the same pixel had just confirmed (flare-edge #148 triage).
    tools/touch-inject writes the pixel itself on a panel, so hardware never
    had this seam."""
    px = max(0, min(size - 1, int(px)))
    return min(AXIS_MAX, math.ceil(px * AXIS_MAX / (size - 1)))


def do_tap(s, f, ax, ay, hold=0.2):
    send_events(s, f, [abs_ev("x", ax), abs_ev("y", ay), btn_ev(True)])
    # Hold the press across several LVGL indev poll periods (33 ms each): an
    # instantaneous press+release lands inside one poll and no click is ever
    # registered.
    time.sleep(hold)
    send_events(s, f, [btn_ev(False)])


def do_swipe(s, f, x1, y1, x2, y2, size, ms=400, steps=None):
    """Drag with interpolated motion.

    LVGL decides a gesture from the movement BETWEEN indev polls, so a press at
    the start and a release at the end (with nothing in between) reads as a
    click on whatever was under the finger, not a scroll. The interpolation
    below is the whole reason a scroll can be tested at all; the step count is
    derived from the duration so the poll period always sees a few pixels of
    travel.
    """
    ax1, ay1 = to_axis(x1, size), to_axis(y1, size)
    ax2, ay2 = to_axis(x2, size), to_axis(y2, size)
    if steps is None:
        steps = max(6, int(ms / 25))
    send_events(s, f, [abs_ev("x", ax1), abs_ev("y", ay1), btn_ev(True)])
    time.sleep(0.05)
    for i in range(1, steps + 1):
        t = i / steps
        send_events(s, f, [
            abs_ev("x", int(ax1 + (ax2 - ax1) * t)),
            abs_ev("y", int(ay1 + (ay2 - ay1) * t)),
        ])
        time.sleep(ms / 1000.0 / steps)
    send_events(s, f, [btn_ev(False)])


class Ctl:
    """The UI's debug channel, reached through run.sh --ctl.

    Inside the VM that channel is a FIFO polled by warden-ui (warden_debug.c);
    rootfs/sbin/init bridges it to a pci-serial port that run.sh exposes as a
    unix socket. One command line in, the reply out, then a sentinel line. On
    real hardware flare-edge tools/warden-ctl reaches the identical FIFO over
    SSH, so the vocabulary here is the vocabulary there: a flow script that
    runs against this rig runs against a panel.

    A missing or silent channel is an infrastructure fault and raises; it is
    never reported as an assertion failure, because a rig that cannot ask is
    not evidence about the UI either way.
    """

    SENTINEL = "<<END>>"

    def __init__(self, path, timeout=15.0):
        self.sock = socket.socket(socket.AF_UNIX)
        self.sock.settimeout(timeout)
        self.sock.connect(path)
        self.buf = b""

    def send(self, cmd):
        self.sock.sendall((cmd + "\n").encode())
        lines = []
        while True:
            nl = self.buf.find(b"\n")
            if nl < 0:
                chunk = self.sock.recv(4096)
                if not chunk:
                    raise RuntimeError("control channel closed")
                self.buf += chunk
                continue
            line = self.buf[:nl].decode("utf-8", "replace").rstrip("\r")
            self.buf = self.buf[nl + 1:]
            if line == self.SENTINEL:
                break
            # The tty is in cooked mode, so the command comes back echoed.
            if line.strip() == cmd.strip():
                continue
            lines.append(line)
        return "\n".join(lines).strip()


HIT_RE = re.compile(r'^hit \d+,\d+: (\S+)(?: text="(.*)" box=(-?\d+),(-?\d+),(\d+)x(\d+))?$')


def parse_hit(reply):
    """-> (cls, text, box) with box as 'x1,y1,WxH', or None for 'nothing';
    raises on an unparseable reply."""
    m = HIT_RE.match(reply)
    if not m:
        raise RuntimeError(f"unparseable hit reply: {reply!r}")
    if m.group(1) == "nothing":
        return None
    box = f"{m.group(3)},{m.group(4)},{m.group(5)}x{m.group(6)}" if m.group(3) else ""
    return m.group(1), m.group(2) or "", box


def ui_exited(console_path):
    """The stage-2 init announces a dead UI on the console; the framebuffer
    does not, it just keeps the last frame. Checked after EVERY step so a crash
    is pinned to the step that caused it, not discovered at the end."""
    if not console_path:
        return None
    try:
        with open(console_path, "rb") as fh:
            data = fh.read()
    except OSError:
        return None
    i = data.find(b"warden-ui EXITED")
    if i < 0:
        return None
    return data[i:i + 80].split(b"\n", 1)[0].decode("utf-8", "replace")


PATH_TOKEN_RE = re.compile(r'^([^\[\]]*)((?:\[\d+\])*)$')
IDX_RE = re.compile(r'\[(\d+)\]')


def resolve_path(doc, path):
    """Walk a dotted PATH (keys and [N] indexes, e.g. 'a.b[0].c') over an
    already-parsed JSON document. -> (value, None) or (None, 'no such path').

    Pure and I/O-free on purpose (contract: testable on its own) -- fetching
    the document is a separate step (fetch_status_json) so this function can
    be unit-tested against a plain dict/list literal with nothing running.
    """
    cur = doc
    for part in path.split("."):
        m = PATH_TOKEN_RE.match(part)
        if not m:
            return None, "no such path"
        key, idxs = m.group(1), m.group(2)
        if key:
            if not isinstance(cur, dict) or key not in cur:
                return None, "no such path"
            cur = cur[key]
        for idx_s in IDX_RE.findall(idxs):
            idx = int(idx_s)
            if not isinstance(cur, list) or idx >= len(cur) or idx < 0:
                return None, "no such path"
            cur = cur[idx]
    return cur, None


def resolve_token(raw):
    """A verb's literal OP argument -> a comparable Python value. A JSON
    literal (a number, true/false/null, or a quoted string) parses as
    itself; anything else -- most values in practice, e.g. `connected` --
    compares as the raw string. This is what lets both
    `assert_json state.net eq connected` and `assert_json state.count eq 3`
    work from one unquoted token, with no escaping convention of its own."""
    try:
        return json.loads(raw)
    except ValueError:
        return raw


def _len_of(v):
    try:
        return len(v)
    except TypeError:
        raise TypeError(f"{v!r} has no length") from None


def apply_op(op, actual, raw_value):
    """The comparison vocabulary shared by assert_json/wait_json/assert_stat:
    eq, ne, contains, len_eq, len_ge, gt, lt. Raises on a combination that
    cannot be judged (contains on a number, gt across incompatible types)
    so the caller turns that into a `fail` detail instead of a driver crash --
    a spec that asks a nonsensical question should be reported, not silently
    True or False."""
    value = resolve_token(raw_value)
    if op == "eq":
        return actual == value
    if op == "ne":
        return actual != value
    if op == "contains":
        if isinstance(actual, str):
            return str(value) in actual
        if isinstance(actual, (list, tuple, dict)):
            return value in actual
        raise TypeError(f"contains: {actual!r} is not a string or list")
    if op == "len_eq":
        return _len_of(actual) == int(value)
    if op == "len_ge":
        return _len_of(actual) >= int(value)
    if op == "gt":
        return actual > value
    if op == "lt":
        return actual < value
    raise ValueError(f"unknown op {op!r}")


def fetch_status_json(ctl):
    """`@cat` the guest's webstatus.c snapshot over the control channel and
    parse it. -> (doc, None) or (None, detail); never raises, so a torn read
    or a not-yet-written file is an ordinary retry/fail for the caller, not a
    crash of the whole driver."""
    reply = ctl.send(f"@cat {STATUS_JSON_PATH}")
    if reply.startswith("bridge: no such file"):
        return None, reply
    try:
        return json.loads(reply), None
    except ValueError as e:
        return None, f"status json unparsable: {e}"


def eval_json(ctl, path, op, value):
    """One fetch + path-resolve + op-apply round for assert_json/wait_json.
    -> (ok, detail); detail is empty on success, otherwise the reason."""
    doc, err = fetch_status_json(ctl)
    if err is not None:
        return False, err
    actual, perr = resolve_path(doc, path)
    if perr is not None:
        return False, perr
    try:
        ok = apply_op(op, actual, value)
    except (TypeError, ValueError) as e:
        return False, f"{op} {value!r} vs {actual!r}: {e}"
    return ok, "" if ok else f"{path} is {actual!r}"


STATS_FIELD_RE = {
    "cpu": re.compile(r'^cpu:\s*(-?\d+(?:\.\d+)?)%?\s*$'),
    "fps": re.compile(r'^fps:\s*(-?\d+(?:\.\d+)?)\s*$'),
    "render": re.compile(r'^render:\s*(-?\d+(?:\.\d+)?)\s*ms/frame\s*$'),
    "idle": re.compile(r'^idle:\s*(\d+)\s*$'),
}


def parse_stats(reply):
    """The `stats` reply (warden_debug.c: 'page: X\\ncpu: N%\\nfps: N\\n
    render: A.BB ms/frame\\nrga: N%\\nidle: N') -> {'cpu'|'fps'|'render'|'idle': float} for
    whichever lines are present. A field the reply lacks is simply absent
    from the result -- the caller reports that as 'no such field', the same
    shape as resolve_path's 'no such path' for the json channel."""
    out = {}
    for ln in reply.splitlines():
        ln = ln.strip()
        for field, rx in STATS_FIELD_RE.items():
            m = rx.match(ln)
            if m:
                out[field] = float(m.group(1))
    return out


def load_refs(path):
    """The region/ocr reference store (see qmp.py's own docstring for the
    schema). Missing file -> {}, same as an empty COVERAGE.yaml: capture_region
    is how one gets created, not a prerequisite to have one already."""
    if not path or not os.path.exists(path):
        return {}
    with open(path) as fh:
        return json.load(fh)


def save_refs(path, refs):
    # Written after every capture_region, not batched at the end: a script
    # that dies three steps later must not lose a reference that was already
    # good.
    with open(path, "w") as fh:
        json.dump(refs, fh, indent=2, sort_keys=True)
        fh.write("\n")


def write_png(img, path):
    """imgtools' (w, h, rgb_bytes) tuple -> a PNG file, because tesseract has
    no raw-RGB input mode. Only assert_ocr needs this; assert_region and
    capture_region work on imgtools' own tuples end to end."""
    if _PILImage is None:
        raise RuntimeError("Pillow (PIL) is not installed")
    w, h, data = img
    _PILImage.frombytes("RGB", (w, h), data).save(path)


def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, refs_path=None):
    os.makedirs(outdir, exist_ok=True)
    with open(script_path) as fh:
        lines = fh.readlines()

    ctl = Ctl(ctl_path) if ctl_path else None
    results = open(os.path.join(outdir, "results.jsonl"), "w")
    counts = {"ok": 0, "fail": 0, "fatal": 0}

    def record(lineno, cmd, status, detail=""):
        counts[status] += 1
        results.write(json.dumps({"line": lineno, "cmd": cmd, "status": status,
                                  "detail": detail}) + "\n")
        results.flush()
        tag = {"ok": "", "fail": "FAIL ", "fatal": "FATAL "}[status]
        print(f"  [{lineno}] {tag}{cmd}{(' -- ' + detail) if detail else ''}", flush=True)

    def need_ctl(lineno, cmd):
        if ctl is None:
            sys.exit(f"FATAL: {script_path}:{lineno}: '{cmd}' needs the control "
                     f"channel; run with --ctl (ui-drive.sh passes it)")

    def need_imgtools(lineno, cmd):
        # A whole-script infrastructure gap (imgtools.py absent), same class
        # as a missing --ctl: no region/ocr verb can do anything without it,
        # so this exits the process rather than recording a per-step fail.
        if imgtools is None:
            sys.exit(f"FATAL: {script_path}:{lineno}: '{cmd}' needs imgtools.py "
                     f"next to qmp.py (vendor/warden-sdk/qemu/tests/imgtools.py)")

    refs_path = refs_path or os.path.join(outdir, "refs.json")
    refs = load_refs(refs_path)

    def fresh_region(name, tag):
        """screendump NOW (never reuse an earlier `shot`) and crop to NAME's
        stored box. Fails closed on an uncaptured NAME: assert_region and
        assert_ocr both need this, capture_region does not."""
        ref = refs.get(name)
        if ref is None:
            return None, None, f"no reference for {name!r} (run capture_region first)"
        path = os.path.join(os.path.abspath(outdir), f"{tag}_{name}.ppm")
        rpc(s, f, {"execute": "screendump", "arguments": {"filename": path}})
        try:
            img = imgtools.load_ppm(path)
            cropped = imgtools.crop(img, ref["x"], ref["y"], ref["w"], ref["h"])
        except (ValueError, OSError, KeyError, TypeError) as e:
            # A box outside this screendump (captured at another --size, or
            # mistyped in the store) is a per-step fatal, never a traceback
            # that ends the run: every later step still gets judged.
            return None, None, f"reference {name!r} unusable: {e}"
        return ref, cropped, None

    for lineno, raw in enumerate(lines, 1):
        line = raw.split("#", 1)[0].strip()
        if not line:
            continue
        parts = line.split()
        cmd, args = parts[0], parts[1:]

        if cmd == "shot":
            name = args[0]
            path = os.path.join(os.path.abspath(outdir), f"{name}.ppm")
            rpc(s, f, {"execute": "screendump", "arguments": {"filename": path}})
            record(lineno, line, "ok")
        elif cmd == "tap":
            x, y = int(args[0]), int(args[1])
            do_tap(s, f, to_axis(x, size), to_axis(y, size))
            record(lineno, line, "ok")
        elif cmd == "swipe":
            ms = int(args[4]) if len(args) > 4 else 400
            do_swipe(s, f, int(args[0]), int(args[1]), int(args[2]), int(args[3]), size, ms)
            record(lineno, line, "ok")
        elif cmd == "fling":
            do_swipe(s, f, int(args[0]), int(args[1]), int(args[2]), int(args[3]),
                     size, ms=120)
            record(lineno, line, "ok")
        elif cmd == "sleep":
            time.sleep(float(args[0]))
            record(lineno, line, "ok")
        elif cmd == "echo":
            print(f"  [{lineno}] {' '.join(args)}", flush=True)
        elif cmd == "nav":
            need_ctl(lineno, cmd)
            reply = ctl.send("nav " + " ".join(args))
            record(lineno, line, "ok" if reply.endswith(": ok") else "fail", reply)
        elif cmd == "wake":
            # Judged, unlike the query verbs: a UI that does not answer the
            # wake is one whose next tap may be swallowed, and that must not
            # read as a passing step.
            need_ctl(lineno, cmd)
            reply = ctl.send("wake")
            record(lineno, line, "ok" if reply == "wake: ok" else "fail", reply)
        elif cmd in ("scroll", "home"):
            # scroll X Y DY: the scrollable under the pixel moves DY pixels
            # with no animation (warden_debug.c), the deterministic stand-in
            # for a swipe whose only job was to bring a control into view.
            # home: close every popup and return to Dashboard/Dashboard, the
            # state a fresh boot starts in, so a live panel can run one flow
            # after another. Both are judged by the channel's own ack.
            need_ctl(lineno, cmd)
            reply = ctl.send(line)
            record(lineno, line, "ok" if reply.startswith(f"{cmd}: ok") else "fail", reply)
        elif cmd in ("page", "hit", "stats", "ctl"):
            # Query verbs: print the reply, never judge it. `ctl` is a raw
            # passthrough for anything the channel grows later.
            need_ctl(lineno, cmd)
            reply = ctl.send(line if cmd != "ctl" else " ".join(args))
            record(lineno, line, "ok", reply)
        elif cmd == "assert_page":
            need_ctl(lineno, cmd)
            want = " ".join(args)
            got = ctl.send("page")
            record(lineno, line, "ok" if got == want else "fail",
                   f"page is {got!r}" if got != want else "")
        elif cmd == "assert_hit":
            # assert_hit X Y CLASS [box=X1,Y1,WxH] [TEXT...]: a real tap at X,Y
            # would land on an object of CLASS, with that exact bounding box if
            # one is given, whose caption contains TEXT. This is the pre-tap
            # self-check: "the button moved" fails HERE, by name, so it is
            # never mistaken for the behaviour behind the button.
            #
            # The box is the strong identity. Two rows of a list share a class,
            # and an icon's caption is a glyph nobody wants in a spec; the
            # geometry the UI itself reports is what tells one instance from
            # another.
            need_ctl(lineno, cmd)
            x, y, cls = args[0], args[1], args[2]
            rest = args[3:]
            want_box = None
            if rest and rest[0].startswith("box="):
                want_box = rest[0][4:]
                rest = rest[1:]
            want_text = " ".join(rest)
            got = parse_hit(ctl.send(f"hit {x} {y}"))
            if got is None:
                record(lineno, line, "fail", "nothing clickable there")
            elif got[0] != cls:
                record(lineno, line, "fail", f"{got[0]} text={got[1]!r} box={got[2]}")
            elif want_box and got[2] != want_box:
                record(lineno, line, "fail", f"{cls} moved: box={got[2]}")
            elif want_text and want_text not in got[1]:
                record(lineno, line, "fail", f"{cls} text={got[1]!r}")
            else:
                record(lineno, line, "ok")
        elif cmd == "assert_json":
            # assert_json PATH OP VALUE: one read of the guest's status JSON
            # (see eval_json). PATH not (yet) in the doc is a `fail`, since
            # the doc is a live snapshot, not a schema.
            need_ctl(lineno, cmd)
            ok, detail = eval_json(ctl, args[0], args[1], args[2])
            record(lineno, line, "ok" if ok else "fail", detail)
        elif cmd == "wait_json":
            # wait_json PATH OP VALUE TIMEOUT_S: poll eval_json every 0.5s
            # until it holds or the deadline passes. The status file is
            # written by a 2s guest timer (webstatus.c), so this exists
            # because a value the doc will reach shortly is not the same
            # fact as a value it never reaches -- assert_json alone cannot
            # tell those apart.
            need_ctl(lineno, cmd)
            timeout_s = float(args[3])
            start = time.monotonic()
            deadline = start + timeout_s
            while True:
                ok, detail = eval_json(ctl, args[0], args[1], args[2])
                if ok or time.monotonic() >= deadline:
                    break
                time.sleep(0.5)
            waited = time.monotonic() - start
            record(lineno, line, "ok" if ok else "fail",
                   f"waited {waited:.1f}s" if ok else detail)
        elif cmd == "assert_stat":
            # assert_stat FIELD OP VALUE: FIELD off a fresh `stats` reply.
            need_ctl(lineno, cmd)
            field, op, value = args[0], args[1], args[2]
            if field not in ("cpu", "fps", "render", "idle"):
                record(lineno, line, "fail", f"unknown stat field {field!r} "
                       f"(known: cpu, fps, render, idle)")
            else:
                stats = parse_stats(ctl.send("stats"))
                if field not in stats:
                    record(lineno, line, "fail", "no such field")
                else:
                    try:
                        ok = apply_op(op, stats[field], value)
                    except (TypeError, ValueError) as e:
                        record(lineno, line, "fail", str(e))
                    else:
                        record(lineno, line, "ok" if ok else "fail",
                               "" if ok else f"{field} is {stats[field]}")
        elif cmd == "capture_region":
            # capture_region NAME X Y W H TOLERANCE: (re)writes NAME in the
            # refs file from a fresh screendump. Both hashes are stored
            # regardless of TOLERANCE so a later spec edit can change
            # tolerance class without a recapture.
            need_imgtools(lineno, cmd)
            name = args[0]
            x, y, w, h = int(args[1]), int(args[2]), int(args[3]), int(args[4])
            tolerance = args[5]
            if tolerance not in ("exact", "loose", "structural"):
                record(lineno, line, "fail", f"unknown tolerance {tolerance!r} "
                       f"(known: exact, loose, structural)")
            else:
                path = os.path.join(os.path.abspath(outdir), f"capture_{name}.ppm")
                rpc(s, f, {"execute": "screendump", "arguments": {"filename": path}})
                try:
                    cropped = imgtools.crop(imgtools.load_ppm(path), x, y, w, h)
                except (ValueError, OSError) as e:
                    record(lineno, line, "fatal", f"cannot capture {name!r}: {e}")
                else:
                    refs[name] = {
                        "x": x, "y": y, "w": w, "h": h, "tolerance": tolerance,
                        "phash": f"{imgtools.phash(cropped):016x}",
                        "structural": imgtools.structural(cropped).hex(),
                    }
                    save_refs(refs_path, refs)
                    record(lineno, line, "ok", f"captured box={x},{y},{w}x{h}")
        elif cmd == "assert_region":
            # assert_region NAME [TOLERANCE]: fresh screendump, crop to
            # NAME's stored box, compare at NAME's stored tolerance. A
            # missing NAME is FATAL for this step: an uncaptured reference
            # is a spec/authoring gap, not a UI defect the run should merely
            # `fail` on. So is a TOLERANCE other than the one NAME was
            # captured under: the script and the reference would be two
            # claims about the same pixels, and judging by either alone
            # would hide that.
            need_imgtools(lineno, cmd)
            name = args[0]
            want_tol = args[1] if len(args) > 1 else None
            ref = refs.get(name)
            if ref is not None and want_tol and want_tol != ref.get("tolerance"):
                record(lineno, line, "fatal",
                       f"reference {name!r} was captured as {ref.get('tolerance')}, "
                       f"script expects {want_tol}: recapture or fix the spec")
            else:
                ref, cropped, err = fresh_region(name, "assert")
                if err:
                    record(lineno, line, "fatal", err)
                else:
                    ok, detail = imgtools.compare(ref, cropped, ref["tolerance"])
                    record(lineno, line, "ok" if ok else "fail", detail)
        elif cmd == "assert_ocr":
            # assert_ocr NAME REGEX: same crop as assert_region, OCR'd
            # through `tesseract`, REGEX searched (re.search, spaces allowed
            # like assert_hit's TEXT) against the extracted text. No
            # tesseract on PATH, or no reference for NAME, is FATAL: pixels
            # were never actually checked, so this must never look like a
            # skipped-but-passing step.
            need_imgtools(lineno, cmd)
            name, pattern = args[0], " ".join(args[1:])
            if shutil.which("tesseract") is None:
                record(lineno, line, "fatal", "tesseract not installed")
            else:
                ref, cropped, err = fresh_region(name, "ocr")
                if err:
                    record(lineno, line, "fatal", err)
                else:
                    png_path = os.path.join(os.path.abspath(outdir), f"ocr_{name}.png")
                    try:
                        write_png(cropped, png_path)
                        text = subprocess.run(
                            ["tesseract", png_path, "stdout"],
                            capture_output=True, text=True, timeout=20, check=True,
                        ).stdout
                    except Exception as e:
                        record(lineno, line, "fatal", f"ocr failed: {e}")
                    else:
                        if re.search(pattern, text):
                            record(lineno, line, "ok")
                        else:
                            record(lineno, line, "fail", f"text was {text.strip()!r}")
        else:
            results.close()  # the rows before this one are still evidence
            sys.exit(f"FATAL: {script_path}:{lineno}: unknown command '{cmd}'")

        crashed = ui_exited(console_path)
        if crashed:
            record(lineno, line, "fatal", crashed)
            break

    results.close()
    print(f"== {counts['ok']} ok, {counts['fail']} failed, {counts['fatal']} fatal "
          f"-> {os.path.join(outdir, 'results.jsonl')}", flush=True)
    if counts["fail"] or counts["fatal"]:
        sys.exit(1)


def main():
    if len(sys.argv) < 3:
        sys.exit(__doc__)
    path, cmd = sys.argv[1], sys.argv[2]
    need = {"screendump": 4, "tap": 5, "quit": 3, "drive": 5}
    if cmd not in need:
        sys.exit(f"unknown command {cmd}\n{__doc__}")
    if len(sys.argv) < need[cmd]:
        sys.exit(f"{cmd}: missing argument(s)\n{__doc__}")

    size = 720
    if "--size" in sys.argv:
        size = int(sys.argv[sys.argv.index("--size") + 1])
    ctl_path = sys.argv[sys.argv.index("--ctl") + 1] if "--ctl" in sys.argv else None
    console_path = sys.argv[sys.argv.index("--console") + 1] if "--console" in sys.argv else None
    refs_path = sys.argv[sys.argv.index("--refs") + 1] if "--refs" in sys.argv else None

    s = socket.socket(socket.AF_UNIX)
    s.connect(path)
    f = s.makefile("r")
    f.readline()  # greeting banner
    rpc(s, f, {"execute": "qmp_capabilities"})

    if cmd == "screendump":
        rpc(s, f, {"execute": "screendump", "arguments": {"filename": sys.argv[3]}})
    elif cmd == "tap":
        do_tap(s, f, int(sys.argv[3]), int(sys.argv[4]))
    elif cmd == "drive":
        drive(s, f, sys.argv[3], sys.argv[4], size, ctl_path, console_path, refs_path)
    elif cmd == "quit":
        s.sendall(b'{"execute":"quit"}\n')


if __name__ == "__main__":
    main()