"""Exercise exact fail-closed branches in the reviewed Hermes image lane."""

from __future__ import annotations

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

import pytest


ROOT = Path(__file__).resolve().parents[2]
RELEASE = ROOT / "ci/scripts/hermes_image_release.py"
HARBOR = 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):
    """Small bounded response for same-process HTTP contract tests."""

    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:
    """Return one reviewed sequence of Harbor API responses."""

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

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

    def request(self, _method, _path, _payload=None):
        return self.responses.pop(0)


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


def _robot(module, *, immutable: bool, duration: int | None = None) -> 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 duration is not None:
        robot["duration"] = duration
    return robot


def test_release_policy_response_edges_are_fail_closed() -> None:
    """Credentials, bounds, page evidence, JSON shape, and redirects stay strict."""
    module = _load(RELEASE, "hermes_release_policy_branch_contract")
    assert module._NoRedirect().redirect_request(None, None, 302, "", {}, "") is None
    with pytest.raises(RuntimeError, match="credentials"):
        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),
        )
    with pytest.raises(RuntimeError, match="valid total count"):
        module._require_complete_rule_page([], {})
    with pytest.raises(RuntimeError, match="invalid shape"):
        module.verify_immutable_policy(
            username="robot",
            password="private",
            opener=lambda *_args: Response(200, b"{}", _count(0)),
        )


def test_release_renderer_exhausts_a_trailing_incomplete_image() -> None:
    """A trailing matching name cannot hide the one complete digest entry."""
    module = _load(RELEASE, "hermes_release_trailing_image_branch")
    digest = "sha256:" + "9" * 64
    source = (
        "images:\n"
        f"  - name: {module.DEFAULT_IMAGE}\n"
        "    digest: sha256:" + "0" * 64 + "\n"
        f"  - name: {module.DEFAULT_IMAGE}\n"
    )
    assert module.render_kustomization(source, digest).count(digest) == 1


def test_harbor_publisher_handles_absent_duration_and_stale_reread(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """An absent optional duration is preserved while stale policy reads retry."""
    module = _load(HARBOR, "harbor_publisher_branch_contract")
    current = _robot(module, immutable=False)
    malformed_persisted = {**_robot(module, immutable=True), "permissions": "bad"}
    exact = _robot(module, immutable=True)
    client = FakeClient(
        [
            (200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1)),
            (200, json.dumps(current).encode(), {}),
            (200, b"", {}),
            (200, json.dumps(malformed_persisted).encode(), {}),
            (200, json.dumps(exact).encode(), {}),
        ]
    )
    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
    assert module.ensure_publisher_can_read_rule(client) == 41


def test_harbor_created_rule_retries_an_id_mismatch(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A stale exact-looking rule with the wrong ID is never accepted."""
    module = _load(HARBOR, "harbor_rule_id_retry_branch")
    path = "/api/v2.0/projects/bstein/immutabletagrules/23"
    wrong = {"id": 22, **module.EXPECTED_RULE}
    exact = {"id": 23, **module.EXPECTED_RULE}
    client = FakeClient(
        [
            (200, b"[]", _count(0)),
            (201, b"", {"Location": path}),
            (200, json.dumps([wrong]).encode(), _count(1)),
            (200, json.dumps([exact]).encode(), _count(1)),
        ]
    )
    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
    assert module.ensure_rule(client) == 23


def test_harbor_main_covers_immediate_success_and_final_retry(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    """Startup exits directly on success and re-raises the final unavailable result."""
    module = _load(HARBOR, "harbor_main_branch_contract")
    password = tmp_path / "password"
    password.write_text("private\n", encoding="utf-8")
    monkeypatch.setenv("HARBOR_API_ORIGIN", module.EXPECTED_ORIGIN)
    monkeypatch.setenv("HARBOR_ADMIN_PASSWORD_FILE", str(password))
    monkeypatch.setattr(module, "HarborClient", lambda *_args: object())
    monkeypatch.setattr(module, "ensure_rule", lambda _client: 17)
    monkeypatch.setattr(module, "ensure_publisher_can_read_rule", lambda _client: 41)
    assert module.main() == 0

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

    monkeypatch.setattr(module, "ensure_rule", unavailable)
    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
    with pytest.raises(module.HarborUnavailable, match="still warming"):
        module.main()