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

Update test_server.py to test direct Hermes anomaly verification

Browse Source
This commit is contained in:
me0nline
2026-09-25 16:15:11 +02:00
parent 2980ad2268
commit 2cd7a990b5
1 changed files with 17 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_server.py
+17 -16
View File
@@ -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"