atlas-iac/testing/tests/test_hermes_oci_promote.py

"""Safety tests for the evidence-bound Hermes OCI release promotion."""

from __future__ import annotations

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

import pytest


ROOT = Path(__file__).resolve().parents[2]
SCRIPT = ROOT / "ci/scripts/hermes_oci_promote.py"
REVISION = "a" * 40
DIGEST = "sha256:" + "b" * 64
DESTINATION = f"registry.bstein.dev/bstein/hermes-agent:git-{REVISION}-build-17"
CONTENT_TYPE = "application/vnd.docker.distribution.manifest.v2+json"
MANIFEST = b'{"schemaVersion":2}'


def _load():
    spec = importlib.util.spec_from_file_location("hermes_oci_promote", SCRIPT)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


class Response(io.BytesIO):
    """Minimal context-managed urllib response."""

    def __init__(self, status: int, body: bytes = b"", headers=None):
        super().__init__(body)
        self.status = status
        self.headers = headers or {}

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        self.close()


def _candidate(headers=None, body: bytes = MANIFEST) -> Response:
    values = {
        "Docker-Content-Digest": DIGEST,
        "Content-Type": CONTENT_TYPE,
    }
    values.update(headers or {})
    return Response(200, body, values)


def _promote(module, opener):
    return module.promote_candidate(
        destination=DESTINATION,
        digest=DIGEST,
        source_revision=REVISION,
        build_number="17",
        username="robot",
        password="private",
        opener=opener,
    )


def test_promotion_copies_exact_candidate_to_release_tag() -> None:
    """Only the exact evidence digest is copied to the Flux-visible tag."""
    module = _load()
    calls: list[urllib.request.Request] = []
    responses = iter(
        [
            _candidate(),
            Response(404),
            Response(201, headers={"Docker-Content-Digest": DIGEST}),
        ]
    )

    def opener(request, timeout):
        calls.append(request)
        assert timeout in {20, 30}
        return next(responses)

    result = _promote(module, opener)
    assert [request.method for request in calls] == ["GET", "HEAD", "PUT"]
    assert calls[0].full_url.endswith(f"/manifests/git-{REVISION}-build-17")
    assert calls[2].full_url.endswith(f"/manifests/git-{REVISION}-build-17-release")
    assert calls[2].data == MANIFEST
    assert result["result"] == "published"
    assert result["digest"] == DIGEST
    assert "private" not in json.dumps(result)


def test_promotion_is_idempotent_for_the_same_digest() -> None:
    """A replay succeeds only when the immutable release already matches."""
    module = _load()
    responses = iter(
        [_candidate(), Response(200, headers={"Docker-Content-Digest": DIGEST})]
    )
    result = _promote(module, lambda *_args: next(responses))
    assert result["result"] == "already-present"


@pytest.mark.parametrize(
    ("changes", "message"),
    [
        ({"destination": "registry.invalid/hermes:latest"}, "destination"),
        ({"source_revision": "c" * 40}, "revision"),
        ({"build_number": "18"}, "build number"),
        ({"digest": "sha256:bad"}, "digest"),
        ({"username": ""}, "credentials"),
    ],
)
def test_promotion_rejects_unbound_inputs(changes: dict, message: str) -> None:
    """Destination, source, build, digest, and credentials fail closed."""
    module = _load()
    kwargs = {
        "destination": DESTINATION,
        "digest": DIGEST,
        "source_revision": REVISION,
        "build_number": "17",
        "username": "robot",
        "password": "private",
        "opener": lambda *_args: _candidate(),
    }
    kwargs.update(changes)
    with pytest.raises((ValueError, RuntimeError), match=message):
        module.promote_candidate(**kwargs)


@pytest.mark.parametrize(
    ("responses", "message"),
    [
        ([Response(401)], "candidate manifest returned HTTP 401"),
        ([_candidate({"Docker-Content-Digest": "sha256:" + "c" * 64})], "digest"),
        ([_candidate({"Content-Type": "text/plain"})], "content type"),
        (
            [
                _candidate(),
                Response(200, headers={"Docker-Content-Digest": "sha256:" + "c" * 64}),
            ],
            "another digest",
        ),
        ([_candidate(), Response(500)], "preflight returned HTTP 500"),
        (
            [_candidate(), Response(404), Response(500)],
            "release manifest returned HTTP 500",
        ),
        (
            [
                _candidate(),
                Response(404),
                Response(201, headers={"Docker-Content-Digest": "sha256:" + "c" * 64}),
            ],
            "changed",
        ),
    ],
)
def test_registry_failures_cannot_create_a_valid_release(
    responses: list[Response], message: str
) -> None:
    """Registry ambiguity or mismatch always fails the final release boundary."""
    module = _load()
    queued = iter(responses)
    with pytest.raises(RuntimeError, match=message):
        _promote(module, lambda *_args: next(queued))


def test_cli_reports_credential_free_success_and_error(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
    """The command emits bounded metadata and converts expected failures to JSON."""
    module = _load()
    digest_file = tmp_path / "digest"
    digest_file.write_text(DIGEST + "\n", encoding="utf-8")
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "promote",
            "--destination",
            DESTINATION,
            "--digest-file",
            str(digest_file),
            "--source-revision",
            REVISION,
            "--build-number",
            "17",
        ],
    )
    monkeypatch.setattr(
        module,
        "promote_candidate",
        lambda **_kwargs: {"result": "published", "digest": DIGEST},
    )
    assert module.main() == 0
    assert json.loads(capsys.readouterr().out)["result"] == "published"
    monkeypatch.setattr(
        module,
        "promote_candidate",
        lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("denied")),
    )
    assert module.main() == 1
    assert json.loads(capsys.readouterr().out) == {"error": "denied"}