pyro-mcp/tests/test_runtime_build.py

from __future__ import annotations

import argparse
import hashlib
import io
import json
import tarfile
import urllib.error
import urllib.request
from email.message import Message
from pathlib import Path

import pytest

from pyro_mcp.runtime_build import (
    _REGISTRY_BEARER_TOKENS,
    DEFAULT_DOCKERHUB_TOKEN_ENV,
    DEFAULT_DOCKERHUB_USERNAME_ENV,
    _build_paths,
    _load_lock,
    _load_oci_layout_manifest,
    _parse_authenticate_parameters,
    _registry_credentials,
    _request_registry,
    _upload_blob,
    build_bundle,
    export_environment_oci_layout,
    generate_manifest,
    main,
    materialize_binaries,
    materialize_sources,
    publish_environment_oci_layout,
    stage_agent,
    stage_binaries,
    stage_kernel,
    stage_rootfs,
    sync_bundle,
    validate_sources,
)


class FakeResponse:
    def __init__(
        self,
        payload: bytes = b"",
        *,
        status: int = 200,
        headers: dict[str, str] | None = None,
    ) -> None:
        self._buffer = io.BytesIO(payload)
        self.status = status
        self.headers = headers or {}

    def read(self, size: int = -1) -> bytes:
        return self._buffer.read(size)

    def __enter__(self) -> FakeResponse:
        return self

    def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
        del exc_type, exc, tb


def _http_headers(headers: dict[str, str]) -> Message:
    message = Message()
    for key, value in headers.items():
        message[key] = value
    return message


def _request_header(request: urllib.request.Request, name: str) -> str | None:
    for key, value in request.header_items():
        if key.lower() == name.lower():
            return value
    return None


def _write_text(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")


def _make_source_tree(tmp_path: Path) -> tuple[Path, Path, Path]:
    source_dir = tmp_path / "runtime_sources"
    platform_root = source_dir / "linux-x86_64"
    _write_text(source_dir / "NOTICE", "notice\n")
    _write_text(platform_root / "bin/firecracker", "firecracker\n")
    _write_text(platform_root / "bin/jailer", "jailer\n")
    _write_text(platform_root / "guest/pyro_guest_agent.py", "#!/usr/bin/env python3\n")
    _write_text(platform_root / "profiles/debian-base/vmlinux", "kernel-base\n")
    _write_text(platform_root / "profiles/debian-base/rootfs.ext4", "rootfs-base\n")
    _write_text(platform_root / "profiles/debian-git/vmlinux", "kernel-git\n")
    _write_text(platform_root / "profiles/debian-git/rootfs.ext4", "rootfs-git\n")
    _write_text(platform_root / "profiles/debian-build/vmlinux", "kernel-build\n")
    _write_text(platform_root / "profiles/debian-build/rootfs.ext4", "rootfs-build\n")
    lock = {
        "bundle_version": "9.9.9",
        "platform": "linux-x86_64",
        "component_versions": {
            "firecracker": "1.0.0",
            "jailer": "1.0.0",
            "kernel": "6.0.0",
            "guest_agent": "0.2.0",
            "base_distro": "debian-12",
        },
        "capabilities": {"vm_boot": True, "guest_exec": True, "guest_network": True},
        "binaries": {"firecracker": "bin/firecracker", "jailer": "bin/jailer"},
        "guest": {"agent": {"path": "guest/pyro_guest_agent.py"}},
        "profiles": {
            "debian-base": {
                "description": "base",
                "kernel": "profiles/debian-base/vmlinux",
                "rootfs": "profiles/debian-base/rootfs.ext4",
            },
            "debian-git": {
                "description": "git",
                "kernel": "profiles/debian-git/vmlinux",
                "rootfs": "profiles/debian-git/rootfs.ext4",
            },
            "debian-build": {
                "description": "build",
                "kernel": "profiles/debian-build/vmlinux",
                "rootfs": "profiles/debian-build/rootfs.ext4",
            },
        },
    }
    (platform_root / "runtime.lock.json").write_text(
        json.dumps(lock, indent=2) + "\n", encoding="utf-8"
    )
    build_dir = tmp_path / "build/runtime_bundle"
    bundle_dir = tmp_path / "bundle_out"
    return source_dir, build_dir, bundle_dir


def test_runtime_build_stages_and_manifest(tmp_path: Path) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized_sources",
        platform="linux-x86_64",
    )
    lock = _load_lock(paths)

    paths.build_platform_root.mkdir(parents=True, exist_ok=True)
    stage_binaries(paths, lock)
    stage_kernel(paths, lock)
    stage_rootfs(paths, lock)
    stage_agent(paths, lock)
    manifest = generate_manifest(paths, lock)

    assert manifest["bundle_version"] == "9.9.9"
    assert manifest["capabilities"]["guest_exec"] is True
    assert manifest["component_versions"]["guest_agent"] == "0.2.0"
    assert (paths.build_platform_root / "guest/pyro_guest_agent.py").exists()


