atlas-iac/testing/tests/test_hermes_webui_hux_backend_e2e.py

"""Exercise current packaged HUX HTTP responses through the browser validators."""

from __future__ import annotations

from http.client import HTTPConnection
import hashlib
import hmac
import json
from pathlib import Path

from hux_node_gate import require_node
import subprocess
import sys
import threading


ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles/hermes-hux-foundation"
NODE = ROOT / "testing/tests/test_hermes_webui_hux_backend_node.js"
if str(FOUNDATION) not in sys.path:
    sys.path.insert(0, str(FOUNDATION))

from hux import contracts, foundation, identity  # noqa: E402
from hux.http import serve  # noqa: E402
from hux.server import build_router  # noqa: E402


def _request(port: int, headers: dict[str, str], method: str, path: str,
             body: dict | None = None, extra: dict[str, str] | None = None) -> tuple[int, dict, dict]:
    connection = HTTPConnection("127.0.0.1", port, timeout=5)
    payload = b"" if body is None else json.dumps(body).encode()
    request_headers = {**headers, **(extra or {})}
    if body is not None:
        request_headers["Content-Type"] = "application/json"
    connection.request(method, path, body=payload, headers=request_headers)
    response = connection.getresponse()
    result = response.status, json.loads(response.read()), dict(response.getheaders())
    connection.close()
    return result


def test_real_backend_http_contracts_are_consumable_by_the_webui(tmp_path: Path) -> None:
    require_node()
    key = b"k" * 32
    key_path = tmp_path / "context-key"
    key_path.write_bytes(key)
    key_path.chmod(0o600)
    evidence_key = tmp_path / "release-key"
    evidence_key.write_text("e" * 32)
    evidence_key.chmod(0o400)
    evidence_policy = tmp_path / "release-policy.json"
    evidence_policy.write_text(json.dumps({
        "schema": "hux.release_evidence_policy.v1", "max_evidence_age_seconds": 3600,
        "workloads": {"hermes-webui": {
            "review_url_prefix": "https://git.bstein.dev/atlas/titan-iac/",
            "jenkins_job_url": "https://jenkins.bstein.dev/job/hermes-webui",
            "image_repository": "registry.bstein.dev/bstein/hermes-webui",
            "flux_kustomization": "hermes-chat", "health_url": "https://chat.bstein.dev/healthz",
        }},
    }))
    evidence_policy.chmod(0o444)
    slot = "slot-3"
    subject = "usr_" + hmac.new(
        key, b"hux.subject.id.v1\0" + slot.encode(), hashlib.sha256
    ).hexdigest()
    flags = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
    env = {
        "HUX_FLAGS": flags,
        "HUX_RELAY_KEY": "relay-key",
        "HUX_CONTEXT_KEY_FILE": str(key_path),
        "HUX_RELEASE_EVIDENCE_KEY_FILE": str(evidence_key),
        "HUX_RELEASE_EVIDENCE_POLICY_FILE": str(evidence_policy),
        "HUX_SWITCHYARD_ROUTE_CATALOG": (
            "atlas/manual/codex/gpt-5,atlas/manual/claude/opus,atlas/manual/local/qwen-14b"
        ),
    }
    router = build_router(tmp_path / "data", env)
    server = serve(router, "127.0.0.1", 0)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    port = server.server_address[1]
    headers = {
        "X-Hermes-Tenant-Identity": slot,
        "X-Hux-Subject": subject,
        "X-Hux-Surface": "chat",
        "X-Hux-Trust": "relay",
        "X-Hux-Relay-Key": "relay-key",
    }
    who = identity.Identity(slot, subject, "chat", "relay")
    raw_session, project_source = "webui-session-e2e", "profile:default"
    scope = {
        "raw_session_id": raw_session,
        "project_source": project_source,
        "session_id": foundation.derive_context_id(key, "session", who, raw_session),
        "conversation_id": foundation.derive_context_id(key, "conversation", who, raw_session),
        "project_id": foundation.derive_context_id(key, "project", who, project_source),
    }
    try:
        status, capabilities, _ = _request(port, headers, "GET", "/hux/v1/capabilities")
        assert status == 200
        bootstrap_status, bootstrap, response_headers = _request(
            port, headers, "POST", "/hux/v1/context/bootstrap", scope,
            {"Idempotency-Key": "webui:e2e:bootstrap"},
        )
        assert bootstrap_status == 201 and response_headers["Cache-Control"] == "no-store"
        project_id, conversation_id = scope["project_id"], scope["conversation_id"]
        base = f"/hux/v1/projects/{project_id}/conversations/{conversation_id}"
        status, mode_page, _ = _request(port, headers, "GET", "/hux/v1/modes")
        assert status == 200
        status, selection, _ = _request(port, headers, "PUT", base + "/mode", {
            "project_id": project_id, "mode": "fast", "advanced": False,
        }, {"If-Match": "0", "Idempotency-Key": "webui:e2e:mode"})
        assert status == 200
        status, media, _ = _request(port, headers, "GET", base + "/multimodal/items")
        assert status == 200
        status, intent, _ = _request(port, headers, "POST", base + "/capture-intents", {
            "project_id": project_id, "source": "screen", "purpose": "Share a reviewed window",
        }, {"If-Match": "0", "Idempotency-Key": "webui:e2e:capture"})
        assert status == 201 and intent["execution_allowed"] is False
        suggestion_base = base + "/suggestions"
        status, evaluation, _ = _request(port, headers, "POST", suggestion_base + "/evaluate", {
            "context": "first_session", "no_store": False,
        }, {"Idempotency-Key": "webui:e2e:suggestion"})
        assert status == 200 and evaluation["stored"] is True
        suggestion_id = evaluation["suggestion"]["id"]
        status, decision, _ = _request(
            port, headers, "POST", suggestion_base + f"/{suggestion_id}/decisions",
            {"decision": "dismissed", "clicked": True},
            {"If-Match": str(evaluation["revision"]), "Idempotency-Key": "webui:e2e:decision"},
        )
        assert status == 200 and decision["decision"] == "dismissed"
        assert _request(port, headers, "GET", suggestion_base + "/states")[0] == 200
        release_base = base + "/releases"
        status, releases, _ = _request(port, headers, "GET", release_base)
        assert status == 200
        bridge = {
            "rawSessionId": raw_session,
            "context": {"schema": "hux.webui_context.v1", "webui_session_id": raw_session,
                **{key_name: scope[key_name] for key_name in ("session_id", "conversation_id", "project_id")},
                "project_source": project_source, "identity": who.record()},
            "bootstrap": {"status": bootstrap_status, "body": bootstrap},
            "capabilities": capabilities, "modes": mode_page, "selection": selection,
            "media": media, "evaluation": evaluation, "releases": releases,
        }
        result = subprocess.run(
            ["node", str(NODE)], input=json.dumps(bridge), text=True,
            capture_output=True, check=False, timeout=10, cwd=ROOT,
        )
        assert result.returncode == 0, result.stdout + result.stderr
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


def test_backend_e2e_sources_remain_bounded() -> None:
    assert len(Path(__file__).read_text().splitlines()) <= 500
    assert len(NODE.read_text().splitlines()) <= 500