diff --git a/qemu/tests/imgtools.py b/qemu/tests/imgtools.py index 4c7e043..73cf6e0 100644 --- a/qemu/tests/imgtools.py +++ b/qemu/tests/imgtools.py @@ -8,6 +8,7 @@ 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 bench [SHOT.ppm] [N] # per-call phash/structural timing, N iterations imgtools.py selftest `compare` and `capture` read/write a reference file: JSON mapping a region @@ -130,24 +131,35 @@ def _dct_basis(n): _DCT32 = _dct_basis(32) +# phash() only ever reads the low-frequency _DCT_KEEP x _DCT_KEEP corner of +# the 32x32 DCT (see its docstring), so _dct2d_32 stops both passes at this +# many frequencies instead of computing all 1024 coefficients -- same 64 +# coefficients out, roughly 6x less pure-Python multiply-add work per call. +_DCT_KEEP = 8 + def _dct2d_32(rows): - """2D DCT-II of a 32x32 list-of-lists via two separable 1D passes.""" + """2D DCT-II of a 32x32 list-of-lists via two separable 1D passes, + truncated to the low-frequency _DCT_KEEP x _DCT_KEEP corner -- the only + coefficients any caller reads.""" 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): + # columns first: tmp[u][x] = DCT of column x at frequency u. Frequencies + # u >= _DCT_KEEP never feed a returned coefficient, so skip them. + tmp = [[0.0] * n for _ in range(_DCT_KEEP)] + for u in range(_DCT_KEEP): 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): + # then rows: out[u][v] = DCT of tmp's row u at frequency v. Same cutoff + # on v; the y/x summations stay full since each kept coefficient still + # needs the whole 32-wide signal. + out = [[0.0] * _DCT_KEEP for _ in range(_DCT_KEEP)] + for u in range(_DCT_KEEP): row = tmp[u] - for v in range(n): + for v in range(_DCT_KEEP): b = _DCT32[v] s = 0.0 for x in range(n): @@ -166,7 +178,7 @@ def phash(img): 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)] + block = [coeffs[u][v] for u in range(_DCT_KEEP) for v in range(_DCT_KEEP)] mean = sum(block[1:]) / (len(block) - 1) bits = 0 for c in block: @@ -350,13 +362,64 @@ def cmd_hash(argv): return 0 +def cmd_bench(argv): + """Time phash()/structural() -- the per-step cost of every + assert_region/wait_region/capture_region in a flow run (wait_region polls + at 0.5s intervals, calling back into these on every poll), so a change to + the DCT size, the downscale filter, or the occupancy thresholds shows up + as a number here instead of only as a slower flow run nobody investigates. + + usage: imgtools.py bench [SHOT.ppm] [N] + + SHOT.ppm: a real screendump to benchmark against a representative crop + of; default is a synthetic 720x720 image (the virt.fragment screendump + size) so bench never depends on a committed fixture or a live rig. + N: iterations per function, default 200. + """ + import timeit + + if len(argv) > 2: + print("usage: imgtools.py bench [SHOT.ppm] [N]", file=sys.stderr) + return 2 + shot_path = next((a for a in argv if not a.isdigit()), None) + n = int(next((a for a in argv if a.isdigit()), "200")) + + if shot_path: + w, h, data = load_ppm(shot_path) + else: + w, h = 720, 720 + data = _make_test_rgb(w, h, (220, 20, 20)) + img = (w, h, data) + # a representative crop, not the whole screen: every real region assert + # crops first, and structural()'s edge filter cost scales with crop + # size even though phash's fixed 32x32 downscale mostly doesn't. + cw, ch = min(200, w), min(200, h) + region = crop(img, 0, 0, cw, ch) + + def percentiles(samples): + s = sorted(samples) + p50 = s[min(len(s) - 1, int(len(s) * 0.50))] + p95 = s[min(len(s) - 1, int(len(s) * 0.95))] + return s[0], p50, p95, s[-1] + + print(f"bench: {cw}x{ch} region, n={n} iterations") + for label, fn in (("phash", lambda: phash(region)), + ("structural", lambda: structural(region))): + samples = timeit.repeat(fn, repeat=n, number=1) + lo, p50, p95, hi = percentiles(samples) + print(f" {label:<10} min={lo * 1000:7.3f}ms p50={p50 * 1000:7.3f}ms " + f"p95={p95 * 1000:7.3f}ms max={hi * 1000:7.3f}ms") + 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).""" + """Black canvas with a coloured square near one corner: the self-test's + synthetic images, and bench's default when it isn't given a real + SHOT.ppm. 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: @@ -373,6 +436,33 @@ def _write_ppm(path, w, h, rgb_bytes): f.write(rgb_bytes) +def _dct2d_32_brute(rows): + """Unoptimised reference DCT: the full 32x32 transform with no early + cutoff. selftest()-only, so _dct2d_32's _DCT_KEEP truncation has + something independent to be checked against -- a future edit that moves + the cutoff on the wrong loop would otherwise change captured phash bits + with nothing catching it.""" + n = 32 + 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 + 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 selftest(): import tempfile @@ -403,6 +493,16 @@ def selftest(): check("crop keeps the drawn pixel in place", cropped[2][px_off:px_off + 3] == bytes((220, 20, 20))) + # _dct2d_32's _DCT_KEEP truncation must land on the exact same + # coefficients an untruncated 32x32 DCT would produce + small = _to_pil((bw, bh, bdata)).convert("L").resize((32, 32), Image.LANCZOS) + px = small.load() + dct_rows = [[px[x, y] for x in range(32)] for y in range(32)] + fast, brute = _dct2d_32(dct_rows), _dct2d_32_brute(dct_rows) + check("_dct2d_32's truncated corner matches the untruncated DCT", + all(fast[u][v] == brute[u][v] + for u in range(_DCT_KEEP) for v in range(_DCT_KEEP))) + # 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))) @@ -434,6 +534,8 @@ def selftest(): 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) + rc = main(["bench", base_path, "3"]) + check("bench runs to completion", rc == 0) if failures: print(f"\n{len(failures)} check(s) failed:") @@ -453,7 +555,7 @@ def main(argv): cmd, rest = argv[0], argv[1:] if cmd == "selftest": return selftest() - handlers = {"compare": cmd_compare, "capture": cmd_capture, "hash": cmd_hash} + handlers = {"compare": cmd_compare, "capture": cmd_capture, "hash": cmd_hash, "bench": cmd_bench} if cmd not in handlers: print(f"unknown command: {cmd!r}", file=sys.stderr) return 2 diff --git a/qemu/tests/qmp.py b/qemu/tests/qmp.py index 385c4c9..f288b22 100755 --- a/qemu/tests/qmp.py +++ b/qemu/tests/qmp.py @@ -43,11 +43,16 @@ a real panel, so these verbs mean the same thing on the rig and on hardware: --rs485-control (ui-drive.sh --rs485-devices) 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 + assert_hit X Y CLASS [box=X1,Y1,WxH] [TEXT...] + a real tap at X,Y lands on an object of CLASS, + with that exact box when one is given, whose caption contains TEXT: the pre-tap self-check that reports a MOVED button as such, not as broken behaviour + wait_hit X Y CLASS TIMEOUT_S [box=X1,Y1,WxH] [TEXT...] + assert_hit, polled every 0.5s (poll_until) until it + holds or TIMEOUT_S elapses -- for a page still + building when the check runs Two more channels reach behavioural state instead of the widget tree. `--refs FILE` (default /refs.json) is the reference store the pixel @@ -85,6 +90,9 @@ math, this file only drives it. 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). + wait_region NAME TOLERANCE TIMEOUT_S + assert_region, polled every 0.5s (poll_until) until + it holds or TIMEOUT_S elapses 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). @@ -288,14 +296,25 @@ def judge_hit(ctl, x, y, cls, rest): def poll_until(check, timeout_s, period=0.5): """Run CHECK (-> (ok, detail)) until it holds or TIMEOUT_S passes; - -> (ok, detail, seconds waited). The wait_* verbs share this so they - all mean the same thing by a timeout.""" + -> (ok, detail, seconds waited, fatal). The wait_* verbs share this so + they all mean the same thing by a timeout. + + A CHECK that raises RuntimeError or OSError -- the control channel + dying mid-poll, e.g. a socket.timeout past Ctl's own 15s per-call + timeout -- stops polling right there instead of retrying a channel + that is probably gone, and comes back with FATAL true so the caller + records a `fatal` row (the same idea as ConsoleWatch.check() turning a + console crash into a labeled row) rather than the exception escaping + drive() and silently dropping every step after it.""" start = time.monotonic() deadline = start + timeout_s while True: - ok, detail = check() + try: + ok, detail = check() + except (RuntimeError, OSError) as e: + return False, str(e), time.monotonic() - start, True if ok or time.monotonic() >= deadline: - return ok, detail, time.monotonic() - start + return ok, detail, time.monotonic() - start, False time.sleep(period) @@ -311,21 +330,52 @@ def parse_hit(reply): 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: +class ConsoleWatch: + """Incremental check for the stage-2 init's console crash marker. + + The framebuffer just keeps the last frame on a crash; only the console + announces it, and drive() checks after EVERY step so a crash is pinned + to the step that caused it, not discovered at the end. But the console + log is not a static boot log -- rootfs/sbin/init pipes warden-ui's own + stdout through `tee` into the same stream the host captures, so it grows + for the life of the run. A fresh open()+read() of the WHOLE file on + every one of those per-step checks costs O(steps * final_size) for + nothing: a marker absent from bytes already scanned cannot retroactively + appear there. This instead keeps the byte offset already scanned and + reads only what was appended since the last check, with a short tail + kept across calls so a marker split across two reads is still caught. + One instance per drive() run.""" + + MARKER = b"warden-ui EXITED" + + def __init__(self, console_path): + self.path = console_path + self.offset = 0 + self.tail = b"" + self.found = None + + def check(self): + """-> the matched line (decoded), once and cached forever after + (nothing later needs another read); None while nothing has matched + yet, or PATH is unset or unreadable.""" + if not self.path or self.found is not None: + return self.found + try: + with open(self.path, "rb") as fh: + fh.seek(self.offset) + chunk = fh.read() + self.offset = fh.tell() + except OSError: + return None + data = self.tail + chunk + i = data.find(self.MARKER) + if i >= 0: + self.found = data[i:i + 80].split(b"\n", 1)[0].decode("utf-8", "replace") + return self.found + # Keep enough of the tail that a marker whose first byte landed in + # this read, but the rest in the next one, is still caught. + self.tail = data[-(len(self.MARKER) - 1):] 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+\])*)$') @@ -409,9 +459,14 @@ def apply_op(op, actual, raw_value): 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.""" + parse it. -> (doc, None) or (None, detail) for a torn read or a + not-yet-written file -- an ordinary retry/fail for the caller, not a + crash of the whole driver. ctl.send() itself is deliberately NOT guarded + here: a dead control channel (RuntimeError on EOF, OSError/socket.timeout + on a stalled read) is left to propagate, so wait_json's poll_until (and + drive()'s own top-level guard for assert_json) can tell a dead channel + apart from a value the document simply doesn't have yet, and report the + former as a fast fatal rather than retrying it out to TIMEOUT_S.""" reply = ctl.send(f"@cat {STATUS_JSON_PATH}") if reply.startswith("bridge: no such file"): return None, reply @@ -490,6 +545,23 @@ def write_png(img, path): _PILImage.frombytes("RGB", (w, h), data).save(path) +def safe_out_path(outdir, name, prefix="", suffix=""): + """/PREFIXnameSUFFIX, or (None, detail) for a NAME that could + write outside outdir. flowc.py does not restrict shot/region/ocr names + beyond forbidding whitespace (see its own `shot` and `one_token` + comments), so a hand-edited or buggy *.drive.txt line -- or a refs.json + entry with a name capture_region itself would never have written -- must + not be allowed to place NAME ahead of outdir in the joined path. An + absolute NAME (e.g. `shot /etc/cron.d/x`) makes os.path.join() discard + outdir entirely and return NAME verbatim; a NAME with any other '/' can + still walk out of outdir with enough '..' segments. Mirrors + flow-run-hw.sh's shot_out_path() for the identical class of bug on the + hardware runner.""" + if not name or "/" in name: + return None, f"{name!r} must not contain '/' (would escape outdir)" + return os.path.join(os.path.abspath(outdir), f"{prefix}{name}{suffix}"), None + + def rs485_send(control_path, line): """One command to mbsim.py's control socket (flare-edge tools/modbus-sim, --control): -> its one-line reply, `ok ...` or `error ...`.""" @@ -500,6 +572,382 @@ def rs485_send(control_path, line): return c.makefile("r").readline().strip() +class Ctx: + """State one drive() run threads through every verb handler below, + replacing the closures drive() used to build fresh on each call (record + stays in drive() itself -- it owns the results.jsonl handle -- but + need_ctl, need_imgtools and fresh_region move here since verb handlers, + not drive(), are what call them now). One instance per run.""" + + def __init__(self, s, f, size, ctl, refs, refs_path, rs485_control, outdir, script_path): + self.s = s + self.f = f + self.size = size + self.ctl = ctl + self.refs = refs + self.refs_path = refs_path + self.rs485_control = rs485_control + self.outdir = outdir + self.script_path = script_path + + def need_ctl(self, lineno, cmd): + if self.ctl is None: + sys.exit(f"FATAL: {self.script_path}:{lineno}: '{cmd}' needs the control " + f"channel; run with --ctl (ui-drive.sh passes it)") + + def need_imgtools(self, 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: {self.script_path}:{lineno}: '{cmd}' needs imgtools.py " + f"next to qmp.py (vendor/warden-sdk/qemu/tests/imgtools.py)") + + def fresh_region(self, 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 = self.refs.get(name) + if ref is None: + return None, None, f"no reference for {name!r} (run capture_region first)" + path, err = safe_out_path(self.outdir, name, prefix=f"{tag}_", suffix=".ppm") + if err: + return None, None, err + rpc(self.s, self.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 + + +# One function per script verb, VERBS-dispatched below instead of a single +# growing if/elif chain: each takes the run's shared Ctx plus the parsed +# line and returns either None (no results.jsonl row -- only echo does +# this) or (status, detail) for drive() to record. Keeping one function per +# verb, the way flowc.py tables its own CHANNELS/ACTIONS vocabulary, is +# what lets a verb be read, tested or reused on its own instead of only as +# one arm of drive(). + +def verb_shot(ctx, lineno, cmd, args, line): + name = args[0] + path, err = safe_out_path(ctx.outdir, name, suffix=".ppm") + if err: + return "fail", f"shot name {err}" + rpc(ctx.s, ctx.f, {"execute": "screendump", "arguments": {"filename": path}}) + return "ok", "" + + +def verb_tap(ctx, lineno, cmd, args, line): + x, y = int(args[0]), int(args[1]) + do_tap(ctx.s, ctx.f, to_axis(x, ctx.size), to_axis(y, ctx.size)) + return "ok", "" + + +def verb_swipe(ctx, lineno, cmd, args, line): + ms = int(args[4]) if len(args) > 4 else 400 + do_swipe(ctx.s, ctx.f, int(args[0]), int(args[1]), int(args[2]), int(args[3]), ctx.size, ms) + return "ok", "" + + +def verb_fling(ctx, lineno, cmd, args, line): + do_swipe(ctx.s, ctx.f, int(args[0]), int(args[1]), int(args[2]), int(args[3]), + ctx.size, ms=120) + return "ok", "" + + +def verb_sleep(ctx, lineno, cmd, args, line): + time.sleep(float(args[0])) + return "ok", "" + + +def verb_echo(ctx, lineno, cmd, args, line): + print(f" [{lineno}] {' '.join(args)}", flush=True) + return None + + +def verb_nav(ctx, lineno, cmd, args, line): + ctx.need_ctl(lineno, cmd) + reply = ctx.ctl.send("nav " + " ".join(args)) + return ("ok" if reply.endswith(": ok") else "fail"), reply + + +def verb_wake(ctx, lineno, cmd, args, line): + # 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. + ctx.need_ctl(lineno, cmd) + reply = ctx.ctl.send("wake") + return ("ok" if reply == "wake: ok" else "fail"), reply + + +def verb_scroll_home(ctx, lineno, cmd, args, line): + # 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. + ctx.need_ctl(lineno, cmd) + reply = ctx.ctl.send(line) + return ("ok" if reply.startswith(f"{cmd}: ok") else "fail"), reply + + +def verb_query(ctx, lineno, cmd, args, line): + # page | hit | stats | ctl: print the reply, never judge it. `ctl` is a + # raw passthrough for anything the channel grows later. + ctx.need_ctl(lineno, cmd) + reply = ctx.ctl.send(line if cmd != "ctl" else " ".join(args)) + return "ok", reply + + +def verb_rs485(ctx, lineno, cmd, args, line): + # rs485 silence|restore ADDR: take a simulated device off the bus (it + # holds its address and answers nothing, exactly an absent unit) or put + # it back, while the guest keeps polling. This is the only runtime lever + # on the roster ui-drive.sh --rs485-devices fixed at boot, and it exists + # so a flow can prove a screen noticing an adopted device go quiet + # (flare-edge #184). No simulated bus in this run is fatal: the step + # could not check anything, and a silent skip would read as a pass. + if ctx.rs485_control is None: + return "fatal", ("no simulated RS485 bus in this run " + "(ui-drive.sh --rs485-devices)") + if len(args) != 2 or args[0] not in ("silence", "restore"): + return "fatal", "expected: rs485 silence|restore ADDR" + try: + reply = rs485_send(ctx.rs485_control, f"{args[0]} {args[1]}") + except OSError as e: + reply = f"error control socket: {e}" + return ("ok" if reply.startswith("ok") else "fail"), reply + + +def verb_assert_page(ctx, lineno, cmd, args, line): + ctx.need_ctl(lineno, cmd) + want = " ".join(args) + got = ctx.ctl.send("page") + return ("ok" if got == want else "fail"), (f"page is {got!r}" if got != want else "") + + +def verb_assert_hit(ctx, lineno, cmd, args, line): + # 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. + ctx.need_ctl(lineno, cmd) + ok, detail = judge_hit(ctx.ctl, args[0], args[1], args[2], args[3:]) + return ("ok" if ok else "fail"), detail + + +def verb_wait_hit(ctx, lineno, cmd, args, line): + # wait_hit X Y CLASS TIMEOUT_S [box=X1,Y1,WxH] [TEXT...]: assert_hit + # polled every 0.5s until it holds or TIMEOUT_S passes. A page still + # building after `nav` answers `hit` with whatever is there at that + # instant, so a check one second in raced the layout (flare-edge #175). + # The timeout is the author's bound on how long settling may take, not a + # sleep: a check that holds early returns early, and the detail says how + # long it took. + ctx.need_ctl(lineno, cmd) + x, y, cls, timeout_s = args[0], args[1], args[2], float(args[3]) + rest = args[4:] + ok, detail, waited, fatal = poll_until( + lambda: judge_hit(ctx.ctl, x, y, cls, rest), timeout_s) + if fatal: + return "fatal", detail + return ("ok" if ok else "fail"), (f"waited {waited:.1f}s" if ok else detail) + + +def verb_assert_json(ctx, lineno, cmd, args, line): + # 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. + ctx.need_ctl(lineno, cmd) + ok, detail = eval_json(ctx.ctl, args[0], args[1], args[2]) + return ("ok" if ok else "fail"), detail + + +def verb_wait_json(ctx, lineno, cmd, args, line): + # wait_json PATH OP VALUE TIMEOUT_S: eval_json polled every 0.5s + # (poll_until) 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. + ctx.need_ctl(lineno, cmd) + timeout_s = float(args[3]) + ok, detail, waited, fatal = poll_until( + lambda: eval_json(ctx.ctl, args[0], args[1], args[2]), timeout_s) + if fatal: + return "fatal", detail + return ("ok" if ok else "fail"), (f"waited {waited:.1f}s" if ok else detail) + + +def verb_assert_stat(ctx, lineno, cmd, args, line): + # assert_stat FIELD OP VALUE: FIELD off a fresh `stats` reply. + ctx.need_ctl(lineno, cmd) + field, op, value = args[0], args[1], args[2] + if field not in ("cpu", "fps", "render", "idle"): + return "fail", (f"unknown stat field {field!r} " + f"(known: cpu, fps, render, idle)") + stats = parse_stats(ctx.ctl.send("stats")) + if field not in stats: + return "fail", "no such field" + try: + ok = apply_op(op, stats[field], value) + except (TypeError, ValueError) as e: + return "fail", str(e) + return ("ok" if ok else "fail"), ("" if ok else f"{field} is {stats[field]}") + + +def verb_capture_region(ctx, lineno, cmd, args, line): + # 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. + ctx.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"): + return "fail", (f"unknown tolerance {tolerance!r} " + f"(known: exact, loose, structural)") + path, err = safe_out_path(ctx.outdir, name, prefix="capture_", suffix=".ppm") + if err: + return "fail", f"region name {err}" + rpc(ctx.s, ctx.f, {"execute": "screendump", "arguments": {"filename": path}}) + try: + cropped = imgtools.crop(imgtools.load_ppm(path), x, y, w, h) + except (ValueError, OSError) as e: + return "fatal", f"cannot capture {name!r}: {e}" + ctx.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(ctx.refs_path, ctx.refs) + return "ok", f"captured box={x},{y},{w}x{h}" + + +def verb_assert_region(ctx, lineno, cmd, args, line): + # 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. + ctx.need_imgtools(lineno, cmd) + name = args[0] + want_tol = args[1] if len(args) > 1 else None + ref = ctx.refs.get(name) + if ref is not None and want_tol and want_tol != ref.get("tolerance"): + return "fatal", (f"reference {name!r} was captured as {ref.get('tolerance')}, " + f"script expects {want_tol}: recapture or fix the spec") + ref, cropped, err = ctx.fresh_region(name, "assert") + if err: + return "fatal", err + ok, detail = imgtools.compare(ref, cropped, ref["tolerance"]) + return ("ok" if ok else "fail"), detail + + +def verb_wait_region(ctx, lineno, cmd, args, line): + # wait_region NAME TOLERANCE TIMEOUT_S: assert_region polled every 0.5s, + # a fresh screendump each time, until the crop matches or the deadline + # passes (flare-edge #175). A missing reference or a foreign tolerance + # is fatal exactly as for assert_region: waiting cannot fix either, so + # polling stops on the first such answer. + ctx.need_imgtools(lineno, cmd) + name, want_tol, timeout_s = args[0], args[1], float(args[2]) + ref = ctx.refs.get(name) + if ref is not None and want_tol != ref.get("tolerance"): + return "fatal", (f"reference {name!r} was captured as {ref.get('tolerance')}, " + f"script expects {want_tol}: recapture or fix the spec") + fault = [] + + def check_region(): + r, cropped, err = ctx.fresh_region(name, "assert") + if err: + fault.append(err) + return True, err + return imgtools.compare(r, cropped, r["tolerance"]) + + ok, detail, waited, poll_fatal = poll_until(check_region, timeout_s) + if fault: + return "fatal", fault[0] + if poll_fatal: + return "fatal", detail + return ("ok" if ok else "fail"), (f"waited {waited:.1f}s" if ok else detail) + + +def verb_assert_ocr(ctx, lineno, cmd, args, line): + # 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. + ctx.need_imgtools(lineno, cmd) + name, pattern = args[0], " ".join(args[1:]) + if shutil.which("tesseract") is None: + return "fatal", "tesseract not installed" + ref, cropped, err = ctx.fresh_region(name, "ocr") + if err: + return "fatal", err + # fresh_region() above already ran NAME through safe_out_path() and + # bailed on `err` if it could escape outdir, so it is safe to build this + # second path from NAME directly. + png_path = os.path.join(os.path.abspath(ctx.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: + return "fatal", f"ocr failed: {e}" + if re.search(pattern, text): + return "ok", "" + return "fail", f"text was {text.strip()!r}" + + +# verb NAME -> handler. Several names share one handler (scroll/home; +# page/hit/stats/ctl) the same way they shared one elif arm before; the +# handler still gets the matched name as `cmd` for messages that name it. +VERBS = { + "shot": verb_shot, + "tap": verb_tap, + "swipe": verb_swipe, + "fling": verb_fling, + "sleep": verb_sleep, + "echo": verb_echo, + "nav": verb_nav, + "wake": verb_wake, + "scroll": verb_scroll_home, + "home": verb_scroll_home, + "page": verb_query, + "hit": verb_query, + "stats": verb_query, + "ctl": verb_query, + "rs485": verb_rs485, + "assert_page": verb_assert_page, + "assert_hit": verb_assert_hit, + "wait_hit": verb_wait_hit, + "assert_json": verb_assert_json, + "wait_json": verb_wait_json, + "assert_stat": verb_assert_stat, + "capture_region": verb_capture_region, + "assert_region": verb_assert_region, + "wait_region": verb_wait_region, + "assert_ocr": verb_assert_ocr, +} + + def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, refs_path=None, rs485_control=None): os.makedirs(outdir, exist_ok=True) @@ -518,40 +966,9 @@ def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, ref 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 + ctx = Ctx(s, f, size, ctl, load_refs(refs_path), refs_path, rs485_control, outdir, script_path) + console = ConsoleWatch(console_path) for lineno, raw in enumerate(lines, 1): line = raw.split("#", 1)[0].strip() @@ -560,269 +977,28 @@ def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, ref 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 == "rs485": - # rs485 silence|restore ADDR: take a simulated device off the bus - # (it holds its address and answers nothing, exactly an absent - # unit) or put it back, while the guest keeps polling. This is - # the only runtime lever on the roster ui-drive.sh --rs485-devices - # fixed at boot, and it exists so a flow can prove a screen - # noticing an adopted device go quiet (flare-edge #184). No - # simulated bus in this run is fatal: the step could not check - # anything, and a silent skip would read as a pass. - if rs485_control is None: - record(lineno, line, "fatal", "no simulated RS485 bus in this run " - "(ui-drive.sh --rs485-devices)") - elif len(args) != 2 or args[0] not in ("silence", "restore"): - record(lineno, line, "fatal", "expected: rs485 silence|restore ADDR") - else: - try: - reply = rs485_send(rs485_control, f"{args[0]} {args[1]}") - except OSError as e: - reply = f"error control socket: {e}" - record(lineno, line, "ok" if reply.startswith("ok") else "fail", 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) - ok, detail = judge_hit(ctl, args[0], args[1], args[2], args[3:]) - record(lineno, line, "ok" if ok else "fail", detail) - elif cmd == "wait_hit": - # wait_hit X Y CLASS TIMEOUT_S [box=X1,Y1,WxH] [TEXT...]: assert_hit - # polled every 0.5s until it holds or TIMEOUT_S passes. A page - # still building after `nav` answers `hit` with whatever is there - # at that instant, so a check one second in raced the layout - # (flare-edge #175). The timeout is the author's bound on how long - # settling may take, not a sleep: a check that holds early returns - # early, and the detail says how long it took. - need_ctl(lineno, cmd) - x, y, cls, timeout_s = args[0], args[1], args[2], float(args[3]) - rest = args[4:] - ok, detail, waited = poll_until(lambda: judge_hit(ctl, x, y, cls, rest), timeout_s) - record(lineno, line, "ok" if ok else "fail", - f"waited {waited:.1f}s" if ok else detail) - 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 == "wait_region": - # wait_region NAME TOLERANCE TIMEOUT_S: assert_region polled every - # 0.5s, a fresh screendump each time, until the crop matches or - # the deadline passes (flare-edge #175). A missing reference or a - # foreign tolerance is fatal exactly as for assert_region: waiting - # cannot fix either, so polling stops on the first such answer. - need_imgtools(lineno, cmd) - name, want_tol, timeout_s = args[0], args[1], float(args[2]) - ref = refs.get(name) - if ref is not None 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: - fault = [] - - def check_region(): - r, cropped, err = fresh_region(name, "assert") - if err: - fault.append(err) - return True, err - return imgtools.compare(r, cropped, r["tolerance"]) - - ok, detail, waited = poll_until(check_region, timeout_s) - if fault: - record(lineno, line, "fatal", fault[0]) - else: - record(lineno, line, "ok" if ok else "fail", - f"waited {waited:.1f}s" if ok else 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: + handler = VERBS.get(cmd) + if handler is None: 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) + try: + result = handler(ctx, lineno, cmd, args, line) + except (RuntimeError, OSError) as e: + # poll_until already turns this into a fatal row for the wait_* + # verbs; every other verb reaches ctl.send() (nav, wake, + # scroll/home, page/hit/stats/ctl, assert_page, assert_hit, + # assert_json, assert_stat) or the QMP socket via rpc() + # (tap/swipe/fling/shot, the region verbs) with no guard of its + # own, so without this a dying channel or a gone QMP socket ends + # the whole run as an unhandled traceback instead of one fatal + # step. Recorded the same as any other fatal row: the run keeps + # going past it. + result = ("fatal", str(e)) + if result is not None: + record(lineno, line, *result) + + crashed = console.check() if crashed: record(lineno, line, "fatal", crashed) break diff --git a/qemu/tests/test_qmp_drive.py b/qemu/tests/test_qmp_drive.py index dfb3078..1736462 100755 --- a/qemu/tests/test_qmp_drive.py +++ b/qemu/tests/test_qmp_drive.py @@ -13,6 +13,7 @@ was never compared with the one the reference was captured under. """ import json import os +import socket import sys import tempfile import time @@ -45,7 +46,10 @@ class FakeCtl: if cmd == "home": return "home: ok" if cmd.startswith("scroll "): - return "scroll: ok y=300 of 900" if not cmd.endswith(" 0") else "scroll: expected X Y DY (DY non-zero pixels)" + arg = cmd[len("scroll "):] + if not cmd.endswith(" 0"): + return "scroll: ok y=300 of 900" + return "scroll: expected X Y DY (DY non-zero pixels), got '%s'" % arg if cmd.startswith("@cat "): return json.dumps({"a": {"b": 1}, "list": [1, 2], "name": "warden"}) if cmd.startswith("hit "): @@ -55,6 +59,121 @@ class FakeCtl: return "" +class DyingCtl: + """A control channel that answers once -- a page still building, not + yet a match, the exact case wait_json/wait_hit poll for -- and then + dies on every later call, the way a real Ctl.send() does on EOF + (RuntimeError) or a socket timeout past its own 15s (socket.timeout, + an OSError). Regression fixture: that used to raise straight out of + drive() and take every later step with it.""" + + def __init__(self, path, timeout=15.0): + self.calls = 0 + + def send(self, cmd): + self.calls += 1 + if self.calls == 1: + if cmd.startswith("@cat "): + return json.dumps({"a": {"b": 0}}) + if cmd.startswith("hit "): + return "hit 47,676: nothing" + return "" + if self.calls == 2: + raise RuntimeError("control channel closed") + raise socket.timeout("timed out") + + +class ImmediateDyingCtl: + """A control channel that is already dead before the first command -- + models a crash that happened between two script steps, not mid-poll. + Regression fixture for drive()'s own top-level guard around handler(): + before that guard existed, any non-wait_* verb (nav, wake, scroll/home, + page/hit/stats/ctl, assert_page, assert_hit, assert_json, assert_stat) + raised this straight out of drive() as an unhandled traceback, ending + the run instead of recording one fatal row and continuing (flare-edge + round-4 review).""" + + def __init__(self, path, timeout=15.0): + pass + + def send(self, cmd): + raise RuntimeError("control channel closed") + + +class FullscreenCtl: + """Models the dashboard's REAL fullscreen state plus warden_debug.c's + `fullscreen`/`home` verbs closely enough to drive + qemu/tests/scripts/fullscreen-toggle-tracks-real-state.txt offline: `home` + always clears real_fs directly (warden_screen_overview_set_fullscreen(false) + in the `home` handler), and `hit 47 676` answers with the gear's box when + real_fs is false or the fullscreen canvas's box when it is true, exactly + like the live dashboard's tap-catcher target (overview.targets.yaml). + + Subclasses disagree only about what `fullscreen toggle` reads to decide + its next state -- the one line of behaviour the underlying bug is about.""" + + def __init__(self, path, timeout=15.0): + self.real_fs = False + + def send(self, cmd): + if cmd == "wake": + return "wake: ok" + if cmd == "page": + return "Dashboard/Dashboard" + if cmd == "home": + self.real_fs = False + return "home: ok" + if cmd.startswith("hit "): + if self.real_fs: + return 'hit 47,676: widget text="" box=0,0,720x720' + return 'hit 47,676: obj text="" box=12,640,72x72' + if cmd in ("fullscreen on", "fullscreen off", "fullscreen toggle"): + want = self._decide(cmd) + self.real_fs = want + self._remember(want) + return f"fullscreen {'on' if want else 'off'}: ok" + return "" + + def _decide(self, cmd): + if cmd == "fullscreen on": + return True + if cmd == "fullscreen off": + return False + return self._toggle_target() + + def _remember(self, want): + pass + + def _toggle_target(self): + raise NotImplementedError + + +class FixedFullscreenCtl(FullscreenCtl): + """The fix: `toggle` reads the SAME ground truth `home` just changed + (warden_screen_overview_get_fullscreen()), so it can never disagree with + what `home` (or a real tap) already did.""" + + def _toggle_target(self): + return not self.real_fs + + +class StaleBeliefFullscreenCtl(FullscreenCtl): + """Pre-fix: `toggle` reads a belief of its own that `home` never touches + (warden_debug.c used to keep a private `static bool on`), so it can drift + from the real state and then toggle AWAY from that stale belief instead of + away from reality.""" + + def __init__(self, path, timeout=15.0): + super().__init__(path, timeout) + self.belief = False + + def _remember(self, want): + self.belief = want + + def _toggle_target(self): + return not self.belief + + def fake_rpc(sock, sock_file, obj): if obj.get("execute") == "screendump": with open(obj["arguments"]["filename"], "wb") as fh: @@ -62,7 +181,7 @@ def fake_rpc(sock, sock_file, obj): return {} -def run_script(text, refs=None, rs485_control=None): +def run_script(text, refs=None, rs485_control=None, ctl_cls=None, rpc_fn=None, console_path=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") @@ -73,12 +192,12 @@ def run_script(text, refs=None, rs485_control=None): with open(refs_path, "w") as fh: json.dump(refs, fh) saved = qmp.rpc, qmp.Ctl - qmp.rpc, qmp.Ctl = fake_rpc, FakeCtl + qmp.rpc, qmp.Ctl = (rpc_fn or fake_rpc), (ctl_cls or FakeCtl) rc = None try: try: qmp.drive(None, None, script, outdir, SIZE, ctl_path="fake", - console_path=None, refs_path=refs_path, rs485_control=rs485_control) + console_path=console_path, refs_path=refs_path, rs485_control=rs485_control) except SystemExit as e: rc = e.code finally: @@ -117,6 +236,32 @@ class PureHelpers(unittest.TestCase): got = qmp.parse_stats(FakeCtl("x").send("stats")) self.assertEqual(got, {"cpu": 12.0, "fps": 10.0, "render": 3.2, "idle": 0.0}) + def test_poll_until_turns_a_channel_fault_into_fatal_not_a_raise(self): + # A CHECK that raises RuntimeError or OSError (Ctl.send on EOF or a + # socket timeout) must stop poll_until() and come back with the + # exception text and fatal=True instead of propagating -- the wait_* + # verbs are the only thing standing between that and drive() dying + # with an unhandled traceback mid-script. + ok, detail, waited, fatal = qmp.poll_until( + lambda: (_ for _ in ()).throw(RuntimeError("control channel closed")), 5) + self.assertFalse(ok) + self.assertEqual(detail, "control channel closed") + self.assertTrue(fatal) + self.assertLess(waited, 1, "a raise must stop the poll immediately, not wait out TIMEOUT_S") + + ok, detail, waited, fatal = qmp.poll_until( + lambda: (_ for _ in ()).throw(socket.timeout("timed out")), 5) + self.assertFalse(ok) + self.assertTrue(fatal, "socket.timeout is an OSError and must be caught the same way") + + # The ordinary paths still return the same 4-tuple shape. + ok, detail, waited, fatal = qmp.poll_until(lambda: (True, ""), 5) + self.assertTrue(ok) + self.assertFalse(fatal) + ok, detail, waited, fatal = qmp.poll_until(lambda: (False, "not yet"), 0.1, period=0.05) + self.assertFalse(ok) + self.assertFalse(fatal) + class DriveVerbs(unittest.TestCase): def test_every_channel_passes_on_a_healthy_ui(self): @@ -163,6 +308,78 @@ class DriveVerbs(unittest.TestCase): self.assertGreaterEqual(time.monotonic() - t0, 2.0, "wait_json and wait_hit must each honour their timeout") + def test_dying_channel_is_fatal_for_the_step_not_a_crash(self): + # A page still building when the control channel itself drops mid- + # poll -- Ctl.send() raising RuntimeError on EOF or socket.timeout + # past its own 15s -- used to escape drive() as an unhandled + # traceback, losing the row for the step that was polling and every + # step after it. wait_json and wait_hit (both poll_until) must + # instead read as one `fatal` row, with the run continuing past it. + rc, by, rows = run_script( + "wait_json a.b eq 1 2\n" + "sleep 0\n", + ctl_cls=DyingCtl, + ) + self.assertEqual(rc, 1) + self.assertEqual([r["status"] for r in rows], ["fatal", "ok"]) + self.assertIn("control channel closed", by["wait_json a.b eq 1 2"]["detail"]) + + rc, by, rows = run_script( + "wait_hit 47 676 obj 2\n" + "sleep 0\n", + ctl_cls=DyingCtl, + ) + self.assertEqual(rc, 1) + self.assertEqual([r["status"] for r in rows], ["fatal", "ok"]) + self.assertIn("control channel closed", by["wait_hit 47 676 obj 2"]["detail"]) + + def test_dying_channel_is_fatal_for_every_verb_not_just_wait(self): + # wait_hit/wait_json route the channel through poll_until, which has + # always caught this. Every other verb that reaches ctl.send() + # directly did not, and used to crash the whole run instead of + # recording one fatal row and moving on. One representative of each + # family, back to back: every one must read as its own `fatal` row + # and the script must still reach the last line. + rc, by, rows = run_script( + "nav Demo/Rows\n" + "wake\n" + "home\n" + "page\n" + "assert_page Demo/Rows\n" + "assert_hit 47 676 obj\n" + "assert_json a.b eq 1\n" + "assert_stat idle eq 0\n" + "sleep 0\n", + ctl_cls=ImmediateDyingCtl, + ) + self.assertEqual(rc, 1) + self.assertEqual([r["status"] for r in rows], + ["fatal"] * 8 + ["ok"], + "a dead channel must not crash the run: every ctl " + "verb gets its own fatal row and sleep still runs") + for row in rows[:-1]: + self.assertIn("control channel closed", row["detail"]) + + def test_dying_qmp_socket_is_fatal_not_a_crash(self): + # tap/swipe/fling/shot (and the region verbs) reach the QMP socket + # through rpc(), a separate channel from ctl.send(), guarded by the + # same top-level try/except in drive(). A RuntimeError from a QMP + # error reply, or an OSError from the socket itself going away, must + # read as a fatal row per step, not crash the run. + def dying_rpc(sock, sock_file, obj): + raise OSError("QMP socket closed") + + rc, by, rows = run_script( + "shot before\n" + "tap 10 10\n" + "sleep 0\n", + rpc_fn=dying_rpc, + ) + self.assertEqual(rc, 1) + self.assertEqual([r["status"] for r in rows], ["fatal", "fatal", "ok"]) + for row in rows[:-1]: + self.assertIn("QMP socket closed", row["detail"]) + 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, @@ -210,6 +427,48 @@ class DriveVerbs(unittest.TestCase): self.assertEqual(rows[0]["status"], "fatal", "no bus in the run must not look like a pass") self.assertIn("no simulated RS485 bus", rows[0]["detail"]) + def test_shot_and_region_names_cannot_escape_outdir(self): + # flowc.py does not restrict shot/region names beyond forbidding + # whitespace (see its own `shot` and `one_token` comments), so an + # absolute or '../'-relative name must fail the step closed instead + # of ever reaching the screendump RPC -- an absolute name used to + # make os.path.join() discard outdir entirely and write there + # verbatim (flare-edge security review). + calls = [] + + def counting_rpc(sock, sock_file, obj): + calls.append(obj) + return fake_rpc(sock, sock_file, obj) + + rc, by, rows = run_script( + "shot /etc/cron.d/evil\n" + "shot ../../../tmp/evil\n" + "capture_region ../evil 0 0 8 8 exact\n", + rpc_fn=counting_rpc, + ) + self.assertEqual(rc, 1) + self.assertEqual([r["status"] for r in rows], ["fail", "fail", "fail"]) + for row in rows: + self.assertIn("must not contain '/'", row["detail"]) + self.assertEqual(calls, [], "a malformed name must never reach a screendump RPC") + + # The ordinary case -- a plain name -- must still work. + rc, by, rows = run_script("shot 01-overview\n", rpc_fn=counting_rpc) + self.assertIsNone(rc) + self.assertEqual(rows[0]["status"], "ok") + + def test_assert_region_refuses_a_refs_entry_whose_name_could_escape_outdir(self): + # capture_region can no longer write such a name (previous test), + # but refs.json is a hand-editable file on disk; assert_region (and + # wait_region, same fresh_region() call) must refuse a bad name from + # there too, not just at capture time. + refs = {"../evil": {"x": 0, "y": 0, "w": 8, "h": 8, "tolerance": "exact", + "phash": "0" * 16, "structural": "00" * 32}} + rc, by, rows = run_script("assert_region ../evil exact\n", refs=refs) + self.assertEqual(rc, 1) + self.assertEqual(rows[0]["status"], "fatal") + self.assertIn("must not contain '/'", rows[0]["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 @@ -220,5 +479,102 @@ class DriveVerbs(unittest.TestCase): self.assertEqual([r["status"] for r in rows], ["ok"]) +class ConsoleWatchTests(unittest.TestCase): + """ConsoleWatch replaced a ui_exited() that reread and rescanned the + WHOLE console log on every one of drive()'s per-step checks -- O(steps * + final_size) for a file that only grows over a run. These pin the two + properties a byte-offset-tracking rewrite must not lose: a marker split + across two reads is still caught, and a match is cached (never rereads + again).""" + + def test_finds_a_marker_split_across_two_reads_and_caches_it(self): + d = tempfile.mkdtemp(prefix="console.") + path = os.path.join(d, "console.log") + with open(path, "wb") as fh: + fh.write(b"boot log line one\nboot log line two\nwarden-ui EXI") + watch = qmp.ConsoleWatch(path) + self.assertIsNone(watch.check(), "no full marker written yet") + + with open(path, "ab") as fh: + fh.write(b"TED at pc=0x1234\n") + got = watch.check() + self.assertIsNotNone(got, "the marker's second half arrived in this read") + self.assertTrue(got.startswith("warden-ui EXITED"), got) + + # Cached: removing the file out from under a later check must not + # un-find the marker or raise. + os.remove(path) + self.assertEqual(watch.check(), got) + + def test_no_console_path_is_always_none(self): + watch = qmp.ConsoleWatch(None) + self.assertIsNone(watch.check()) + self.assertIsNone(watch.check()) + + def test_missing_file_returns_none_without_raising(self): + watch = qmp.ConsoleWatch(os.path.join(tempfile.mkdtemp(), "does-not-exist.log")) + self.assertIsNone(watch.check()) + + +class DriveConsoleCheck(unittest.TestCase): + def test_a_console_crash_is_a_fatal_row_and_stops_the_run(self): + # End-to-end through drive(), not just ConsoleWatch on its own: the + # stage-2 init's marker on the console must still stop the script at + # the step that caused it (module docstring), the same contract + # ui_exited() made before this became an incremental read. + d = tempfile.mkdtemp(prefix="console.") + console_path = os.path.join(d, "console.log") + with open(console_path, "wb") as fh: + fh.write(b"warden-ui EXITED signal=11\n") + rc, by, rows = run_script("wake\nsleep 0\n", console_path=console_path) + self.assertEqual(rc, 1) + # drive() records the step's own result (wake's "ok") and then a + # second row for the crash it finds right after -- both pinned to + # line 1 -- and never reaches the `sleep 0` on line 2. + self.assertEqual(len(rows), 2, "drive() must stop at the crashing step") + self.assertEqual(rows[-1]["status"], "fatal") + self.assertIn("warden-ui EXITED", rows[-1]["detail"]) + self.assertTrue(all(r["line"] == 1 for r in rows)) + + +class FullscreenToggleTracksRealState(unittest.TestCase): + """qemu/tests/scripts/fullscreen-toggle-tracks-real-state.txt, offline: + `home` clears the dashboard's real fullscreen state directly, and a + `fullscreen toggle` right after it must read THAT, not a belief `home` + bypassed (warden_debug.c's `fullscreen` verb). Run twice against the same + script to pin both sides: the fix passes it, the pre-fix shape fails on + exactly the step the bug breaks.""" + + SCRIPT = ( + "wake\n" + "assert_page Dashboard/Dashboard\n" + "assert_hit 47 676 obj box=12,640,72x72\n" + "ctl fullscreen on\n" + "assert_hit 47 676 widget box=0,0,720x720\n" + "home\n" + "assert_hit 47 676 obj box=12,640,72x72\n" + "ctl fullscreen toggle\n" + "assert_hit 47 676 widget box=0,0,720x720\n" + ) + + def test_fixed_channel_passes_every_step(self): + rc, by, rows = run_script(self.SCRIPT, ctl_cls=FixedFullscreenCtl) + self.assertIsNone(rc, [r for r in rows if r["status"] != "ok"]) + self.assertEqual(len(rows), 9) + self.assertTrue(all(r["status"] == "ok" for r in rows)) + + def test_stale_belief_channel_fails_the_toggle_after_home(self): + rc, by, rows = run_script(self.SCRIPT, ctl_cls=StaleBeliefFullscreenCtl) + self.assertEqual(rc, 1) + # Every step up to and including the post-home reset still passes -- + # the drift is invisible until the toggle right after it. + self.assertEqual([r["status"] for r in rows[:7]], ["ok"] * 7) + self.assertEqual(rows[7]["cmd"], "ctl fullscreen toggle") + self.assertEqual(rows[8]["status"], "fail", + "toggle read the stale belief (still 'on' from the earlier " + "explicit call) and flipped away from it, landing back on " + "the already-off real state instead of turning it on") + + if __name__ == "__main__": unittest.main(verbosity=1) diff --git a/qemu/tests/ui-drive.sh b/qemu/tests/ui-drive.sh index 067bc06..1ac4d04 100755 --- a/qemu/tests/ui-drive.sh +++ b/qemu/tests/ui-drive.sh @@ -47,7 +47,7 @@ while [ $# -gt 0 ]; do --refs) REFS_FILE="${2:?--refs needs a path}"; shift 2 ;; --rs485-devices) RS485_DEVICES="${2:?--rs485-devices needs ADDR:SLUG[,...]}"; shift 2 ;; --) shift; ARGS+=("$@"); break ;; - -*) echo "FATAL: unknown option '$1' (usage: $0 [--seed FILE] [--refs FILE]