def test_runtime_build_bundle_syncs_output(tmp_path: Path) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized_sources",
        platform="linux-x86_64",
    )

    manifest = build_bundle(paths, sync=True)

    assert manifest["profiles"]["debian-git"]["description"] == "git"
    assert (bundle_dir / "NOTICE").exists()
    assert (bundle_dir / "linux-x86_64/manifest.json").exists()
    assert (bundle_dir / "linux-x86_64/guest/pyro_guest_agent.py").exists()


def test_runtime_build_rejects_guest_capabilities_for_placeholder_sources(tmp_path: Path) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized_sources",
        platform="linux-x86_64",
    )
    lock_path = source_dir / "linux-x86_64/runtime.lock.json"
    _write_text(
        source_dir / "linux-x86_64/bin/firecracker",
        "#!/usr/bin/env bash\n"
        "echo 'bundled firecracker shim'\n",
    )
    _write_text(
        source_dir / "linux-x86_64/profiles/debian-base/rootfs.ext4",
        "placeholder-rootfs\n",
    )
    lock = json.loads(lock_path.read_text(encoding="utf-8"))
    lock["capabilities"] = {"vm_boot": True, "guest_exec": True, "guest_network": True}
    lock_path.write_text(json.dumps(lock, indent=2) + "\n", encoding="utf-8")

    with pytest.raises(RuntimeError, match="guest-capable features"):
        validate_sources(paths, _load_lock(paths))


def test_runtime_build_materializes_firecracker_release(tmp_path: Path) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    archive_path = tmp_path / "firecracker-v1.12.1-x86_64.tgz"
    release_dir = "release-v1.12.1-x86_64"
    with tarfile.open(archive_path, "w:gz") as archive:
        firecracker_path = tmp_path / "firecracker-bin"
        jailer_path = tmp_path / "jailer-bin"
        firecracker_path.write_text("real-firecracker\n", encoding="utf-8")
        jailer_path.write_text("real-jailer\n", encoding="utf-8")
        archive.add(firecracker_path, arcname=f"{release_dir}/firecracker-v1.12.1-x86_64")
        archive.add(jailer_path, arcname=f"{release_dir}/jailer-v1.12.1-x86_64")

    digest = hashlib.sha256(archive_path.read_bytes()).hexdigest()
    lock_path = source_dir / "linux-x86_64/runtime.lock.json"
    lock = json.loads(lock_path.read_text(encoding="utf-8"))
    lock["upstream"] = {
        "firecracker_release": {
            "archive_url": archive_path.resolve().as_uri(),
            "archive_sha256": digest,
            "firecracker_member": f"{release_dir}/firecracker-v1.12.1-x86_64",
            "jailer_member": f"{release_dir}/jailer-v1.12.1-x86_64",
        }
    }
    lock_path.write_text(json.dumps(lock, indent=2) + "\n", encoding="utf-8")

    paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized_sources",
        platform="linux-x86_64",
    )
    materialize_binaries(paths, _load_lock(paths))

    assert (paths.materialized_platform_root / "bin/firecracker").read_text(encoding="utf-8") == (
        "real-firecracker\n"
    )
    assert (paths.materialized_platform_root / "bin/jailer").read_text(encoding="utf-8") == (
        "real-jailer\n"
    )


def test_export_environment_oci_layout_writes_boot_artifacts(tmp_path: Path) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized_sources",
        platform="linux-x86_64",
    )

    result = export_environment_oci_layout(
        paths,
        environment="debian:12-base",
        output_dir=tmp_path / "oci_layouts",
    )

    layout_dir = Path(str(result["layout_dir"]))
    assert layout_dir == tmp_path / "oci_layouts" / "debian_12-base"
    assert json.loads((layout_dir / "oci-layout").read_text(encoding="utf-8")) == {
        "imageLayoutVersion": "1.0.0"
    }
    index = json.loads((layout_dir / "index.json").read_text(encoding="utf-8"))
    manifest_descriptor = index["manifests"][0]
    assert manifest_descriptor["annotations"]["org.opencontainers.image.ref.name"] == "1.0.0"
    assert manifest_descriptor["platform"] == {"os": "linux", "architecture": "amd64"}

    manifest_digest = str(result["manifest_digest"]).split(":", 1)[1]
    manifest = json.loads(
        (layout_dir / "blobs" / "sha256" / manifest_digest).read_text(encoding="utf-8")
    )
    assert manifest["config"]["mediaType"] == "application/vnd.oci.image.config.v1+json"
    assert len(manifest["layers"]) == 3

    extracted_files: dict[str, str] = {}
    for layer in manifest["layers"]:
        layer_digest = str(layer["digest"]).split(":", 1)[1]
        layer_path = layout_dir / "blobs" / "sha256" / layer_digest
        with tarfile.open(layer_path, "r:") as archive:
            members = archive.getmembers()
            assert len(members) == 1
            member = members[0]
            extracted = archive.extractfile(member)
            if extracted is None:
                raise AssertionError("expected layer member content")
            extracted_files[member.name] = extracted.read().decode("utf-8")

    assert extracted_files["vmlinux"] == "kernel-base\n"
    assert extracted_files["rootfs.ext4"] == "rootfs-base\n"
    metadata = json.loads(extracted_files["environment.json"])
    assert metadata["environment"] == "debian:12-base"
    assert metadata["source_profile"] == "debian-base"


