From ca3c78ef577f5d8435dd10db0296c3e2fd8e69e2 Mon Sep 17 00:00:00 2001 From: historia Date: Tue, 25 Aug 2026 20:10:40 -0400 Subject: feat: stop server and exit after generating --- app/tests/test_backends_servers.py | 60 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) (limited to 'app/tests/test_backends_servers.py') diff --git a/app/tests/test_backends_servers.py b/app/tests/test_backends_servers.py index 201d5b6..8eaf53f 100644 --- a/app/tests/test_backends_servers.py +++ b/app/tests/test_backends_servers.py @@ -1,5 +1,7 @@ """Tests for the server lifecycle module (backends/servers.py).""" +import os +import signal import tempfile import unittest from pathlib import Path @@ -225,6 +227,64 @@ class StopTests(unittest.TestCase): self.assertFalse((self.dir / "test-server.pid").exists()) +class ReapTests(unittest.TestCase): + """_reap_exited: an exited child must stop counting as alive.""" + + def test_true_when_child_exited(self): + with patch("os.waitpid", return_value=(4242, 0)) as mk: + self.assertTrue(servers._reap_exited(4242)) + mk.assert_called_once_with(4242, os.WNOHANG) + + def test_false_while_still_running(self): + # (0, 0) is WNOHANG's "still running" answer. + with patch("os.waitpid", return_value=(0, 0)): + self.assertFalse(servers._reap_exited(4242)) + + def test_false_when_not_our_child(self): + with patch("os.waitpid", side_effect=ChildProcessError): + self.assertFalse(servers._reap_exited(4242)) + + +class KillPidTests(unittest.TestCase): + """_kill_pid: the reap check ends the grace wait before SIGKILL.""" + + def test_reaped_child_ends_wait_without_sigkill(self): + with patch("os.getpgid", return_value=4242), \ + patch("os.killpg") as mk_killpg, \ + patch("os.waitpid", return_value=(4242, 0)) as mk_waitpid, \ + patch("time.sleep") as mk_sleep: + ok = servers._kill_pid(4242) + self.assertTrue(ok) + mk_killpg.assert_called_once_with(4242, signal.SIGTERM) + mk_waitpid.assert_called_once_with(4242, os.WNOHANG) + mk_sleep.assert_not_called() + + def test_escalates_to_sigkill_when_child_stays_alive(self): + with patch("os.getpgid", return_value=4242), \ + patch("os.killpg") as mk_killpg, \ + patch("os.waitpid", return_value=(0, 0)), \ + patch("time.sleep"): + ok = servers._kill_pid(4242) + self.assertTrue(ok) + calls = mk_killpg.call_args_list + self.assertEqual(calls[0].args, (4242, signal.SIGTERM)) + self.assertEqual(calls[-1].args, (4242, signal.SIGKILL)) + + def test_foreign_child_falls_back_to_group_probe(self): + # ChildProcessError from waitpid (not our child / already reaped): + # the killpg(0) probe decides; a vanished group ends the wait. + with patch("os.getpgid", return_value=4242), \ + patch("os.killpg", + side_effect=[None, ProcessLookupError]) as mk_killpg, \ + patch("os.waitpid", side_effect=ChildProcessError), \ + patch("time.sleep"): + ok = servers._kill_pid(4242) + self.assertTrue(ok) + calls = mk_killpg.call_args_list + self.assertEqual(calls[0].args, (4242, signal.SIGTERM)) + self.assertEqual(calls[-1].args, (4242, 0)) + + class ManagesTests(unittest.TestCase): """manages(): a live recorded pid marks a server as ours.""" -- cgit v1.2.3