docs(netvm): document watchdog, swarm-worker, and alert relay in README; track approvals CLI

tests/test_approvals.py

@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""
+test_approvals.py — Unit and integration tests for:
+1. approvals.py module (inspect, check_fleet, auto_approve, allow/deny structure)
+2. box approvals CLI command (check, list, auto, --json)
+3. box-ctl.py RPC actions (approval-check, approval-auto)
+4. gravity.py loop break detection (approval_blocked taxonomy)
+5. muse-chat-api.py account loading and connection
+"""
+
+import json
+import os
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+
+REPO_ROOT = Path("/home/super/Projects/NetVM")
+BIN_DIR = REPO_ROOT / "bin"
+sys.path.insert(0, str(BIN_DIR))
+
+import approvals
+import gravity
+
+
+class TestApprovalsModule(unittest.TestCase):
+    """Test approvals.py core module functionality."""
+
+    def test_trusted_ips_configuration(self):
+        self.assertIn("34.139.37.135", approvals.TRUSTED_IPS)
+        self.assertIn("100.123.153.75", approvals.TRUSTED_IPS)
+
+    def test_get_node_connection_info(self):
+        info = approvals.get_node_connection_info("pip")
+        self.assertEqual(info["node"], "pip")
+        self.assertEqual(info["cdp_port"], 9420)
+        self.assertTrue(info["peer_ip"].startswith("10.201."))
+
+    def test_check_fleet_approvals_structure(self):
+        res = approvals.check_fleet_approvals(nodes=["pip", "muse"])
+        self.assertIsInstance(res, list)
+        self.assertEqual(len(res), 2)
+        for it in res:
+            self.assertIn("node", it)
+            self.assertIn("status", it)
+            self.assertIn("has_pending", it)
+            self.assertIn("buttons", it)
+            self.assertIn("is_trusted", it)
+
+    def test_auto_approve_fleet_structure(self):
+        res = approvals.auto_approve_fleet(nodes=["pip"])
+        self.assertTrue(res.get("ok"))
+        self.assertIn("auto_approved", res)
+        self.assertIn("untrusted_pending", res)
+        self.assertIn("clear_nodes", res)
+
+
+class TestBoxApprovalsCli(unittest.TestCase):
+    """Test 'box approvals' and 'box approval' CLI commands."""
+
+    def test_box_approvals_check_json(self):
+        cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "approvals", "check", "--json"]
+        r = subprocess.run(cmd, capture_output=True, text=True)
+        self.assertEqual(r.returncode, 0)
+        data = json.loads(r.stdout)
+        self.assertTrue(data.get("ok"))
+        self.assertIn("approvals", data)
+        self.assertIsInstance(data["approvals"], list)
+
+    def test_box_approval_alias(self):
+        cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "approval", "--json"]
+        r = subprocess.run(cmd, capture_output=True, text=True)
+        self.assertEqual(r.returncode, 0)
+        data = json.loads(r.stdout)
+        self.assertTrue(data.get("ok"))
+
+    def test_box_approvals_auto_json(self):
+        cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "approvals", "auto", "--node", "pip", "--json"]
+        r = subprocess.run(cmd, capture_output=True, text=True)
+        self.assertEqual(r.returncode, 0)
+        data = json.loads(r.stdout)
+        self.assertTrue(data.get("ok"))
+
+
+class TestBoxCtlApprovals(unittest.TestCase):
+    """Test box-ctl.py allowlisted RPC actions."""
+
+    def test_box_ctl_approval_check(self):
+        cmd = [sys.executable, str(BIN_DIR / "box-ctl.py"), "approval-check"]
+        r = subprocess.run(cmd, capture_output=True, text=True)
+        self.assertEqual(r.returncode, 0)
+        data = json.loads(r.stdout)
+        self.assertTrue(data.get("ok"))
+        self.assertIn("approvals", data)
+
+    def test_box_ctl_approval_auto(self):
+        cmd = [sys.executable, str(BIN_DIR / "box-ctl.py"), "approval-auto", "pip"]
+        r = subprocess.run(cmd, capture_output=True, text=True)
+        self.assertEqual(r.returncode, 0)
+        data = json.loads(r.stdout)
+        self.assertTrue(data.get("ok"))
+
+
+class TestGravityApprovalIntegration(unittest.TestCase):
+    """Test loop break diagnostics for approvals."""
+
+    def test_diagnose_breaks_includes_approval_check(self):
+        breaks = gravity.diagnose_breaks()
+        self.assertIsInstance(breaks, list)
+        # Verify schema
+        for b in breaks:
+            self.assertIn("type", b)
+            self.assertIn("severity", b)
+            self.assertIn("detail", b)
+
+
+class TestMuseChatApiConnection(unittest.TestCase):
+    """Test muse-chat-api.py account mapping and connection."""
+
+    def test_muse_chat_api_approvals_command(self):
+        cmd = [sys.executable, str(BIN_DIR / "muse-chat-api.py"), "--account", "pip", "approvals"]
+        r = subprocess.run(cmd, capture_output=True, text=True)
+        self.assertEqual(r.returncode, 0)
+        self.assertIn("No pending approvals", r.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()