atlas-iac/testing/tests/test_hermes_gitea_pr_integration.py

"""Runtime and Flux integration contracts for the safe Atlas PR client."""

from __future__ import annotations

import importlib.util
import io
import json
import sys
import urllib.error
from pathlib import Path

import pytest
import yaml

ROOT = Path(__file__).parents[2]
CLIENT_PATH = ROOT / "services/hermes/scm-common/scripts/gitea_api.py"
HEAD_SHA = "465cf9146b05c174a2a8d310aff6c64be58277b6"
if str(CLIENT_PATH.parent) not in sys.path:
    sys.path.insert(0, str(CLIENT_PATH.parent))


def _load():
    spec = importlib.util.spec_from_file_location(
        "safe_gitea_api_integration", CLIENT_PATH
    )
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def test_dry_run_has_no_file_input_and_uses_no_token_or_network(monkeypatch, capsys):
    client = _load()
    monkeypatch.setattr(client, "read_token", lambda: pytest.fail("read token"))
    monkeypatch.setattr(
        client, "_safe_urlopen", lambda *_a, **_k: pytest.fail("network")
    )

    assert (
        client.main(
            [
                "--dry-run",
                "create-draft",
                "cassandra",
                "--base",
                "main",
                "--head",
                "hermes/fix",
                "--head-sha",
                HEAD_SHA,
                "--title",
                "Repair",
                "--body",
                "Evidence only",
            ]
        )
        == 0
    )
    output = json.loads(capsys.readouterr().out)
    assert output["operation"] == "create-draft"
    assert output["owner"] == "titan"
    assert "Evidence only" not in json.dumps(output)
    with pytest.raises(SystemExit):
        client.parse_args(
            [
                "create-draft",
                "cassandra",
                "--base",
                "main",
                "--head",
                "hermes/fix",
                "--head-sha",
                HEAD_SHA,
                "--title",
                "Repair",
                "--body-file",
                "/runtime-access/gitea-token",
            ]
        )


def test_http_error_path_redacts_token(monkeypatch, capsys):
    client = _load()
    monkeypatch.setattr(client, "read_token", lambda: "do-not-leak")

    def fail(*_args, **_kwargs):
        raise urllib.error.HTTPError(
            "https://scm.bstein.dev/api/v1/repos/titan/cassandra",
            403,
            "forbidden",
            {},
            io.BytesIO(b"Authorization: token do-not-leak"),
        )

    monkeypatch.setattr(client, "read", lambda *_a, **_k: fail())

    assert client.main(["read", "/api/v1/repos/titan/cassandra"]) == 1
    captured = capsys.readouterr()
    assert "do-not-leak" not in captured.err
    assert "credential was disclosed" in captured.err


def test_flux_manifest_isolates_vault_token_in_separate_broker_only():
    client_source = CLIENT_PATH.read_text(encoding="utf-8")
    assert "scm_broker_client" in client_source
    assert "GITEA_TOKEN" not in client_source

    deployment = yaml.safe_load(
        (ROOT / "services/hermes/agent-deployment.yaml").read_text(encoding="utf-8")
    )
    template = deployment["spec"]["template"]
    annotations = template["metadata"]["annotations"]
    assert not any("gitea" in key.lower() for key in annotations)
    runtime = next(
        volume
        for volume in template["spec"]["volumes"]
        if volume["name"] == "runtime-access"
    )
    assert runtime["emptyDir"]["medium"] == "Memory"

    expected = {"hermes", "terminal", "cli-lane-runner"}
    mounted = {
        container["name"]
        for container in template["spec"]["containers"]
        if any(
            mount["name"] == "atlas-pr-skill"
            and mount["mountPath"]
            == "/opt/data/workspace/skills/manage-atlas-pull-requests"
            and mount.get("readOnly") is True
            for mount in container.get("volumeMounts", [])
        )
    }
    assert mounted == expected

    kustomization = yaml.safe_load(
        (ROOT / "services/hermes/kustomization.yaml").read_text(encoding="utf-8")
    )
    common = yaml.safe_load(
        (ROOT / "services/hermes/scm-common/kustomization.yaml").read_text(
            encoding="utf-8"
        )
    )
    boundary = common["configMapGenerator"][0]
    assert boundary["name"] == "hermes-scm-boundary-v2"
    assert "gitea_api.py=scripts/gitea_api.py" in boundary["files"]
    assert "gitea_api_policy.py=scripts/gitea_api_policy.py" in boundary["files"]
    assert "scm_broker_client.py=scripts/scm_broker_client.py" in boundary["files"]
    assert not any("gitea_askpass" in item for item in boundary["files"])
    generator = next(
        item
        for item in kustomization["configMapGenerator"]
        if item["name"] == "hermes-atlas-pr-skill"
    )
    assert generator["files"] == [
        "SKILL.md=skills/manage-atlas-pull-requests/SKILL.md",
        "openai.yaml=skills/manage-atlas-pull-requests/agents/openai.yaml",
    ]

    broker = yaml.safe_load(
        (ROOT / "services/hermes-scm-broker/deployment.yaml").read_text(
            encoding="utf-8"
        )
    )
    assert broker["metadata"]["namespace"] == "hermes-scm"
    pod = broker["spec"]["template"]
    assert pod["spec"]["serviceAccountName"] == "hermes-scm-broker"
    assert pod["metadata"]["annotations"][
        "vault.hashicorp.com/agent-inject-secret-gitea-token"
    ] == "kv/data/atlas/hermes/developer-gitea"
    ledger = next(
        volume for volume in pod["spec"]["volumes"] if volume["name"] == "task-ledger"
    )
    assert ledger["persistentVolumeClaim"]["claimName"] == "hermes-scm-task-ledger"
    assert not any(volume.get("hostPath") for volume in pod["spec"]["volumes"])