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

Update test_pipeline.py for direct Hermes anomaly verification

Browse Source
This commit is contained in:
me0nline
2026-09-25 16:15:21 +02:00
parent 2cd7a990b5
commit b7da54feb2
1 changed files with 5 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_pipeline.py
+5 -5
View File
@@ -146,7 +146,7 @@ def run_tests():
resp_bytes.extend(chunk) resp_bytes.extend(chunk)
return json.loads(resp_bytes.decode("utf-8")) return json.loads(resp_bytes.decode("utf-8"))
print("\n=== [3] Testing Temporal Persistence & 4-Run Rule for Warnings over mTLS ===") print("\n=== [3] Testing Temporal Persistence & Direct Anomaly Verification over mTLS ===")
test_signature = "TestServiceDegraded" test_signature = "TestServiceDegraded"
candidate_log = [{ candidate_log = [{
"server": client_id, "server": client_id,
@@ -161,10 +161,10 @@ def run_tests():
assert resp.get("status") == "success", f"Run {run_num} failed: {resp}" assert resp.get("status") == "success", f"Run {run_num} failed: {resp}"
promoted = resp.get("promoted_verified", 0) promoted = resp.get("promoted_verified", 0)
print(f"[Run {run_num}/4] Ingested successfully via mTLS. Promoted to verified: {promoted}") print(f"[Run {run_num}/4] Ingested successfully via mTLS. Promoted to verified: {promoted}")
if run_num < 4: if run_num == 1:
assert promoted == 0, f"Expected 0 promoted on run {run_num} for warning, got {promoted}" assert promoted == 1, f"Expected 1 promoted on run 1 for warning, got {promoted}"
else: else:
assert promoted == 1, f"Expected 1 promoted on run 4 for warning, got {promoted}" assert promoted == 0, f"Expected 0 newly promoted on run {run_num} (already verified), got {promoted}"
# Inspect SQLite database directly # Inspect SQLite database directly
conn = sqlite3.connect(server_conf.get("db_path", "logar_state.db")) conn = sqlite3.connect(server_conf.get("db_path", "logar_state.db"))
@@ -178,7 +178,7 @@ def run_tests():
print(f"[DB Verification] Issue '{test_signature}' -> run_count: {run_count}, status: {status}") print(f"[DB Verification] Issue '{test_signature}' -> run_count: {run_count}, status: {status}")
assert run_count >= 4, f"Expected run_count >= 4, got {run_count}" assert run_count >= 4, f"Expected run_count >= 4, got {run_count}"
assert status == "VERIFIED", f"Expected status 'VERIFIED', got {status}" assert status == "VERIFIED", f"Expected status 'VERIFIED', got {status}"
print("[OK] 4-Run Rule verified: Warning promoted to VERIFIED anomaly on 4th run over mTLS!") print("[OK] Direct verification confirmed: Warning immediately verified for Hermes with run count tracking over mTLS!")
print("\n=== [4] Testing Immediate Pass for Errors over mTLS ===") print("\n=== [4] Testing Immediate Pass for Errors over mTLS ===")
error_signature = "TestServiceCrashImmediate" error_signature = "TestServiceCrashImmediate"