atlas-iac/testing/tests/test_hermes_image_builder_coverage.py

"""Boundary coverage for the three security-critical image-lane helpers."""

from __future__ import annotations

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

import pytest


REPO_ROOT = Path(__file__).resolve().parents[2]
RUNNER = REPO_ROOT / "dockerfiles/hermes-kaniko-heredoc-runner.py"
DOCKERFILE = REPO_ROOT / "dockerfiles/Dockerfile.hermes-agent"
HARBOR = (
    REPO_ROOT
    / "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py"
)
TRIGGER = REPO_ROOT / "services/hermes/scripts/jenkins_image_build_trigger.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 {}

    def __enter__(self):
        return self

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


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


def _robot_list(module, *, robot_id=41, name=None):
    return json.dumps(
        [{"id": robot_id, "name": name or module.PUBLISH_ROBOT}]
    ).encode()


def test_runner_rejects_directive_and_heredoc_boundaries(tmp_path: Path) -> None:
    """Malformed parser state, bodies, sizes, and indices all fail closed."""
    module = _load(RUNNER, "runner_boundary_coverage")
    assert module._escape_character([""]) == "\\"
    with pytest.raises(ValueError, match="multiple"):
        module._escape_character(["# escape=\\", "# escape=\\"])
    with pytest.raises(ValueError, match="unsupported"):
        module._escape_character(["# escape=^"])
    with pytest.raises(ValueError, match="unterminated.*continuation"):
        module._logical_instruction(["R\\"], 0, "\\")
    with pytest.raises(ValueError, match="unterminated node"):
        module.extract_blocks("RUN node <<'NODE'\nbody")

    source = DOCKERFILE.read_text(encoding="utf-8")
    marker = "RUN node <<'NODE'\n"
    body_start = source.index(marker) + len(marker)
    body_end = source.index("\nNODE", body_start)
    empty = source[:body_start] + source[body_end + 1 :]
    with pytest.raises(ValueError, match="empty node"):
        module.extract_blocks(empty)

    dockerfile = tmp_path / "Dockerfile"
    dockerfile.write_text("", encoding="utf-8")
    with pytest.raises(ValueError, match="size"):
        module.replay(dockerfile, 1)
    with dockerfile.open("wb") as stream:
        stream.truncate(module.MAX_DOCKERFILE_BYTES + 1)
    with pytest.raises(ValueError, match="size"):
        module.replay(dockerfile, 1)
    dockerfile.write_text(source, encoding="utf-8")
    for index in (0, 10):
        with pytest.raises(ValueError, match="index"):
            module.replay(dockerfile, index)


