diff --git a/qemu/tests/qmp.py b/qemu/tests/qmp.py index 4e2e8fc..6f37389 100755 --- a/qemu/tests/qmp.py +++ b/qemu/tests/qmp.py @@ -217,26 +217,36 @@ def do_tap(s, f, ax, ay, hold=TAP_HOLD_S): send_events(s, f, [btn_ev(False)]) -def do_observed_tap(ctx, ax, ay): - """Press until the guest reports consuming it, then release.""" - before = parse_stats(ctx.ctl.send("stats")).get("presses") +def wait_input_count(ctx, field, before): + """Wait until one guest input counter advances.""" if before is None: + time.sleep(0.05) + return True + deadline = time.monotonic() + TAP_OBSERVE_TIMEOUT_S + while time.monotonic() < deadline: + now = parse_stats(ctx.ctl.send("stats")).get(field) + if now is not None and now != before: + return True + time.sleep(0.02) + return False + + +def do_observed_tap(ctx, ax, ay): + """Wait until the guest consumes both halves of a tap.""" + before = parse_stats(ctx.ctl.send("stats")) + if before.get("presses") is None: do_tap(ctx.s, ctx.f, ax, ay) return True send_events(ctx.s, ctx.f, [abs_ev("x", ax), abs_ev("y", ay), btn_ev(True)]) - observed = False - deadline = time.monotonic() + TAP_OBSERVE_TIMEOUT_S + pressed = False try: - while time.monotonic() < deadline: - now = parse_stats(ctx.ctl.send("stats")).get("presses") - if now is not None and now != before: - observed = True - break - time.sleep(0.02) + pressed = wait_input_count(ctx, "presses", before.get("presses")) finally: send_events(ctx.s, ctx.f, [btn_ev(False)]) - return observed + if not pressed: + return False + return wait_input_count(ctx, "releases", before.get("releases")) def _move_swipe(s, f, ax1, ay1, ax2, ay2, ms, steps): @@ -278,27 +288,18 @@ def do_observed_swipe(ctx, x1, y1, x2, y2, ms=400, steps=None): if steps is None: steps = max(6, int(ms / 25)) - before = parse_stats(ctx.ctl.send("stats")).get("presses") + before = parse_stats(ctx.ctl.send("stats")) send_events(ctx.s, ctx.f, [abs_ev("x", ax1), abs_ev("y", ay1), btn_ev(True)]) - observed = before is None + pressed = False try: - if observed: - time.sleep(0.05) - else: - deadline = time.monotonic() + TAP_OBSERVE_TIMEOUT_S - while time.monotonic() < deadline: - now = parse_stats(ctx.ctl.send("stats")).get("presses") - if now is not None and now != before: - observed = True - break - time.sleep(0.02) - if not observed: + pressed = wait_input_count(ctx, "presses", before.get("presses")) + if not pressed: return False _move_swipe(ctx.s, ctx.f, ax1, ay1, ax2, ay2, ms, steps) - return True finally: send_events(ctx.s, ctx.f, [btn_ev(False)]) + return wait_input_count(ctx, "releases", before.get("releases")) class Ctl: @@ -593,6 +594,7 @@ STATS_FIELD_RE = { "render": re.compile(r'^render:\s*(-?\d+(?:\.\d+)?)\s*ms/frame\s*$'), "idle": re.compile(r'^idle:\s*(\d+)\s*$'), "presses": re.compile(r'^presses:\s*(\d+)\s*$'), + "releases": re.compile(r'^releases:\s*(\d+)\s*$'), "termbusy": re.compile(r'^termbusy:\s*(\d+)\s*$'), "termintr": re.compile(r'^termintr:\s*(\d+)\s*$'), "termfg": re.compile(r'^termfg:\s*(-?\d+)\s*$'), @@ -755,20 +757,23 @@ def verb_shot(ctx, lineno, cmd, args, line): def verb_tap(ctx, lineno, cmd, args, line): x, y = int(args[0]), int(args[1]) observed = do_observed_tap(ctx, to_axis(x, ctx.size), to_axis(y, ctx.size)) - return ("ok", "") if observed else ("fail", "press was not consumed within 2 seconds") + return (("ok", "") if observed else + ("fail", "input press or release was not consumed within 2 seconds")) def verb_swipe(ctx, lineno, cmd, args, line): ms = int(args[4]) if len(args) > 4 else 400 observed = do_observed_swipe( ctx, int(args[0]), int(args[1]), int(args[2]), int(args[3]), ms) - return ("ok", "") if observed else ("fail", "press was not consumed within 2 seconds") + return (("ok", "") if observed else + ("fail", "input press or release was not consumed within 2 seconds")) def verb_fling(ctx, lineno, cmd, args, line): observed = do_observed_swipe( ctx, int(args[0]), int(args[1]), int(args[2]), int(args[3]), ms=120) - return ("ok", "") if observed else ("fail", "press was not consumed within 2 seconds") + return (("ok", "") if observed else + ("fail", "input press or release was not consumed within 2 seconds")) def verb_sleep(ctx, lineno, cmd, args, line): diff --git a/qemu/tests/test_qmp_drive.py b/qemu/tests/test_qmp_drive.py index a3a399e..1bc4f18 100755 --- a/qemu/tests/test_qmp_drive.py +++ b/qemu/tests/test_qmp_drive.py @@ -48,7 +48,8 @@ class FakeCtl: if cmd == "stats": self.stats_calls += 1 return ("page: Demo/Rows\ncpu: 12%\nfps: 10\nrender: 3.20 ms/frame\n" - f"rga: 0%\nidle: 0\npresses: {self.stats_calls}\ntermbusy: 1\n" + f"rga: 0%\nidle: 0\npresses: {self.stats_calls}\n" + f"releases: {self.stats_calls}\ntermbusy: 1\n" "termintr: 2\ntermfg: -1\ntermsig: 3") if cmd == "wake": return "wake: ok" @@ -241,7 +242,9 @@ class PureHelpers(unittest.TestCase): def send(self, cmd): self.calls += 1 - return f"presses: {0 if self.calls < 3 else 1}" + presses = 0 if self.calls < 3 else 1 + releases = 0 if self.calls < 4 else 1 + return f"presses: {presses}\nreleases: {releases}" saved_send, saved_sleep = qmp.send_events, qmp.time.sleep qmp.send_events = lambda _s, _f, events: calls.append(events) @@ -264,7 +267,7 @@ class PureHelpers(unittest.TestCase): class UnobservingCtl: def send(self, cmd): - return "presses: 0" + return "presses: 0\nreleases: 0" saved_send, saved_sleep, saved_monotonic = ( qmp.send_events, qmp.time.sleep, qmp.time.monotonic) @@ -291,7 +294,9 @@ class PureHelpers(unittest.TestCase): def send(self, cmd): self.calls += 1 - return f"presses: {0 if self.calls < 3 else 1}" + presses = 0 if self.calls < 3 else 1 + releases = 0 if self.calls < 4 else 1 + return f"presses: {presses}\nreleases: {releases}" saved_send, saved_sleep = qmp.send_events, qmp.time.sleep qmp.send_events = lambda _s, _f, events: calls.append(events) @@ -314,7 +319,7 @@ class PureHelpers(unittest.TestCase): class UnobservingCtl: def send(self, cmd): - return "presses: 0" + return "presses: 0\nreleases: 0" saved_send, saved_sleep, saved_monotonic = ( qmp.send_events, qmp.time.sleep, qmp.time.monotonic) @@ -378,7 +383,7 @@ class PureHelpers(unittest.TestCase): 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, "presses": 1.0, + "idle": 0.0, "presses": 1.0, "releases": 1.0, "termbusy": 1.0, "termintr": 2.0, "termfg": -1.0, "termsig": 3.0})