Harden QEMU input and credentials (#516 #519)

qemu/tests/test_qmp_drive.py

@@ -22,6 +22,7 @@ import tempfile
 import threading
 import time
 import unittest
+from types import SimpleNamespace
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, HERE)
@@ -39,12 +40,16 @@ class FakeCtl:
 
     def __init__(self, path, timeout=15.0):
         self.path = path
+        self.stats_calls = 0
 
     def send(self, cmd):
         if cmd == "page":
             return "Demo/Rows"
         if cmd == "stats":
-            return "page: Demo/Rows\ncpu: 12%\nfps: 10\nrender: 3.20 ms/frame\nrga: 0%\nidle: 0"
+            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"
+                    "termintr: 2\ntermfg: -1\ntermsig: 3")
         if cmd == "wake":
             return "wake: ok"
         if cmd == "home":
@@ -212,6 +217,71 @@ def run_script(text, refs=None, rs485_control=None, ctl_cls=None, rpc_fn=None, c
 
 
 class PureHelpers(unittest.TestCase):
+    def test_tap_holds_press_across_loaded_guest_polls(self):
+        calls = []
+        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: calls.append(seconds)
+        try:
+            qmp.do_tap(None, None, 123, 456)
+        finally:
+            qmp.send_events, qmp.time.sleep = saved_send, saved_sleep
+
+        self.assertEqual(calls[0], [qmp.abs_ev("x", 123), qmp.abs_ev("y", 456), qmp.btn_ev(True)])
+        self.assertGreaterEqual(calls[1], 0.25)
+        self.assertLess(calls[1], 0.4)
+        self.assertEqual(calls[2], [qmp.btn_ev(False)])
+
+    def test_observed_tap_releases_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_tap(
+                SimpleNamespace(s=None, f=None, ctl=ObservingCtl()), 123, 456)
+        finally:
+            qmp.send_events, qmp.time.sleep = saved_send, saved_sleep
+
+        self.assertTrue(observed)
+        self.assertEqual(calls, [
+            [qmp.abs_ev("x", 123), qmp.abs_ev("y", 456), qmp.btn_ev(True)],
+            [qmp.btn_ev(False)],
+        ])
+
+    def test_observed_tap_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_tap(
+                SimpleNamespace(s=None, f=None, ctl=UnobservingCtl()), 123, 456)
+        finally:
+            qmp.send_events = saved_send
+            qmp.time.sleep = saved_sleep
+            qmp.time.monotonic = saved_monotonic
+
+        self.assertFalse(observed)
+        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
@@ -255,7 +325,10 @@ 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})
+        self.assertEqual(got, {"cpu": 12.0, "fps": 10.0, "render": 3.2,
+                               "idle": 0.0, "presses": 1.0,
+                               "termbusy": 1.0, "termintr": 2.0,
+                               "termfg": -1.0, "termsig": 3.0})
 
     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