#!/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 socket 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%\nidle: 0" if cmd == "wake": return "wake: ok" if cmd == "home": return "home: ok" if cmd.startswith("scroll "): 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 "): return 'hit 47,676: obj text="" box=12,640,72x72' if cmd.startswith("nav "): return cmd + ": ok" 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: fh.write(ppm_bytes()) return {} 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") 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 = (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=console_path, refs_path=refs_path, rs485_control=rs485_control) 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_every_pixel_round_trips_through_lvgl_calibration(self): # lv_evdev.c _evdev_calibrate: px = axis * (width - 1) / AXIS_MAX, # integer division. A tap requested at px must land at px, for every # px, or a hit-confirmed target can be missed by one pixel. for size in (720, 480, 1024): for px in range(size): axis = qmp.to_axis(px, size) self.assertTrue(0 <= axis <= qmp.AXIS_MAX) back = axis * (size - 1) // qmp.AXIS_MAX self.assertEqual(back, px, f"size {size}: px {px} -> axis {axis} -> {back}") 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, "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): rc, by, rows = run_script( "wake\n" "assert_page Demo/Rows\n" "assert_hit 47 676 obj box=12,640,72x72\n" "assert_stat idle eq 0\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" "wait_hit 47 676 obj 2 box=12,640,72x72\n" "wait_region r1 exact 2\n" "scroll 360 400 300\n" "home\n" ) self.assertIsNone(rc, [r for r in rows if r["status"] != "ok"]) self.assertEqual(len(rows), 16) self.assertIn("waited", by["wait_hit 47 676 obj 2 box=12,640,72x72"]["detail"]) 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" "scroll 360 400 0\n" "wait_hit 47 676 obj 1 box=0,0,1x1\n" "assert_page Demo/Rows\n" ) self.assertEqual(rc, 1) self.assertEqual([r["status"] for r in rows], ["fail", "fail", "fail", "fail", "fail", "fail", "fail", "ok"]) self.assertIn("moved", by["assert_hit 47 676 obj box=0,0,1x1"]["detail"]) self.assertIn("moved", by["wait_hit 47 676 obj 1 box=0,0,1x1"]["detail"]) 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, # 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_rs485_verb_talks_to_the_simulator_or_is_fatal(self): sent = [] def fake_send(path, line): sent.append((path, line)) return "ok 5 silent" if line.startswith("silence") else "error no simulated device at 9" saved = qmp.rs485_send qmp.rs485_send = fake_send try: rc, by, rows = run_script("rs485 silence 5\nrs485 restore 9\nrs485 bounce 5\n", rs485_control="fake.ctl") finally: qmp.rs485_send = saved self.assertEqual([r["status"] for r in rows], ["ok", "fail", "fatal"]) self.assertEqual(sent, [("fake.ctl", "silence 5"), ("fake.ctl", "restore 9")]) rc, by, rows = run_script("rs485 silence 5\n") 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 # 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"]) 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)