diff --git a/tests/test_server.py b/tests/test_server.py index 8a80619..e62c5f4 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -75,7 +75,7 @@ class TestServerComponent(unittest.TestCase): if os.path.exists(client_out): os.remove(client_out) - def test_4_run_rule_and_12h_window(self): + def test_direct_hermes_anomaly_and_12h_window(self): Server.init_db(self.test_db) log_entry = { "server": "app-worker-01.corp.local", @@ -89,23 +89,24 @@ class TestServerComponent(unittest.TestCase): "logs": [log_entry] } - # Runs 1 to 3: WARNING should remain TRANSIENT - for run_idx in range(1, 4): - res = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=4) - self.assertEqual(res["status"], "success") - self.assertEqual(res["promoted_verified"], 0) + # Run 1: WARNING is immediately VERIFIED for Hermes AI reasoning + res1 = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=1) + self.assertEqual(res1["status"], "success") + self.assertEqual(res1["promoted_verified"], 1) conn = sqlite3.connect(self.test_db) c = conn.cursor() c.execute("SELECT run_count, status FROM active_issues WHERE signature = ?", ("PostgresConnWarning",)) row = c.fetchone() conn.close() - self.assertEqual(row[0], 3) - self.assertEqual(row[1], "TRANSIENT") + self.assertEqual(row[0], 1) + self.assertEqual(row[1], "VERIFIED") - # Run 4: promotes WARNING to VERIFIED! - res4 = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=4) - self.assertEqual(res4["promoted_verified"], 1) + # Runs 2 to 4: Increments consecutive run count while remaining VERIFIED + for run_idx in range(2, 5): + res = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=1) + self.assertEqual(res["status"], "success") + self.assertEqual(res["promoted_verified"], 0) conn = sqlite3.connect(self.test_db) c = conn.cursor() @@ -130,7 +131,7 @@ class TestServerComponent(unittest.TestCase): } # Run 1: ERROR must immediately promote to VERIFIED - res = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=4) + res = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=1) self.assertEqual(res["status"], "success") self.assertEqual(res["promoted_verified"], 1) @@ -156,7 +157,7 @@ class TestServerComponent(unittest.TestCase): {"server": "app-worker-01", "signature": "SigTrace", "severity": "TRACE", "message": "Trace msg", "os_type": "linux"} ] } - res = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=4) + res = Server.process_ingested_logs(payload, self.test_db, window_hours=12, min_runs=1) self.assertEqual(res["status"], "success") conn = sqlite3.connect(self.test_db) @@ -171,10 +172,10 @@ class TestServerComponent(unittest.TestCase): self.assertNotIn("SigDebug", rows) self.assertNotIn("SigTrace", rows) - # SigErr is immediately VERIFIED; SigWarn and SigInfo are TRANSIENT on run 1 + # All actionable issues (ERROR, WARNING, INFO) are promoted to VERIFIED immediately for Hermes reasoning self.assertEqual(rows["SigErr"], "VERIFIED") - self.assertEqual(rows["SigWarn"], "TRANSIENT") - self.assertEqual(rows["SigInfo"], "TRANSIENT") + self.assertEqual(rows["SigWarn"], "VERIFIED") + self.assertEqual(rows["SigInfo"], "VERIFIED") def test_license_schema_and_pki_generation(self): secret = "test-secret-12345"