net-kingdom/bounded-executor/tests/test_clients.py

"""approval-engine and audit-core clients, and the CLI's setup failure path (NK-WP-0056-T13). Fakes only."""

from __future__ import annotations

import importlib.util
import json
import pathlib
import sys
import tempfile
import threading
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

HERE = pathlib.Path(__file__).resolve().parent


def load(name):
    spec = importlib.util.spec_from_file_location(name, HERE.parent / f"{name}.py")
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


C = load("clients")
CLI = load("cli")


class Fake:
    def __init__(self):
        self.requests = []
        self.status = 200


def serve(fake):
    class H(BaseHTTPRequestHandler):
        def handle_one(self, method):
            raw = self.rfile.read(int(self.headers.get("Content-Length", "0") or 0))
            self.server_fake.requests.append((method, self.path, dict(self.headers), json.loads(raw) if raw else None))
            body = json.dumps({"id": "apr-1", "status": "approved"} if method == "GET" else
                              {"status": "consumed", "idempotent": False}).encode()
            self.send_response(self.server_fake.status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def do_GET(self):
            self.handle_one("GET")

        def do_POST(self):
            self.handle_one("POST")

        def log_message(self, *a):
            return

    H.server_fake = fake
    server = ThreadingHTTPServer(("127.0.0.1", 0), H)
    threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.02}, daemon=True).start()
    return server


class ClientTests(unittest.TestCase):
    def setUp(self):
        self.fake = Fake()
        self.server = serve(self.fake)
        self.addCleanup(self.server.shutdown)
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        self.token = pathlib.Path(tmp.name) / "token"
        self.token.write_text("synthetic-bearer\n")
        self.base = f"http://127.0.0.1:{self.server.server_port}"

    def test_approval_engine_get_and_consume(self):
        ae = C.ApprovalEngine(self.base, str(self.token))
        self.assertEqual(ae.get("apr-1")["status"], "approved")
        ae.consume("apr-1", "sha256:" + "a" * 64, "run-1")
        (_, p1, h1, _), (_, p2, _, b2) = self.fake.requests
        self.assertEqual((p1, p2), ("/v1/approvals/apr-1", "/v1/approvals/apr-1/consume"))
        self.assertEqual(h1["Authorization"], "Bearer synthetic-bearer")
        self.assertEqual(b2, {"request_digest": "sha256:" + "a" * 64, "decision_id": "run-1"})

    def test_approval_id_cannot_escape_the_path(self):
        ae = C.ApprovalEngine(self.base, str(self.token))
        for bad in ("../admin", "", "a/b"):
            with self.subTest(bad=bad), self.assertRaises(C.ClientError):
                ae.get(bad)
        self.assertEqual(self.fake.requests, [])

    def test_audit_core_idempotency_key_equals_event_id(self):
        event = {"id": "run:r1:started", "type": "execution.run.started", "source": "s", "subject": "x",
                 "tenant": "tenant:platform", "correlation_id": "r1", "occurred_at": "2026-11-03T09:30:00+00:00",
                 "data": {"a": 1}}
        C.AuditCore(self.base, str(self.token)).emit(event)
        ((method, path, headers, body),) = self.fake.requests
        self.assertEqual((method, path, headers["Idempotency-Key"]), ("POST", "/v1/events", "run:r1:started"))
        self.assertEqual(body, event)

    def test_errors_carry_status_only(self):
        self.fake.status = 403
        with self.assertRaises(C.ClientError) as caught:
            C.ApprovalEngine(self.base, str(self.token)).get("apr-1")
        self.assertEqual(str(caught.exception), "GET 403")


class CliTests(unittest.TestCase):
    def test_exit_codes(self):
        self.assertEqual(CLI.exit_code({"outcome": "completed"}), 0)
        self.assertEqual(CLI.exit_code({"outcome": "reverted"}), 3)
        self.assertEqual(CLI.exit_code({"outcome": "refused", "reason": "approval_entry"}), 4)
        self.assertEqual(CLI.exit_code({"outcome": "revert_failed"}), 5)
        self.assertEqual(CLI.exit_code({"outcome": "stopped", "reason": "audit_unavailable"}), 6)
        self.assertEqual(CLI.exit_code({"outcome": "completed", "reason": "receipt_not_in_custody"}), 6)

    def test_setup_failure_refuses_before_any_effect(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = pathlib.Path(tmp)
            (root / "jwt").write_text("synthetic-jwt")
            cfg = {"openbao": {"addr": "http://127.0.0.1:9", "k8s_role": "r", "jwt_file": str(root / "jwt")}}
            (root / "cfg.json").write_text(json.dumps(cfg))
            admission = CLI.X.REPO / "examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml"
            import contextlib, io
            wait = CLI.OPS.wait_reachable
            self.addCleanup(setattr, CLI.OPS, "wait_reachable", wait)
            CLI.OPS.wait_reachable = lambda addr, **kw: wait(addr, **{**kw, "timeout": 0.2, "interval": 0.1})
            with contextlib.redirect_stdout(io.StringIO()) as out:
                code = CLI.main(["--admission", str(admission), "--approval-id", "apr-1", "--config", str(root / "cfg.json")])
            self.assertEqual(code, 4)
            self.assertEqual(json.loads(out.getvalue())["reason"].split(":")[0], "setup")


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