atlas-iac/testing/tests/test_hermes_gitea_pr_integration.py
2026-08-16 20:18:49 -03:00

148 lines
4.4 KiB
Python

"""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/scripts/gitea_api.py"
HEAD_SHA = "465cf9146b05c174a2a8d310aff6c64be58277b6"
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"] == "atlas"
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/atlas/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/atlas/cassandra"]) == 1
captured = capsys.readouterr()
assert "do-not-leak" not in captured.err
assert "HTTP 403" in captured.err
def test_flux_manifest_projects_runtime_vault_token_and_skill_only():
client_source = CLIENT_PATH.read_text(encoding="utf-8")
assert "/runtime-access/gitea-token" 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 annotations["vault.hashicorp.com/agent-inject-secret-gitea-token"] == (
"kv/data/atlas/hermes/developer-gitea"
)
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")
)
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",
]