operator
129 lines
4.6 KiB
Python
129 lines
4.6 KiB
Python
#!/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()
|