From d37191e302ea56db13824fb5d76946b6a64229bc Mon Sep 17 00:00:00 2001
From: Maximilian Eibl <maxi2003.eibl@gmail.com>
Date: Fri, 4 Sep 2026 16:10:36 +0200
Subject: [PATCH] Update unit tests to verify 4-run rule on warnings and
 immediate pass on errors

---
 tests/test_server.py | 62 ++++++++++++++++++++++++++++++++++----------
 1 file changed, 48 insertions(+), 14 deletions(-)

diff --git a/tests/test_server.py b/tests/test_server.py
index ac884b7..12bd9b2 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -79,9 +79,9 @@ class TestServerComponent(unittest.TestCase):
         Server.init_db(self.test_db)
         log_entry = {
             "server": "app-worker-01.corp.local",
-            "signature": "PostgresConnTimeout",
-            "severity": "ERROR",
-            "message": "Connection to database pool timed out after 30s",
+            "signature": "PostgresConnWarning",
+            "severity": "WARNING",
+            "message": "Connection to database pool near capacity: 85%",
             "os_type": "linux"
         }
         payload = {
@@ -89,7 +89,7 @@ class TestServerComponent(unittest.TestCase):
             "logs": [log_entry]
         }
 
-        # Runs 1 to 3: should remain TRANSIENT
+        # 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")
@@ -97,24 +97,53 @@ class TestServerComponent(unittest.TestCase):
 
         conn = sqlite3.connect(self.test_db)
         c = conn.cursor()
-        c.execute("SELECT run_count, status FROM active_issues WHERE signature = ?", ("PostgresConnTimeout",))
+        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")
 
-        # Run 4: promotes to 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)
 
         conn = sqlite3.connect(self.test_db)
         c = conn.cursor()
-        c.execute("SELECT run_count, status FROM active_issues WHERE signature = ?", ("PostgresConnTimeout",))
+        c.execute("SELECT run_count, status FROM active_issues WHERE signature = ?", ("PostgresConnWarning",))
         row = c.fetchone()
         conn.close()
         self.assertEqual(row[0], 4)
         self.assertEqual(row[1], "VERIFIED")
 
+    def test_error_immediate_pass(self):
+        Server.init_db(self.test_db)
+        log_entry = {
+            "server": "app-worker-01.corp.local",
+            "signature": "KernelPanicCritical",
+            "severity": "ERROR",
+            "message": "Kernel panic - not syncing: Fatal hardware error",
+            "os_type": "linux"
+        }
+        payload = {
+            "server": "app-worker-01.corp.local",
+            "logs": [log_entry]
+        }
+
+        # Run 1: ERROR must immediately promote to VERIFIED
+        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"], 1)
+
+        conn = sqlite3.connect(self.test_db)
+        c = conn.cursor()
+        c.execute("SELECT run_count, status, severity FROM active_issues WHERE signature = ?", ("KernelPanicCritical",))
+        row = c.fetchone()
+        conn.close()
+        self.assertIsNotNone(row)
+        self.assertEqual(row[0], 1)
+        self.assertEqual(row[1], "VERIFIED")
+        self.assertEqual(row[2], "ERROR")
+
     def test_server_severity_filtering(self):
         Server.init_db(self.test_db)
         payload = {
@@ -132,15 +161,20 @@ class TestServerComponent(unittest.TestCase):
 
         conn = sqlite3.connect(self.test_db)
         c = conn.cursor()
-        c.execute("SELECT signature FROM active_issues ORDER BY signature")
-        sigs = [r[0] for r in c.fetchall()]
+        c.execute("SELECT signature, status FROM active_issues ORDER BY signature")
+        rows = dict(c.fetchall())
         conn.close()
 
-        self.assertIn("SigInfo", sigs)
-        self.assertIn("SigWarn", sigs)
-        self.assertIn("SigErr", sigs)
-        self.assertNotIn("SigDebug", sigs)
-        self.assertNotIn("SigTrace", sigs)
+        self.assertIn("SigInfo", rows)
+        self.assertIn("SigWarn", rows)
+        self.assertIn("SigErr", rows)
+        self.assertNotIn("SigDebug", rows)
+        self.assertNotIn("SigTrace", rows)
+
+        # SigErr is immediately VERIFIED; SigWarn and SigInfo are TRANSIENT on run 1
+        self.assertEqual(rows["SigErr"], "VERIFIED")
+        self.assertEqual(rows["SigWarn"], "TRANSIENT")
+        self.assertEqual(rows["SigInfo"], "TRANSIENT")
 
 
 if __name__ == "__main__":