"""U-040 tests: c_deploy_supervisor_monitor. Real HTTP server for http_health/ canary_task signals (same pattern as test_deploy_health_check.py), real sqlite for the validator_approval_rate derivation and window-closure logic. Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v """ import json import os import sys import threading import unittest import uuid sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from supervisor import deploy_monitor as dm from src import backend_server, schema def _ensure_task(conn, project_id, task_id): if conn.execute("SELECT 1 FROM task WHERE task_id = ?", (task_id,)).fetchone() is None: conn.execute( "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " "VALUES (?, ?, 'g', 'goal', 'validating', 'validator')", (task_id, project_id), ) conn.commit() def _status_event(conn, project_id, task_id, to_status, created_at=None): _ensure_task(conn, project_id, task_id) conn.execute( "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload, created_at) " "VALUES (?, ?, ?, 'task_status_changed', ?, COALESCE(?, CURRENT_TIMESTAMP))", (str(uuid.uuid4()), project_id, task_id, json.dumps({"to_status": to_status}), created_at), ) conn.commit() def _validator_bug(conn, task_id): conn.execute( "INSERT INTO bug_record (bug_id, task_id, source, description, root_cause_category, status) " "VALUES (?, ?, 'validator', 'x', 'coder', 'routed')", (str(uuid.uuid4()), task_id), ) conn.commit() class ValidatorApprovalRateTestBase(unittest.TestCase): def setUp(self): self.conn = schema.connect(":memory:") schema.create_schema(self.conn) self.conn.execute( "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" ) self.conn.commit() self.addCleanup(self.conn.close) class ComputeValidatorApprovalRateTest(ValidatorApprovalRateTestBase): def test_no_validating_events_returns_none(self): self.assertIsNone(dm.compute_validator_approval_rate(self.conn, "proj-1", "2000-01-01 00:00:00")) def test_all_approved_gives_rate_1(self): _status_event(self.conn, "proj-1", "t1", "validating") _status_event(self.conn, "proj-1", "t2", "validating") rate, n = dm.compute_validator_approval_rate(self.conn, "proj-1", "2000-01-01 00:00:00") self.assertEqual((rate, n), (1.0, 2)) def test_bug_record_marks_task_as_not_approved(self): _status_event(self.conn, "proj-1", "t1", "validating") _status_event(self.conn, "proj-1", "t2", "validating") _validator_bug(self.conn, "t1") rate, n = dm.compute_validator_approval_rate(self.conn, "proj-1", "2000-01-01 00:00:00") self.assertEqual((rate, n), (0.5, 2)) def test_non_validator_bug_record_does_not_count_against_approval(self): _status_event(self.conn, "proj-1", "t1", "validating") self.conn.execute( "INSERT INTO bug_record (bug_id, task_id, source, description, status) " "VALUES ('b1', 't1', 'human', 'x', 'open')" ) self.conn.commit() rate, n = dm.compute_validator_approval_rate(self.conn, "proj-1", "2000-01-01 00:00:00") self.assertEqual((rate, n), (1.0, 1)) def test_events_before_since_are_excluded(self): _status_event(self.conn, "proj-1", "t1", "validating", created_at="2000-01-01 00:00:00") result = dm.compute_validator_approval_rate(self.conn, "proj-1", "2020-01-01 00:00:00") self.assertIsNone(result) class ComputeBaselineTest(ValidatorApprovalRateTestBase): def test_baseline_looks_back_from_before_text(self): _status_event(self.conn, "proj-1", "t-old", "validating", created_at="2026-01-01 00:00:00") _status_event(self.conn, "proj-1", "t-too-old", "validating", created_at="2020-01-01 00:00:00") rate, n = dm.compute_baseline_validator_approval_rate( self.conn, "proj-1", "2026-01-02 00:00:00", lookback_hours=24 ) self.assertEqual(n, 1) # only t-old falls in the 24h lookback window 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 RunMonitorCycleTest(RealServerTestBase): def setUp(self): super().setUp() self.conn = schema.connect(":memory:") schema.create_schema(self.conn) self.conn.execute( "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" ) self.conn.execute( "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, " "previous_instance_dir, cutover_at) " "VALUES ('dep-1', 'c1', 'c0', 'monitoring', '/prev', '2020-01-01 00:00:00')" ) self.conn.commit() self.addCleanup(self.conn.close) def test_all_clear_and_window_not_yet_closed_stays_monitoring(self): result = dm.run_monitor_cycle( self.conn, "dep-1", os.getpid(), self.port, "proj-1", baseline_validator_rate=None, window_start_text="2099-01-01 00:00:00", # future -> 0 elapsed hours now=lambda: dm._parse_sqlite_timestamp("2099-01-01 00:00:00") + 60, ) self.assertEqual(result["status"], "monitoring") row = self.conn.execute("SELECT status FROM deployment WHERE deployment_id = 'dep-1'").fetchone() self.assertEqual(row[0], "monitoring") def test_window_closes_via_elapsed_hours(self): result = dm.run_monitor_cycle( self.conn, "dep-1", os.getpid(), self.port, "proj-1", baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00", now=lambda: dm._parse_sqlite_timestamp("2020-01-01 00:00:00") + 25 * 3600, window_hours=24, ) self.assertEqual(result["status"], "monitoring_window_clean") row = self.conn.execute( "SELECT status, monitoring_closed_at FROM deployment WHERE deployment_id = 'dep-1'" ).fetchone() self.assertEqual(row[0], "monitoring_window_clean") self.assertIsNotNone(row[1]) def test_window_closes_via_real_task_count(self): for i in range(10): self.conn.execute( "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " "VALUES (?, 'proj-1', 'g', 'goal', 'queued', 'planner')", ("t{}".format(i),) ) self.conn.commit() result = dm.run_monitor_cycle( self.conn, "dep-1", os.getpid(), self.port, "proj-1", baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00", now=lambda: dm._parse_sqlite_timestamp("2020-01-01 00:00:00") + 60, window_hours=999999, min_task_count=10, ) self.assertEqual(result["status"], "monitoring_window_clean") def test_single_dead_process_reading_alone_does_not_yet_anomaly(self): # design doc ยง6: process_alive's anomaly condition is MAX_CRASH_COUNT # crashes within CRASH_WINDOW, not a single dead reading -- a lone # blip records the signal but must NOT immediately roll back. result = dm.run_monitor_cycle( self.conn, "dep-1", 2**30, self.port, "proj-1", baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00", window_hours=999999, ) self.assertNotEqual(result["status"], "anomaly_detected") signal_row = self.conn.execute( "SELECT status FROM deploy_health_signal WHERE deployment_id = 'dep-1' AND signal_type = 'process_alive'" ).fetchone() self.assertEqual(signal_row[0], "fail") def test_repeated_process_crashes_within_window_trigger_anomaly(self): # Pre-seed one prior process_alive fail -- this cycle's fail makes 2, # meeting MAX_CRASH_COUNT=2 even though the process IS alive right now. self.conn.execute( "INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status) " "VALUES (?, 'dep-1', CURRENT_TIMESTAMP, 'process_alive', 'fail')", (str(uuid.uuid4()),) ) self.conn.commit() result = dm.run_monitor_cycle( self.conn, "dep-1", 2**30, self.port, "proj-1", baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00", crash_window_samples=5, ) self.assertEqual(result["status"], "anomaly_detected") def test_consecutive_http_failures_trigger_anomaly(self): for _ in range(3): self.conn.execute( "INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status) " "VALUES (?, 'dep-1', CURRENT_TIMESTAMP, 'http_health', 'fail')", (str(uuid.uuid4()),) ) self.conn.commit() self.server.conn.close() # this cycle's own http_health check also fails result = dm.run_monitor_cycle( self.conn, "dep-1", os.getpid(), self.port, "proj-1", baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00", consecutive_http_fail_limit=4, ) self.assertEqual(result["status"], "anomaly_detected") self.assertEqual(result["latest_signal"]["signal_type"], "http_health") def test_validator_rate_deviation_triggers_anomaly(self): for i in range(4): _status_event(self.conn, "proj-1", "t{}".format(i), "validating") _validator_bug(self.conn, "t{}".format(i)) # 0% approval this window result = dm.run_monitor_cycle( self.conn, "dep-1", os.getpid(), self.port, "proj-1", baseline_validator_rate=0.9, window_start_text="2020-01-01 00:00:00", ) self.assertEqual(result["status"], "anomaly_detected") self.assertEqual(result["latest_signal"]["signal_type"], "validator_approval_rate") def test_no_baseline_yet_never_flags_validator_anomaly(self): for i in range(4): _status_event(self.conn, "proj-1", "t{}".format(i), "validating") _validator_bug(self.conn, "t{}".format(i)) result = dm.run_monitor_cycle( self.conn, "dep-1", os.getpid(), self.port, "proj-1", baseline_validator_rate=None, window_start_text="2020-01-01 00:00:00", ) self.assertNotEqual(result["status"], "anomaly_detected") if __name__ == "__main__": unittest.main()