box/tests/test_approvals.py

#!/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)
        self.assertTrue(approvals.is_trusted_target("34.139.37.135"))
        self.assertTrue(approvals.is_trusted_target("status.muse-dev.online"))
        self.assertTrue(approvals.is_trusted_target("1.1.1.1"))
        self.assertFalse(approvals.is_trusted_target("8.8.8.8"))
        self.assertFalse(approvals.is_trusted_target("malicious-site.com"))
        self.assertFalse(approvals.is_trusted_target("evilmuse-dev.online"))
        self.assertFalse(approvals.is_trusted_target(
            "evil.com", "operator-main wants to reach evil.com for status.muse-dev.online"))
        self.assertFalse(approvals.is_trusted_target(None, "status.muse-dev.online"))

    def test_redact_sensitive(self):
        s1 = "Connecting with Bearer ya29.a0AfH6SMAKskd9238jdf"
        self.assertEqual(approvals.redact_sensitive(s1), "Connecting with Bearer [REDACTED]")
        s2 = "My api_key: secret12345678"
        self.assertEqual(approvals.redact_sensitive(s2), "My api_key: [REDACTED]")
        s3 = "JWT token eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0"
        self.assertIn("[JWT-REDACTED]", approvals.redact_sensitive(s3))

    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)
            self.assertIn("input_waits", 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)


class TestApprovalsReplySafety(unittest.TestCase):
    """Test reply policy enforcement on Main Chat."""

    def test_reply_main_chat_refusal(self):
        cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "approvals", "reply", "pip", "hello test"]
        r = subprocess.run(cmd, capture_output=True, text=True)
        # Should refuse with exit code 2 when target is Main Chat without --allow-main-chat
        self.assertEqual(r.returncode, 2)
        self.assertIn("Refusing reply by sidechat-first policy", r.stderr + r.stdout)


class TestKeyApprovalsAndPasskey(unittest.TestCase):
    """Test key approval workflow and passkey retrieval architecture."""

    def test_request_and_resolve_key_approval(self):
        req = approvals.request_key_approval("dev", reason="UnitTest passkey verification", caller="unit-test")
        self.assertTrue(req.get("ok"))
        self.assertEqual(req.get("node"), "dev")

        # Verify active in check_node_key_request
        pending = approvals.check_node_key_request("dev")
        self.assertIsNotNone(pending)
        self.assertEqual(pending.get("reason"), "UnitTest passkey verification")

        # Inspect should report KEY_APPROVAL
        info = approvals.inspect_node_approvals("dev")
        self.assertEqual(info.get("status"), "KEY_APPROVAL")
        self.assertTrue(info.get("has_pending"))

        # Resolve via allow_node_approval
        res = approvals.allow_node_approval("dev", caller="unit-test")
        self.assertTrue(res.get("ok"))
        self.assertEqual(res.get("type"), "key_approval")
        self.assertEqual(res.get("decision"), "allow")

        # After resolution, pending should be cleared
        cleared = approvals.check_node_key_request("dev")
        self.assertIsNone(cleared)

    def test_box_passkey_info_json(self):
        cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "passkey", "--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.assertEqual(data.get("operator_pin"), "3128")
        self.assertIn("key_location", data)
        self.assertIn("canonical_path", data["key_location"])
        self.assertEqual(data["key_location"]["canonical_path"], "/srv/box/passkey.txt")

    def test_box_passkey_fetch_json(self):
        cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "passkey", "fetch", "--json"]
        r = subprocess.run(cmd, capture_output=True, text=True)
        self.assertEqual(r.returncode, 0)
        data = json.loads(r.stdout)
        self.assertIn("operator_pin", data)
        self.assertEqual(data.get("operator_pin"), "3128")
        self.assertEqual(data.get("vm_host"), "34.139.37.135")
        self.assertEqual(data.get("path"), "/srv/box/passkey.txt")
        self.assertIn("operator_command", data)

    def test_box_lookup_key(self):
        cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "lookup", "key", "--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.assertEqual(data.get("operator_pin"), "3128")

    def test_cli_request_key_lifecycle(self):
        # 1. Request key
        r_req = subprocess.run([
            sys.executable, str(BIN_DIR / "super-cli.py"),
            "approvals", "request-key", "dev", "--reason", "CLI lifecycle test", "--json"
        ], capture_output=True, text=True)
        self.assertEqual(r_req.returncode, 0)
        req_data = json.loads(r_req.stdout)
        self.assertTrue(req_data.get("ok"))

        # 2. Check shows KEY_APPROVAL
        r_check = subprocess.run([
            sys.executable, str(BIN_DIR / "super-cli.py"),
            "approvals", "check", "--node", "dev", "--json"
        ], capture_output=True, text=True)
        self.assertEqual(r_check.returncode, 0)
        check_data = json.loads(r_check.stdout)
        dev_app = next(a for a in check_data["approvals"] if a["node"] == "dev")
        self.assertEqual(dev_app["status"], "KEY_APPROVAL")

        # 3. Deny key
        r_deny = subprocess.run([
            sys.executable, str(BIN_DIR / "super-cli.py"),
            "approvals", "deny", "dev", "--json"
        ], capture_output=True, text=True)
        self.assertEqual(r_deny.returncode, 0)
        deny_data = json.loads(r_deny.stdout)
        self.assertTrue(deny_data.get("ok"))
        self.assertEqual(deny_data.get("decision"), "deny")


if __name__ == "__main__":
    unittest.main()