atlas-iac/testing/tests/test_hermes_mirror_project_ensure.py

"""Boundary coverage for the Harbor ``mirror`` project bootstrap helper.

The Hermes agent image build pulls its base from the in-cluster Harbor
``mirror`` project instead of docker.io; this exercises the tiny fail-closed
helper that guarantees the project exists (and is public) before the sibling
``skopeo copy --all`` step pushes into it. No network is touched.
"""

from __future__ import annotations

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

import pytest


REPO_ROOT = Path(__file__).resolve().parents[2]
SCRIPT = REPO_ROOT / "services/harbor/scripts/harbor_mirror_project_ensure.py"


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

    def __init__(self, status: int, body: bytes = b"") -> None:
        super().__init__(body)
        self.status = status

    def __enter__(self):
        return self

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


class FakeOpener:
    """Ordered opener fake matching urllib's build_opener().open() contract."""

    def __init__(self, results) -> None:
        self.results = list(results)
        self.requests = []

    def open(self, req, timeout=None):  # noqa: A003 - urllib signature
        self.requests.append((req.get_method(), req.full_url, req.data))
        result = self.results.pop(0)
        if isinstance(result, Exception):
            raise result
        return result


def _http_error(code: int, body: bytes = b"boom") -> urllib.error.HTTPError:
    return urllib.error.HTTPError(
        "http://harbor/api", code, "err", {}, io.BytesIO(body)
    )


def _password_file(tmp_path: Path, contents: str) -> Path:
    path = tmp_path / "harbor-admin-password"
    path.write_text(contents, encoding="utf-8")
    return path


# --------------------------------------------------------------------------- #
# _admin_password / _api_origin
# --------------------------------------------------------------------------- #
def test_admin_password_reads_and_strips_the_injected_file(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    module = _load("mirror_pw_ok")
    monkeypatch.setenv(
        "HARBOR_ADMIN_PASSWORD_FILE", str(_password_file(tmp_path, "  s3cret \n"))
    )
    assert module._admin_password() == "s3cret"


def test_admin_password_rejects_empty_secret(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    module = _load("mirror_pw_empty")
    monkeypatch.setenv(
        "HARBOR_ADMIN_PASSWORD_FILE", str(_password_file(tmp_path, "   \n"))
    )
    with pytest.raises(SystemExit):
        module._admin_password()


def test_api_origin_defaults_and_honours_override(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load("mirror_origin")
    monkeypatch.delenv("HARBOR_API_ORIGIN", raising=False)
    assert module._api_origin().endswith("harbor.svc.cluster.local/api/v2.0")
    monkeypatch.setenv("HARBOR_API_ORIGIN", "http://example/api/v2.0/")
    assert module._api_origin() == "http://example/api/v2.0"


# --------------------------------------------------------------------------- #
# _request
# --------------------------------------------------------------------------- #
def test_request_returns_parsed_body_and_records_method() -> None:
    module = _load("mirror_req_ok")
    opener = FakeOpener([Response(200, json.dumps([{"name": "mirror"}]).encode())])
    result = module._request(opener, "http://h", "auth", "GET", "/projects")
    assert result == [{"name": "mirror"}]
    assert opener.requests[0][0] == "GET"


def test_request_posts_payload_and_tolerates_empty_body() -> None:
    module = _load("mirror_req_post")
    opener = FakeOpener([Response(201, b"")])
    result = module._request(
        opener, "http://h", "auth", "POST", "/projects", {"a": 1}, ok=(201, 409)
    )
    assert result is None
    method, _url, data = opener.requests[0]
    assert method == "POST" and data == json.dumps({"a": 1}).encode()


def test_request_rejects_oversized_response(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load("mirror_req_big")
    monkeypatch.setattr(module, "MAX_RESPONSE", 4)
    opener = FakeOpener([Response(200, b"xxxxxxxx")])
    with pytest.raises(SystemExit):
        module._request(opener, "http://h", "auth", "GET", "/projects")


def test_request_rejects_unexpected_status() -> None:
    module = _load("mirror_req_status")
    opener = FakeOpener([Response(500, b"nope")])
    with pytest.raises(SystemExit):
        module._request(opener, "http://h", "auth", "GET", "/projects")


def test_request_maps_http_error_status_to_ok_or_failure() -> None:
    module = _load("mirror_req_httperror")
    tolerated = FakeOpener([_http_error(409)])
    assert (
        module._request(
            tolerated, "http://h", "auth", "POST", "/projects", {"a": 1}, ok=(201, 409)
        )
        is None
    )
    fatal = FakeOpener([_http_error(403)])
    with pytest.raises(SystemExit):
        module._request(fatal, "http://h", "auth", "POST", "/projects", {"a": 1})


# --------------------------------------------------------------------------- #
# main
# --------------------------------------------------------------------------- #
def _prime(module, monkeypatch, tmp_path, results):
    monkeypatch.setenv(
        "HARBOR_ADMIN_PASSWORD_FILE", str(_password_file(tmp_path, "pw"))
    )
    monkeypatch.setenv("HARBOR_API_ORIGIN", "http://harbor/api/v2.0")
    opener = FakeOpener(results)
    monkeypatch.setattr(
        module.urllib.request, "build_opener", lambda *_a, **_k: opener
    )
    return opener


def test_main_is_a_noop_when_public_project_exists(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    module = _load("mirror_main_present")
    body = json.dumps(
        [{"name": "mirror", "metadata": {"public": "true"}}]
    ).encode()
    opener = _prime(module, monkeypatch, tmp_path, [Response(200, body)])
    module.main()
    assert [call[0] for call in opener.requests] == ["GET"]


def test_main_fails_closed_on_a_private_collision(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    module = _load("mirror_main_private")
    body = json.dumps(
        [{"name": "mirror", "metadata": {"public": "false"}}]
    ).encode()
    _prime(module, monkeypatch, tmp_path, [Response(200, body)])
    with pytest.raises(SystemExit):
        module.main()


@pytest.mark.parametrize(
    "listing",
    [b"[]", json.dumps([{"name": "other", "metadata": {"public": "true"}}]).encode()],
)
def test_main_creates_the_project_when_absent(
    listing: bytes, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    module = _load("mirror_main_create")
    opener = _prime(
        module,
        monkeypatch,
        tmp_path,
        [Response(200, listing), Response(201, b"")],
    )
    module.main()
    methods = [call[0] for call in opener.requests]
    assert methods == ["GET", "POST"]
    assert opener.requests[1][1].endswith("/projects")


# --------------------------------------------------------------------------- #
# misc
# --------------------------------------------------------------------------- #
def test_no_redirect_handler_refuses_to_replay_credentials() -> None:
    module = _load("mirror_noredirect")
    handler = module.NoRedirect()
    assert handler.redirect_request(None, None, 302, "m", {}, "http://elsewhere") is None