blueflare/bfe-core1106-sdk
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Wait for QEMU input releases (#546)

Browse Source
This commit is contained in:
Noah
2026-09-11 11:15:02 -06:00
parent 55660b7c01
commit 2efe36f582
2 changed files with 45 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
qemu/tests/qmp.py
+34 -29
View File
@@ -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):
qemu/tests/test_qmp_drive.py
+11 -6
View File
@@ -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})