qemu: add wait_stat verb for polling stat checks
assert_stat samples a stat field once. warden-ui's fps counter is a rolling one-second window, so a single sample right after a page opens can read 0 even though the UI is live (flare-edge #44). Add wait_stat FIELD OP VALUE TIMEOUT, the same polling pattern as wait_hit/wait_json, built on a shared eval_stat() that assert_stat now uses too. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SMRwnkPp1upR6QFWCZouE7
This commit is contained in:
Noah
@@ -387,6 +387,24 @@ class PureHelpers(unittest.TestCase):
|
||||
"termbusy": 1.0, "termintr": 2.0,
|
||||
"termfg": -1.0, "termsig": 3.0})
|
||||
|
||||
def test_eval_stat_rejects_unknown_fields_and_catches_bad_ops(self):
|
||||
# eval_stat's own whitelist is narrower than parse_stats(): 'releases'
|
||||
# is a real field in a `stats` reply (see test_parse_stats) but not
|
||||
# one assert_stat/wait_stat expose. And it must catch apply_op's
|
||||
# (TypeError, ValueError) the same way eval_json does, so a malformed
|
||||
# OP reads as an ordinary fail, not a driver crash, for both the
|
||||
# single-read and the polling verb built on top of it.
|
||||
ctl = FakeCtl("x")
|
||||
ok, detail = qmp.eval_stat(ctl, "releases", "eq", "1")
|
||||
self.assertFalse(ok)
|
||||
self.assertIn("unknown stat field", detail)
|
||||
ok, detail = qmp.eval_stat(ctl, "fps", "bogus", "1")
|
||||
self.assertFalse(ok)
|
||||
self.assertIn("bogus", detail)
|
||||
ok, detail = qmp.eval_stat(ctl, "fps", "eq", "10")
|
||||
self.assertTrue(ok)
|
||||
self.assertEqual(detail, "")
|
||||
|
||||
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
|
||||
@@ -425,6 +443,7 @@ class DriveVerbs(unittest.TestCase):
|
||||
"wait_json list len_ge 2 2\n"
|
||||
"assert_json name eq warden\n"
|
||||
"assert_stat fps gt 0\n"
|
||||
"wait_stat fps gt 0 2\n"
|
||||
"nav Demo/Rows\n"
|
||||
"capture_region r1 0 0 8 8 exact\n"
|
||||
"assert_region r1 exact\n"
|
||||
@@ -435,8 +454,9 @@ class DriveVerbs(unittest.TestCase):
|
||||
"home\n"
|
||||
)
|
||||
self.assertIsNone(rc, [r for r in rows if r["status"] != "ok"])
|
||||
self.assertEqual(len(rows), 16)
|
||||
self.assertEqual(len(rows), 17)
|
||||
self.assertIn("waited", by["wait_hit 47 676 obj 2 box=12,640,72x72"]["detail"])
|
||||
self.assertIn("waited", by["wait_stat fps gt 0 2"]["detail"])
|
||||
self.assertTrue(all(r["status"] == "ok" for r in rows))
|
||||
|
||||
def test_mismatches_are_fails_not_stops(self):
|
||||
@@ -447,17 +467,19 @@ class DriveVerbs(unittest.TestCase):
|
||||
"wait_json a.b eq 2 1\n"
|
||||
"assert_json a.zz eq 1\n"
|
||||
"assert_stat fps lt 0\n"
|
||||
"wait_stat fps lt 0 1\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"])
|
||||
["fail", "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")
|
||||
self.assertIn("fps is 10.0", by["wait_stat fps lt 0 1"]["detail"])
|
||||
self.assertGreaterEqual(time.monotonic() - t0, 3.0,
|
||||
"wait_json, wait_stat 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-
|
||||
@@ -484,13 +506,23 @@ class DriveVerbs(unittest.TestCase):
|
||||
self.assertEqual([r["status"] for r in rows], ["fatal", "ok"])
|
||||
self.assertIn("control channel closed", by["wait_hit 47 676 obj 2"]["detail"])
|
||||
|
||||
rc, by, rows = run_script(
|
||||
"wait_stat fps gt 0 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_stat fps gt 0 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.
|
||||
# wait_hit/wait_json/wait_stat 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"
|
||||
@@ -500,12 +532,13 @@ class DriveVerbs(unittest.TestCase):
|
||||
"assert_hit 47 676 obj\n"
|
||||
"assert_json a.b eq 1\n"
|
||||
"assert_stat idle eq 0\n"
|
||||
"wait_stat fps gt 0 1\n"
|
||||
"sleep 0\n",
|
||||
ctl_cls=ImmediateDyingCtl,
|
||||
)
|
||||
self.assertEqual(rc, 1)
|
||||
self.assertEqual([r["status"] for r in rows],
|
||||
["fatal"] * 8 + ["ok"],
|
||||
["fatal"] * 9 + ["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]:
|
||||
@@ -534,12 +567,12 @@ class DriveVerbs(unittest.TestCase):
|
||||
def test_malformed_numeric_argument_is_fatal_for_the_step_not_a_crash(self):
|
||||
# Several verbs parse their own arguments with bare
|
||||
# int()/float()/positional indexing before any handler-local guard
|
||||
# (tap, swipe, fling, sleep, wait_hit, wait_json, capture_region,
|
||||
# wait_region). A typo'd coordinate or a missing argument -- exactly
|
||||
# what a hand-edited *.txt script or a flowc.py bug can produce --
|
||||
# used to raise ValueError/IndexError straight out of drive(),
|
||||
# losing every row from that line onward instead of reading as its
|
||||
# own fatal row (flare-edge #244).
|
||||
# (tap, swipe, fling, sleep, wait_hit, wait_json, wait_stat,
|
||||
# capture_region, wait_region). A typo'd coordinate or a missing
|
||||
# argument -- exactly what a hand-edited *.txt script or a flowc.py
|
||||
# bug can produce -- used to raise ValueError/IndexError straight
|
||||
# out of drive(), losing every row from that line onward instead of
|
||||
# reading as its own fatal row (flare-edge #244).
|
||||
rc, by, rows = run_script(
|
||||
"tap 10 abc\n"
|
||||
"sleep 0\n",
|
||||
@@ -565,6 +598,15 @@ class DriveVerbs(unittest.TestCase):
|
||||
self.assertIn("invalid literal",
|
||||
by["capture_region r1 0 0 8 notanumber exact"]["detail"])
|
||||
|
||||
rc, by, rows = run_script(
|
||||
"wait_stat fps gt 0 notanumber\n"
|
||||
"sleep 0\n",
|
||||
)
|
||||
self.assertEqual(rc, 1)
|
||||
self.assertEqual([r["status"] for r in rows], ["fatal", "ok"])
|
||||
self.assertIn("could not convert string to float",
|
||||
by["wait_stat fps gt 0 notanumber"]["detail"])
|
||||
|
||||
def test_bad_op_reads_as_a_fail_row_not_a_crash(self):
|
||||
# apply_op's error paths (unknown OP -> ValueError, 'contains'
|
||||
# against the wrong type -> TypeError) are caught by both callers
|
||||
|
||||
Reference in New Issue
Block a user