mirror of
https://github.com/MCKero6423/uv-k5-v3-emulator.git
synced 2026-10-02 03:15:36 +00:00
mckero
5b53cbbc82
The server now owns the QEMU process by default but does not launch it. You open the page to a dark screen and press On, which is the behaviour asked for: like walking up to a machine rather than finding it already booted. This inverts the flag from the plan. Owning the process has to be the default, since it is the only way On/Off can work at all; --attach is the opt-in for joining a run.sh instance, where Off is refused. Two tests guard the intent rather than the wiring: one asserts main() has --attach and not --own-emulator, another asserts main() never calls power_on(), so a future edit cannot quietly restore auto-boot. Verified on the live server with no QEMU running beforehand: startup 0 QEMU processes, powered=false, frame.png 503, page 200 On 1 QEMU process, powered=true, frame.png 200 (2920 bytes) Off 0 QEMU processes, frame.png 503, page still 200, keys 409 On again 1 QEMU process, powered=true, frame.png 200 The web server stays up across Off, which is what you asked for: the screen goes dark and waits for the next person to press On.
476 lines
18 KiB
Python
476 lines
18 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_measures_press_duration_in_the_browser(self):
|
|
self.assertIn("performance.now()", 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_enforces_a_minimum_hold(self):
|
|
"""A very fast click still has to clear the 20 ms debounce."""
|
|
self.assertIn("MIN_HOLD_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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|