mirror of
https://github.com/MCKero6423/uv-k5-v3-emulator.git
synced 2026-10-02 03:15:36 +00:00
mckero
dcf9dc00f5
The page edits a working copy of the flash image on purpose -- the file it was first pointed at may be a real calibration dump -- but a restart pointed at a different image switched to that file's copy instead, and the games installed through the page looked like they had vanished (measured: the app table came back holding an older slot). work/run-webui.ps1 now prefers the working copy the page has been editing, ahead of the original, and the page's own hint says which image it is editing and that the loaded one is never modified. A test asserts the order in the script.
322 lines
11 KiB
Python
322 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Unit tests for the QEMU supervisor. Uses a fake launcher, not real QEMU."""
|
|
import os
|
|
import socket
|
|
import unittest
|
|
|
|
import uvk5_socket
|
|
import unittest.mock
|
|
|
|
from uvk5_supervisor import Supervisor
|
|
|
|
|
|
class FakeProc:
|
|
def __init__(self):
|
|
self.terminated = False
|
|
self.killed = False
|
|
self._alive = True
|
|
self.stderr = None
|
|
|
|
def poll(self):
|
|
return None if self._alive else 0
|
|
|
|
def terminate(self):
|
|
self.terminated = True
|
|
self._alive = False
|
|
|
|
def wait(self, timeout=None):
|
|
self._alive = False
|
|
return 0
|
|
|
|
def kill(self):
|
|
self.killed = True
|
|
self._alive = False
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self):
|
|
self.commands = []
|
|
self.closed = False
|
|
|
|
def command(self, name, **args):
|
|
self.commands.append(name)
|
|
if name == "query-status":
|
|
return {"status": "running", "running": True}
|
|
return {}
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
class TestSupervisor(unittest.TestCase):
|
|
def setUp(self):
|
|
self.procs = []
|
|
self.clients = []
|
|
|
|
def launch():
|
|
proc = FakeProc()
|
|
self.procs.append(proc)
|
|
return proc
|
|
|
|
def connect():
|
|
client = FakeClient()
|
|
self.clients.append(client)
|
|
return client
|
|
|
|
self.sup = Supervisor(launch=launch, connect=connect)
|
|
|
|
def test_starts_powered_off_when_nothing_is_running(self):
|
|
"""The user presses On; the server does not boot it for them."""
|
|
self.assertFalse(self.sup.is_running())
|
|
self.assertIsNone(self.sup.client())
|
|
self.assertEqual(self.procs, [])
|
|
|
|
def test_power_on_launches_and_connects(self):
|
|
self.assertTrue(self.sup.power_on())
|
|
self.assertTrue(self.sup.is_running())
|
|
self.assertEqual(len(self.procs), 1)
|
|
self.assertIsNotNone(self.sup.client())
|
|
|
|
def test_power_on_twice_does_not_launch_twice(self):
|
|
self.sup.power_on()
|
|
self.assertFalse(self.sup.power_on())
|
|
self.assertEqual(len(self.procs), 1)
|
|
|
|
def test_power_off_quits_via_qmp_then_stops_the_process(self):
|
|
self.sup.power_on()
|
|
client = self.sup.client()
|
|
self.assertTrue(self.sup.power_off())
|
|
self.assertIn("quit", client.commands)
|
|
self.assertTrue(client.closed)
|
|
self.assertFalse(self.sup.is_running())
|
|
self.assertIsNone(self.sup.client())
|
|
|
|
def test_power_cycle_boots_a_fresh_process(self):
|
|
"""Off then On is a cold boot, like mains power cut and restored."""
|
|
self.sup.power_on()
|
|
self.sup.power_off()
|
|
self.sup.power_on()
|
|
self.assertEqual(len(self.procs), 2)
|
|
self.assertTrue(self.sup.is_running())
|
|
|
|
def test_reset_uses_system_reset_when_running(self):
|
|
self.sup.power_on()
|
|
client = self.sup.client()
|
|
self.sup.reset()
|
|
self.assertIn("system_reset", client.commands)
|
|
self.assertEqual(len(self.procs), 1, "reset must not respawn QEMU")
|
|
|
|
def test_reset_powers_on_when_stopped(self):
|
|
self.sup.reset()
|
|
self.assertTrue(self.sup.is_running())
|
|
self.assertEqual(len(self.procs), 1)
|
|
|
|
def test_pause_and_resume(self):
|
|
self.sup.power_on()
|
|
client = self.sup.client()
|
|
self.assertTrue(self.sup.pause())
|
|
self.assertTrue(self.sup.resume())
|
|
self.assertIn("stop", client.commands)
|
|
self.assertIn("cont", client.commands)
|
|
|
|
def test_pause_when_off_is_refused(self):
|
|
self.assertFalse(self.sup.pause())
|
|
self.assertFalse(self.sup.resume())
|
|
|
|
def test_power_off_when_already_off_is_harmless(self):
|
|
self.assertFalse(self.sup.power_off())
|
|
self.assertFalse(self.sup.is_running())
|
|
|
|
def test_power_off_survives_a_quit_that_raises(self):
|
|
"""The socket usually drops mid-quit; that is success, not failure."""
|
|
class Rude(FakeClient):
|
|
def command(self, name, **args):
|
|
if name == "quit":
|
|
raise RuntimeError("connection reset")
|
|
return super().command(name, **args)
|
|
|
|
sup = Supervisor(launch=lambda: FakeProc(), connect=Rude)
|
|
sup.power_on()
|
|
self.assertTrue(sup.power_off())
|
|
self.assertFalse(sup.is_running())
|
|
|
|
def test_dead_process_is_reported_as_not_running(self):
|
|
self.sup.power_on()
|
|
self.procs[0]._alive = False # QEMU crashed on its own
|
|
self.assertFalse(self.sup.is_running())
|
|
|
|
def test_adopting_an_external_emulator_does_not_own_the_process(self):
|
|
"""Attaching to a run.sh instance must not let power_off kill it."""
|
|
client = FakeClient()
|
|
sup = Supervisor(launch=lambda: self.fail("must not launch"),
|
|
connect=lambda: client)
|
|
sup.adopt(client)
|
|
self.assertTrue(sup.is_running())
|
|
self.assertFalse(sup.owns_process())
|
|
|
|
def test_owns_process_is_true_after_power_on(self):
|
|
self.sup.power_on()
|
|
self.assertTrue(self.sup.owns_process())
|
|
|
|
|
|
class TestSupervisorLogging(unittest.TestCase):
|
|
def setUp(self):
|
|
from uvk5_logs import LogBuffer
|
|
self.log = LogBuffer(capacity=50)
|
|
self.procs = []
|
|
|
|
def launch():
|
|
proc = FakeProc()
|
|
self.procs.append(proc)
|
|
return proc
|
|
|
|
self.sup = Supervisor(launch=launch, connect=FakeClient, log=self.log)
|
|
|
|
def texts(self):
|
|
return [e["text"].lower() for e in self.log.entries()]
|
|
|
|
def test_power_on_is_logged(self):
|
|
self.sup.power_on()
|
|
self.assertTrue(any("power on" in t for t in self.texts()), self.texts())
|
|
|
|
def test_power_off_is_logged(self):
|
|
self.sup.power_on()
|
|
self.sup.power_off()
|
|
self.assertTrue(any("power off" in t for t in self.texts()), self.texts())
|
|
|
|
def test_reset_is_logged(self):
|
|
self.sup.power_on()
|
|
self.sup.reset()
|
|
self.assertTrue(any("reset" in t for t in self.texts()), self.texts())
|
|
|
|
def test_events_are_tagged_as_power(self):
|
|
self.sup.power_on()
|
|
self.assertTrue(any(e["source"] == "power"
|
|
for e in self.log.entries()))
|
|
|
|
def test_works_without_a_log(self):
|
|
"""The log is optional; a supervisor with none must not crash."""
|
|
sup = Supervisor(launch=lambda: FakeProc(), connect=FakeClient)
|
|
sup.power_on()
|
|
sup.power_off()
|
|
|
|
|
|
class TestWaitForSocket(unittest.TestCase):
|
|
"""A leftover socket file must not be mistaken for a listening emulator.
|
|
|
|
Hit for real: a killed QEMU left /tmp/uvk5-qmp.sock behind, wait_for_socket
|
|
returned immediately because the path existed, and the connect then failed with
|
|
ECONNREFUSED -- which the user saw as power on returning HTTP 500.
|
|
"""
|
|
|
|
def setUp(self):
|
|
import tempfile
|
|
self.dir = tempfile.mkdtemp()
|
|
self.path = os.path.join(self.dir, "qmp.sock")
|
|
|
|
def tearDown(self):
|
|
import shutil
|
|
shutil.rmtree(self.dir, ignore_errors=True)
|
|
|
|
@unittest.skipUnless(uvk5_socket.can_use_unix(),
|
|
"needs unix sockets, which a Windows QEMU cannot create")
|
|
def test_returns_false_for_a_stale_socket_file(self):
|
|
from uvk5_supervisor import wait_for_socket
|
|
# A socket file with nothing listening: bind then close.
|
|
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
srv.bind(self.path)
|
|
srv.close()
|
|
self.assertTrue(os.path.exists(self.path), "need a leftover file")
|
|
self.assertFalse(wait_for_socket(self.path, timeout=0.5))
|
|
|
|
@unittest.skipUnless(uvk5_socket.can_use_unix(),
|
|
"needs unix sockets, which a Windows QEMU cannot create")
|
|
def test_returns_true_when_something_is_listening(self):
|
|
from uvk5_supervisor import wait_for_socket
|
|
srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
srv.bind(self.path)
|
|
srv.listen(1)
|
|
self.addCleanup(srv.close)
|
|
self.assertTrue(wait_for_socket(self.path, timeout=2))
|
|
|
|
def test_returns_false_when_the_path_never_appears(self):
|
|
from uvk5_supervisor import wait_for_socket
|
|
self.assertFalse(wait_for_socket(self.path + ".missing", timeout=0.3))
|
|
|
|
|
|
class TestRecoversFromAnExternalKill(unittest.TestCase):
|
|
"""Power on must work again after the emulator dies behind the supervisor's back.
|
|
|
|
Hit for real, twice: a stray cleanup killed the QEMU that webui.py owned. The
|
|
supervisor kept its QMP client object, so is_running() reported a broken pipe and
|
|
power_on() returned False immediately without relaunching -- the Power button was
|
|
dead until the whole service was restarted.
|
|
"""
|
|
|
|
def _supervisor(self, clients):
|
|
launched = []
|
|
|
|
def launch():
|
|
launched.append(1)
|
|
proc = unittest.mock.MagicMock()
|
|
proc.poll.return_value = None
|
|
proc.stderr = None
|
|
return proc
|
|
|
|
def connect():
|
|
return clients.pop(0)
|
|
|
|
sup = Supervisor(launch, connect)
|
|
return sup, launched
|
|
|
|
def test_power_on_relaunches_when_the_client_is_dead(self):
|
|
dead = unittest.mock.MagicMock()
|
|
dead.command.side_effect = BrokenPipeError("dead")
|
|
fresh = unittest.mock.MagicMock()
|
|
fresh.command.return_value = {"status": "running"}
|
|
|
|
sup, launched = self._supervisor([dead, fresh])
|
|
self.assertTrue(sup.power_on())
|
|
self.assertEqual(len(launched), 1)
|
|
|
|
# An external kill: the process is gone, so poll() reports an exit status.
|
|
sup._proc.poll.return_value = -15
|
|
self.assertFalse(sup.is_running())
|
|
|
|
# Power on must notice the client is unusable and start a new emulator.
|
|
self.assertTrue(sup.power_on(), "power_on refused to relaunch a dead guest")
|
|
self.assertEqual(len(launched), 2, "no new emulator was launched")
|
|
self.assertTrue(sup.is_running())
|
|
|
|
def test_power_on_still_refuses_when_genuinely_running(self):
|
|
live = unittest.mock.MagicMock()
|
|
live.command.return_value = {"status": "running"}
|
|
sup, launched = self._supervisor([live])
|
|
self.assertTrue(sup.power_on())
|
|
self.assertFalse(sup.power_on(), "started a second emulator over a live one")
|
|
self.assertEqual(len(launched), 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class TestLauncherPrefersTheWorkingCopy(unittest.TestCase):
|
|
"""A restart must not look like the page's installs vanished."""
|
|
|
|
def test_the_script_lists_the_working_copy_before_the_original(self):
|
|
import os as _os
|
|
path = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)),
|
|
"..", "work", "run-webui.ps1")
|
|
if not _os.path.exists(path):
|
|
self.skipTest("work/run-webui.ps1 is not in this checkout")
|
|
with open(path, encoding="utf-8") as fh:
|
|
text = fh.read()
|
|
# the working copy the page edits comes before the file it was first pointed at
|
|
working = text.find("work\\firmware\\flash-current.img")
|
|
original = text.find("work\\user-flash.img")
|
|
self.assertNotEqual(working, -1, "the working copy is not in the list")
|
|
self.assertNotEqual(original, -1)
|
|
self.assertLess(working, original,
|
|
"a restart would fall back to the original image")
|