guanghu-ice-heart/server-tools/enterprise-lighthouse/test_lighthouse.py

#!/usr/bin/env python3
import json
import os
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(__file__).parent
NODE_MAP = ROOT.parents[1] / "deployment" / "navigation-maps" / "AW-GZ-001.json"
BASE = "http://127.0.0.1:48031"
HEADERS = {
    "X-Lighthouse-Admin-Token": "test-token",
    "X-Guanghu-Principal-Id": "ICE-GL-ZY001:TEST-INSTANCE",
    "X-Guanghu-Subject-Kind": "persona",
    "X-Guanghu-Human-Anchor": "ICE-GL-INF",
}


def post(path, payload, headers=HEADERS):
    request = urllib.request.Request(
        BASE + path,
        data=json.dumps(payload).encode(),
        method="POST",
        headers=headers,
    )
    return json.load(urllib.request.urlopen(request))


def expect_error(code, path, payload, headers=HEADERS):
    try:
        post(path, payload, headers)
    except urllib.error.HTTPError as error:
        assert error.code == code, (error.code, json.load(error))
        return json.load(error)
    raise AssertionError(f"{path} unexpectedly succeeded")


with tempfile.TemporaryDirectory() as temp:
    env = {
        **os.environ,
        "LIGHTHOUSE_DB": f"{temp}/lighthouse.db",
        "LIGHTHOUSE_ADMIN_TOKEN": "test-token",
        "LIGHTHOUSE_PORT": "48031",
        "LIGHTHOUSE_NODE_MAP": str(NODE_MAP),
    }
    process = subprocess.Popen(
        [sys.executable, "lighthouse.py"],
        cwd=ROOT,
        env=env,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    try:
        for _ in range(30):
            try:
                health = json.load(urllib.request.urlopen(BASE + "/health", timeout=1))
                assert health["execution"] == "disabled"
                assert health["mode"] == "map-intent-agent-gate"
                break
            except OSError:
                time.sleep(.1)
        else:
            raise AssertionError("server did not start")

        status = json.load(urllib.request.urlopen(BASE + "/v1/status"))
        assert status["node_id"] == "AW-GZ-001"
        assert status["host_state"] == "ONLINE"
        assert {domain["state"] for domain in status["domains"] if domain["id"] != "DOMAIN-FIFTH"} == {
            "HOSTED_READ_ONLY"
        }
        assert {
            domain["steward_state"] for domain in status["domains"] if domain["id"] != "DOMAIN-FIFTH"
        } == {"UNBOUND"}
        assert {
            domain["mutation_state"] for domain in status["domains"] if domain["id"] != "DOMAIN-FIFTH"
        } == {"BLOCKED_UNTIL_PERSONA_STEWARD_BOUND"}
        assert all(
            domain["human_steward_id"] is None and domain["persona_steward_id"] is None
            for domain in status["domains"]
            if domain["id"] != "DOMAIN-FIFTH"
        )
        fifth = next(domain for domain in status["domains"] if domain["id"] == "DOMAIN-FIFTH")
        assert fifth["human_steward_id"] == "ICE-GL∞"
        assert fifth["persona_steward_id"] == "ICE-P-ZY001"
        assert fifth["responsibility_state"] == "EXTERNAL_SOVEREIGN_BOUND"
        assert status["node_counts"]["ACTIVE"] == 4

        nodes = json.load(urllib.request.urlopen(BASE + "/v1/nodes"))["nodes"]
        hosted = {node["domain_id"]: node for node in nodes}
        assert set(hosted) == {"DOMAIN-MAIN", "DOMAIN-SUB", "DOMAIN-ZERO", "DOMAIN-ZS"}
        assert all(node["state"] == "ACTIVE" for node in hosted.values())
        assert all(node["display_name"].startswith("AW-GZ-001") for node in hosted.values())

        intake = {
            "human_name": "Test",
            "email": "test@example.invalid",
            "server_ip": "203.0.113.8",
            "domain_id": "DOMAIN-ZS",
        }
        expect_error(423, "/v1/intakes", intake)

        nav = json.load(urllib.request.urlopen(BASE + "/v2/global-navigation-map"))
        assert nav["navigation_map"]["schema"] == "guanghu.enterprise-global-navigation-map/v2"
        assert nav["navigation_map"]["node_map"]["node_id"] == "AW-GZ-001"
        assert post("/v1/navigation-map/ack", {"map_hash": nav["map_hash"]})["ok"] is True

        expect_error(423, "/v1/intakes", intake)
        capsule = {
            "schema": "guanghu.intent-state-capsule/v1",
            "human_anchor": "ICE-GL-INF",
            "persona_id": "ICE-GL-ZY001",
            "task_intent": "Test the enterprise navigation and Agent gate.",
            "identity_boundary": "The public operations system is not a persona.",
            "established_facts": ["AW map is current."],
            "decisions": ["Use only a registered Agent."],
            "rejected_routes": ["raw shell"],
            "authorization_state": "test-only",
            "current_checkpoint": "map acknowledged",
            "next_action": "unlock inspector",
            "completion_definition": ["preflight is gated"],
            "evidence": ["deployment/navigation-maps/AW-GZ-001.json"],
        }
        restored = post("/v1/intent-state/restore", capsule)
        assert restored["ok"] is True
        assert json.load(urllib.request.urlopen(urllib.request.Request(
            BASE + "/v1/intakes",
            data=json.dumps(intake).encode(),
            method="POST",
            headers=HEADERS,
        )))["state"] == "PENDING_REVIEW"

        bootstrap_error = expect_error(423, "/v1/nodes/bootstrap", {
            "id": "AW-GZ-001",
            "domain_id": "DOMAIN-ZS",
            "display_name": "Enterprise root node",
            "server_ip": "203.0.113.8",
        })
        assert bootstrap_error["required"] == "ACTIVE_HUMAN_BACKED_PERSONA_STEWARD"

        preflight = {"action": "health_check", "target_node_id": "AW-GZ-001", "agent_id": "AW-INSPECTOR"}
        expect_error(423, "/v1/preflight", preflight)
        unlocked = post("/v1/agents/unlock", {
            "agent_id": "AW-INSPECTOR",
            "action": "health_check",
            "target_node_id": "AW-GZ-001",
        })
        preflight["unlock_id"] = unlocked["unlock_id"]
        expect_error(404, "/v1/preflight", preflight)
        assert post("/v1/agents/consume", {
            **preflight,
            "unlock_id": unlocked["unlock_id"],
        })["consumed"] is True
        expect_error(423, "/v1/preflight", preflight)

        expect_error(403, "/v1/nodes/bootstrap", {
            "id": "NODE-FIFTH-001",
            "domain_id": "DOMAIN-FIFTH",
            "display_name": "Not allowed",
            "server_ip": "203.0.113.8",
        })
        expect_error(400, "/v1/intakes", {"cmd": "unsafe"})

        other_headers = {
            **HEADERS,
            "X-Guanghu-Principal-Id": "ANOTHER-PERSONA:TEST",
        }
        expect_error(423, "/v1/intent-state/restore", capsule, other_headers)
        assert post("/v1/navigation-map/ack", {"map_hash": nav["map_hash"]}, other_headers)["ok"] is True
        expect_error(400, "/v1/intent-state/restore", {**capsule, "token": "forbidden"}, other_headers)
    finally:
        process.terminate()
        process.wait(timeout=5)

print("enterprise lighthouse tests passed")