qemu: json/stat/region channels, seeding, driver tests
- rootfs/sbin/init: the control bridge answers `@cat PATH` locally so the driver can read /tmp/warden-web-status.json out of the guest. That file has no trailing newline; the bridge adds one so the sentinel stays on its own line and the line-based reader never blocks. - tests/qmp.py: wait_json/assert_json (dotted paths, eq/ne/contains/ len_eq/len_ge/gt/lt), assert_stat off the FIFO's stats reply, capture_region, assert_region NAME [TOLERANCE] and assert_ocr. A tolerance other than the captured one, a reference box that does not fit the screendump, a missing reference or a missing tesseract is FATAL for that step and the run continues (flare-edge #147). - tests/imgtools.py: P6 reader, crop, perceptual and structural hashes, compare, with a self-test. - tests/test_qmp_drive.py: drive() with QMP and the control channel faked, pinning the per-step ok/fail/fatal contract. - mkimage.sh SEED_DIR and ui-drive.sh --seed/--refs: settings fixtures staged into userdata before warden-ui starts, and a reference store handed to the driver. - ci: the driver tests and the imgtools self-test run in qemu-tools. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013aHKWzT5EF86RFKRMtAv9n
This commit is contained in:
Noah
Executable
+176
@@ -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)
|
||||
Reference in New Issue
Block a user