mirror of
https://github.com/MCKero6423/uv-k5-v3-emulator.git
synced 2026-10-02 03:15:36 +00:00
mckero
65e50c0065
Two faults that combined to take the web UI down twice in one session, each time
surfacing to the user as a 502 through the reverse proxy.
`pkill -f 'M uv-k5-v3'` in run.sh and trace_run.sh matched far more than intended.
-f tests the whole command line, so it also matched the shell running the pkill
(the pattern sits in its own argv), any script mentioning the machine type, and the
QEMU child of a running webui.py. Cleanup now lives in tools/lib_kill_emulator.sh:
pgrep -x on the binary name, confirm uv-k5-v3 in /proc/PID/cmdline, and optionally
scope to one QMP socket so a caller only stops the instance it owns.
The supervisor then could not recover from it. power_on() began with
`if self._client is not None: return False`, but a client object is not proof of a
live guest -- after an external kill the stale client made power_on refuse forever,
so the Power button was dead until the whole service was restarted. It now checks
whether the process actually exited and relaunches, logging why.
Tests:
- tools/test_kill_emulator.sh checks a plain process, a process whose command line
merely mentions uv-k5-v3, and the calling script all survive; that a real emulator
on a named socket is stopped; that one on another socket is not; and that an
unscoped call still clears everything. Verified it leaves a live webui.py alone.
- Two supervisor unit tests cover relaunch-after-external-kill and the case that
must still refuse, so this cannot regress into starting two emulators at once.
Verified end to end against the running web UI: kill the emulator from outside,
status reports unreachable, and pressing Power brings it back to
{"status":"running"} where before it stayed dead.
While writing the first version of the test I modelled the failure as a client
raising BrokenPipeError, which is not what is_running() looks at -- it checks
poll(). The mock was wrong, not the code; the test now has the process report an
exit status, which is what really happens.
296 lines
9.9 KiB
Python
296 lines
9.9 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 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)
|
|
|
|
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))
|
|
|
|
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()
|