diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e607325..d19c783 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -111,6 +111,14 @@ jobs: shellcheck -x qemu/*.sh qemu/tests/*.sh build/*.sh \ qemu/rootfs/etc/warden-lib.sh qemu/rootfs/etc/rc \ qemu/rootfs/sbin/init qemu/rootfs/init + - name: ui-drive driver and image tools (offline) + # qmp.py's drive() with QMP and the control channel faked, plus + # imgtools' self-test: the per-step ok/fail/fatal contract and the + # region reference math, no VM needed. + run: | + sudo apt-get install -y -qq python3-pil + python3 qemu/tests/imgtools.py selftest + python3 qemu/tests/test_qmp_drive.py - name: cache pinned busybox uses: actions/cache@v4 with: diff --git a/qemu/mkimage.sh b/qemu/mkimage.sh index dad66a9..c1de0a5 100755 --- a/qemu/mkimage.sh +++ b/qemu/mkimage.sh @@ -15,8 +15,14 @@ # flare-edge build's rootfs.img/oem.img matched pair) into slot A instead of # the busybox skeleton; slot B keeps the skeleton as a known-good fallback. # Env: -# BUSYBOX path to a local busybox binary (skips the download; still verified) -# OUT output dir (default: qemu/out); image at $OUT/disk.img +# BUSYBOX path to a local busybox binary (skips the download; still verified) +# OUT output dir (default: qemu/out); image at $OUT/disk.img +# SEED_DIR a directory of pre-built userdata/warden files (one file per +# settings key, e.g. flare-edge tools/seed-fixtures.py's output) +# copied in VERBATIM, after --portal-url/--state. Lets a caller +# seed an arbitrary key set -- a flow spec's setup.seed can name +# anything settings.c reads -- without growing --state into a +# multi-value flag; last one written wins, same as --state. set -euo pipefail QEMU_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" @@ -85,6 +91,17 @@ for kv in ${STATE_KV[@]+"${STATE_KV[@]}"}; do printf '%s\n' "${kv#*=}" > "$UDATA/warden/${kv%%=*}" done +# SEED_DIR: a pre-built set of settings files (see the Env note above), copied +# in whole rather than re-parsed here -- seed-fixtures.py already wrote them +# in the exact format settings.c reads (filename = key, mode 0600 for a +# secret), so re-deriving that here would be a second place to keep in sync +# with settings.c. Applied after --state so a seeded file can override a +# same-named --state value; `cp -a` preserves the 0600 on a secret entry. +if [ -n "${SEED_DIR:-}" ]; then + [ -d "$SEED_DIR" ] || { echo "FATAL: SEED_DIR '$SEED_DIR' is not a directory" >&2; exit 1; } + cp -a "$SEED_DIR"/. "$UDATA/warden/" +fi + mkdir -p "$SCRATCH/empty" # mkfs an ext4 partition image of exactly $2 bytes from staged dir $1. diff --git a/qemu/rootfs/sbin/init b/qemu/rootfs/sbin/init index 80059c7..52f3420 100755 --- a/qemu/rootfs/sbin/init +++ b/qemu/rootfs/sbin/init @@ -129,9 +129,16 @@ fi # equivalent seam is a second 16550 that run.sh --ctl exposes as a unix socket # (pci-serial, the same device the RS485 bridge already rides). One command # per line in, the FIFO's reply out, and a sentinel line so the reader knows -# the reply is complete without a timeout. The vocabulary is identical on both -# sides of that seam, which is what lets one flow script run against the sim -# and against a panel. +# the reply is complete without a timeout. The FIFO vocabulary itself is +# identical on both sides of that seam, which is what lets one flow script +# run against the sim and against a panel. +# +# One exception, answered by the bridge itself and never forwarded to the +# FIFO: `@cat PATH` replies with PATH's contents (or one "bridge: no such +# file: PATH" line if it is missing), then the same sentinel. This is how the +# json flow channel reads webstatus.c's /tmp/warden-web-status.json snapshot +# from OUTSIDE the VM -- on a panel that file is just as reachable over the +# SSH session tools/warden-ctl already has, so hardware needs no equivalent. # # run.sh lists the ctl port before any other pci-serial, so it is always the # first 8250, and it says so with warden.ctl on the command line. The marker, @@ -149,16 +156,42 @@ if grep -qw warden.ctl /proc/cmdline && [ -c /dev/ttyS0 ]; then exec 3<> "$ctl" while IFS= read -r cmd <&3; do [ -n "$cmd" ] || continue - if [ -p /tmp/warden-ui.ctl ]; then - printf '%s\n' "$cmd" > /tmp/warden-ui.ctl - # The UI polls its FIFO every 100 ms and truncates the reply - # file on each command, so a short settle then a read is the - # same protocol warden-ctl uses over SSH. - sleep 0.3 - cat /tmp/warden-ui.dbg 2>/dev/null >&3 - else - echo "bridge: warden-ui control FIFO not present" >&3 - fi + case "$cmd" in + "@cat "*) + # A bridge-local command, never forwarded to warden-ui's + # FIFO: `@cat PATH` reads PATH directly off the GUEST's + # own filesystem and answers with it, which is how the + # json flow channel gets webstatus.c's snapshot out to + # the host driving the VM from outside. `-f` so a + # directory or device node reports as missing rather than + # cat hanging or erroring oddly. + path="${cmd#@cat }" + if [ -f "$path" ]; then + cat "$path" >&3 + # Force a newline after the file's own bytes: the + # status json (webstatus.c) is written with NO + # trailing newline, and without this the sentinel + # below would land on the SAME line as the content + # and the reader (qmp.py Ctl.send, line-based) would + # block forever waiting for a line that never comes. + echo >&3 + else + echo "bridge: no such file: $path" >&3 + fi + ;; + *) + if [ -p /tmp/warden-ui.ctl ]; then + printf '%s\n' "$cmd" > /tmp/warden-ui.ctl + # The UI polls its FIFO every 100 ms and truncates the + # reply file on each command, so a short settle then a + # read is the same protocol warden-ctl uses over SSH. + sleep 0.3 + cat /tmp/warden-ui.dbg 2>/dev/null >&3 + else + echo "bridge: warden-ui control FIFO not present" >&3 + fi + ;; + esac echo "<>" >&3 done ) & diff --git a/qemu/tests/imgtools.py b/qemu/tests/imgtools.py new file mode 100644 index 0000000..ff61f56 --- /dev/null +++ b/qemu/tests/imgtools.py @@ -0,0 +1,444 @@ +#!/usr/bin/env python3 +"""PPM loading, cropping and image comparison for UI flow region asserts. + +Standalone (no VM, no other module in this repo needed): a screendump is +QEMU's own P6 PPM file, and everything below works straight off that file +plus a JSON reference. Pure Python plus Pillow only. + + imgtools.py compare REFS NAME SHOT.ppm + imgtools.py capture REFS NAME SHOT.ppm X Y W H TOL # TOL: exact|loose|structural + imgtools.py hash SHOT.ppm + imgtools.py selftest + +`compare` and `capture` read/write a reference file: JSON mapping a region +name to {"x","y","w","h","tolerance","phash","structural"}. `capture` always +fills in both hashes so an existing entry's tolerance class can be changed +later without re-shooting the region. + +Public API (imported directly by the driver, not just via the CLI above): + + load_ppm(path) -> (w, h, rgb_bytes) + crop(img, x, y, w, h) -> img + phash(img) -> int 8x8 DCT of a 32x32 grey downscale, 64-bit + structural(img) -> bytes 16x16 edge/nonblack occupancy mask, 32 bytes + compare(ref_entry, cur_img, tolerance) -> (ok, detail) + +`img` is always the (w, h, rgb_bytes) triple load_ppm/crop return -- there is +no separate image type. rgb_bytes is tightly packed row-major RGB, 3 bytes +per pixel, no padding, matching both PPM P6 and PIL's "RGB" raw layout. +""" +import math +import os +import sys +import json + +from PIL import Image, ImageFilter + +# occupancy thresholds for structural(): a downsampled cell counts as +# "occupied" if it is meaningfully brighter than black, or sits on an edge. +# Both are 0..255 greyscale/edge-magnitude averages over the cell. +NONBLACK_THRESHOLD = 10 +EDGE_THRESHOLD = 10 + + +# --------------------------------------------------------------------------- +# PPM +# --------------------------------------------------------------------------- + +def load_ppm(path): + """Read a binary PPM (P6), return (w, h, rgb_bytes). + + QEMU's screendump writes a comment-free header ("P6\\nW H\\n255\\n" + then raw bytes) but the token reader here handles the general P6 + grammar -- whitespace runs and '#' comments -- since that costs + nothing extra and means any other producer of P6 files just works. + """ + with open(path, "rb") as f: + data = f.read() + + def skip_ws_comments(p): + while p < len(data): + c = data[p] + if c in b" \t\r\n": + p += 1 + elif c == ord("#"): + while p < len(data) and data[p] != ord("\n"): + p += 1 + else: + break + return p + + def read_token(p): + p = skip_ws_comments(p) + start = p + while p < len(data) and data[p] not in b" \t\r\n": + p += 1 + return data[start:p], p + + pos = 0 + magic, pos = read_token(pos) + if magic != b"P6": + raise ValueError(f"{path}: not a P6 PPM (magic={magic!r})") + w_tok, pos = read_token(pos) + h_tok, pos = read_token(pos) + maxval_tok, pos = read_token(pos) + w, h, maxval = int(w_tok), int(h_tok), int(maxval_tok) + if maxval != 255: + raise ValueError(f"{path}: unsupported PPM maxval {maxval} (only 255 handled)") + # exactly one whitespace byte separates the header from the binary data + pos += 1 + need = w * h * 3 + pixels = data[pos:pos + need] + if len(pixels) != need: + raise ValueError(f"{path}: truncated PPM, want {need} bytes got {len(pixels)}") + return w, h, pixels + + +def crop(img, x, y, w, h): + """Crop (W,H,rgb) to the x,y,w,h box, return a new (w,h,rgb) image.""" + width, height, data = img + if x < 0 or y < 0 or w <= 0 or h <= 0 or x + w > width or y + h > height: + raise ValueError(f"crop box {x},{y},{w}x{h} outside image {width}x{height}") + row_bytes = w * 3 + out = bytearray(h * row_bytes) + for row in range(h): + src_off = ((y + row) * width + x) * 3 + dst_off = row * row_bytes + out[dst_off:dst_off + row_bytes] = data[src_off:src_off + row_bytes] + return w, h, bytes(out) + + +def _to_pil(img): + w, h, data = img + return Image.frombytes("RGB", (w, h), data) + + +# --------------------------------------------------------------------------- +# perceptual hash (phash): 8x8 DCT of a 32x32 grey downscale +# --------------------------------------------------------------------------- + +def _dct_basis(n): + """n x n DCT-II basis matrix. Unnormalised -- fine since phash only + thresholds coefficients against each other, never compares magnitudes + across images.""" + return [[math.cos(math.pi / n * (col + 0.5) * row) for col in range(n)] for row in range(n)] + + +_DCT32 = _dct_basis(32) + + +def _dct2d_32(rows): + """2D DCT-II of a 32x32 list-of-lists via two separable 1D passes.""" + n = 32 + # columns first: tmp[u][x] = DCT of column x at frequency u + tmp = [[0.0] * n for _ in range(n)] + for u in range(n): + b = _DCT32[u] + for x in range(n): + s = 0.0 + for y in range(n): + s += rows[y][x] * b[y] + tmp[u][x] = s + # then rows: out[u][v] = DCT of tmp's row u at frequency v + out = [[0.0] * n for _ in range(n)] + for u in range(n): + row = tmp[u] + for v in range(n): + b = _DCT32[v] + s = 0.0 + for x in range(n): + s += row[x] * b[x] + out[u][v] = s + return out + + +def phash(img): + """8x8-DCT perceptual hash of a 32x32 greyscale downscale, as a 64-bit + int. Classic pHash recipe: downscale, DCT, keep the low-frequency 8x8 + corner, threshold each coefficient against the block's mean (DC term + excluded from the mean -- it is just overall brightness and would bias + every bit the same way).""" + small = _to_pil(img).convert("L").resize((32, 32), Image.LANCZOS) + px = small.load() + rows = [[px[x, y] for x in range(32)] for y in range(32)] + coeffs = _dct2d_32(rows) + block = [coeffs[u][v] for u in range(8) for v in range(8)] + mean = sum(block[1:]) / (len(block) - 1) + bits = 0 + for c in block: + bits = (bits << 1) | (1 if c > mean else 0) + return bits + + +def hamming(a, b): + """Bit-differences between two phash ints.""" + return bin(a ^ b).count("1") + + +# --------------------------------------------------------------------------- +# structural occupancy hash: 16x16 edge/nonblack mask +# --------------------------------------------------------------------------- + +def structural(img): + """16x16 binary occupancy mask as 32 bytes (256 bits, MSB first, + row-major). A cell is "occupied" if it is meaningfully non-black or + sits on an edge, so the mask is robust to a recolour (still occupied) + but sensitive to a shape disappearing (goes from occupied to empty).""" + grey = _to_pil(img).convert("L") + edges = grey.filter(ImageFilter.FIND_EDGES) + grey_small = grey.resize((16, 16), Image.BOX) + edge_small = edges.resize((16, 16), Image.BOX) + gpx, epx = grey_small.load(), edge_small.load() + bits = bytearray(32) + idx = 0 + for y in range(16): + for x in range(16): + occupied = gpx[x, y] > NONBLACK_THRESHOLD or epx[x, y] > EDGE_THRESHOLD + if occupied: + bits[idx // 8] |= 1 << (7 - (idx % 8)) + idx += 1 + return bytes(bits) + + +def _structural_diff(a, b): + """Count of differing bits between two 32-byte occupancy masks.""" + return sum(bin(x ^ y).count("1") for x, y in zip(a, b)) + + +# --------------------------------------------------------------------------- +# compare +# --------------------------------------------------------------------------- + +def compare(ref_entry, cur_img, tolerance): + """Compare cur_img (already cropped to the reference's box) against + ref_entry under the given tolerance class. Returns (ok, detail); detail + explains the verdict either way, for results.jsonl. + + exact: phash hamming distance == 0 + loose: phash hamming distance <= 6 + structural: occupancy mask bytes equal (no distance, no tolerance) + """ + if tolerance in ("exact", "loose"): + if "phash" not in ref_entry: + return False, "reference missing 'phash' field" + try: + ref_hash = int(ref_entry["phash"], 16) + except (TypeError, ValueError): + return False, f"reference 'phash' is not valid hex: {ref_entry.get('phash')!r}" + cur_hash = phash(cur_img) + dist = hamming(ref_hash, cur_hash) + limit = 0 if tolerance == "exact" else 6 + ok = dist <= limit + detail = f"phash hamming={dist} limit={limit} ref={ref_hash:016x} cur={cur_hash:016x}" + return ok, detail + + if tolerance == "structural": + if "structural" not in ref_entry: + return False, "reference missing 'structural' field" + try: + ref_bits = bytes.fromhex(ref_entry["structural"]) + except (TypeError, ValueError): + return False, f"reference 'structural' is not valid hex: {ref_entry.get('structural')!r}" + cur_bits = structural(cur_img) + ok = ref_bits == cur_bits + if ok: + detail = "structural mask matches (0/256 cells differ)" + else: + diff = _structural_diff(ref_bits, cur_bits) + detail = f"structural mask differs in {diff}/256 cells" + return ok, detail + + return False, f"unknown tolerance class: {tolerance!r}" + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +def _load_refs(path): + if not os.path.exists(path): + return {} + with open(path) as f: + return json.load(f) + + +def _save_refs(path, refs): + with open(path, "w") as f: + json.dump(refs, f, indent=2, sort_keys=True) + f.write("\n") + + +def cmd_compare(argv): + if len(argv) != 3: + print("usage: imgtools.py compare REFS NAME SHOT.ppm", file=sys.stderr) + return 2 + refs_path, name, shot_path = argv + refs = _load_refs(refs_path) + if name not in refs: + print(f"FAIL {name}: no such reference (run capture first)") + return 1 + entry = refs[name] + for key in ("x", "y", "w", "h", "tolerance"): + if key not in entry: + print(f"FAIL {name}: reference entry missing '{key}'") + return 1 + w, h, data = load_ppm(shot_path) + cropped = crop((w, h, data), entry["x"], entry["y"], entry["w"], entry["h"]) + ok, detail = compare(entry, cropped, entry["tolerance"]) + print(f"{'PASS' if ok else 'FAIL'} {name}: {detail}") + return 0 if ok else 1 + + +def cmd_capture(argv): + if len(argv) != 8: + print("usage: imgtools.py capture REFS NAME SHOT.ppm X Y W H TOL", file=sys.stderr) + return 2 + refs_path, name, shot_path, x, y, w, h, tol = argv + if tol not in ("exact", "loose", "structural"): + print(f"usage: TOL must be exact|loose|structural, got {tol!r}", file=sys.stderr) + return 2 + x, y, w, h = int(x), int(y), int(w), int(h) + iw, ih, data = load_ppm(shot_path) + cropped = crop((iw, ih, data), x, y, w, h) + entry = { + "x": x, "y": y, "w": w, "h": h, + "tolerance": tol, + "phash": f"{phash(cropped):016x}", + "structural": structural(cropped).hex(), + } + refs = _load_refs(refs_path) + refs[name] = entry + _save_refs(refs_path, refs) + print(f"captured {name}: {entry}") + return 0 + + +def cmd_hash(argv): + if len(argv) != 1: + print("usage: imgtools.py hash SHOT.ppm", file=sys.stderr) + return 2 + w, h, data = load_ppm(argv[0]) + img = (w, h, data) + print(f"{w}x{h} phash={phash(img):016x} structural={structural(img).hex()}") + return 0 + + +# --------------------------------------------------------------------------- +# selftest +# --------------------------------------------------------------------------- + +def _make_test_rgb(w, h, colour, shift=(0, 0)): + """Black canvas with a coloured square near one corner, used only by + the self-test. colour=None means no square at all (removed shape).""" + from PIL import ImageDraw + img = Image.new("RGB", (w, h), (0, 0, 0)) + if colour is not None: + draw = ImageDraw.Draw(img) + ox, oy = shift + x0, y0 = 8 + ox, 8 + oy + draw.rectangle([x0, y0, x0 + 20, y0 + 20], fill=colour) + return img.tobytes() + + +def _write_ppm(path, w, h, rgb_bytes): + with open(path, "wb") as f: + f.write(f"P6\n{w} {h}\n255\n".encode("ascii")) + f.write(rgb_bytes) + + +def selftest(): + import tempfile + + w, h = 128, 128 + failures = [] + + def check(label, cond): + print(f"[{'ok' if cond else 'FAIL'}] {label}") + if not cond: + failures.append(label) + + with tempfile.TemporaryDirectory() as tmp: + refs_path = os.path.join(tmp, "refs.json") + + base_path = os.path.join(tmp, "base.ppm") + _write_ppm(base_path, w, h, _make_test_rgb(w, h, (220, 20, 20))) + + # capture the same corner box under all three tolerance classes + for tol in ("exact", "loose", "structural"): + rc = main(["capture", refs_path, f"corner-{tol}", base_path, "4", "4", "32", "32", tol]) + check(f"capture corner-{tol} succeeds", rc == 0) + + # load_ppm/crop sanity: the drawn pixel lands where expected + bw, bh, bdata = load_ppm(base_path) + check("load_ppm reads back the written size", (bw, bh) == (w, h)) + cropped = crop((bw, bh, bdata), 4, 4, 32, 32) + px_off = ((8 - 4) * 32 + (8 - 4)) * 3 + check("crop keeps the drawn pixel in place", + cropped[2][px_off:px_off + 3] == bytes((220, 20, 20))) + + # identical shot: exact must pass + same_path = os.path.join(tmp, "same.ppm") + _write_ppm(same_path, w, h, _make_test_rgb(w, h, (220, 20, 20))) + rc = main(["compare", refs_path, "corner-exact", same_path]) + check("exact passes for an identical shot", rc == 0) + + # small shift: exact must fail, loose must still pass + shift_path = os.path.join(tmp, "shift.ppm") + _write_ppm(shift_path, w, h, _make_test_rgb(w, h, (220, 20, 20), shift=(1, 1))) + rc = main(["compare", refs_path, "corner-exact", shift_path]) + check("exact fails for a slight shift", rc == 1) + rc = main(["compare", refs_path, "corner-loose", shift_path]) + check("loose passes for a slight shift", rc == 0) + + # colour change, same footprint: structural must still pass + colour_path = os.path.join(tmp, "colour.ppm") + _write_ppm(colour_path, w, h, _make_test_rgb(w, h, (20, 20, 220))) + rc = main(["compare", refs_path, "corner-structural", colour_path]) + check("structural passes for a colour change", rc == 0) + + # shape removed entirely: structural must fail + removed_path = os.path.join(tmp, "removed.ppm") + _write_ppm(removed_path, w, h, _make_test_rgb(w, h, None)) + rc = main(["compare", refs_path, "corner-structural", removed_path]) + check("structural fails when the shape is removed", rc == 1) + + # bad-path coverage + rc = main(["compare", refs_path, "no-such-name", base_path]) + check("compare on an unknown reference name fails loudly", rc == 1) + rc = main(["hash", base_path]) + check("hash on a real ppm succeeds", rc == 0) + + if failures: + print(f"\n{len(failures)} check(s) failed:") + for label in failures: + print(f" - {label}") + return 1 + print("\nall checks passed") + return 0 + + +# --------------------------------------------------------------------------- + +def main(argv): + if not argv: + print(__doc__, file=sys.stderr) + return 2 + cmd, rest = argv[0], argv[1:] + if cmd == "selftest": + return selftest() + handlers = {"compare": cmd_compare, "capture": cmd_capture, "hash": cmd_hash} + if cmd not in handlers: + print(f"unknown command: {cmd!r}", file=sys.stderr) + return 2 + # A missing file, unreadable ref JSON, or malformed PPM/crop box is a + # prerequisite failure -- report it plainly and fail loudly (exit 1), + # never let it fall through as a silent pass or a raw traceback. + try: + return handlers[cmd](rest) + except (OSError, ValueError, json.JSONDecodeError) as exc: + print(f"FATAL: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/qemu/tests/qmp.py b/qemu/tests/qmp.py index 0af8c8a..d60b172 100755 --- a/qemu/tests/qmp.py +++ b/qemu/tests/qmp.py @@ -35,6 +35,52 @@ a real panel, so these verbs mean the same thing on the rig and on hardware: reports a MOVED button as such, not as broken behaviour +Two more channels reach behavioural state instead of the widget tree. +`--refs FILE` (default /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 or render, 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 /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 @@ -46,13 +92,38 @@ 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 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()) @@ -198,8 +269,168 @@ def ui_exited(console_path): return data[i:i + 80].split(b"\n", 1)[0].decode("utf-8", "replace") -def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None): - import os +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*$'), +} + + +def parse_stats(reply): + """The `stats` reply (warden_debug.c: 'page: X\\ncpu: N%\\nfps: N\\n + render: A.BB ms/frame\\nrga: N%') -> {'cpu'|'fps'|'render': 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() @@ -221,6 +452,36 @@ def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=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: @@ -296,7 +557,134 @@ def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None): 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"): + record(lineno, line, "fail", f"unknown stat field {field!r} " + f"(known: cpu, fps, render)") + 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) @@ -326,6 +714,7 @@ def main(): 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) @@ -338,7 +727,7 @@ def main(): 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) + 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') diff --git a/qemu/tests/test_qmp_drive.py b/qemu/tests/test_qmp_drive.py new file mode 100755 index 0000000..c6e4261 --- /dev/null +++ b/qemu/tests/test_qmp_drive.py @@ -0,0 +1,176 @@ +#!/usr/bin/env python3 +"""Offline tests for qmp.py's drive(): the QMP socket and the control channel +are faked, so this runs in well under two seconds with no VM. + +What is worth pinning is the contract the docstring makes: every step gets a +results.jsonl row of ok / fail / fatal and the run CONTINUES, so one run +reports every broken expectation. The region verbs are where that was once +false (flare-edge issue #147): a reference whose box did not fit the +screendump raised out of drive() as a traceback, and a tolerance on the line +was never compared with the one the reference was captured under. + + python3 test_qmp_drive.py +""" +import json +import os +import sys +import tempfile +import time +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) +import qmp # noqa: E402 + +SIZE = 64 + + +def ppm_bytes(fill=0): + return b"P6\n%d %d\n255\n" % (SIZE, SIZE) + bytes([fill]) * (SIZE * SIZE * 3) + + +class FakeCtl: + """Canned replies in the shapes warden_debug.c actually produces.""" + + def __init__(self, path, timeout=15.0): + self.path = path + + def send(self, cmd): + if cmd == "page": + return "Demo/Rows" + if cmd == "stats": + return "page: Demo/Rows\ncpu: 12%\nfps: 10\nrender: 3.20 ms/frame\nrga: 0%" + if cmd.startswith("@cat "): + return json.dumps({"a": {"b": 1}, "list": [1, 2], "name": "warden"}) + if cmd.startswith("hit "): + return 'hit 47,676: obj text="" box=12,640,72x72' + if cmd.startswith("nav "): + return cmd + ": ok" + return "" + + +def fake_rpc(sock, sock_file, obj): + if obj.get("execute") == "screendump": + with open(obj["arguments"]["filename"], "wb") as fh: + fh.write(ppm_bytes()) + return {} + + +def run_script(text, refs=None): + """-> (exit code or None, {cmd: row}, rows) for one drive() over TEXT.""" + outdir = tempfile.mkdtemp(prefix="qmpdrive.") + script = os.path.join(outdir, "s.txt") + with open(script, "w") as fh: + fh.write(text) + refs_path = os.path.join(outdir, "refs.json") + if refs is not None: + with open(refs_path, "w") as fh: + json.dump(refs, fh) + saved = qmp.rpc, qmp.Ctl + qmp.rpc, qmp.Ctl = fake_rpc, FakeCtl + rc = None + try: + try: + qmp.drive(None, None, script, outdir, SIZE, ctl_path="fake", + console_path=None, refs_path=refs_path) + except SystemExit as e: + rc = e.code + finally: + qmp.rpc, qmp.Ctl = saved + with open(os.path.join(outdir, "results.jsonl")) as fh: + rows = [json.loads(line) for line in fh if line.strip()] + return rc, {r["cmd"]: r for r in rows}, rows + + +class PureHelpers(unittest.TestCase): + def test_resolve_path_and_ops(self): + doc = {"a": {"b": [5, 6]}, "s": "connected"} + self.assertEqual(qmp.resolve_path(doc, "a.b[1]"), (6, None)) + self.assertIsNotNone(qmp.resolve_path(doc, "a.c")[1]) + self.assertTrue(qmp.apply_op("eq", 1, "1")) + self.assertTrue(qmp.apply_op("ne", 1, "2")) + self.assertTrue(qmp.apply_op("contains", "connected", "nect")) + self.assertTrue(qmp.apply_op("len_ge", [1, 2], "2")) + self.assertTrue(qmp.apply_op("len_eq", [1, 2], "2")) + self.assertTrue(qmp.apply_op("gt", 3.0, "2")) + self.assertTrue(qmp.apply_op("lt", 1, "2")) + self.assertTrue(qmp.apply_op("eq", "connected", "connected")) + + def test_parse_stats(self): + got = qmp.parse_stats(FakeCtl("x").send("stats")) + self.assertEqual(got, {"cpu": 12.0, "fps": 10.0, "render": 3.2}) + + +class DriveVerbs(unittest.TestCase): + def test_every_channel_passes_on_a_healthy_ui(self): + rc, by, rows = run_script( + "assert_page Demo/Rows\n" + "assert_hit 47 676 obj box=12,640,72x72\n" + "wait_json a.b eq 1 2\n" + "wait_json list len_ge 2 2\n" + "assert_json name eq warden\n" + "assert_stat fps gt 0\n" + "nav Demo/Rows\n" + "capture_region r1 0 0 8 8 exact\n" + "assert_region r1 exact\n" + "assert_region r1\n" + ) + self.assertIsNone(rc, [r for r in rows if r["status"] != "ok"]) + self.assertEqual(len(rows), 10) + self.assertTrue(all(r["status"] == "ok" for r in rows)) + + def test_mismatches_are_fails_not_stops(self): + t0 = time.monotonic() + rc, by, rows = run_script( + "assert_page Demo/Other\n" + "assert_hit 47 676 obj box=0,0,1x1\n" + "wait_json a.b eq 2 1\n" + "assert_json a.zz eq 1\n" + "assert_stat fps lt 0\n" + "assert_page Demo/Rows\n" + ) + self.assertEqual(rc, 1) + self.assertEqual([r["status"] for r in rows], + ["fail", "fail", "fail", "fail", "fail", "ok"]) + self.assertIn("moved", by["assert_hit 47 676 obj box=0,0,1x1"]["detail"]) + self.assertGreaterEqual(time.monotonic() - t0, 1.0, "wait_json must honour its timeout") + + def test_region_faults_are_per_step_fatal(self): + # A pre-seeded reference whose box does not fit a 64x64 screendump, + # a name with no reference, a tolerance other than the captured one, + # and a capture box off the screen: each is FATAL for its own step + # and the assert_page after them still runs. + refs = {"big": {"x": 10, "y": 10, "w": 1000, "h": 1000, "tolerance": "exact", + "phash": "0" * 16, "structural": "00" * 32}} + rc, by, rows = run_script( + "assert_region big exact\n" + "assert_region nope exact\n" + "capture_region r1 0 0 8 8 exact\n" + "assert_region r1 loose\n" + "capture_region huge 0 0 999 999 exact\n" + "capture_region r2 0 0 8 8 fuzzy\n" + "assert_page Demo/Rows\n", + refs=refs, + ) + self.assertEqual(rc, 1) + self.assertEqual([r["status"] for r in rows], + ["fatal", "fatal", "ok", "fatal", "fatal", "fail", "ok"]) + self.assertIn("unusable", by["assert_region big exact"]["detail"]) + self.assertIn("no reference", by["assert_region nope exact"]["detail"]) + self.assertIn("captured as exact, script expects loose", + by["assert_region r1 loose"]["detail"]) + self.assertIn("cannot capture", by["capture_region huge 0 0 999 999 exact"]["detail"]) + self.assertIn("unknown tolerance", by["capture_region r2 0 0 8 8 fuzzy"]["detail"]) + + def test_unknown_verb_is_fatal_for_the_run(self): + # A silently-ignored line is a test that proves nothing, so this one + # is the documented exception to "the run continues": drive() exits + # with the message rather than recording a row. + rc, by, rows = run_script("assert_page Demo/Rows\nfrobnicate 1 2\n") + self.assertIsInstance(rc, str) + self.assertIn("unknown command 'frobnicate'", rc) + self.assertEqual([r["status"] for r in rows], ["ok"]) + + +if __name__ == "__main__": + unittest.main(verbosity=1) diff --git a/qemu/tests/ui-drive.sh b/qemu/tests/ui-drive.sh index 93bcb45..d38e1b2 100755 --- a/qemu/tests/ui-drive.sh +++ b/qemu/tests/ui-drive.sh @@ -2,8 +2,8 @@ # Drive the LVGL UI through a scripted interaction and collect screenshots. # # ui-shot.sh proves touch reaches the UI in one tap; this is the same rig for -# work that needs a SEQUENCE — swipe through the app rows, open a submenu, tap a -# tab, bring up the keyboard — with a screendump wherever the script asks for +# work that needs a SEQUENCE (swipe through the app rows, open a submenu, tap a +# tab, bring up the keyboard) with a screendump wherever the script asks for # one. One boot serves the whole script, because booting per step (TCG, no KVM) # costs about a minute and a real interaction is thirty steps. # @@ -14,23 +14,53 @@ # # FAILS CLOSED on missing prerequisites. # -# Usage: ui-drive.sh