blueflare/bfe-core1106-sdk
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
eb16a301c046ad062e2e4a313ba602b65bc952c8
bfe-core1106-sdk/qemu/tests/qmp.py
T
NoahandClaude Fable 5.1 eb16a301c0 qemu: wake verb, idle stat, and no fbcon cursor in the framebuffer 
- tests/qmp.py: `wake` is a judged verb (the reply must be the channel's
  own ack) and `assert_stat` knows `idle` (0 awake, 1 dimmed, 2 asleep).
  warden-ui's sleep.c swallows the touch that wakes a dimmed or blanked
  panel, so every compiled flow script now opens with wake (flare-edge
  #148); tests cover both.
- run.sh: vt.global_cursor_default=0. fbcon shares the virtio-gpu
  framebuffer with warden-ui and its cursor blinked an 8x2 block onto the
  top-left corner in some boots and not others, which made a structural
  pixel reference fail by one cell (#18). The panel has no fbcon on its
  display, so this only makes the rig match it.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013aHKWzT5EF86RFKRMtAv9n
2026-09-07 23:39:07 -06:00

750 lines
32 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""Tiny QMP client for the qemu/ device-sim tests.
qmp.py <socket> screendump <out.ppm>
qmp.py <socket> tap <x> <y> # absolute 0..32767 (virtio-tablet)
qmp.py <socket> quit
qmp.py <socket> drive <script> <outdir> [--size N]
`drive` runs a whole interaction on ONE connection, so the VM boots once and
the scenario costs a couple of seconds per step instead of a full boot. Its
script is one command per line, '#' comments and blank lines ignored, and
coordinates are PANEL PIXELS (0..size-1, default 720) rather than the
0..32767 absolute axis the raw `tap` takes -- the point of the mode is to write
"tap the gear at 47,676" straight off a screenshot.
shot NAME screendump to <outdir>/NAME.ppm
tap X Y press, hold past several LVGL poll periods, release
swipe X1 Y1 X2 Y2 [MS] drag, interpolated so LVGL sees real motion
fling X1 Y1 X2 Y2 swipe fast enough to leave momentum behind
sleep SECONDS let animations settle or timers fire
echo TEXT progress marker in the driver's own output
With `--ctl SOCK` (the unix socket run.sh --ctl creates; ui-drive.sh passes
it) the script can also ask the UI itself, through the debug channel in
warden_debug.c. The same channel is what tools/warden-ctl reaches over SSH on
a real panel, so these verbs mean the same thing on the rig and on hardware:
nav MENU[/TAB] jump to a page; fails if the page is unknown
wake wake the UI as if touched, minus the swallow
(sleep.c dims and then blanks an idle panel and
eats the touch that ends either); every compiled
flow script opens with it
page | stats | hit X Y print the reply, judge nothing
ctl WORDS... raw passthrough for anything the channel grows
assert_page MENU/TAB the active page is exactly this
assert_hit X Y CLASS [TEXT...]
a real tap at X,Y lands on an object of CLASS whose
caption contains TEXT: the pre-tap self-check that
reports a MOVED button as such, not as broken
behaviour
Two more channels reach behavioural state instead of the widget tree.
`--refs FILE` (default <outdir>/refs.json) is the reference store the pixel
ones read and write; imgtools.py in this directory does the actual image
math, this file only drives it.
assert_json PATH OP VALUE
wait_json PATH OP VALUE TIMEOUT_S
PATH is dotted with optional [N] indexes (e.g.
network.scan.ranges[0]) into the guest's
/tmp/warden-web-status.json, fetched fresh on
every check with `@cat` over the SAME control
channel as nav/hit (see the bridge added to
rootfs/sbin/init). OP is one of eq, ne, contains,
len_eq, len_ge, gt, lt. `wait_json` polls every
0.5s until OP holds or TIMEOUT_S elapses and
`assert_json` reads once. A PATH the document
doesn't have is a `fail`, not a crash -- that
document is a live snapshot and the key might
simply not be built yet.
assert_stat FIELD OP VALUE
FIELD is cpu, fps, render or idle (0 awake,
1 dimmed, 2 asleep), read off a fresh `stats`
reply (warden_debug.c). Same OP vocabulary as
assert_json.
capture_region NAME X Y W H TOLERANCE
screendump now, crop to X,Y,WxH, and (over)write
NAME in the refs file with both a phash and a
structural hash plus TOLERANCE (exact, loose or
structural). Run this once, by hand, to author a
reference -- the same way targets are authored,
see flows/README.md.
assert_region NAME [TOLERANCE]
fresh screendump, crop to NAME's stored box,
compare against its stored hash at its stored
tolerance. TOLERANCE, when given, must be the one
NAME was captured under (flowc.py always emits it).
assert_ocr NAME REGEX same crop, run through `tesseract`, REGEX searched
against the extracted text (spaces in REGEX are
fine, same as assert_hit's TEXT).
assert_region/assert_ocr against a NAME with no captured reference, and
assert_ocr with no `tesseract` on PATH, are FATAL for that one step: a
silently-skipped check reads as coverage that was never actually there. So
are a TOLERANCE that differs from the captured one and a stored box that does
not fit the screendump: both are the reference's fault, not the UI's, and one
bad reference must not take the rest of the script with it.
Every step is recorded to <outdir>/results.jsonl as ok / fail / fatal. An
assertion mismatch is a `fail` and the script continues, so one run reports
every broken expectation. With `--console LOG` the console is checked after
EVERY step for the stage-2 init's "warden-ui EXITED" line, so a crash is a
`fatal` pinned to the step that caused it rather than a stale picture found at
the end. The exit status is non-zero if anything failed.
A step that names an unknown command is a FATAL error rather than a skip: a
silently-ignored line in a scenario is a test that proves nothing.
"""
import json
import os
import re
import shutil
import socket
import subprocess
import sys
import time
# imgtools.py lives beside this file and does the actual pixel math (phash,
# structural hash, crop, compare) for the region/ocr verbs. Imported at
# module load but never let a missing/broken imgtools take down the verbs
# that don't need it: nav/tap/page/... must keep working while it is being
# written, so the failure is deferred to need_imgtools() at first use.
try:
import imgtools
except ImportError:
imgtools = None
# Pillow, likewise, is only needed to hand tesseract an image file (it has no
# raw-RGB stdin mode); a host without it still runs every other verb.
try:
from PIL import Image as _PILImage
except ImportError:
_PILImage = None
AXIS_MAX = 32767
# Where webstatus.c (ui-src/src/warden/webstatus.c) publishes its atomic
# snapshot inside the guest. assert_json/wait_json `@cat` this path over the
# control channel; see the bridge in rootfs/sbin/init.
STATUS_JSON_PATH = "/tmp/warden-web-status.json"
def rpc(sock, sock_file, obj):
sock.sendall((json.dumps(obj) + "\n").encode())
while True:
line = sock_file.readline()
if not line:
raise RuntimeError("QMP connection closed")
msg = json.loads(line)
if "return" in msg:
return msg["return"]
if "error" in msg:
raise RuntimeError(f"QMP error: {msg['error']}")
# asynchronous events are interleaved; skip them
def send_events(s, f, events):
rpc(s, f, {"execute": "input-send-event", "arguments": {"events": events}})
def abs_ev(axis, value):
return {"type": "abs", "data": {"axis": axis, "value": value}}
def btn_ev(down):
return {"type": "btn", "data": {"down": down, "button": "left"}}
def to_axis(px, size):
"""Panel pixels -> the tablet's absolute axis, clamped to the panel."""
px = max(0, min(size - 1, int(px)))
return int(px * AXIS_MAX / (size - 1))
def do_tap(s, f, ax, ay, hold=0.2):
send_events(s, f, [abs_ev("x", ax), abs_ev("y", ay), btn_ev(True)])
# Hold the press across several LVGL indev poll periods (33 ms each): an
# instantaneous press+release lands inside one poll and no click is ever
# registered.
time.sleep(hold)
send_events(s, f, [btn_ev(False)])
def do_swipe(s, f, x1, y1, x2, y2, size, ms=400, steps=None):
"""Drag with interpolated motion.
LVGL decides a gesture from the movement BETWEEN indev polls, so a press at
the start and a release at the end (with nothing in between) reads as a
click on whatever was under the finger, not a scroll. The interpolation
below is the whole reason a scroll can be tested at all; the step count is
derived from the duration so the poll period always sees a few pixels of
travel.
"""
ax1, ay1 = to_axis(x1, size), to_axis(y1, size)
ax2, ay2 = to_axis(x2, size), to_axis(y2, size)
if steps is None:
steps = max(6, int(ms / 25))
send_events(s, f, [abs_ev("x", ax1), abs_ev("y", ay1), btn_ev(True)])
time.sleep(0.05)
for i in range(1, steps + 1):
t = i / steps
send_events(s, f, [
abs_ev("x", int(ax1 + (ax2 - ax1) * t)),
abs_ev("y", int(ay1 + (ay2 - ay1) * t)),
])
time.sleep(ms / 1000.0 / steps)
send_events(s, f, [btn_ev(False)])
class Ctl:
"""The UI's debug channel, reached through run.sh --ctl.
Inside the VM that channel is a FIFO polled by warden-ui (warden_debug.c);
rootfs/sbin/init bridges it to a pci-serial port that run.sh exposes as a
unix socket. One command line in, the reply out, then a sentinel line. On
real hardware flare-edge tools/warden-ctl reaches the identical FIFO over
SSH, so the vocabulary here is the vocabulary there: a flow script that
runs against this rig runs against a panel.
A missing or silent channel is an infrastructure fault and raises; it is
never reported as an assertion failure, because a rig that cannot ask is
not evidence about the UI either way.
"""
SENTINEL = "<<END>>"
def __init__(self, path, timeout=15.0):
self.sock = socket.socket(socket.AF_UNIX)
self.sock.settimeout(timeout)
self.sock.connect(path)
self.buf = b""
def send(self, cmd):
self.sock.sendall((cmd + "\n").encode())
lines = []
while True:
nl = self.buf.find(b"\n")
if nl < 0:
chunk = self.sock.recv(4096)
if not chunk:
raise RuntimeError("control channel closed")
self.buf += chunk
continue
line = self.buf[:nl].decode("utf-8", "replace").rstrip("\r")
self.buf = self.buf[nl + 1:]
if line == self.SENTINEL:
break
# The tty is in cooked mode, so the command comes back echoed.
if line.strip() == cmd.strip():
continue
lines.append(line)
return "\n".join(lines).strip()
HIT_RE = re.compile(r'^hit \d+,\d+: (\S+)(?: text="(.*)" box=(-?\d+),(-?\d+),(\d+)x(\d+))?$')
def parse_hit(reply):
"""-> (cls, text, box) with box as 'x1,y1,WxH', or None for 'nothing';
raises on an unparseable reply."""
m = HIT_RE.match(reply)
if not m:
raise RuntimeError(f"unparseable hit reply: {reply!r}")
if m.group(1) == "nothing":
return None
box = f"{m.group(3)},{m.group(4)},{m.group(5)}x{m.group(6)}" if m.group(3) else ""
return m.group(1), m.group(2) or "", box
def ui_exited(console_path):
"""The stage-2 init announces a dead UI on the console; the framebuffer
does not, it just keeps the last frame. Checked after EVERY step so a crash
is pinned to the step that caused it, not discovered at the end."""
if not console_path:
return None
try:
with open(console_path, "rb") as fh:
data = fh.read()
except OSError:
return None
i = data.find(b"warden-ui EXITED")
if i < 0:
return None
return data[i:i + 80].split(b"\n", 1)[0].decode("utf-8", "replace")
PATH_TOKEN_RE = re.compile(r'^([^\[\]]*)((?:\[\d+\])*)$')
IDX_RE = re.compile(r'\[(\d+)\]')
def resolve_path(doc, path):
"""Walk a dotted PATH (keys and [N] indexes, e.g. 'a.b[0].c') over an
already-parsed JSON document. -> (value, None) or (None, 'no such path').
Pure and I/O-free on purpose (contract: testable on its own) -- fetching
the document is a separate step (fetch_status_json) so this function can
be unit-tested against a plain dict/list literal with nothing running.
"""
cur = doc
for part in path.split("."):
m = PATH_TOKEN_RE.match(part)
if not m:
return None, "no such path"
key, idxs = m.group(1), m.group(2)
if key:
if not isinstance(cur, dict) or key not in cur:
return None, "no such path"
cur = cur[key]
for idx_s in IDX_RE.findall(idxs):
idx = int(idx_s)
if not isinstance(cur, list) or idx >= len(cur) or idx < 0:
return None, "no such path"
cur = cur[idx]
return cur, None
def resolve_token(raw):
"""A verb's literal OP argument -> a comparable Python value. A JSON
literal (a number, true/false/null, or a quoted string) parses as
itself; anything else -- most values in practice, e.g. `connected` --
compares as the raw string. This is what lets both
`assert_json state.net eq connected` and `assert_json state.count eq 3`
work from one unquoted token, with no escaping convention of its own."""
try:
return json.loads(raw)
except ValueError:
return raw
def _len_of(v):
try:
return len(v)
except TypeError:
raise TypeError(f"{v!r} has no length") from None
def apply_op(op, actual, raw_value):
"""The comparison vocabulary shared by assert_json/wait_json/assert_stat:
eq, ne, contains, len_eq, len_ge, gt, lt. Raises on a combination that
cannot be judged (contains on a number, gt across incompatible types)
so the caller turns that into a `fail` detail instead of a driver crash --
a spec that asks a nonsensical question should be reported, not silently
True or False."""
value = resolve_token(raw_value)
if op == "eq":
return actual == value
if op == "ne":
return actual != value
if op == "contains":
if isinstance(actual, str):
return str(value) in actual
if isinstance(actual, (list, tuple, dict)):
return value in actual
raise TypeError(f"contains: {actual!r} is not a string or list")
if op == "len_eq":
return _len_of(actual) == int(value)
if op == "len_ge":
return _len_of(actual) >= int(value)
if op == "gt":
return actual > value
if op == "lt":
return actual < value
raise ValueError(f"unknown op {op!r}")
def fetch_status_json(ctl):
"""`@cat` the guest's webstatus.c snapshot over the control channel and
parse it. -> (doc, None) or (None, detail); never raises, so a torn read
or a not-yet-written file is an ordinary retry/fail for the caller, not a
crash of the whole driver."""
reply = ctl.send(f"@cat {STATUS_JSON_PATH}")
if reply.startswith("bridge: no such file"):
return None, reply
try:
return json.loads(reply), None
except ValueError as e:
return None, f"status json unparsable: {e}"
def eval_json(ctl, path, op, value):
"""One fetch + path-resolve + op-apply round for assert_json/wait_json.
-> (ok, detail); detail is empty on success, otherwise the reason."""
doc, err = fetch_status_json(ctl)
if err is not None:
return False, err
actual, perr = resolve_path(doc, path)
if perr is not None:
return False, perr
try:
ok = apply_op(op, actual, value)
except (TypeError, ValueError) as e:
return False, f"{op} {value!r} vs {actual!r}: {e}"
return ok, "" if ok else f"{path} is {actual!r}"
STATS_FIELD_RE = {
"cpu": re.compile(r'^cpu:\s*(-?\d+(?:\.\d+)?)%?\s*$'),
"fps": re.compile(r'^fps:\s*(-?\d+(?:\.\d+)?)\s*$'),
"render": re.compile(r'^render:\s*(-?\d+(?:\.\d+)?)\s*ms/frame\s*$'),
"idle": re.compile(r'^idle:\s*(\d+)\s*$'),
}
def parse_stats(reply):
"""The `stats` reply (warden_debug.c: 'page: X\\ncpu: N%\\nfps: N\\n
render: A.BB ms/frame\\nrga: N%\\nidle: N') -> {'cpu'|'fps'|'render'|'idle': float} for
whichever lines are present. A field the reply lacks is simply absent
from the result -- the caller reports that as 'no such field', the same
shape as resolve_path's 'no such path' for the json channel."""
out = {}
for ln in reply.splitlines():
ln = ln.strip()
for field, rx in STATS_FIELD_RE.items():
m = rx.match(ln)
if m:
out[field] = float(m.group(1))
return out
def load_refs(path):
"""The region/ocr reference store (see qmp.py's own docstring for the
schema). Missing file -> {}, same as an empty COVERAGE.yaml: capture_region
is how one gets created, not a prerequisite to have one already."""
if not path or not os.path.exists(path):
return {}
with open(path) as fh:
return json.load(fh)
def save_refs(path, refs):
# Written after every capture_region, not batched at the end: a script
# that dies three steps later must not lose a reference that was already
# good.
with open(path, "w") as fh:
json.dump(refs, fh, indent=2, sort_keys=True)
fh.write("\n")
def write_png(img, path):
"""imgtools' (w, h, rgb_bytes) tuple -> a PNG file, because tesseract has
no raw-RGB input mode. Only assert_ocr needs this; assert_region and
capture_region work on imgtools' own tuples end to end."""
if _PILImage is None:
raise RuntimeError("Pillow (PIL) is not installed")
w, h, data = img
_PILImage.frombytes("RGB", (w, h), data).save(path)
def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, refs_path=None):
os.makedirs(outdir, exist_ok=True)
with open(script_path) as fh:
lines = fh.readlines()
ctl = Ctl(ctl_path) if ctl_path else None
results = open(os.path.join(outdir, "results.jsonl"), "w")
counts = {"ok": 0, "fail": 0, "fatal": 0}
def record(lineno, cmd, status, detail=""):
counts[status] += 1
results.write(json.dumps({"line": lineno, "cmd": cmd, "status": status,
"detail": detail}) + "\n")
results.flush()
tag = {"ok": "", "fail": "FAIL ", "fatal": "FATAL "}[status]
print(f" [{lineno}] {tag}{cmd}{(' -- ' + detail) if detail else ''}", flush=True)
def need_ctl(lineno, cmd):
if ctl is None:
sys.exit(f"FATAL: {script_path}:{lineno}: '{cmd}' needs the control "
f"channel; run with --ctl (ui-drive.sh passes it)")
def need_imgtools(lineno, cmd):
# A whole-script infrastructure gap (imgtools.py absent), same class
# as a missing --ctl: no region/ocr verb can do anything without it,
# so this exits the process rather than recording a per-step fail.
if imgtools is None:
sys.exit(f"FATAL: {script_path}:{lineno}: '{cmd}' needs imgtools.py "
f"next to qmp.py (vendor/warden-sdk/qemu/tests/imgtools.py)")
refs_path = refs_path or os.path.join(outdir, "refs.json")
refs = load_refs(refs_path)
def fresh_region(name, tag):
"""screendump NOW (never reuse an earlier `shot`) and crop to NAME's
stored box. Fails closed on an uncaptured NAME: assert_region and
assert_ocr both need this, capture_region does not."""
ref = refs.get(name)
if ref is None:
return None, None, f"no reference for {name!r} (run capture_region first)"
path = os.path.join(os.path.abspath(outdir), f"{tag}_{name}.ppm")
rpc(s, f, {"execute": "screendump", "arguments": {"filename": path}})
try:
img = imgtools.load_ppm(path)
cropped = imgtools.crop(img, ref["x"], ref["y"], ref["w"], ref["h"])
except (ValueError, OSError, KeyError, TypeError) as e:
# A box outside this screendump (captured at another --size, or
# mistyped in the store) is a per-step fatal, never a traceback
# that ends the run: every later step still gets judged.
return None, None, f"reference {name!r} unusable: {e}"
return ref, cropped, None
for lineno, raw in enumerate(lines, 1):
line = raw.split("#", 1)[0].strip()
if not line:
continue
parts = line.split()
cmd, args = parts[0], parts[1:]
if cmd == "shot":
name = args[0]
path = os.path.join(os.path.abspath(outdir), f"{name}.ppm")
rpc(s, f, {"execute": "screendump", "arguments": {"filename": path}})
record(lineno, line, "ok")
elif cmd == "tap":
x, y = int(args[0]), int(args[1])
do_tap(s, f, to_axis(x, size), to_axis(y, size))
record(lineno, line, "ok")
elif cmd == "swipe":
ms = int(args[4]) if len(args) > 4 else 400
do_swipe(s, f, int(args[0]), int(args[1]), int(args[2]), int(args[3]), size, ms)
record(lineno, line, "ok")
elif cmd == "fling":
do_swipe(s, f, int(args[0]), int(args[1]), int(args[2]), int(args[3]),
size, ms=120)
record(lineno, line, "ok")
elif cmd == "sleep":
time.sleep(float(args[0]))
record(lineno, line, "ok")
elif cmd == "echo":
print(f" [{lineno}] {' '.join(args)}", flush=True)
elif cmd == "nav":
need_ctl(lineno, cmd)
reply = ctl.send("nav " + " ".join(args))
record(lineno, line, "ok" if reply.endswith(": ok") else "fail", reply)
elif cmd == "wake":
# Judged, unlike the query verbs: a UI that does not answer the
# wake is one whose next tap may be swallowed, and that must not
# read as a passing step.
need_ctl(lineno, cmd)
reply = ctl.send("wake")
record(lineno, line, "ok" if reply == "wake: ok" else "fail", reply)
elif cmd in ("page", "hit", "stats", "ctl"):
# Query verbs: print the reply, never judge it. `ctl` is a raw
# passthrough for anything the channel grows later.
need_ctl(lineno, cmd)
reply = ctl.send(line if cmd != "ctl" else " ".join(args))
record(lineno, line, "ok", reply)
elif cmd == "assert_page":
need_ctl(lineno, cmd)
want = " ".join(args)
got = ctl.send("page")
record(lineno, line, "ok" if got == want else "fail",
f"page is {got!r}" if got != want else "")
elif cmd == "assert_hit":
# assert_hit X Y CLASS [box=X1,Y1,WxH] [TEXT...]: a real tap at X,Y
# would land on an object of CLASS, with that exact bounding box if
# one is given, whose caption contains TEXT. This is the pre-tap
# self-check: "the button moved" fails HERE, by name, so it is
# never mistaken for the behaviour behind the button.
#
# The box is the strong identity. Two rows of a list share a class,
# and an icon's caption is a glyph nobody wants in a spec; the
# geometry the UI itself reports is what tells one instance from
# another.
need_ctl(lineno, cmd)
x, y, cls = args[0], args[1], args[2]
rest = args[3:]
want_box = None
if rest and rest[0].startswith("box="):
want_box = rest[0][4:]
rest = rest[1:]
want_text = " ".join(rest)
got = parse_hit(ctl.send(f"hit {x} {y}"))
if got is None:
record(lineno, line, "fail", "nothing clickable there")
elif got[0] != cls:
record(lineno, line, "fail", f"{got[0]} text={got[1]!r} box={got[2]}")
elif want_box and got[2] != want_box:
record(lineno, line, "fail", f"{cls} moved: box={got[2]}")
elif want_text and want_text not in got[1]:
record(lineno, line, "fail", f"{cls} text={got[1]!r}")
else:
record(lineno, line, "ok")
elif cmd == "assert_json":
# assert_json PATH OP VALUE: one read of the guest's status JSON
# (see eval_json). PATH not (yet) in the doc is a `fail`, since
# the doc is a live snapshot, not a schema.
need_ctl(lineno, cmd)
ok, detail = eval_json(ctl, args[0], args[1], args[2])
record(lineno, line, "ok" if ok else "fail", detail)
elif cmd == "wait_json":
# wait_json PATH OP VALUE TIMEOUT_S: poll eval_json every 0.5s
# until it holds or the deadline passes. The status file is
# written by a 2s guest timer (webstatus.c), so this exists
# because a value the doc will reach shortly is not the same
# fact as a value it never reaches -- assert_json alone cannot
# tell those apart.
need_ctl(lineno, cmd)
timeout_s = float(args[3])
start = time.monotonic()
deadline = start + timeout_s
while True:
ok, detail = eval_json(ctl, args[0], args[1], args[2])
if ok or time.monotonic() >= deadline:
break
time.sleep(0.5)
waited = time.monotonic() - start
record(lineno, line, "ok" if ok else "fail",
f"waited {waited:.1f}s" if ok else detail)
elif cmd == "assert_stat":
# assert_stat FIELD OP VALUE: FIELD off a fresh `stats` reply.
need_ctl(lineno, cmd)
field, op, value = args[0], args[1], args[2]
if field not in ("cpu", "fps", "render", "idle"):
record(lineno, line, "fail", f"unknown stat field {field!r} "
f"(known: cpu, fps, render, idle)")
else:
stats = parse_stats(ctl.send("stats"))
if field not in stats:
record(lineno, line, "fail", "no such field")
else:
try:
ok = apply_op(op, stats[field], value)
except (TypeError, ValueError) as e:
record(lineno, line, "fail", str(e))
else:
record(lineno, line, "ok" if ok else "fail",
"" if ok else f"{field} is {stats[field]}")
elif cmd == "capture_region":
# capture_region NAME X Y W H TOLERANCE: (re)writes NAME in the
# refs file from a fresh screendump. Both hashes are stored
# regardless of TOLERANCE so a later spec edit can change
# tolerance class without a recapture.
need_imgtools(lineno, cmd)
name = args[0]
x, y, w, h = int(args[1]), int(args[2]), int(args[3]), int(args[4])
tolerance = args[5]
if tolerance not in ("exact", "loose", "structural"):
record(lineno, line, "fail", f"unknown tolerance {tolerance!r} "
f"(known: exact, loose, structural)")
else:
path = os.path.join(os.path.abspath(outdir), f"capture_{name}.ppm")
rpc(s, f, {"execute": "screendump", "arguments": {"filename": path}})
try:
cropped = imgtools.crop(imgtools.load_ppm(path), x, y, w, h)
except (ValueError, OSError) as e:
record(lineno, line, "fatal", f"cannot capture {name!r}: {e}")
else:
refs[name] = {
"x": x, "y": y, "w": w, "h": h, "tolerance": tolerance,
"phash": f"{imgtools.phash(cropped):016x}",
"structural": imgtools.structural(cropped).hex(),
}
save_refs(refs_path, refs)
record(lineno, line, "ok", f"captured box={x},{y},{w}x{h}")
elif cmd == "assert_region":
# assert_region NAME [TOLERANCE]: fresh screendump, crop to
# NAME's stored box, compare at NAME's stored tolerance. A
# missing NAME is FATAL for this step: an uncaptured reference
# is a spec/authoring gap, not a UI defect the run should merely
# `fail` on. So is a TOLERANCE other than the one NAME was
# captured under: the script and the reference would be two
# claims about the same pixels, and judging by either alone
# would hide that.
need_imgtools(lineno, cmd)
name = args[0]
want_tol = args[1] if len(args) > 1 else None
ref = refs.get(name)
if ref is not None and want_tol and want_tol != ref.get("tolerance"):
record(lineno, line, "fatal",
f"reference {name!r} was captured as {ref.get('tolerance')}, "
f"script expects {want_tol}: recapture or fix the spec")
else:
ref, cropped, err = fresh_region(name, "assert")
if err:
record(lineno, line, "fatal", err)
else:
ok, detail = imgtools.compare(ref, cropped, ref["tolerance"])
record(lineno, line, "ok" if ok else "fail", detail)
elif cmd == "assert_ocr":
# assert_ocr NAME REGEX: same crop as assert_region, OCR'd
# through `tesseract`, REGEX searched (re.search, spaces allowed
# like assert_hit's TEXT) against the extracted text. No
# tesseract on PATH, or no reference for NAME, is FATAL: pixels
# were never actually checked, so this must never look like a
# skipped-but-passing step.
need_imgtools(lineno, cmd)
name, pattern = args[0], " ".join(args[1:])
if shutil.which("tesseract") is None:
record(lineno, line, "fatal", "tesseract not installed")
else:
ref, cropped, err = fresh_region(name, "ocr")
if err:
record(lineno, line, "fatal", err)
else:
png_path = os.path.join(os.path.abspath(outdir), f"ocr_{name}.png")
try:
write_png(cropped, png_path)
text = subprocess.run(
["tesseract", png_path, "stdout"],
capture_output=True, text=True, timeout=20, check=True,
).stdout
except Exception as e:
record(lineno, line, "fatal", f"ocr failed: {e}")
else:
if re.search(pattern, text):
record(lineno, line, "ok")
else:
record(lineno, line, "fail", f"text was {text.strip()!r}")
else:
results.close() # the rows before this one are still evidence
sys.exit(f"FATAL: {script_path}:{lineno}: unknown command '{cmd}'")
crashed = ui_exited(console_path)
if crashed:
record(lineno, line, "fatal", crashed)
break
results.close()
print(f"== {counts['ok']} ok, {counts['fail']} failed, {counts['fatal']} fatal "
f"-> {os.path.join(outdir, 'results.jsonl')}", flush=True)
if counts["fail"] or counts["fatal"]:
sys.exit(1)
def main():
if len(sys.argv) < 3:
sys.exit(__doc__)
path, cmd = sys.argv[1], sys.argv[2]
need = {"screendump": 4, "tap": 5, "quit": 3, "drive": 5}
if cmd not in need:
sys.exit(f"unknown command {cmd}\n{__doc__}")
if len(sys.argv) < need[cmd]:
sys.exit(f"{cmd}: missing argument(s)\n{__doc__}")
size = 720
if "--size" in sys.argv:
size = int(sys.argv[sys.argv.index("--size") + 1])
ctl_path = sys.argv[sys.argv.index("--ctl") + 1] if "--ctl" in sys.argv else None
console_path = sys.argv[sys.argv.index("--console") + 1] if "--console" in sys.argv else None
refs_path = sys.argv[sys.argv.index("--refs") + 1] if "--refs" in sys.argv else None
s = socket.socket(socket.AF_UNIX)
s.connect(path)
f = s.makefile("r")
f.readline() # greeting banner
rpc(s, f, {"execute": "qmp_capabilities"})
if cmd == "screendump":
rpc(s, f, {"execute": "screendump", "arguments": {"filename": sys.argv[3]}})
elif cmd == "tap":
do_tap(s, f, int(sys.argv[3]), int(sys.argv[4]))
elif cmd == "drive":
drive(s, f, sys.argv[3], sys.argv[4], size, ctl_path, console_path, refs_path)
elif cmd == "quit":
s.sendall(b'{"execute":"quit"}\n')
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink