qemu: wait_hit and wait_region poll a settling page

A hit or region check fired once, so a check one second after `nav`
raced a page still laying itself out (flare-edge #175). wait_hit X Y CLASS
TIMEOUT_S [box=...] [TEXT] and wait_region NAME TOLERANCE TIMEOUT_S run
the same judgement every 0.5 s until it holds or the deadline passes, and
report how long they waited. The assert_hit judgement moves into
judge_hit so both verbs mean the same thing by a match; a missing
reference or foreign tolerance stays fatal on the first answer, since
waiting cannot fix either. Offline tests cover both verbs and the timeout.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N3G6m9Aw5RyVY4ZowtKzEj

qemu/tests/qmp.py

@@ -261,6 +261,40 @@ class Ctl:
 HIT_RE = re.compile(r'^hit \d+,\d+: (\S+)(?: text="(.*)" box=(-?\d+),(-?\d+),(\d+)x(\d+))?$')
 
 
+def judge_hit(ctl, x, y, cls, rest):
+    """One `hit` round trip judged the way assert_hit documents: -> (ok, detail).
+    REST is the optional box=X1,Y1,WxH followed by the caption words."""
+    rest = list(rest)
+    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:
+        return False, "nothing clickable there"
+    if got[0] != cls:
+        return False, f"{got[0]} text={got[1]!r} box={got[2]}"
+    if want_box and got[2] != want_box:
+        return False, f"{cls} moved: box={got[2]}"
+    if want_text and want_text not in got[1]:
+        return False, f"{cls} text={got[1]!r}"
+    return True, ""
+
+
+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."""
+    start = time.monotonic()
+    deadline = start + timeout_s
+    while True:
+        ok, detail = check()
+        if ok or time.monotonic() >= deadline:
+            return ok, detail, time.monotonic() - start
+        time.sleep(period)
+
+
 def parse_hit(reply):
     """-> (cls, text, box) with box as 'x1,y1,WxH', or None for 'nothing';
     raises on an unparseable reply."""
@@ -578,24 +612,22 @@ def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, ref
             # 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]
+            ok, detail = judge_hit(ctl, args[0], args[1], args[2], args[3:])
-            rest = args[3:]
+            record(lineno, line, "ok" if ok else "fail", detail)
-            want_box = None
+        elif cmd == "wait_hit":
-            if rest and rest[0].startswith("box="):
+            # wait_hit X Y CLASS TIMEOUT_S [box=X1,Y1,WxH] [TEXT...]: assert_hit
-                want_box = rest[0][4:]
+            # polled every 0.5s until it holds or TIMEOUT_S passes. A page
-                rest = rest[1:]
+            # still building after `nav` answers `hit` with whatever is there
-            want_text = " ".join(rest)
+            # at that instant, so a check one second in raced the layout
-            got = parse_hit(ctl.send(f"hit {x} {y}"))
+            # (flare-edge #175). The timeout is the author's bound on how long
-            if got is None:
+            # settling may take, not a sleep: a check that holds early returns
-                record(lineno, line, "fail", "nothing clickable there")
+            # early, and the detail says how long it took.
-            elif got[0] != cls:
+            need_ctl(lineno, cmd)
-                record(lineno, line, "fail", f"{got[0]} text={got[1]!r} box={got[2]}")
+            x, y, cls, timeout_s = args[0], args[1], args[2], float(args[3])
-            elif want_box and got[2] != want_box:
+            rest = args[4:]
-                record(lineno, line, "fail", f"{cls} moved: box={got[2]}")
+            ok, detail, waited = poll_until(lambda: judge_hit(ctl, x, y, cls, rest), timeout_s)
-            elif want_text and want_text not in got[1]:
+            record(lineno, line, "ok" if ok else "fail",
-                record(lineno, line, "fail", f"{cls} text={got[1]!r}")
+                   f"waited {waited:.1f}s" if ok else detail)
-            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
@@ -692,6 +724,35 @@ def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None, ref
                 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

qemu/tests/test_qmp_drive.py

@@ -133,11 +133,14 @@ class DriveVerbs(unittest.TestCase):
             "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), 14)
+        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):
@@ -149,13 +152,16 @@ class DriveVerbs(unittest.TestCase):
             "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", "ok"])
+                         ["fail", "fail", "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")
+        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_region_faults_are_per_step_fatal(self):
         # A pre-seeded reference whose box does not fit a 64x64 screendump,