def test_export_environment_oci_layout_rejects_missing_profile(tmp_path: Path) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized_sources",
        platform="linux-x86_64",
    )
    lock_path = source_dir / "linux-x86_64/runtime.lock.json"
    lock = json.loads(lock_path.read_text(encoding="utf-8"))
    del lock["profiles"]["debian-base"]
    lock_path.write_text(json.dumps(lock, indent=2) + "\n", encoding="utf-8")

    with pytest.raises(RuntimeError, match="does not define source profile"):
        export_environment_oci_layout(
            paths,
            environment="debian:12-base",
            output_dir=tmp_path / "oci_layouts",
        )


def test_runtime_build_main_dispatches_export(
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
    tmp_path: Path,
) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    runtime_paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized",
        platform="linux-x86_64",
    )

    class StubParser:
        def parse_args(self) -> argparse.Namespace:
            return argparse.Namespace(
                command="export-environment-oci",
                platform="linux-x86_64",
                source_dir=str(source_dir),
                build_dir=str(build_dir),
                bundle_dir=str(bundle_dir),
                materialized_dir=str(tmp_path / "materialized"),
                environment="debian:12-base",
                output_dir=str(tmp_path / "oci_layouts"),
                reference="1.0.0",
            )

    monkeypatch.setattr("pyro_mcp.runtime_build._build_parser", lambda: StubParser())
    monkeypatch.setattr(
        "pyro_mcp.runtime_build._load_lock",
        lambda paths: _load_lock(runtime_paths),
    )
    monkeypatch.setattr(
        "pyro_mcp.runtime_build.export_environment_oci_layout",
        lambda paths, **kwargs: {
            "environment": kwargs["environment"],
            "layout_dir": str(Path(kwargs["output_dir"]) / "debian_12-base"),
            "manifest_digest": "sha256:test",
        },
    )

    main()

    payload = json.loads(capsys.readouterr().out)
    assert payload["environment"] == "debian:12-base"
    assert payload["manifest_digest"] == "sha256:test"


def test_runtime_build_main_dispatches_publish_with_implicit_export(
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
    tmp_path: Path,
) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    layout_root = tmp_path / "oci_layouts"
    calls: list[str] = []

    class StubParser:
        def parse_args(self) -> argparse.Namespace:
            return argparse.Namespace(
                command="publish-environment-oci",
                platform="linux-x86_64",
                source_dir=str(source_dir),
                build_dir=str(build_dir),
                bundle_dir=str(bundle_dir),
                materialized_dir=str(tmp_path / "materialized"),
                environment="debian:12-base",
                output_dir=str(layout_root),
                layout_root=str(layout_root),
                registry=None,
                repository=None,
                reference="1.0.0",
                username_env="OCI_REGISTRY_USERNAME",
                password_env="OCI_REGISTRY_PASSWORD",
            )

    monkeypatch.setattr("pyro_mcp.runtime_build._build_parser", lambda: StubParser())
    def fake_export(paths: object, **kwargs: object) -> dict[str, str]:
        del paths
        environment = str(kwargs["environment"])
        output_dir = Path(str(kwargs["output_dir"]))
        calls.append(f"export:{environment}")
        return {
            "environment": environment,
            "layout_dir": str(output_dir / "debian_12-base"),
            "manifest_digest": "sha256:exported",
        }

    def fake_publish(**kwargs: object) -> dict[str, str]:
        environment = str(kwargs["environment"])
        layout_root = Path(str(kwargs["layout_root"]))
        calls.append(f"publish:{environment}")
        return {
            "environment": environment,
            "layout_dir": str(layout_root / "debian_12-base"),
            "manifest_digest": "sha256:published",
        }

    monkeypatch.setattr("pyro_mcp.runtime_build.export_environment_oci_layout", fake_export)
    monkeypatch.setattr("pyro_mcp.runtime_build.publish_environment_oci_layout", fake_publish)

    main()

    payload = json.loads(capsys.readouterr().out)
    assert calls == ["export:debian:12-base", "publish:debian:12-base"]
    assert payload["environment"] == "debian:12-base"
    assert payload["manifest_digest"] == "sha256:published"


