97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
test_loop_health_remediation.py — Unit and integration tests for:
|
||
|
|
1. Reconstructing intrinsic loops from followups.json and dm-log.jsonl
|
||
|
|
2. Fleet loop health metrics computation and threshold evaluation
|
||
|
|
3. Progressive remediation:
|
||
|
|
- Soft break healing (auto-resolving answered follow-ups)
|
||
|
|
- Deadline re-arming for pending expired nudges
|
||
|
|
4. Hard break diagnostic escalations (silent agent, auth rot, systemd failure)
|
||
|
|
5. Audit event logging in job-log.jsonl
|
||
|
|
"""
|
||
|
|
import unittest
|
||
|
|
import subprocess
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
import shutil
|
||
|
|
from pathlib import Path
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
|
||
|
|
REPO_ROOT = Path("/home/super/Projects/NetVM")
|
||
|
|
BIN_DIR = REPO_ROOT / "bin"
|
||
|
|
BOX_CTL = BIN_DIR / "box-ctl.py"
|
||
|
|
|
||
|
|
sys.path.insert(0, str(BIN_DIR))
|
||
|
|
import gravity
|
||
|
|
|
||
|
|
|
||
|
|
class TestLoopDiagnosticsAndRemediation(unittest.TestCase):
|
||
|
|
"""Test gravity.py loop health and progressive remediation."""
|
||
|
|
|
||
|
|
def test_fleet_loop_health_calculation(self):
|
||
|
|
health = gravity.get_fleet_loop_health(threshold=0.5)
|
||
|
|
self.assertIn("agents", health)
|
||
|
|
self.assertIn("summary", health)
|
||
|
|
self.assertIn("healthy", health["summary"])
|
||
|
|
self.assertIn("overall_health", health["summary"])
|
||
|
|
|
||
|
|
def test_reconstruct_loops(self):
|
||
|
|
loops = gravity.reconstruct_loops(limit=10)
|
||
|
|
self.assertIsInstance(loops, list)
|
||
|
|
if loops:
|
||
|
|
l = loops[0]
|
||
|
|
self.assertIn("loop_id", l)
|
||
|
|
self.assertIn("agent", l)
|
||
|
|
self.assertIn("state", l)
|
||
|
|
|
||
|
|
def test_diagnose_breaks_structure(self):
|
||
|
|
breaks = gravity.diagnose_breaks()
|
||
|
|
self.assertIsInstance(breaks, list)
|
||
|
|
for b in breaks:
|
||
|
|
self.assertIn("type", b)
|
||
|
|
self.assertIn("severity", b)
|
||
|
|
self.assertIn("detail", b)
|
||
|
|
|
||
|
|
def test_remediate_breaks_dry_run(self):
|
||
|
|
res = gravity.remediate_breaks(dry_run=True)
|
||
|
|
self.assertTrue(res.get("ok"))
|
||
|
|
self.assertTrue(res.get("dry_run"))
|
||
|
|
self.assertIsInstance(res.get("remediated"), list)
|
||
|
|
self.assertIsInstance(res.get("escalated"), list)
|
||
|
|
|
||
|
|
|
||
|
|
class TestLoopRpc(unittest.TestCase):
|
||
|
|
"""Test box-ctl.py allowlisted RPC actions for loops."""
|
||
|
|
|
||
|
|
def test_box_ctl_loop_health(self):
|
||
|
|
cmd = [sys.executable, str(BOX_CTL), "loop-health"]
|
||
|
|
res = subprocess.run(cmd, capture_output=True, text=True)
|
||
|
|
self.assertEqual(res.returncode, 0)
|
||
|
|
data = json.loads(res.stdout)
|
||
|
|
self.assertTrue(data.get("ok"))
|
||
|
|
self.assertIn("agents", data)
|
||
|
|
self.assertIn("summary", data)
|
||
|
|
|
||
|
|
|
||
|
|
def test_box_ctl_loop_status(self):
|
||
|
|
cmd = [sys.executable, str(BOX_CTL), "loop-status", "--limit", "5"]
|
||
|
|
res = subprocess.run(cmd, capture_output=True, text=True)
|
||
|
|
self.assertEqual(res.returncode, 0)
|
||
|
|
data = json.loads(res.stdout)
|
||
|
|
self.assertTrue(data.get("ok"))
|
||
|
|
self.assertIn("loops", data)
|
||
|
|
|
||
|
|
def test_box_ctl_loop_remediate_dry_run(self):
|
||
|
|
cmd = [sys.executable, str(BOX_CTL), "loop-remediate", "--dry-run"]
|
||
|
|
res = subprocess.run(cmd, capture_output=True, text=True)
|
||
|
|
self.assertEqual(res.returncode, 0)
|
||
|
|
data = json.loads(res.stdout)
|
||
|
|
self.assertTrue(data.get("ok"))
|
||
|
|
self.assertTrue(data.get("dry_run"))
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|