758 lines
27 KiB
Python
758 lines
27 KiB
Python
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 (
|
|
_build_paths,
|
|
_load_lock,
|
|
_load_oci_layout_manifest,
|
|
_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_request_registry_retries_with_bearer_token(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
requested_auth: list[str | None] = []
|
|
|
|
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://ghcr.io/token?"):
|
|
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://ghcr.io/token",'
|
|
'service="ghcr.io",'
|
|
'scope="repository:thaloco/pyro-environments/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://ghcr.io/v2/thaloco/pyro-environments/debian-12-base/test",
|
|
method="GET",
|
|
repository="thaloco/pyro-environments/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"
|
|
|
|
|
|
def test_request_registry_requires_auth_challenge(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
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://ghcr.io/v2/thaloco/pyro-environments/debian-12-base/test",
|
|
method="GET",
|
|
repository="thaloco/pyro-environments/debian-12-base",
|
|
scope_actions="pull,push",
|
|
)
|
|
|
|
|
|
def test_request_registry_returns_allowed_retry_status(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
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://ghcr.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://ghcr.io/token",'
|
|
'service="ghcr.io",'
|
|
'scope="repository:thaloco/pyro-environments/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://ghcr.io/v2/thaloco/pyro-environments/debian-12-base/test",
|
|
method="GET",
|
|
repository="thaloco/pyro-environments/debian-12-base",
|
|
scope_actions="pull,push",
|
|
allow_statuses=(404,),
|
|
)
|
|
|
|
assert status == 404
|
|
assert payload == b"missing"
|
|
assert headers["docker-content-digest"] == "sha256:missing"
|
|
|
|
|
|
@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="ghcr.io",
|
|
repository="thaloco/pyro-environments/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="ghcr.io",
|
|
repository="thaloco/pyro-environments/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",
|
|
)
|
|
|
|
uploads: list[tuple[str, int]] = []
|
|
manifest_publish: list[tuple[str, str, int]] = []
|
|
upload_counter = 0
|
|
|
|
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 == "PUT" and "/upload/session-" in url:
|
|
digest = url.split("digest=", 1)[1]
|
|
body = request.data
|
|
if isinstance(body, bytes):
|
|
payload = body
|
|
elif body is not None and hasattr(body, "read"):
|
|
payload = body.read()
|
|
else:
|
|
raise AssertionError("expected upload body")
|
|
uploads.append((digest, len(payload)))
|
|
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="ghcr.io",
|
|
repository="thaloco/pyro-environments/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(uploads) == 4
|
|
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_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",
|
|
]
|