atlas-iac/testing/tests/test_hermes_image_builder_branch_edges.py

"""Branch-edge coverage for the Hermes release and Harbor policy scripts."""

from __future__ import annotations

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

import pytest


REPO_ROOT = Path(__file__).resolve().parents[2]
RELEASE = REPO_ROOT / "ci/scripts/hermes_image_release.py"
HARBOR = (
    REPO_ROOT
    / "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py"
)


def _load(path: Path, name: str):
    spec = importlib.util.spec_from_file_location(name, path)
    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):
    """Context-managed urllib response used without network access."""

    def __init__(
        self,
        status: int,
        body: bytes = b"",
        headers: dict[str, str] | None = None,
    ) -> None:
        super().__init__(body)
        self.status = status
        self.headers = headers or {}


class FakeClient:
    """Ordered Harbor response fake with the pinned API origin."""

    origin = "https://registry.bstein.dev/api/v2.0"

    def __init__(self, responses) -> None:
        self.responses = list(responses)
        self.calls = []

    def request(self, method, path, payload=None):
        self.calls.append((method, path, payload))
        return self.responses.pop(0)


def _count(value: int) -> dict[str, str]:
    return {"X-Total-Count": str(value)}


def _publisher(module, *, immutable: bool = False, with_duration: bool = True) -> dict:
    access = [
        {"resource": "repository", "action": "pull"},
        {"resource": "repository", "action": "push"},
    ]
    if immutable:
        access.append({"resource": "immutable-tag", "action": "list"})
    robot = {
        "id": 41,
        "name": module.PUBLISH_ROBOT,
        "description": "publisher",
        "level": "system",
        "editable": True,
        "disable": False,
        "permissions": [
            {"kind": "project", "namespace": module.PROJECT, "access": access}
        ],
    }
    if with_duration:
        robot["duration"] = -1
    return robot


def test_release_redirect_handler_never_forwards_credentials() -> None:
    """The registry opener refuses to follow any redirect target."""
    module = _load(RELEASE, "release_redirect_edges")
    handler = module._NoRedirect()
    assert handler.redirect_request(None, None, 0, None, None, None) is None


def test_release_policy_read_requires_credentials_and_bounded_body() -> None:
    """The policy preflight rejects empty credentials and oversized responses."""
    module = _load(RELEASE, "release_policy_read_edges")
    with pytest.raises(RuntimeError, match="credentials are empty"):
        module._immutable_rules_response(username="", password="private")
    with pytest.raises(RuntimeError, match="size limit"):
        module._immutable_rules_response(
            username="robot",
            password="private",
            opener=lambda *_args: Response(200, b"x" * 1_048_577),
        )


@pytest.mark.parametrize("headers", [{}, {"X-Total-Count": "many"}])
def test_release_policy_rejects_missing_or_invalid_count(headers: dict) -> None:
    """A proxy that strips or mangles count evidence cannot prove completeness."""
    module = _load(RELEASE, "release_policy_count_edges")
    with pytest.raises(RuntimeError, match="total count"):
        module.verify_immutable_policy(
            username="robot",
            password="private",
            opener=lambda *_args, evidence=headers: Response(200, b"[]", evidence),
        )


@pytest.mark.parametrize("body", [b"{}", b"[1]"])
def test_release_policy_rejects_non_rule_list_shapes(body: bytes) -> None:
    """Valid JSON that is not a list of rule objects fails the preflight."""
    module = _load(RELEASE, "release_policy_shape_edges")
    with pytest.raises(RuntimeError, match="invalid shape"):
        module.verify_immutable_policy(
            username="robot",
            password="private",
            opener=lambda *_args, value=body: Response(200, value, _count(1)),
        )