def test_request_registry_retries_with_bearer_token(monkeypatch: pytest.MonkeyPatch) -> None:
    requested_auth: list[str | None] = []
    requested_token_urls: list[str] = []
    _REGISTRY_BEARER_TOKENS.clear()

    def fake_urlopen(request: object, timeout: int = 90) -> FakeResponse:
        del timeout
        url = request.full_url if isinstance(request, urllib.request.Request) else str(request)
        if url.startswith("https://auth.docker.io/token?"):
            requested_token_urls.append(url)
            authorization = (
                _request_header(request, "Authorization")
                if isinstance(request, urllib.request.Request)
                else None
            )
            requested_auth.append(authorization if isinstance(authorization, str) else None)
            return FakeResponse(b'{"token":"secret-token"}')
        if isinstance(request, urllib.request.Request):
            authorization = _request_header(request, "Authorization")
        else:
            authorization = None
        requested_auth.append(authorization if isinstance(authorization, str) else None)
        if authorization is None:
            raise urllib.error.HTTPError(
                url,
                401,
                "Unauthorized",
                _http_headers(
                    {
                        "WWW-Authenticate": (
                            'Bearer realm="https://auth.docker.io/token",'
                            'service="registry.docker.io",'
                            'scope="repository:thalesmaciel/pyro-environment-debian-12-base:pull,push"'
                        )
                    }
                ),
                io.BytesIO(b""),
            )
        return FakeResponse(b"ok", headers={"Docker-Content-Digest": "sha256:abc123"})

    monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)

    status, payload, headers = _request_registry(
        "https://registry-1.docker.io/v2/thalesmaciel/pyro-environment-debian-12-base/test",
        method="GET",
        repository="thalesmaciel/pyro-environment-debian-12-base",
        scope_actions="pull,push",
        username="user",
        password="token",
    )

    assert status == 200
    assert payload == b"ok"
    assert headers["docker-content-digest"] == "sha256:abc123"
    assert requested_auth[0] is None
    assert requested_auth[1] is not None and requested_auth[1].startswith("Basic ")
    assert requested_auth[2] == "Bearer secret-token"
    assert requested_token_urls == [
        (
            "https://auth.docker.io/token?service=registry.docker.io&"
            "scope=repository%3Athalesmaciel%2Fpyro-environment-debian-12-base%3Apull%2Cpush"
        )
    ]


def test_parse_authenticate_parameters_preserves_quoted_commas() -> None:
    params = _parse_authenticate_parameters(
        'realm="https://auth.docker.io/token",'
        'service="registry.docker.io",'
        'scope="repository:thalesmaciel/pyro-environment-debian-12:pull,push"'
    )

    assert params == {
        "realm": "https://auth.docker.io/token",
        "service": "registry.docker.io",
        "scope": "repository:thalesmaciel/pyro-environment-debian-12:pull,push",
    }


def test_registry_credentials_fall_back_to_dockerhub_envs(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("OCI_REGISTRY_USERNAME", raising=False)
    monkeypatch.delenv("OCI_REGISTRY_PASSWORD", raising=False)
    monkeypatch.setenv(DEFAULT_DOCKERHUB_USERNAME_ENV, "docker-user")
    monkeypatch.setenv(DEFAULT_DOCKERHUB_TOKEN_ENV, "docker-token")

    username, password = _registry_credentials()

    assert username == "docker-user"
    assert password == "docker-token"


def test_request_registry_requires_auth_challenge(monkeypatch: pytest.MonkeyPatch) -> None:
    _REGISTRY_BEARER_TOKENS.clear()

    def fake_urlopen(request: object, timeout: int = 90) -> FakeResponse:
        del timeout
        url = request.full_url if isinstance(request, urllib.request.Request) else str(request)
        raise urllib.error.HTTPError(
            url,
            401,
            "Unauthorized",
            _http_headers({}),
            io.BytesIO(b""),
        )

    monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)

    with pytest.raises(RuntimeError, match="denied access without an auth challenge"):
        _request_registry(
            "https://registry-1.docker.io/v2/thalesmaciel/pyro-environment-debian-12-base/test",
            method="GET",
            repository="thalesmaciel/pyro-environment-debian-12-base",
            scope_actions="pull,push",
        )


