me0nline/LOGAR
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

Update unit tests to verify 4-run rule on warnings and immediate pass on errors

Browse Source
This commit is contained in:
me0nline
2026-09-04 16:10:36 +02:00
parent 35a736dacb
commit d37191e302
1 changed files with 48 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_server.py
+48 -14
View File
@@ -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__":