diff --git a/qemu/tests/qmp.py b/qemu/tests/qmp.py index 4973a10..b79e55f 100755 --- a/qemu/tests/qmp.py +++ b/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] - 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") + ok, detail = judge_hit(ctl, args[0], args[1], args[2], args[3:]) + record(lineno, line, "ok" if ok else "fail", detail) + elif cmd == "wait_hit": + # wait_hit X Y CLASS TIMEOUT_S [box=X1,Y1,WxH] [TEXT...]: assert_hit + # polled every 0.5s until it holds or TIMEOUT_S passes. A page + # still building after `nav` answers `hit` with whatever is there + # at that instant, so a check one second in raced the layout + # (flare-edge #175). The timeout is the author's bound on how long + # settling may take, not a sleep: a check that holds early returns + # early, and the detail says how long it took. + need_ctl(lineno, cmd) + x, y, cls, timeout_s = args[0], args[1], args[2], float(args[3]) + rest = args[4:] + ok, detail, waited = poll_until(lambda: judge_hit(ctl, x, y, cls, rest), timeout_s) + record(lineno, line, "ok" if ok else "fail", + f"waited {waited:.1f}s" if ok else detail) 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 diff --git a/qemu/tests/test_qmp_drive.py b/qemu/tests/test_qmp_drive.py index e3f59bf..c5162f0 100755 --- a/qemu/tests/test_qmp_drive.py +++ b/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,