feat(tests): add comprehensive unit test suite for variables, strategy modulation, and loop health

2026-10-04 16:50:13 +00:00
parent a2c5832aff
commit 5b3d77fb60
5 changed files with 1475 additions and 0 deletions
@@ -0,0 +1,96 @@
+#!/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()