"""U-038 tests: c_deploy_supervisor_health_check. Real HTTP server (in-process thread, same pattern src/test_backend_server.py already uses), real project/ task creation round trips for the canary check -- nothing here is mocked except process liveness timing (injectable sleep/now, same convention as work_node_run's own quota-wait tests). Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v """ import os import sys import unittest from unittest.mock import patch sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import threading from supervisor import deploy_health_check as dhc from src import backend_server, schema class RealServerTestBase(unittest.TestCase): def setUp(self): self.server = backend_server.create_server(db_path=":memory:") self.port = self.server.server_address[1] self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) class CheckProcessAliveTest(unittest.TestCase): def test_alive_process_returns_ok_immediately(self): signal = dhc.check_process_alive(os.getpid()) self.assertEqual(signal, {"signal_type": "process_alive", "status": "ok", "detail": None}) def test_dead_pid_times_out_and_fails(self): clock = {"t": 0.0} sleeps = [] signal = dhc.check_process_alive( 2**30, timeout=5, sleep=lambda s: (sleeps.append(s), clock.__setitem__("t", clock["t"] + s)), now=lambda: clock["t"], ) self.assertEqual(signal["signal_type"], "process_alive") self.assertEqual(signal["status"], "fail") self.assertTrue(sleeps) def test_process_that_becomes_alive_mid_poll_still_passes(self): calls = {"n": 0} def fake_is_alive(pid): calls["n"] += 1 return calls["n"] >= 3 with patch("supervisor.deploy_supervisor.is_alive", side_effect=fake_is_alive): signal = dhc.check_process_alive(99999, timeout=5, sleep=lambda s: None, now=lambda: 0.0) self.assertEqual(signal["status"], "ok") self.assertEqual(calls["n"], 3) class CheckHttpHealthTest(RealServerTestBase): def test_healthy_server_returns_ok(self): signal = dhc.check_http_health(self.port) self.assertEqual(signal, {"signal_type": "http_health", "status": "ok", "detail": None}) def test_broken_connection_returns_fail(self): self.server.conn.close() signal = dhc.check_http_health(self.port) self.assertEqual(signal["signal_type"], "http_health") self.assertEqual(signal["status"], "fail") def test_nothing_listening_on_port_returns_fail(self): clock = {"t": 0.0} signal = dhc.check_http_health( 1, timeout=5, # privileged/unbound port, nothing listening sleep=lambda s: clock.__setitem__("t", clock["t"] + s), now=lambda: clock["t"], ) self.assertEqual(signal["status"], "fail") def test_connection_refused_polls_until_server_actually_listening(self): # 2026-08-03 regression: a real self-hosting run showed # process_alive can read 'ok' before the candidate has finished # binding its listen socket -- check_http_health must tolerate # that same startup window instead of failing on the first # ECONNREFUSED, same as check_process_alive already tolerates a # not-yet-existing pid. server = backend_server.create_server(db_path=":memory:") real_port = server.server_address[1] self.addCleanup(server.server_close) def start_server_after_delay(): thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() self.addCleanup(lambda: (server.shutdown(), thread.join(timeout=5))) timer = threading.Timer(0.3, start_server_after_delay) timer.start() self.addCleanup(timer.cancel) signal = dhc.check_http_health(real_port, timeout=5) self.assertEqual(signal, {"signal_type": "http_health", "status": "ok", "detail": None}) def test_real_http_error_response_fails_immediately_without_polling(self): # a genuine non-2xx response IS a definitive answer from a live # server -- must not be treated the same as "not up yet". sleeps = [] with patch.object(dhc.urllib.request, "urlopen") as mock_urlopen: error = dhc.urllib.error.HTTPError("url", 500, "Internal Server Error", {}, None) mock_urlopen.side_effect = error signal = dhc.check_http_health(self.port, timeout=5, sleep=lambda s: sleeps.append(s), now=lambda: 0.0) self.assertEqual(signal["status"], "fail") self.assertIn("500", signal["detail"]) self.assertEqual(sleeps, []) mock_urlopen.assert_called_once() class CheckCanaryTaskTest(RealServerTestBase): def test_real_project_and_task_round_trip_passes(self): signal = dhc.check_canary_task(self.port) self.assertEqual(signal, {"signal_type": "canary_task", "status": "ok", "detail": None}) conn = self.server.conn count = conn.execute("SELECT COUNT(*) FROM task WHERE goal_text LIKE 'deploy-supervisor canary%'").fetchone()[0] self.assertEqual(count, 1) def test_canary_project_is_archived_after_a_passing_check(self): # 2026-08-03 regression: found via 5 real leaked 'deploy-canary' # rows sitting permanently active in the real ~/.graphlemon/ # graphlemon.db -- every real health check used to grow the # candidate's DB forever. The task itself is left alone (asserted # above), only the project should end up archived. dhc.check_canary_task(self.port) conn = self.server.conn row = conn.execute("SELECT status FROM project WHERE name = 'deploy-canary'").fetchone() self.assertEqual(row[0], "archived") def test_canary_project_is_archived_even_when_task_creation_fails(self): # The project is created before the task -- a failure on the task # step must not leak the project either. real_post_json = dhc._post_json def fake_post_json(port, path, payload, timeout): if path == "/api/tasks": return 500, {"error": "boom"} return real_post_json(port, path, payload, timeout) with patch.object(dhc, "_post_json", side_effect=fake_post_json): signal = dhc.check_canary_task(self.port) self.assertEqual(signal["status"], "fail") conn = self.server.conn row = conn.execute("SELECT status FROM project WHERE name = 'deploy-canary'").fetchone() self.assertEqual(row[0], "archived") def test_nothing_listening_returns_fail(self): signal = dhc.check_canary_task(1) self.assertEqual(signal["signal_type"], "canary_task") self.assertEqual(signal["status"], "fail") class RunHealthCheckTest(RealServerTestBase): def setUp(self): super().setUp() self.conn = schema.connect(":memory:") schema.create_schema(self.conn) self.conn.execute( "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, " "previous_instance_dir) VALUES ('dep-1', 'c1', 'c0', 'candidate_starting', '/prev')" ) self.conn.commit() self.addCleanup(self.conn.close) def test_all_checks_pass_marks_health_check_passed(self): result = dhc.run_health_check(self.conn, "dep-1", os.getpid(), self.port) self.assertEqual(result["status"], "health_check_passed") self.assertEqual([s["signal_type"] for s in result["signals"]], ["process_alive", "http_health", "canary_task"]) self.assertTrue(all(s["status"] == "ok" for s in result["signals"])) row = self.conn.execute("SELECT status FROM deployment WHERE deployment_id = 'dep-1'").fetchone() self.assertEqual(row[0], "health_check_passed") signal_count = self.conn.execute( "SELECT COUNT(*) FROM deploy_health_signal WHERE deployment_id = 'dep-1'" ).fetchone()[0] self.assertEqual(signal_count, 3) def test_process_alive_failure_stops_before_http_and_canary_checks(self): result = dhc.run_health_check( self.conn, "dep-1", 2**30, self.port, startup_timeout=0, sleep=lambda s: None, now=lambda: 0.0, ) self.assertEqual(result["status"], "health_check_failed") self.assertEqual(len(result["signals"]), 1) self.assertEqual(result["signals"][0]["signal_type"], "process_alive") row = self.conn.execute("SELECT status FROM deployment WHERE deployment_id = 'dep-1'").fetchone() self.assertEqual(row[0], "health_check_failed") signal_count = self.conn.execute( "SELECT COUNT(*) FROM deploy_health_signal WHERE deployment_id = 'dep-1'" ).fetchone()[0] self.assertEqual(signal_count, 1) # http_health/canary_task never ran def test_http_health_failure_stops_before_canary_check(self): self.server.conn.close() result = dhc.run_health_check(self.conn, "dep-1", os.getpid(), self.port) self.assertEqual(result["status"], "health_check_failed") self.assertEqual([s["signal_type"] for s in result["signals"]], ["process_alive", "http_health"]) if __name__ == "__main__": unittest.main()