atlas-iac/testing/tests/test_hermes_cluster_read_plugin.py

"""Read-only cluster tool: GET-only, secrets-refusing, bounded."""

from __future__ import annotations

import importlib.util
import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
PLUGIN = ROOT / "services/hermes/plugins/cluster-read"
SPEC = importlib.util.spec_from_file_location("hermes_cluster_read", PLUGIN / "__init__.py")
assert SPEC and SPEC.loader
mod = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = mod
SPEC.loader.exec_module(mod)


def test_paths_are_shaped_and_bounded():
    assert mod._build_path({"resource": "pods"}) == "/api/v1/pods?limit=200"
    assert mod._build_path({"resource": "pods", "namespace": "hermes"}) == "/api/v1/namespaces/hermes/pods?limit=200"
    assert mod._build_path({"resource": "pods", "namespace": "hermes", "name": "x-1"}) == "/api/v1/namespaces/hermes/pods/x-1"
    assert mod._build_path({"resource": "deployments", "group": "apps"}) == "/apis/apps/v1/deployments?limit=200"
    assert (
        mod._build_path({"resource": "kustomizations", "group": "kustomize.toolkit.fluxcd.io", "version": "v1"})
        == "/apis/kustomize.toolkit.fluxcd.io/v1/kustomizations?limit=200"
    )
    assert "labelSelector=app%3Dx" in mod._build_path({"resource": "pods", "label_selector": "app=x"})


def test_secrets_are_refused_everywhere():
    import pytest

    for resource in ("secrets", "secret-things", "mysecrets"):
        with pytest.raises(PermissionError):
            mod._build_path({"resource": resource})
    body = json.loads(mod._handle_cluster_read({"resource": "secrets"}))
    assert "not readable" in body["error"]


def test_malformed_segments_are_rejected():
    import pytest

    # Falsy values mean "omitted" by contract; every present value is validated.
    for bad in ("Pods", "a/b", "a b", "x" * 300, "../etc"):
        with pytest.raises((ValueError, PermissionError)):
            mod._build_path({"resource": "pods", "namespace": bad})
    with pytest.raises(ValueError):
        mod._build_path({"resource": "pods", "label_selector": "a b?c\n"})


def test_handler_is_get_only_and_bounded(monkeypatch, tmp_path):
    token = tmp_path / "token"
    token.write_text("tok")
    ca = tmp_path / "ca.crt"
    ca.write_text("cert")
    monkeypatch.setattr(mod, "TOKEN_FILE", token)
    monkeypatch.setattr(mod, "CA_FILE", ca)
    seen = {}

    class FakeResponse:
        def __init__(self, data): self._data = data
        def read(self, n): return self._data[:n]
        def __enter__(self): return self
        def __exit__(self, *a): return False

    def fake_open(request, timeout=None, context=None):
        seen["method"] = request.get_method()
        seen["url"] = request.full_url
        seen["auth"] = request.get_header("Authorization")
        payload = {"kind": "PodList", "items": [{"metadata": {"name": "p", "managedFields": [{"x": 1}],
                   "annotations": {"kubectl.kubernetes.io/last-applied-configuration": "big"}}}]}
        return FakeResponse(json.dumps(payload).encode())

    monkeypatch.setattr(mod, "ssl", type("S", (), {"create_default_context": staticmethod(lambda cafile=None: None)}))
    monkeypatch.setattr(mod.urllib.request, "urlopen", fake_open)
    out = json.loads(mod._handle_cluster_read({"resource": "pods", "namespace": "hermes"}))
    assert seen["method"] == "GET" and seen["auth"] == "Bearer tok"
    assert seen["url"].startswith("https://kubernetes.default.svc/api/v1/namespaces/hermes/pods")
    assert out["kind"] == "PodList"
    assert "managedFields" not in json.dumps(out)
    assert "last-applied" not in json.dumps(out)

    def huge_open(request, timeout=None, context=None):
        return FakeResponse(b'{"kind":"List","items":["' + b"x" * (mod.MAX_RESPONSE_BYTES + 100) + b'"]}')

    monkeypatch.setattr(mod.urllib.request, "urlopen", huge_open)
    out = json.loads(mod._handle_cluster_read({"resource": "pods"}))
    assert out.get("truncated") is True or out.get("error")


def test_errors_never_raise(monkeypatch, tmp_path):
    token = tmp_path / "token"
    token.write_text("tok")
    ca = tmp_path / "ca.crt"
    ca.write_text("cert")
    monkeypatch.setattr(mod, "TOKEN_FILE", token)
    monkeypatch.setattr(mod, "CA_FILE", ca)
    monkeypatch.setattr(mod, "ssl", type("S", (), {"create_default_context": staticmethod(lambda cafile=None: None)}))

    def down(request, timeout=None, context=None):
        raise OSError("down")

    monkeypatch.setattr(mod.urllib.request, "urlopen", down)
    out = json.loads(mod._handle_cluster_read({"resource": "pods"}))
    assert "unavailable" in out["error"]


def test_registration_shape_and_availability(monkeypatch, tmp_path):
    calls = {}

    class Ctx:
        def register_tool(self, **kwargs): calls.update(kwargs)

    mod.register(Ctx())
    assert calls["name"] == "cluster_read" and calls["toolset"] == "cluster"
    assert calls["is_async"] is False and calls["requires_env"] == []
    assert "Secrets" in calls["description"]
    monkeypatch.setattr(mod, "TOKEN_FILE", tmp_path / "missing")
    assert calls["check_fn"]() is False


def test_rbac_manifest_is_read_only_and_secretless():
    rbac = (ROOT / "services/hermes/chat-cluster-read-rbac.yaml").read_text()
    assert "secrets" not in rbac
    for verb in ("create", "update", "patch", "delete", "escalate", "impersonate"):
        assert verb not in rbac
    assert "name: view" in rbac and "hermes-chat" in rbac
    config = (ROOT / "services/hermes/chat-configmap.yaml").read_text()
    assert "- cluster-read" in config and "cluster," in config