def test_renderer_ignores_trailing_entry_without_digest_at_eof() -> None:
    """A dangling final image entry cannot shadow the one complete digest."""
    module = _load(RELEASE, "release_render_eof_edges")
    digest = "sha256:" + "6" * 64
    source = (
        "images:\n"
        f"  - name: {module.DEFAULT_IMAGE}\n"
        "    digest: sha256:" + "0" * 64 + "\n"
        f"  - name: {module.DEFAULT_IMAGE}\n"
        "    newTag: dangling\n"
    )
    rendered = module.render_kustomization(source, digest)
    assert f"digest: {digest}\n" in rendered
    assert rendered.endswith("newTag: dangling\n")


def test_harbor_verified_publisher_rejects_exact_identity_with_bad_scope() -> None:
    """An exact robot identity still fails verification on a malformed scope."""
    module = _load(HARBOR, "harbor_verified_scope_edges")
    robot = _publisher(module)
    robot["permissions"] = "bad"
    assert module._verified_publisher(robot) is False


def test_harbor_publisher_update_omits_absent_duration(monkeypatch) -> None:
    """A robot without a configured duration is preserved without inventing one."""
    module = _load(HARBOR, "harbor_duration_edges")
    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
    verified = _publisher(module, immutable=True)
    client = FakeClient(
        [
            (200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1)),
            (200, json.dumps(_publisher(module, with_duration=False)).encode(), {}),
            (200, b"", {}),
            (200, json.dumps(verified).encode(), {}),
        ]
    )
    assert module.ensure_publisher_can_read_rule(client) == 41
    method, path, payload = client.calls[2]
    assert (method, path) == ("PUT", "/robots/41")
    assert "duration" not in payload


def test_harbor_publisher_verify_retries_past_invalid_json(monkeypatch) -> None:
    """One stale unreadable reread retries instead of passing or failing outright."""
    module = _load(HARBOR, "harbor_verify_retry_edges")
    sleeps = []
    monkeypatch.setattr(module.time, "sleep", sleeps.append)
    verified = _publisher(module, immutable=True)
    client = FakeClient(
        [
            (200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1)),
            (200, json.dumps(_publisher(module)).encode(), {}),
            (200, b"", {}),
            (200, b"not-json", {}),
            (200, json.dumps(verified).encode(), {}),
        ]
    )
    assert module.ensure_publisher_can_read_rule(client) == 41
    assert sleeps == [1]


def test_harbor_rule_verify_requires_the_created_rule_id(monkeypatch) -> None:
    """A matching rule under a different ID is a stale read, not creation proof."""
    module = _load(HARBOR, "harbor_rule_id_retry_edges")
    sleeps = []
    monkeypatch.setattr(module.time, "sleep", sleeps.append)
    location = "/api/v2.0/projects/bstein/immutabletagrules/23"
    client = FakeClient(
        [
            (200, b"[]", _count(0)),
            (201, b"", {"Location": location}),
            (200, json.dumps([{"id": 99, **module.EXPECTED_RULE}]).encode(), _count(1)),
            (200, json.dumps([{"id": 23, **module.EXPECTED_RULE}]).encode(), _count(1)),
        ]
    )
    assert module.ensure_rule(client) == 23
    assert sleeps == [1]


def test_harbor_main_reraises_after_bounded_unavailable_retries(
    monkeypatch, tmp_path: Path
) -> None:
    """Persistent Harbor unavailability surfaces instead of looping forever."""
    module = _load(HARBOR, "harbor_main_retry_edges")
    password_file = tmp_path / "password"
    password_file.write_text("private\n", encoding="utf-8")
    monkeypatch.setenv("HARBOR_API_ORIGIN", module.EXPECTED_ORIGIN)
    monkeypatch.setenv("HARBOR_ADMIN_PASSWORD_FILE", str(password_file))
    monkeypatch.setattr(module, "HarborClient", lambda *_args: object())

    def never_ready(_client):
        raise module.HarborUnavailable("still warming")

    monkeypatch.setattr(module, "ensure_rule", never_ready)
    sleeps = []
    monkeypatch.setattr(module.time, "sleep", sleeps.append)
    with pytest.raises(module.HarborUnavailable, match="still warming"):
        module.main()
    assert len(sleeps) == 11