mirror of
https://github.com/MCKero6423/uv-k5-v3-emulator.git
synced 2026-10-02 11:07:31 +00:00
mckero
20e7577257
Two changes, both aimed only at latency, since that is what the link makes expensive. 1. Send on pointerdown instead of pointerup. Waiting for release left the network idle for the entire click. The duration is unknown at that moment, so the speculative request asks for a short press; holding past LONG_PRESS_AFTER_MS sends a second, deliberately long press, which is how held events stay reachable. 2. TAP_MS 200 -> 60 ms. The server blocks for hold_ms before replying, so this is latency the user pays directly. 200 ms was a guess that gave back most of what change 1 saved. The 60 ms is measured, and the sample size mattered: at 4 trials per value 30 ms looked reliable, but at 12 trials 20 ms registered only 5/12 while 30 ms was 12/12. The nominal 20 ms debounce is not sufficient alone because KEYBOARD_Poll samples each column 8 times wanting 2 matching reads. 60 ms is double the proven floor. Click-to-visible at 400 ms RTT, where one round trip is an unavoidable 400 ms: original (2 requests, on release) 2525 ms (+2125 over the floor) one request, on release 657 ms (+257) current (1 request, on press) 505 ms (+105) Long press still works: a 60 ms press opens the menu (gScreenToDisplay 0 -> 1) while a 900 ms press is treated as held and correctly does not, so the firmware still distinguishes them. Also drops MIN_HOLD_MS, now dead: the browser no longer measures press duration, so there is no measurement to clamp. Its test asserted only that the string appeared, which would have kept passing over dead code.
511 lines
19 KiB
Python
511 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Unit tests for the web UI. Stubs the QMP client, so no emulator needed."""
|
|
import time
|
|
import unittest
|
|
|
|
import webui
|
|
|
|
|
|
class StubClient:
|
|
"""Records commands; writes the files memsave would write."""
|
|
|
|
def __init__(self):
|
|
self.sent = []
|
|
|
|
def command(self, name, **args):
|
|
self.sent.append((name, args))
|
|
if name == "query-status":
|
|
return {"status": "running", "running": True}
|
|
if name == "memsave":
|
|
with open(args["filename"], "wb") as fh:
|
|
fh.write(bytes(args["size"]))
|
|
return {}
|
|
if name == "pmemsave":
|
|
raise AssertionError("pmemsave reads physical addresses; use memsave")
|
|
return {}
|
|
|
|
def presses(self):
|
|
"""The sequence of values written to the keypad press property."""
|
|
return [a["value"] for n, a in self.sent if n == "qom-set"]
|
|
|
|
|
|
def make_app():
|
|
client = StubClient()
|
|
app = webui.create_app(client, frame_addr=0x1000, status_addr=0x2000)
|
|
app.config.update(TESTING=True)
|
|
return client, app.test_client()
|
|
|
|
|
|
class TestStatus(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client, self.http = make_app()
|
|
|
|
def test_status_reports_running(self):
|
|
resp = self.http.get("/api/status")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertEqual(resp.get_json()["status"], "running")
|
|
|
|
def test_index_serves_html_with_the_screen_and_keypad(self):
|
|
resp = self.http.get("/")
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = resp.get_data(as_text=True)
|
|
self.assertIn("/stream", body)
|
|
self.assertIn("MENU", body)
|
|
|
|
|
|
class TestKeyEndpoint(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client, self.http = make_app()
|
|
|
|
def test_down_holds_the_key(self):
|
|
resp = self.http.post("/api/key", json={"key": "MENU", "action": "down"})
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertEqual(self.client.presses(), ["MENU"])
|
|
|
|
def test_up_releases(self):
|
|
self.http.post("/api/key", json={"key": "MENU", "action": "up"})
|
|
self.assertEqual(self.client.presses(), [""])
|
|
|
|
def test_tap_holds_then_releases(self):
|
|
resp = self.http.post("/api/key", json={"key": "UP", "action": "tap"})
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertEqual(self.client.presses(), ["UP", ""])
|
|
|
|
def test_tap_duration_is_a_short_press(self):
|
|
"""Must land above the 20 ms debounce and below the 400 ms hold."""
|
|
self.assertGreater(webui.TAP_MS, 20)
|
|
self.assertLess(webui.TAP_MS, 400)
|
|
|
|
def test_writes_the_keypad_press_property(self):
|
|
self.http.post("/api/key", json={"key": "MENU", "action": "down"})
|
|
name, args = self.client.sent[-1]
|
|
self.assertEqual(name, "qom-set")
|
|
self.assertEqual(args["path"], "/machine/keypad")
|
|
self.assertEqual(args["property"], "press")
|
|
|
|
def test_rejects_unknown_key(self):
|
|
resp = self.http.post("/api/key", json={"key": "PTT", "action": "tap"})
|
|
self.assertEqual(resp.status_code, 400)
|
|
self.assertEqual(self.client.presses(), [])
|
|
|
|
def test_rejects_unknown_action(self):
|
|
resp = self.http.post("/api/key", json={"key": "UP", "action": "wiggle"})
|
|
self.assertEqual(resp.status_code, 400)
|
|
self.assertEqual(self.client.presses(), [])
|
|
|
|
def test_accepts_lowercase_key(self):
|
|
resp = self.http.post("/api/key", json={"key": "menu", "action": "down"})
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertEqual(self.client.presses(), ["MENU"])
|
|
|
|
def test_release_all_clears_every_key(self):
|
|
self.http.post("/api/key", json={"key": "MENU", "action": "down"})
|
|
resp = self.http.post("/api/release-all")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertEqual(self.client.presses()[-1], "")
|
|
|
|
|
|
class TestStream(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client, self.http = make_app()
|
|
|
|
def test_stream_is_multipart_and_yields_a_png_part(self):
|
|
resp = self.http.get("/stream?limit=1")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn("multipart/x-mixed-replace", resp.headers["Content-Type"])
|
|
body = resp.get_data()
|
|
self.assertIn(b"Content-Type: image/png", body)
|
|
self.assertIn(b"\x89PNG\r\n\x1a\n", body)
|
|
|
|
def test_single_frame_endpoint_returns_a_png(self):
|
|
resp = self.http.get("/frame.png")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertEqual(resp.headers["Content-Type"], "image/png")
|
|
self.assertTrue(resp.get_data().startswith(b"\x89PNG\r\n\x1a\n"))
|
|
|
|
def test_stream_reads_frames_with_memsave(self):
|
|
self.http.get("/stream?limit=1")
|
|
self.assertIn("memsave", [n for n, _ in self.client.sent])
|
|
|
|
|
|
class TestFrontEnd(unittest.TestCase):
|
|
def setUp(self):
|
|
_, http = make_app()
|
|
self.body = http.get("/").get_data(as_text=True)
|
|
|
|
def test_every_model_key_has_a_button(self):
|
|
for key in webui.KEYS:
|
|
self.assertIn(f'data-key="{key}"', self.body)
|
|
|
|
def test_binds_pointer_and_keyboard_input(self):
|
|
self.assertIn("pointerdown", self.body)
|
|
self.assertIn("keydown", self.body)
|
|
|
|
def test_releases_on_blur_so_keys_cannot_stick(self):
|
|
self.assertIn("blur", self.body)
|
|
|
|
def test_does_not_offer_ptt(self):
|
|
self.assertNotIn('data-key="PTT"', self.body)
|
|
|
|
|
|
class TestHoldMs(unittest.TestCase):
|
|
"""Press duration must be produced by the server, not by request timing.
|
|
|
|
At 400 ms RTT the gap between a `down` request and an `up` request is itself
|
|
~400 ms, which the firmware reads as a held key (key_repeat_delay_10ms = 40).
|
|
Measured against the real server: an intended tap arrived as 407 ms. Holding
|
|
server-side is what makes a short press possible over a slow link.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.client, self.http = make_app()
|
|
|
|
def test_hold_ms_presses_and_releases(self):
|
|
resp = self.http.post("/api/key", json={"key": "MENU", "hold_ms": 120})
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertEqual(self.client.presses(), ["MENU", ""])
|
|
|
|
def test_hold_ms_is_honoured_server_side(self):
|
|
start = time.monotonic()
|
|
self.http.post("/api/key", json={"key": "MENU", "hold_ms": 150})
|
|
elapsed = (time.monotonic() - start) * 1000
|
|
self.assertGreaterEqual(elapsed, 140)
|
|
self.assertLess(elapsed, 400)
|
|
|
|
def test_hold_ms_defaults_to_a_short_press(self):
|
|
resp = self.http.post("/api/key", json={"key": "UP"})
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertEqual(resp.get_json()["hold_ms"], webui.TAP_MS)
|
|
|
|
def test_hold_ms_is_clamped(self):
|
|
resp = self.http.post("/api/key", json={"key": "UP", "hold_ms": 99999})
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertLessEqual(resp.get_json()["hold_ms"], webui.MAX_HOLD_MS)
|
|
|
|
def test_hold_ms_rejects_nonsense(self):
|
|
resp = self.http.post("/api/key", json={"key": "UP", "hold_ms": "soon"})
|
|
self.assertEqual(resp.status_code, 400)
|
|
self.assertEqual(self.client.presses(), [])
|
|
|
|
def test_hold_ms_rejects_negative(self):
|
|
resp = self.http.post("/api/key", json={"key": "UP", "hold_ms": -5})
|
|
self.assertEqual(resp.status_code, 400)
|
|
|
|
def test_long_hold_is_preserved_not_clamped_to_a_tap(self):
|
|
"""A deliberate long press must stay long, or hold events break."""
|
|
resp = self.http.post("/api/key", json={"key": "MENU", "hold_ms": 900})
|
|
self.assertEqual(resp.get_json()["hold_ms"], 900)
|
|
|
|
|
|
class TestFrontEndHoldMs(unittest.TestCase):
|
|
"""The page must send one request per key, carrying a measured duration."""
|
|
|
|
def setUp(self):
|
|
_, http = make_app()
|
|
self.body = http.get("/").get_data(as_text=True)
|
|
|
|
def test_sends_hold_ms(self):
|
|
self.assertIn("hold_ms", self.body)
|
|
|
|
def test_does_not_send_separate_down_and_up_for_taps(self):
|
|
"""Two requests per key double the latency and break at 400 ms RTT."""
|
|
self.assertNotIn("send(key, 'down')", self.body)
|
|
self.assertNotIn("send(key, 'up')", self.body)
|
|
|
|
def test_uses_a_measured_tap_length(self):
|
|
"""The tap length is a server-side constant, not a browser measurement."""
|
|
self.assertIn("TAP_MS", self.body)
|
|
|
|
|
|
class TestStreamUsesPump(unittest.TestCase):
|
|
"""Serving frames must not cost QMP reads: the pump owns the grabbing."""
|
|
|
|
def _app(self):
|
|
client = StubClient()
|
|
app = webui.create_app(client, frame_addr=0x1000, status_addr=0x2000)
|
|
app.config.update(TESTING=True)
|
|
return client, app, app.test_client()
|
|
|
|
def _wait(self, app, timeout=3.0):
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if app.config["PUMP"].latest() is not None:
|
|
return True
|
|
time.sleep(0.02)
|
|
return False
|
|
|
|
def test_exposes_the_pump(self):
|
|
_, app, _ = self._app()
|
|
self.assertIn("PUMP", app.config)
|
|
|
|
def test_serving_frames_costs_no_qmp_reads(self):
|
|
client, app, http = self._app()
|
|
self.assertTrue(self._wait(app), "pump produced no frame")
|
|
|
|
before = len([n for n, _ in client.sent if n == "memsave"])
|
|
http.get("/stream?limit=1")
|
|
http.get("/frame.png")
|
|
after = len([n for n, _ in client.sent if n == "memsave"])
|
|
# The pump keeps grabbing in the background, so allow a little drift;
|
|
# what must not happen is a read per request.
|
|
self.assertLess(after - before, 8,
|
|
"serving a frame triggered fresh QMP reads")
|
|
|
|
def test_frame_png_still_returns_a_png(self):
|
|
_, app, http = self._app()
|
|
self.assertTrue(self._wait(app))
|
|
resp = http.get("/frame.png")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertTrue(resp.get_data().startswith(b"\x89PNG\r\n\x1a\n"))
|
|
|
|
def test_frame_png_returns_503_when_there_is_no_frame(self):
|
|
"""No frame is a state, not a crash: the emulator may be powered off."""
|
|
app = webui.create_app(None, frame_addr=0x1000, status_addr=0x2000)
|
|
app.config.update(TESTING=True)
|
|
self.assertEqual(app.test_client().get("/frame.png").status_code, 503)
|
|
|
|
|
|
class FakeSupervisor:
|
|
"""Minimal supervisor double: records calls, tracks powered state."""
|
|
|
|
def __init__(self, client=None, owns=True):
|
|
self._client = client
|
|
self._owns = owns
|
|
self.calls = []
|
|
|
|
def is_running(self):
|
|
return self._client is not None
|
|
|
|
def owns_process(self):
|
|
return self._owns
|
|
|
|
def client(self):
|
|
return self._client
|
|
|
|
def power_on(self):
|
|
self.calls.append("power_on")
|
|
if self._client is None:
|
|
self._client = StubClient()
|
|
return True
|
|
|
|
def power_off(self):
|
|
self.calls.append("power_off")
|
|
self._client = None
|
|
return True
|
|
|
|
def reset(self):
|
|
self.calls.append("reset")
|
|
if self._client is not None:
|
|
self._client.command("system_reset")
|
|
return True
|
|
|
|
def pause(self):
|
|
self.calls.append("pause")
|
|
if self._client is None:
|
|
return False
|
|
self._client.command("stop")
|
|
return True
|
|
|
|
def resume(self):
|
|
self.calls.append("resume")
|
|
if self._client is None:
|
|
return False
|
|
self._client.command("cont")
|
|
return True
|
|
|
|
|
|
def make_supervised(owns=True):
|
|
client = StubClient()
|
|
sup = FakeSupervisor(client, owns=owns)
|
|
app = webui.create_app(client, frame_addr=0x1000, status_addr=0x2000,
|
|
supervisor=sup)
|
|
app.config.update(TESTING=True)
|
|
return client, sup, app.test_client()
|
|
|
|
|
|
class TestPowerEndpoints(unittest.TestCase):
|
|
def test_reset_issues_system_reset(self):
|
|
client, sup, http = make_supervised()
|
|
resp = http.post("/api/power/reset")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn("reset", sup.calls)
|
|
self.assertIn("system_reset", [n for n, _ in client.sent])
|
|
|
|
def test_pause_and_resume(self):
|
|
client, sup, http = make_supervised()
|
|
self.assertEqual(http.post("/api/power/pause").status_code, 200)
|
|
self.assertEqual(http.post("/api/power/resume").status_code, 200)
|
|
names = [n for n, _ in client.sent]
|
|
self.assertIn("stop", names)
|
|
self.assertIn("cont", names)
|
|
|
|
def test_off_then_on(self):
|
|
_, sup, http = make_supervised()
|
|
self.assertEqual(http.post("/api/power/off").status_code, 200)
|
|
self.assertFalse(http.get("/api/status").get_json()["powered"])
|
|
self.assertEqual(http.post("/api/power/on").status_code, 200)
|
|
self.assertTrue(http.get("/api/status").get_json()["powered"])
|
|
self.assertEqual(sup.calls, ["power_off", "power_on"])
|
|
|
|
def test_unknown_action_is_rejected(self):
|
|
_, sup, http = make_supervised()
|
|
resp = http.post("/api/power/explode")
|
|
self.assertEqual(resp.status_code, 400)
|
|
self.assertEqual(sup.calls, [])
|
|
|
|
def test_off_is_refused_for_an_adopted_emulator(self):
|
|
"""We did not start that process, so killing it is not ours to do."""
|
|
_, sup, http = make_supervised(owns=False)
|
|
resp = http.post("/api/power/off")
|
|
self.assertEqual(resp.status_code, 409)
|
|
self.assertEqual(sup.calls, [])
|
|
|
|
def test_reset_is_allowed_for_an_adopted_emulator(self):
|
|
"""system_reset does not kill anything, so it is fine either way."""
|
|
_, sup, http = make_supervised(owns=False)
|
|
self.assertEqual(http.post("/api/power/reset").status_code, 200)
|
|
self.assertIn("reset", sup.calls)
|
|
|
|
def test_power_without_a_supervisor_is_409_not_500(self):
|
|
_, http = make_app()
|
|
resp = http.post("/api/power/on")
|
|
self.assertEqual(resp.status_code, 409)
|
|
|
|
|
|
class TestStatusReportsPower(unittest.TestCase):
|
|
def test_status_includes_powered(self):
|
|
_, http = make_app()
|
|
self.assertIn("powered", http.get("/api/status").get_json())
|
|
|
|
def test_status_when_powered_off_does_not_error(self):
|
|
"""The page must load with the emulator off, not 500."""
|
|
app = webui.create_app(None, frame_addr=0x1000, status_addr=0x2000)
|
|
app.config.update(TESTING=True)
|
|
http = app.test_client()
|
|
self.assertEqual(http.get("/").status_code, 200)
|
|
resp = http.get("/api/status")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertFalse(resp.get_json()["powered"])
|
|
|
|
def test_status_reports_unreachable_when_qmp_raises(self):
|
|
class Broken:
|
|
def command(self, name, **args):
|
|
raise RuntimeError("socket gone")
|
|
|
|
app = webui.create_app(Broken(), frame_addr=0x1000, status_addr=0x2000)
|
|
app.config.update(TESTING=True)
|
|
body = app.test_client().get("/api/status").get_json()
|
|
self.assertFalse(body["powered"])
|
|
self.assertIn("unreachable", body["status"])
|
|
|
|
|
|
class TestKeyWhenPoweredOff(unittest.TestCase):
|
|
def test_key_is_refused_with_no_emulator(self):
|
|
"""Pressing a key on a dark screen is a 409, not a crash."""
|
|
app = webui.create_app(None, frame_addr=0x1000, status_addr=0x2000)
|
|
app.config.update(TESTING=True)
|
|
resp = app.test_client().post("/api/key", json={"key": "MENU"})
|
|
self.assertEqual(resp.status_code, 409)
|
|
|
|
|
|
class TestPowerBar(unittest.TestCase):
|
|
def setUp(self):
|
|
_, http = make_app()
|
|
self.body = http.get("/").get_data(as_text=True)
|
|
|
|
def test_has_power_buttons(self):
|
|
for action in ("on", "off", "reset"):
|
|
self.assertIn(f'data-power="{action}"', self.body)
|
|
|
|
def test_power_bar_is_above_the_screen(self):
|
|
"""The user asked for it on top."""
|
|
self.assertLess(self.body.index('data-power="on"'),
|
|
self.body.index('id="screen"'))
|
|
|
|
def test_power_off_asks_for_confirmation(self):
|
|
"""Off kills the emulator; a stray click should not do that silently."""
|
|
self.assertIn("confirm(", self.body)
|
|
|
|
def test_shows_the_powered_state(self):
|
|
self.assertIn("powerstate", self.body)
|
|
|
|
def test_dims_the_screen_when_off(self):
|
|
"""A dark screen is the signal that the machine is off."""
|
|
self.assertIn("screen-off", self.body)
|
|
|
|
|
|
class TestStartsPoweredOff(unittest.TestCase):
|
|
"""The emulator must not be running until the user asks for it."""
|
|
|
|
def test_app_with_no_client_serves_a_dark_screen(self):
|
|
app = webui.create_app(None, frame_addr=0x1000, status_addr=0x2000)
|
|
app.config.update(TESTING=True)
|
|
http = app.test_client()
|
|
|
|
self.assertEqual(http.get("/").status_code, 200)
|
|
self.assertFalse(http.get("/api/status").get_json()["powered"])
|
|
# No frame yet, and that is a state rather than an error.
|
|
self.assertEqual(http.get("/frame.png").status_code, 503)
|
|
|
|
def test_page_offers_an_on_button_while_off(self):
|
|
app = webui.create_app(None, frame_addr=0x1000, status_addr=0x2000)
|
|
app.config.update(TESTING=True)
|
|
body = app.test_client().get("/").get_data(as_text=True)
|
|
self.assertIn('data-power="on"', body)
|
|
|
|
def test_main_has_an_attach_flag_not_an_own_flag(self):
|
|
"""Owning the process is the default; attaching is the opt-in."""
|
|
import inspect
|
|
src = inspect.getsource(webui.main)
|
|
self.assertIn("--attach", src)
|
|
self.assertNotIn("--own-emulator", src)
|
|
|
|
def test_main_does_not_power_on_at_startup(self):
|
|
"""Arriving at a dark screen is the point; do not boot it for them."""
|
|
import inspect
|
|
src = inspect.getsource(webui.main)
|
|
# power_on may only appear under the attach branch, never unconditionally.
|
|
self.assertNotIn("supervisor.power_on()", src)
|
|
|
|
|
|
class TestOptimisticSend(unittest.TestCase):
|
|
"""The request must leave on pointerdown, not on release.
|
|
|
|
Waiting for pointerup spends the whole click duration with the network idle:
|
|
on a 400 ms link a 120 ms click cost 650 ms click-to-visible instead of 530 ms,
|
|
because nothing was in flight while the button was down.
|
|
"""
|
|
|
|
def setUp(self):
|
|
_, http = make_app()
|
|
self.body = http.get("/").get_data(as_text=True)
|
|
|
|
def test_sends_from_pointerdown(self):
|
|
# down() must issue the request itself rather than only recording a time.
|
|
down_fn = self.body.split("function down(key)")[1].split("function up(")[0]
|
|
self.assertIn("sendKey(", down_fn,
|
|
"down() must fire the request immediately")
|
|
|
|
def test_up_does_not_send_the_press(self):
|
|
up_fn = self.body.split("function up(key)")[1].split("}")[0]
|
|
self.assertNotIn("sendKey(", up_fn,
|
|
"up() must not be where the press is sent")
|
|
|
|
def test_long_press_is_still_reachable(self):
|
|
"""Holding must still produce a held event, or long-press breaks."""
|
|
self.assertIn("LONG_PRESS_MS", self.body)
|
|
self.assertIn("LONG_PRESS_AFTER_MS", self.body)
|
|
|
|
def test_long_press_threshold_is_past_the_firmware_boundary(self):
|
|
"""Must exceed 400 ms or the firmware will not call it held."""
|
|
self.assertGreater(webui.LONG_PRESS_MS, 400)
|
|
self.assertGreaterEqual(webui.LONG_PRESS_AFTER_MS, 400)
|
|
|
|
def test_optimistic_hold_is_a_short_press(self):
|
|
"""The speculative press must stay under the held threshold."""
|
|
self.assertLess(webui.TAP_MS, 400)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|