def test_request_registry_returns_allowed_retry_status(monkeypatch: pytest.MonkeyPatch) -> None:
    _REGISTRY_BEARER_TOKENS.clear()

    def fake_urlopen(request: object, timeout: int = 90) -> FakeResponse:
        del timeout
        url = request.full_url if isinstance(request, urllib.request.Request) else str(request)
        if url.startswith("https://auth.docker.io/token?"):
            return FakeResponse(b'{"token":"secret-token"}')
        authorization = (
            _request_header(request, "Authorization")
            if isinstance(request, urllib.request.Request)
            else None
        )
        if authorization is None:
            raise urllib.error.HTTPError(
                url,
                401,
                "Unauthorized",
                _http_headers(
                    {
                        "WWW-Authenticate": (
                            'Bearer realm="https://auth.docker.io/token",'
                            'service="registry.docker.io",'
                            'scope="repository:thalesmaciel/pyro-environment-debian-12-base:pull,push"'
                        )
                    }
                ),
                io.BytesIO(b""),
            )
        raise urllib.error.HTTPError(
            url,
            404,
            "Not Found",
            _http_headers({"Docker-Content-Digest": "sha256:missing"}),
            io.BytesIO(b"missing"),
        )

    monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)

    status, payload, headers = _request_registry(
        "https://registry-1.docker.io/v2/thalesmaciel/pyro-environment-debian-12-base/test",
        method="GET",
        repository="thalesmaciel/pyro-environment-debian-12-base",
        scope_actions="pull,push",
        allow_statuses=(404,),
    )

    assert status == 404
    assert payload == b"missing"
    assert headers["docker-content-digest"] == "sha256:missing"


def test_request_registry_reuses_cached_bearer_token(monkeypatch: pytest.MonkeyPatch) -> None:
    requested_auth: list[str | None] = []
    requested_token_urls: list[str] = []
    _REGISTRY_BEARER_TOKENS.clear()

    def fake_urlopen(request: object, timeout: int = 90) -> FakeResponse:
        del timeout
        url = request.full_url if isinstance(request, urllib.request.Request) else str(request)
        authorization = (
            _request_header(request, "Authorization")
            if isinstance(request, urllib.request.Request)
            else None
        )
        requested_auth.append(authorization if isinstance(authorization, str) else None)
        if url.startswith("https://auth.docker.io/token?"):
            requested_token_urls.append(url)
            return FakeResponse(b'{"token":"secret-token"}')
        if url.endswith("/first"):
            if authorization is None:
                raise urllib.error.HTTPError(
                    url,
                    401,
                    "Unauthorized",
                    _http_headers(
                        {
                            "WWW-Authenticate": (
                                'Bearer realm="https://auth.docker.io/token",'
                                'service="registry.docker.io",'
                                'scope="repository:thalesmaciel/pyro-environment-debian-12:pull,push"'
                            )
                        }
                    ),
                    io.BytesIO(b""),
                )
            return FakeResponse(b"ok-first")
        if url.endswith("/second"):
            if authorization != "Bearer secret-token":
                raise AssertionError("expected cached bearer token on second request")
            return FakeResponse(b"ok-second")
        raise AssertionError(f"unexpected registry request: {url}")

    monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)

    first_status, first_payload, _ = _request_registry(
        "https://registry-1.docker.io/v2/thalesmaciel/pyro-environment-debian-12/first",
        method="GET",
        repository="thalesmaciel/pyro-environment-debian-12",
        scope_actions="pull,push",
        username="user",
        password="token",
    )
    second_status, second_payload, _ = _request_registry(
        "https://registry-1.docker.io/v2/thalesmaciel/pyro-environment-debian-12/second",
        method="PUT",
        repository="thalesmaciel/pyro-environment-debian-12",
        scope_actions="pull,push",
        username="user",
        password="token",
    )

    assert first_status == 200
    assert first_payload == b"ok-first"
    assert second_status == 200
    assert second_payload == b"ok-second"
    assert requested_token_urls == [
        (
            "https://auth.docker.io/token?service=registry.docker.io&"
            "scope=repository%3Athalesmaciel%2Fpyro-environment-debian-12%3Apull%2Cpush"
        )
    ]


@pytest.mark.parametrize(
    ("index_payload", "message"),
    [
        ({}, "exactly one manifest descriptor"),
        ({"manifests": ["bad"]}, "descriptor is malformed"),
        ({"manifests": [{}]}, "missing a digest"),
    ],
)
def test_load_oci_layout_manifest_rejects_invalid_index(
    tmp_path: Path,
    index_payload: dict[str, object],
    message: str,
) -> None:
    layout_dir = tmp_path / "layout"
    layout_dir.mkdir(parents=True)
    (layout_dir / "index.json").write_text(json.dumps(index_payload), encoding="utf-8")

    with pytest.raises(RuntimeError, match=message):
        _load_oci_layout_manifest(layout_dir)