def test_runner_main_dispatches_the_exact_path_and_index(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """The command-line wrapper cannot omit either bounded input."""
    module = _load(RUNNER, "runner_main_coverage")
    dockerfile = tmp_path / "Dockerfile"
    seen = []
    monkeypatch.setattr(module, "replay", lambda path, index: seen.append((path, index)))
    monkeypatch.setattr(
        sys,
        "argv",
        ["runner", "--dockerfile", str(dockerfile), "--block-index", "4"],
    )
    assert module.main() == 0
    assert seen == [(dockerfile, 4)]


def test_trigger_redirect_wrapper_accepts_only_plugin_303(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The low-level opener returns a real response or the one expected redirect."""
    module = _load(TRIGGER, "trigger_opener_coverage")
    assert module._NoRedirect().redirect_request(None, None, 0, None, None, None) is None
    request = module.urllib.request.Request(module.JENKINS_BUILD_URL)

    class Opener:
        def __init__(self, result) -> None:
            self.result = result

        def open(self, _request, timeout):
            assert timeout == 9
            if isinstance(self.result, BaseException):
                raise self.result
            return self.result

    success = Response(201)
    monkeypatch.setattr(
        module.urllib.request, "build_opener", lambda *_args: Opener(success)
    )
    assert module._open_without_redirect(request, 9) is success
    redirect = module.urllib.error.HTTPError(
        request.full_url, 303, "queued", {"Location": "/queue/item/1/"}, None
    )
    monkeypatch.setattr(
        module.urllib.request, "build_opener", lambda *_args: Opener(redirect)
    )
    assert module._open_without_redirect(request, 9) is redirect
    denied = module.urllib.error.HTTPError(request.full_url, 403, "denied", {}, None)
    monkeypatch.setattr(
        module.urllib.request, "build_opener", lambda *_args: Opener(denied)
    )
    with pytest.raises(urllib.error.HTTPError):
        module._open_without_redirect(request, 9)


def test_trigger_main_reports_safe_success_and_errors(
    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
    """CLI output contains safe metadata and converts expected failures to JSON."""
    module = _load(TRIGGER, "trigger_main_coverage")
    revision = "a" * 40
    monkeypatch.setattr(sys, "argv", ["trigger", revision])
    monkeypatch.setattr(
        module,
        "trigger_build",
        lambda value: {"job": module.JOB_NAME, "source_revision": value, "status": 201},
    )
    assert module.main() == 0
    assert json.loads(capsys.readouterr().out)["source_revision"] == revision
    monkeypatch.setattr(
        module, "trigger_build", lambda _value: (_ for _ in ()).throw(OSError("closed"))
    )
    assert module.main() == 1
    assert json.loads(capsys.readouterr().out) == {"error": "closed"}


def test_harbor_http_client_bounds_requests_and_failures(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Credentials stay same-origin and all response/error paths remain bounded."""
    module = _load(HARBOR, "harbor_client_coverage")
    assert module.NoRedirect().redirect_request(None, None, 0, None, None, None) is None

    class Opener:
        result = Response(200, b"ok", {"X-Test": "yes"})

        def open(self, request, timeout):
            self.request = request
            assert timeout == 20
            if isinstance(self.result, BaseException):
                raise self.result
            return self.result

    opener = Opener()
    monkeypatch.setattr(module.urllib.request, "build_opener", lambda *_args: opener)
    client = module.HarborClient(module.EXPECTED_ORIGIN + "/", "admin", "private")
    status, body, headers = client.request("POST", "/rules", {"value": 1})
    assert (status, body, headers["X-Test"]) == (200, b"ok", "yes")
    assert opener.request.get_header("Content-type") == "application/json"

    opener.result = module.urllib.error.HTTPError(
        module.EXPECTED_ORIGIN, 404, "missing", {}, io.BytesIO(b"missing")
    )
    assert client.request("GET", "/missing")[:2] == (404, b"missing")
    opener.result = module.urllib.error.URLError("offline")
    with pytest.raises(module.HarborUnavailable):
        client.request("GET", "/rules")
    opener.result = Response(200, b"x" * (module.MAX_RESPONSE + 1))
    with pytest.raises(RuntimeError, match="size"):
        client.request("GET", "/rules")


@pytest.mark.parametrize("body", [b"\xff", b"not-json", b"{}", b"[1]"])
def test_harbor_json_lists_reject_invalid_shapes(body: bytes) -> None:
    """Malformed encodings, JSON, objects, and scalar members are never lists."""
    module = _load(HARBOR, f"harbor_json_{body!r}")
    with pytest.raises(RuntimeError, match="invalid"):
        module._json_list(body, "fixture")


@pytest.mark.parametrize(
    "robot,match",
    [
        ({"permissions": "bad"}, "permissions"),
        ({"permissions": [1]}, "permissions"),
        ({"permissions": []}, "one existing"),
        (
            {
                "permissions": [
                    {"kind": "project", "namespace": "bstein", "access": []},
                    {"kind": "project", "namespace": "bstein", "access": []},
                ]
            },
            "one existing",
        ),
        (
            {
                "permissions": [
                    {"kind": "project", "namespace": "bstein", "access": "bad"}
                ]
            },
            "access",
        ),
        (
            {
                "permissions": [
                    {"kind": "project", "namespace": "bstein", "access": []}
                ]
            },
            "pull/push",
        ),
    ],
)
def test_publisher_scope_rejects_malformed_or_incomplete_permissions(
    robot: dict, match: str
) -> None:
    """Robot mutations require one structurally exact existing push scope."""
    module = _load(HARBOR, f"harbor_scope_{match}")
    with pytest.raises(RuntimeError, match=match):
        module._publisher_scope(robot)
    assert module._verified_publisher(robot) is False


@pytest.mark.parametrize(
    "responses,match,unavailable",
    [
        ([(503, b"", {})], "503", True),
        ([(403, b"", {})], "403", False),
        ([(200, b"[]", _count(0))], "exactly one", False),
        ([(200, b"[]", _count(0))], "exactly one", False),
        ([(200, b'[{\"name\":\"robot$jenkins-pipelines\",\"id\":0}]', _count(1))], "valid ID", False),
    ],
)
def test_publisher_list_rejects_status_count_and_id(
    responses, match: str, unavailable: bool
) -> None:
    """Publisher discovery distinguishes readiness from permanent policy errors."""
    module = _load(HARBOR, f"harbor_publisher_list_{match}_{unavailable}")
    error = module.HarborUnavailable if unavailable else RuntimeError
    with pytest.raises(error, match=match):
        module.ensure_publisher_can_read_rule(FakeClient(responses))


@pytest.mark.parametrize(
    "second,match,unavailable",
    [
        ((503, b"", {}), "503", True),
        ((403, b"", {}), "403", False),
        ((200, b"not-json", {}), "invalid robot JSON", False),
        ((200, b"[]", {}), "invalid shape", False),
        ((200, b'{}', {}), "not active and exact", False),
    ],
)
def test_publisher_read_rejects_status_and_identity(
    second, match: str, unavailable: bool
) -> None:
    """The detailed robot reread must be available, valid, and exact."""
    module = _load(HARBOR, f"harbor_publisher_read_{match}_{unavailable}")
    responses = [(200, _robot_list(module), _count(1)), second]
    error = module.HarborUnavailable if unavailable else RuntimeError
    with pytest.raises(error, match=match):
        module.ensure_publisher_can_read_rule(FakeClient(responses))


def test_publisher_update_rejects_duration_reference_and_status(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Only the preserved scope and a successful policy PUT can be accepted."""
    module = _load(HARBOR, "harbor_publisher_update_coverage")
    for duration in ("forever",):
        responses = [
            (200, _robot_list(module), _count(1)),
            (200, json.dumps(_robot(module, duration=duration)).encode(), {}),
        ]
        with pytest.raises(RuntimeError, match="duration"):
            module.ensure_publisher_can_read_rule(FakeClient(responses))

    robot = _robot(module)
    monkeypatch.setattr(module, "_publisher_scope", lambda _robot: ({}, []))
    responses = [
        (200, _robot_list(module), _count(1)),
        (200, json.dumps(robot).encode(), {}),
    ]
    with pytest.raises(RuntimeError, match="normalization"):
        module.ensure_publisher_can_read_rule(FakeClient(responses))

    monkeypatch.undo()
    for status, error in ((503, module.HarborUnavailable), (409, RuntimeError)):
        responses = [
            (200, _robot_list(module), _count(1)),
            (200, json.dumps(_robot(module)).encode(), {}),
            (status, b"", {}),
        ]
        with pytest.raises(error):
            module.ensure_publisher_can_read_rule(FakeClient(responses))


def test_publisher_and_rule_verification_retries_are_bounded(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Stale reads may retry, but never silently become successful evidence."""
    module = _load(HARBOR, "harbor_retry_coverage")
    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
    responses = [
        (200, _robot_list(module), _count(1)),
        (200, json.dumps(_robot(module)).encode(), {}),
        (200, b"", {}),
        *((500, b"", {}) for _ in range(5)),
    ]
    with pytest.raises(RuntimeError, match="did not verify"):
        module.ensure_publisher_can_read_rule(FakeClient(responses))

    rule_path = "/api/v2.0/projects/bstein/immutabletagrules/23"
    responses = [
        (200, b"[]", _count(0)),
        (201, b"", {"Location": rule_path}),
        *((200, b"[]", _count(0)) for _ in range(5)),
    ]
    with pytest.raises(RuntimeError, match="did not verify"):
        module.ensure_rule(FakeClient(responses))


def test_rule_rejects_ids_create_status_and_location_suffix() -> None:
    """Neither existing nor newly created rules can omit their exact positive ID."""
    module = _load(HARBOR, "harbor_rule_id_coverage")
    bad_rule = {"id": 0, **module.EXPECTED_RULE}
    with pytest.raises(RuntimeError, match="valid ID"):
        module.ensure_rule(
            FakeClient([(200, json.dumps([bad_rule]).encode(), _count(1))])
        )
    with pytest.raises(module.HarborUnavailable):
        module.ensure_rule(
            FakeClient([(200, b"[]", _count(0)), (503, b"", {})])
        )
    with pytest.raises(RuntimeError, match="invalid ID"):
        module.ensure_rule(
            FakeClient(
                [
                    (200, b"[]", _count(0)),
                    (
                        201,
                        b"",
                        {
                            "location": "/api/v2.0/projects/bstein/immutabletagrules/nope"
                        },
                    ),
                ]
            )
        )


def test_harbor_main_retries_then_reports_exact_ids(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """Runtime credentials are required and transient startup failures are bounded."""
    module = _load(HARBOR, "harbor_main_coverage")
    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())
    attempts = iter([module.HarborUnavailable("warming"), 17])

    def ensure_rule(_client):
        result = next(attempts)
        if isinstance(result, BaseException):
            raise result
        return result

    monkeypatch.setattr(module, "ensure_rule", ensure_rule)
    monkeypatch.setattr(module, "ensure_publisher_can_read_rule", lambda _client: 41)
    monkeypatch.setattr(module.time, "sleep", lambda seconds: None)
    assert module.main() == 0
    assert "id=17" in capsys.readouterr().out

    password_file.write_text("\n", encoding="utf-8")
    with pytest.raises(RuntimeError, match="empty"):
        module.main()