Observe QEMU swipe presses (#546)

qemu/tests/qmp.py

@@ -239,6 +239,16 @@ def do_observed_tap(ctx, ax, ay):
     return observed
 
 
+def _move_swipe(s, f, ax1, ay1, ax2, ay2, ms, steps):
+    for i in range(1, steps + 1):
+        t = i / steps
+        send_events(s, f, [
+            abs_ev("x", int(ax1 + (ax2 - ax1) * t)),
+            abs_ev("y", int(ay1 + (ay2 - ay1) * t)),
+        ])
+        time.sleep(ms / 1000.0 / steps)
+
+
 def do_swipe(s, f, x1, y1, x2, y2, size, ms=400, steps=None):
     """Drag with interpolated motion.
 
@@ -254,17 +264,43 @@ def do_swipe(s, f, x1, y1, x2, y2, size, ms=400, steps=None):
     if steps is None:
         steps = max(6, int(ms / 25))
     send_events(s, f, [abs_ev("x", ax1), abs_ev("y", ay1), btn_ev(True)])
+    try:
         time.sleep(0.05)
-    for i in range(1, steps + 1):
-        t = i / steps
-        send_events(s, f, [
-            abs_ev("x", int(ax1 + (ax2 - ax1) * t)),
-            abs_ev("y", int(ay1 + (ay2 - ay1) * t)),
-        ])
-        time.sleep(ms / 1000.0 / steps)
+        _move_swipe(s, f, ax1, ay1, ax2, ay2, ms, steps)
+    finally:
         send_events(s, f, [btn_ev(False)])
 
 
+def do_observed_swipe(ctx, x1, y1, x2, y2, ms=400, steps=None):
+    """Start a drag only after the guest has consumed its press."""
+    ax1, ay1 = to_axis(x1, ctx.size), to_axis(y1, ctx.size)
+    ax2, ay2 = to_axis(x2, ctx.size), to_axis(y2, ctx.size)
+    if steps is None:
+        steps = max(6, int(ms / 25))
+
+    before = parse_stats(ctx.ctl.send("stats")).get("presses")
+    send_events(ctx.s, ctx.f, [abs_ev("x", ax1), abs_ev("y", ay1), btn_ev(True)])
+    observed = before is None
+    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:
+            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)])
+
+
 class Ctl:
     """The UI's debug channel, reached through run.sh --ctl.
 
@@ -724,14 +760,15 @@ def verb_tap(ctx, lineno, cmd, args, line):
 
 def verb_swipe(ctx, lineno, cmd, args, line):
     ms = int(args[4]) if len(args) > 4 else 400
-    do_swipe(ctx.s, ctx.f, int(args[0]), int(args[1]), int(args[2]), int(args[3]), ctx.size, ms)
-    return "ok", ""
+    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")
 
 
 def verb_fling(ctx, lineno, cmd, args, line):
-    do_swipe(ctx.s, ctx.f, int(args[0]), int(args[1]), int(args[2]), int(args[3]),
-             ctx.size, ms=120)
-    return "ok", ""
+    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")
 
 
 def verb_sleep(ctx, lineno, cmd, args, line):

qemu/tests/test_qmp_drive.py

@@ -282,6 +282,58 @@ class PureHelpers(unittest.TestCase):
         self.assertFalse(observed)
         self.assertEqual(calls[-1], [qmp.btn_ev(False)])
 
+    def test_observed_swipe_moves_after_guest_consumes_press(self):
+        calls = []
+
+        class ObservingCtl:
+            def __init__(self):
+                self.calls = 0
+
+            def send(self, cmd):
+                self.calls += 1
+                return f"presses: {0 if self.calls < 3 else 1}"
+
+        saved_send, saved_sleep = qmp.send_events, qmp.time.sleep
+        qmp.send_events = lambda _s, _f, events: calls.append(events)
+        qmp.time.sleep = lambda _seconds: None
+        try:
+            observed = qmp.do_observed_swipe(
+                SimpleNamespace(s=None, f=None, ctl=ObservingCtl(), size=720),
+                10, 20, 110, 120, ms=50, steps=2)
+        finally:
+            qmp.send_events, qmp.time.sleep = saved_send, saved_sleep
+
+        self.assertTrue(observed)
+        self.assertEqual(calls[0][-1], qmp.btn_ev(True))
+        self.assertEqual(calls[-1], [qmp.btn_ev(False)])
+        self.assertEqual(len(calls), 4)
+
+    def test_observed_swipe_timeout_still_releases_press(self):
+        calls = []
+        clock = iter((0.0, 0.0, qmp.TAP_OBSERVE_TIMEOUT_S + 0.1))
+
+        class UnobservingCtl:
+            def send(self, cmd):
+                return "presses: 0"
+
+        saved_send, saved_sleep, saved_monotonic = (
+            qmp.send_events, qmp.time.sleep, qmp.time.monotonic)
+        qmp.send_events = lambda _s, _f, events: calls.append(events)
+        qmp.time.sleep = lambda _seconds: None
+        qmp.time.monotonic = lambda: next(clock)
+        try:
+            observed = qmp.do_observed_swipe(
+                SimpleNamespace(s=None, f=None, ctl=UnobservingCtl(), size=720),
+                10, 20, 110, 120, ms=50, steps=2)
+        finally:
+            qmp.send_events = saved_send
+            qmp.time.sleep = saved_sleep
+            qmp.time.monotonic = saved_monotonic
+
+        self.assertFalse(observed)
+        self.assertEqual(len(calls), 2)
+        self.assertEqual(calls[-1], [qmp.btn_ev(False)])
+
     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