def test_load_oci_layout_manifest_rejects_missing_index(tmp_path: Path) -> None:
    with pytest.raises(RuntimeError, match="OCI layout index not found"):
        _load_oci_layout_manifest(tmp_path / "missing-layout")


def test_upload_blob_skips_existing_blob(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    blob_path = tmp_path / "blob.tar"
    blob_path.write_bytes(b"blob-data")
    monkeypatch.setattr("pyro_mcp.runtime_build._blob_exists", lambda **_: True)

    result = _upload_blob(
        registry="registry-1.docker.io",
        repository="thalesmaciel/pyro-environment-debian-12-base",
        digest="sha256:abc123",
        blob_path=blob_path,
        username=None,
        password=None,
    )

    assert result == {
        "digest": "sha256:abc123",
        "size": len(b"blob-data"),
        "uploaded": False,
    }


def test_upload_blob_requires_location(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    blob_path = tmp_path / "blob.tar"
    blob_path.write_bytes(b"blob-data")
    monkeypatch.setattr("pyro_mcp.runtime_build._blob_exists", lambda **_: False)
    monkeypatch.setattr(
        "pyro_mcp.runtime_build._request_registry",
        lambda *args, **kwargs: (202, b"", {}) if kwargs["method"] == "POST" else (201, b"", {}),
    )

    with pytest.raises(RuntimeError, match="did not return a blob upload location"):
        _upload_blob(
            registry="registry-1.docker.io",
            repository="thalesmaciel/pyro-environment-debian-12-base",
            digest="sha256:abc123",
            blob_path=blob_path,
            username=None,
            password=None,
        )


def test_upload_blob_uses_patch_chunks_and_finalize_put(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    blob_path = tmp_path / "blob.tar"
    blob_path.write_bytes(b"abcdefghij")
    calls: list[tuple[str, str, int, int | None]] = []
    monkeypatch.setenv("PYRO_OCI_UPLOAD_CHUNK_SIZE_BYTES", "4")
    monkeypatch.setenv("PYRO_OCI_UPLOAD_TIMEOUT_SECONDS", "123")
    monkeypatch.setattr("pyro_mcp.runtime_build._blob_exists", lambda **_: False)

    def fake_request_registry(
        url: str,
        *,
        method: str,
        repository: str,
        scope_actions: str,
        headers: dict[str, str] | None = None,
        data: object = None,
        allow_statuses: tuple[int, ...] = (),
        username: str | None = None,
        password: str | None = None,
        timeout_seconds: int | None = None,
    ) -> tuple[int, bytes, dict[str, str]]:
        del repository, scope_actions, allow_statuses, username, password
        if isinstance(data, bytes):
            payload_length = len(data)
        else:
            payload_length = 0
        calls.append((method, url, payload_length, timeout_seconds))
        if method == "POST":
            return 202, b"", {"location": "/upload/session-1"}
        if method == "PATCH":
            return 202, b"", {"location": "/upload/session-1"}
        if method == "PUT":
            assert isinstance(headers, dict)
            assert headers["Content-Length"] == "0"
            return 201, b"", {}
        raise AssertionError(f"unexpected method: {method}")

    monkeypatch.setattr("pyro_mcp.runtime_build._request_registry", fake_request_registry)

    result = _upload_blob(
        registry="registry-1.docker.io",
        repository="thalesmaciel/pyro-environment-debian-12-base",
        digest="sha256:abc123",
        blob_path=blob_path,
        username=None,
        password=None,
    )

    assert result == {"digest": "sha256:abc123", "size": 10, "uploaded": True}
    assert calls == [
        (
            "POST",
            "https://registry-1.docker.io/v2/thalesmaciel/pyro-environment-debian-12-base/blobs/uploads/",
            0,
            None,
        ),
        ("PATCH", "https://registry-1.docker.io/upload/session-1", 4, 123),
        ("PATCH", "https://registry-1.docker.io/upload/session-1", 4, 123),
        ("PATCH", "https://registry-1.docker.io/upload/session-1", 2, 123),
        ("PUT", "https://registry-1.docker.io/upload/session-1?digest=sha256%3Aabc123", 0, 123),
    ]


def test_upload_blob_reports_patch_offset_on_failure(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    blob_path = tmp_path / "blob.tar"
    blob_path.write_bytes(b"abcdefghij")
    monkeypatch.setenv("PYRO_OCI_UPLOAD_CHUNK_SIZE_BYTES", "4")
    monkeypatch.setattr("pyro_mcp.runtime_build._blob_exists", lambda **_: False)

    def fake_request_registry(
        url: str,
        *,
        method: str,
        repository: str,
        scope_actions: str,
        headers: dict[str, str] | None = None,
        data: object = None,
        allow_statuses: tuple[int, ...] = (),
        username: str | None = None,
        password: str | None = None,
        timeout_seconds: int | None = None,
    ) -> tuple[int, bytes, dict[str, str]]:
        del (
            url,
            repository,
            scope_actions,
            headers,
            data,
            allow_statuses,
            username,
            password,
            timeout_seconds,
        )
        if method == "POST":
            return 202, b"", {"location": "/upload/session-1"}
        if method == "PATCH":
            raise RuntimeError(
                "registry request failed for "
                "https://registry-1.docker.io/upload/session-1: timed out"
            )
        raise AssertionError(f"unexpected method: {method}")

    monkeypatch.setattr("pyro_mcp.runtime_build._request_registry", fake_request_registry)

    with pytest.raises(
        RuntimeError,
        match=r"patch failed.*byte offset 0 \(chunk 0\)",
    ):
        _upload_blob(
            registry="registry-1.docker.io",
            repository="thalesmaciel/pyro-environment-debian-12-base",
            digest="sha256:abc123",
            blob_path=blob_path,
            username=None,
            password=None,
        )


def test_publish_environment_oci_layout_rejects_missing_registry_target(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.setattr(
        "pyro_mcp.runtime_build.get_environment",
        lambda environment: type(
            "Spec",
            (),
            {
                "name": environment,
                "version": "1.0.0",
                "oci_registry": None,
                "oci_repository": None,
                "oci_reference": None,
            },
        )(),
    )

    with pytest.raises(RuntimeError, match="does not define an OCI registry target"):
        publish_environment_oci_layout(
            environment="debian:12-base",
            layout_root=tmp_path / "oci_layouts",
        )


def test_publish_environment_oci_layout_uploads_blobs_and_manifest(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized_sources",
        platform="linux-x86_64",
    )
    export_environment_oci_layout(
        paths,
        environment="debian:12-base",
        output_dir=tmp_path / "oci_layouts",
    )

    patch_uploads: list[tuple[str, int]] = []
    finalize_uploads: list[str] = []
    manifest_publish: list[tuple[str, str, int]] = []
    upload_counter = 0
    monkeypatch.setenv("OCI_REGISTRY_USERNAME", "registry-user")
    monkeypatch.setenv("OCI_REGISTRY_PASSWORD", "registry-token")
    monkeypatch.setenv("PYRO_OCI_UPLOAD_CHUNK_SIZE_BYTES", "1048576")

    def fake_urlopen(request: object, timeout: int = 90) -> FakeResponse:
        del timeout
        nonlocal upload_counter
        if not isinstance(request, urllib.request.Request):
            raise AssertionError("expected urllib request")
        url = request.full_url
        method = request.get_method()
        if method == "HEAD" and "/blobs/" in url:
            raise urllib.error.HTTPError(
                url,
                404,
                "Not Found",
                _http_headers({}),
                io.BytesIO(b""),
            )
        if method == "POST" and url.endswith("/blobs/uploads/"):
            upload_counter += 1
            return FakeResponse(
                status=202,
                headers={"Location": f"/upload/session-{upload_counter}"},
            )
        if method == "PATCH" and "/upload/session-" in url:
            body = request.data
            if isinstance(body, bytes):
                payload = body
            else:
                raise AssertionError("expected upload body")
            patch_uploads.append((url, len(payload)))
            session_id = url.rsplit("-", 1)[1]
            return FakeResponse(status=202, headers={"Location": f"/upload/session-{session_id}"})
        if method == "PUT" and "/upload/session-" in url:
            body = request.data
            if body != b"":
                raise AssertionError("expected finalize PUT to have an empty body")
            finalize_uploads.append(url)
            return FakeResponse(status=201)
        if method == "PUT" and "/manifests/" in url:
            body = request.data
            if not isinstance(body, bytes):
                raise AssertionError("expected manifest bytes")
            manifest_publish.append(
                (
                    url,
                    str(_request_header(request, "Content-Type")),
                    len(body),
                )
            )
            return FakeResponse(
                status=201,
                headers={"Docker-Content-Digest": "sha256:published-manifest"},
            )
        raise AssertionError(f"unexpected registry request: {method} {url}")

    monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)

    result = publish_environment_oci_layout(
        environment="debian:12-base",
        layout_root=tmp_path / "oci_layouts",
        registry="registry-1.docker.io",
        repository="thalesmaciel/pyro-environment-debian-12-base",
        reference="1.0.0",
    )

    assert result["manifest_digest"] == "sha256:published-manifest"
    assert len(result["uploaded_blobs"]) == 4
    assert all(bool(entry["uploaded"]) for entry in result["uploaded_blobs"])
    assert len(patch_uploads) == 4
    assert len(finalize_uploads) == 4
    assert all(
        url.startswith("https://registry-1.docker.io/upload/session-")
        for url in finalize_uploads
    )
    assert all("digest=sha256%3A" in url for url in finalize_uploads)
    assert len(manifest_publish) == 1
    assert manifest_publish[0][0].endswith("/manifests/1.0.0")
    assert manifest_publish[0][1] == "application/vnd.oci.image.manifest.v1+json"


def test_publish_environment_oci_layout_requires_credentials(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized_sources",
        platform="linux-x86_64",
    )
    export_environment_oci_layout(
        paths,
        environment="debian:12-base",
        output_dir=tmp_path / "oci_layouts",
    )
    monkeypatch.delenv("OCI_REGISTRY_USERNAME", raising=False)
    monkeypatch.delenv("OCI_REGISTRY_PASSWORD", raising=False)
    monkeypatch.delenv(DEFAULT_DOCKERHUB_USERNAME_ENV, raising=False)
    monkeypatch.delenv(DEFAULT_DOCKERHUB_TOKEN_ENV, raising=False)

    with pytest.raises(
        RuntimeError,
        match=(
            "OCI registry credentials are not configured; set "
            "OCI_REGISTRY_USERNAME/OCI_REGISTRY_PASSWORD or "
            "DOCKERHUB_USERNAME/DOCKERHUB_TOKEN"
        ),
    ):
        publish_environment_oci_layout(
            environment="debian:12-base",
            layout_root=tmp_path / "oci_layouts",
            registry="registry-1.docker.io",
            repository="thalesmaciel/pyro-environment-debian-12-base",
            reference="1.0.0",
        )


def test_sync_bundle_replaces_existing_bundle_dir(tmp_path: Path) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized_sources",
        platform="linux-x86_64",
    )
    paths.build_platform_root.mkdir(parents=True, exist_ok=True)
    (paths.build_root / "NOTICE").write_text("built notice\n", encoding="utf-8")
    (paths.build_platform_root / "manifest.json").write_text("{}", encoding="utf-8")
    existing_dir = bundle_dir / "linux-x86_64"
    existing_dir.mkdir(parents=True, exist_ok=True)
    (existing_dir / "stale.txt").write_text("stale\n", encoding="utf-8")

    sync_bundle(paths)

    assert not (existing_dir / "stale.txt").exists()
    assert (existing_dir / "manifest.json").exists()
    assert (bundle_dir / "NOTICE").read_text(encoding="utf-8") == "built notice\n"


def test_build_bundle_rejects_platform_mismatch(tmp_path: Path) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    lock_path = source_dir / "linux-x86_64/runtime.lock.json"
    lock = json.loads(lock_path.read_text(encoding="utf-8"))
    lock["platform"] = "linux-aarch64"
    lock_path.write_text(json.dumps(lock, indent=2) + "\n", encoding="utf-8")
    paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized_sources",
        platform="linux-x86_64",
    )

    with pytest.raises(RuntimeError, match="does not match requested platform"):
        build_bundle(paths, sync=False)


def test_materialize_sources_dispatches_steps(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    source_dir, build_dir, bundle_dir = _make_source_tree(tmp_path)
    paths = _build_paths(
        source_dir=source_dir,
        build_dir=build_dir,
        bundle_dir=bundle_dir,
        materialized_dir=tmp_path / "materialized_sources",
        platform="linux-x86_64",
    )
    lock = _load_lock(paths)
    calls: list[str] = []
    monkeypatch.setattr("pyro_mcp.runtime_build._load_lock", lambda _: lock)
    monkeypatch.setattr(
        "pyro_mcp.runtime_build.materialize_binaries",
        lambda runtime_paths, runtime_lock: calls.append(
            f"binaries:{runtime_paths.platform}:{runtime_lock.platform}"
        ),
    )
    monkeypatch.setattr(
        "pyro_mcp.runtime_build.materialize_kernel",
        lambda runtime_paths, runtime_lock: calls.append(
            f"kernel:{runtime_paths.platform}:{runtime_lock.platform}"
        ),
    )
    monkeypatch.setattr(
        "pyro_mcp.runtime_build.materialize_rootfs",
        lambda runtime_paths, runtime_lock: calls.append(
            f"rootfs:{runtime_paths.platform}:{runtime_lock.platform}"
        ),
    )

    materialize_sources(paths)

    assert calls == [
        "binaries:linux-x86_64:linux-x86_64",
        "kernel:linux-x86_64:linux-x86_64",
        "rootfs:linux-x86_64:linux-x86_64",
    ]