Export bootable environments as OCI layouts

This commit is contained in:
Thales Maciel 2026-03-08 18:17:25 -03:00
parent 89f3d6f012
commit f6d3bf0e90
3 changed files with 450 additions and 6 deletions

View file

@ -8,8 +8,10 @@ RUNTIME_SOURCE_DIR ?= runtime_sources
RUNTIME_BUILD_DIR ?= build/runtime_bundle RUNTIME_BUILD_DIR ?= build/runtime_bundle
RUNTIME_BUNDLE_DIR ?= src/pyro_mcp/runtime_bundle RUNTIME_BUNDLE_DIR ?= src/pyro_mcp/runtime_bundle
RUNTIME_MATERIALIZED_DIR ?= build/runtime_sources RUNTIME_MATERIALIZED_DIR ?= build/runtime_sources
RUNTIME_OCI_LAYOUT_DIR ?= build/oci_layouts
RUNTIME_ENVIRONMENT ?= debian:12-base
.PHONY: help setup lint format typecheck test check dist-check demo network-demo doctor ollama ollama-demo run-server install-hooks runtime-bundle runtime-binaries runtime-kernel runtime-rootfs runtime-agent runtime-validate runtime-manifest runtime-sync runtime-clean runtime-fetch-binaries runtime-build-kernel-real runtime-build-rootfs-real runtime-materialize runtime-boot-check runtime-network-check .PHONY: help setup lint format typecheck test check dist-check demo network-demo doctor ollama ollama-demo run-server install-hooks runtime-bundle runtime-binaries runtime-kernel runtime-rootfs runtime-agent runtime-validate runtime-manifest runtime-sync runtime-clean runtime-fetch-binaries runtime-build-kernel-real runtime-build-rootfs-real runtime-materialize runtime-export-environment-oci runtime-boot-check runtime-network-check
help: help:
@printf '%s\n' \ @printf '%s\n' \
@ -40,6 +42,7 @@ help:
' runtime-build-kernel-real Materialize the real guest kernel' \ ' runtime-build-kernel-real Materialize the real guest kernel' \
' runtime-build-rootfs-real Materialize the real guest rootfs images' \ ' runtime-build-rootfs-real Materialize the real guest rootfs images' \
' runtime-materialize Run all real-source materialization steps' \ ' runtime-materialize Run all real-source materialization steps' \
' runtime-export-environment-oci Export one environment as a local OCI layout' \
' runtime-boot-check Validate direct Firecracker boot from the bundled runtime' \ ' runtime-boot-check Validate direct Firecracker boot from the bundled runtime' \
' runtime-network-check Validate outbound guest networking from the bundled runtime' \ ' runtime-network-check Validate outbound guest networking from the bundled runtime' \
' runtime-clean Remove generated runtime build artifacts' ' runtime-clean Remove generated runtime build artifacts'
@ -126,6 +129,9 @@ runtime-build-rootfs-real:
runtime-materialize: runtime-materialize:
uv run python -m pyro_mcp.runtime_build materialize --platform "$(RUNTIME_PLATFORM)" --source-dir "$(RUNTIME_SOURCE_DIR)" --build-dir "$(RUNTIME_BUILD_DIR)" --bundle-dir "$(RUNTIME_BUNDLE_DIR)" --materialized-dir "$(RUNTIME_MATERIALIZED_DIR)" uv run python -m pyro_mcp.runtime_build materialize --platform "$(RUNTIME_PLATFORM)" --source-dir "$(RUNTIME_SOURCE_DIR)" --build-dir "$(RUNTIME_BUILD_DIR)" --bundle-dir "$(RUNTIME_BUNDLE_DIR)" --materialized-dir "$(RUNTIME_MATERIALIZED_DIR)"
runtime-export-environment-oci:
uv run python -m pyro_mcp.runtime_build export-environment-oci --platform "$(RUNTIME_PLATFORM)" --source-dir "$(RUNTIME_SOURCE_DIR)" --build-dir "$(RUNTIME_BUILD_DIR)" --bundle-dir "$(RUNTIME_BUNDLE_DIR)" --materialized-dir "$(RUNTIME_MATERIALIZED_DIR)" --environment "$(RUNTIME_ENVIRONMENT)" --output-dir "$(RUNTIME_OCI_LAYOUT_DIR)"
runtime-boot-check: runtime-boot-check:
uv run python -m pyro_mcp.runtime_boot_check uv run python -m pyro_mcp.runtime_boot_check

View file

@ -4,6 +4,7 @@ from __future__ import annotations
import argparse import argparse
import hashlib import hashlib
import io
import json import json
import shutil import shutil
import subprocess import subprocess
@ -11,16 +12,25 @@ import tarfile
import urllib.request import urllib.request
import uuid import uuid
from dataclasses import dataclass from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from pyro_mcp.runtime import DEFAULT_PLATFORM from pyro_mcp.runtime import DEFAULT_PLATFORM
from pyro_mcp.vm_environments import get_environment
DEFAULT_RUNTIME_SOURCE_DIR = Path("runtime_sources") DEFAULT_RUNTIME_SOURCE_DIR = Path("runtime_sources")
DEFAULT_RUNTIME_BUILD_DIR = Path("build/runtime_bundle") DEFAULT_RUNTIME_BUILD_DIR = Path("build/runtime_bundle")
DEFAULT_RUNTIME_BUNDLE_DIR = Path("src/pyro_mcp/runtime_bundle") DEFAULT_RUNTIME_BUNDLE_DIR = Path("src/pyro_mcp/runtime_bundle")
DEFAULT_RUNTIME_MATERIALIZED_DIR = Path("build/runtime_sources") DEFAULT_RUNTIME_MATERIALIZED_DIR = Path("build/runtime_sources")
DEFAULT_RUNTIME_OCI_LAYOUT_DIR = Path("build/oci_layouts")
DOWNLOAD_CHUNK_SIZE = 1024 * 1024 DOWNLOAD_CHUNK_SIZE = 1024 * 1024
OCI_IMAGE_MANIFEST_MEDIA_TYPE = "application/vnd.oci.image.manifest.v1+json"
OCI_IMAGE_CONFIG_MEDIA_TYPE = "application/vnd.oci.image.config.v1+json"
OCI_IMAGE_LAYER_MEDIA_TYPE = "application/vnd.oci.image.layer.v1.tar"
OCI_IMAGE_INDEX_MEDIA_TYPE = "application/vnd.oci.image.index.v1+json"
OCI_LAYOUT_VERSION = "1.0.0"
VALIDATION_READ_LIMIT = 1024 * 1024
@dataclass(frozen=True) @dataclass(frozen=True)
@ -120,21 +130,139 @@ def _run(command: list[str]) -> None: # pragma: no cover - integration helper
raise RuntimeError(f"command {' '.join(command)!r} failed: {stderr}") raise RuntimeError(f"command {' '.join(command)!r} failed: {stderr}")
def _sha256_bytes(payload: bytes) -> str:
return hashlib.sha256(payload).hexdigest()
def _path_contains_marker(
path: Path,
marker: bytes,
*,
read_limit: int = VALIDATION_READ_LIMIT,
) -> bool:
bytes_remaining = read_limit
overlap = max(len(marker) - 1, 0)
previous_tail = b""
with path.open("rb") as fp:
while bytes_remaining > 0:
chunk = fp.read(min(DOWNLOAD_CHUNK_SIZE, bytes_remaining))
if chunk == b"":
break
payload = previous_tail + chunk
if marker in payload:
return True
previous_tail = payload[-overlap:] if overlap > 0 else b""
bytes_remaining -= len(chunk)
return False
def _environment_slug(environment: str) -> str:
return environment.replace(":", "_").replace("/", "_")
def _platform_to_oci_platform(platform: str) -> tuple[str, str]:
os_name, separator, architecture = platform.partition("-")
if separator == "" or os_name == "" or architecture == "":
raise RuntimeError(f"unsupported runtime platform format: {platform}")
architecture_aliases = {"x86_64": "amd64", "aarch64": "arm64"}
return os_name, architecture_aliases.get(architecture, architecture)
def _blob_path(blobs_dir: Path, digest: str) -> Path:
algorithm, separator, value = digest.partition(":")
if separator == "" or value == "":
raise RuntimeError(f"invalid OCI digest: {digest}")
return blobs_dir / algorithm / value
def _write_blob_bytes(
blobs_dir: Path,
payload: bytes,
) -> tuple[str, int, Path]:
digest = f"sha256:{_sha256_bytes(payload)}"
blob_path = _blob_path(blobs_dir, digest)
blob_path.parent.mkdir(parents=True, exist_ok=True)
blob_path.write_bytes(payload)
return digest, len(payload), blob_path
def _normalized_tar_info(name: str, size: int, *, mode: int = 0o644) -> tarfile.TarInfo:
info = tarfile.TarInfo(name=name)
info.size = size
info.mode = mode
info.mtime = 0
info.uid = 0
info.gid = 0
info.uname = ""
info.gname = ""
return info
def _write_tar_blob_from_path(
blobs_dir: Path,
*,
source_path: Path,
arcname: str,
) -> tuple[str, int, Path]:
temp_dir = blobs_dir / "sha256"
temp_dir.mkdir(parents=True, exist_ok=True)
temp_path = temp_dir / f".tmp-{uuid.uuid4().hex}.tar"
try:
with temp_path.open("wb") as fp:
with tarfile.open(fileobj=fp, mode="w") as archive:
file_size = source_path.stat().st_size
info = _normalized_tar_info(arcname, file_size)
with source_path.open("rb") as source_fp:
archive.addfile(info, source_fp)
digest = f"sha256:{_sha256(temp_path)}"
blob_path = _blob_path(blobs_dir, digest)
blob_path.parent.mkdir(parents=True, exist_ok=True)
size = temp_path.stat().st_size
temp_path.replace(blob_path)
return digest, size, blob_path
except Exception:
temp_path.unlink(missing_ok=True)
raise
def _write_tar_blob_from_bytes(
blobs_dir: Path,
*,
payload: bytes,
arcname: str,
) -> tuple[str, int, Path]:
temp_dir = blobs_dir / "sha256"
temp_dir.mkdir(parents=True, exist_ok=True)
temp_path = temp_dir / f".tmp-{uuid.uuid4().hex}.tar"
try:
with temp_path.open("wb") as fp:
with tarfile.open(fileobj=fp, mode="w") as archive:
info = _normalized_tar_info(arcname, len(payload))
archive.addfile(info, io.BytesIO(payload))
digest = f"sha256:{_sha256(temp_path)}"
blob_path = _blob_path(blobs_dir, digest)
blob_path.parent.mkdir(parents=True, exist_ok=True)
size = temp_path.stat().st_size
temp_path.replace(blob_path)
return digest, size, blob_path
except Exception:
temp_path.unlink(missing_ok=True)
raise
def validate_sources(paths: RuntimeBuildPaths, lock: RuntimeBuildLock) -> None: def validate_sources(paths: RuntimeBuildPaths, lock: RuntimeBuildLock) -> None:
firecracker_source = _resolved_source_path(paths, lock.binaries["firecracker"]) firecracker_source = _resolved_source_path(paths, lock.binaries["firecracker"])
jailer_source = _resolved_source_path(paths, lock.binaries["jailer"]) jailer_source = _resolved_source_path(paths, lock.binaries["jailer"])
firecracker_text = firecracker_source.read_text(encoding="utf-8", errors="ignore")
jailer_text = jailer_source.read_text(encoding="utf-8", errors="ignore")
has_shim_binaries = ( has_shim_binaries = (
"bundled firecracker shim" in firecracker_text or "bundled jailer shim" in jailer_text _path_contains_marker(firecracker_source, b"bundled firecracker shim")
or _path_contains_marker(jailer_source, b"bundled jailer shim")
) )
has_placeholder_profiles = False has_placeholder_profiles = False
for profile in lock.profiles.values(): for profile in lock.profiles.values():
for kind in ("kernel", "rootfs"): for kind in ("kernel", "rootfs"):
source = _resolved_source_path(paths, profile[kind]) source = _resolved_source_path(paths, profile[kind])
text = source.read_text(encoding="utf-8", errors="ignore") if _path_contains_marker(source, b"placeholder-"):
if "placeholder-" in text:
has_placeholder_profiles = True has_placeholder_profiles = True
break break
if has_placeholder_profiles: if has_placeholder_profiles:
@ -365,6 +493,170 @@ def stage_agent(paths: RuntimeBuildPaths, lock: RuntimeBuildLock) -> None:
dest.chmod(dest.stat().st_mode | 0o111) dest.chmod(dest.stat().st_mode | 0o111)
def export_environment_oci_layout(
paths: RuntimeBuildPaths,
*,
environment: str,
output_dir: Path,
reference: str | None = None,
) -> dict[str, Any]:
lock = _load_lock(paths)
validate_sources(paths, lock)
spec = get_environment(environment)
try:
profile = lock.profiles[spec.source_profile]
except KeyError as exc:
raise RuntimeError(
f"runtime lock does not define source profile {spec.source_profile!r} "
f"for environment {environment!r}"
) from exc
kernel_path = _resolved_source_path(paths, profile["kernel"])
rootfs_path = _resolved_source_path(paths, profile["rootfs"])
if not kernel_path.exists() or not rootfs_path.exists():
raise RuntimeError(
f"missing artifacts for environment {environment!r}; expected "
f"{kernel_path} and {rootfs_path}"
)
layout_dir = output_dir / _environment_slug(environment)
if layout_dir.exists():
shutil.rmtree(layout_dir)
blobs_dir = layout_dir / "blobs"
blob_sha_dir = blobs_dir / "sha256"
blob_sha_dir.mkdir(parents=True, exist_ok=True)
metadata_payload = {
"environment": spec.name,
"version": spec.version,
"platform": spec.platform,
"distribution": spec.distribution,
"distribution_version": spec.distribution_version,
"description": spec.description,
"default_packages": list(spec.default_packages),
"source_profile": spec.source_profile,
"bundle_version": lock.bundle_version,
"component_versions": lock.component_versions,
"capabilities": lock.capabilities,
}
metadata_bytes = json.dumps(metadata_payload, indent=2, sort_keys=True).encode("utf-8") + b"\n"
kernel_digest, kernel_size, _ = _write_tar_blob_from_path(
blobs_dir,
source_path=kernel_path,
arcname="vmlinux",
)
rootfs_digest, rootfs_size, _ = _write_tar_blob_from_path(
blobs_dir,
source_path=rootfs_path,
arcname="rootfs.ext4",
)
metadata_digest, metadata_size, _ = _write_tar_blob_from_bytes(
blobs_dir,
payload=metadata_bytes,
arcname="environment.json",
)
created_at = datetime.now(UTC).isoformat().replace("+00:00", "Z")
ref_name = reference or spec.version
os_name, architecture = _platform_to_oci_platform(spec.platform)
labels = {
"io.pyro.environment": spec.name,
"io.pyro.environment.version": spec.version,
"io.pyro.source_profile": spec.source_profile,
"org.opencontainers.image.title": spec.name,
"org.opencontainers.image.version": spec.version,
}
config_payload = {
"created": created_at,
"architecture": architecture,
"os": os_name,
"config": {"Labels": labels},
"rootfs": {
"type": "layers",
"diff_ids": [kernel_digest, rootfs_digest, metadata_digest],
},
"history": [
{"created": created_at, "created_by": "pyro runtime_build export-environment-oci"}
],
}
config_bytes = json.dumps(config_payload, indent=2, sort_keys=True).encode("utf-8") + b"\n"
config_digest, config_size, _ = _write_blob_bytes(blobs_dir, config_bytes)
manifest_payload = {
"schemaVersion": 2,
"mediaType": OCI_IMAGE_MANIFEST_MEDIA_TYPE,
"config": {
"mediaType": OCI_IMAGE_CONFIG_MEDIA_TYPE,
"digest": config_digest,
"size": config_size,
},
"layers": [
{
"mediaType": OCI_IMAGE_LAYER_MEDIA_TYPE,
"digest": kernel_digest,
"size": kernel_size,
"annotations": {"org.opencontainers.image.title": "vmlinux"},
},
{
"mediaType": OCI_IMAGE_LAYER_MEDIA_TYPE,
"digest": rootfs_digest,
"size": rootfs_size,
"annotations": {"org.opencontainers.image.title": "rootfs.ext4"},
},
{
"mediaType": OCI_IMAGE_LAYER_MEDIA_TYPE,
"digest": metadata_digest,
"size": metadata_size,
"annotations": {"org.opencontainers.image.title": "environment.json"},
},
],
"annotations": labels,
}
manifest_bytes = json.dumps(manifest_payload, indent=2, sort_keys=True).encode("utf-8") + b"\n"
manifest_digest, manifest_size, _ = _write_blob_bytes(blobs_dir, manifest_bytes)
index_payload = {
"schemaVersion": 2,
"mediaType": OCI_IMAGE_INDEX_MEDIA_TYPE,
"manifests": [
{
"mediaType": OCI_IMAGE_MANIFEST_MEDIA_TYPE,
"digest": manifest_digest,
"size": manifest_size,
"annotations": {
"org.opencontainers.image.ref.name": ref_name,
"org.opencontainers.image.title": spec.name,
},
"platform": {"os": os_name, "architecture": architecture},
}
],
}
(layout_dir / "oci-layout").write_text(
json.dumps({"imageLayoutVersion": OCI_LAYOUT_VERSION}, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
(layout_dir / "index.json").write_text(
json.dumps(index_payload, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
return {
"environment": spec.name,
"version": spec.version,
"reference": ref_name,
"layout_dir": str(layout_dir),
"manifest_digest": manifest_digest,
"config_digest": config_digest,
"layers": [
{"title": "vmlinux", "digest": kernel_digest, "size": kernel_size},
{"title": "rootfs.ext4", "digest": rootfs_digest, "size": rootfs_size},
{"title": "environment.json", "digest": metadata_digest, "size": metadata_size},
],
}
def generate_manifest(paths: RuntimeBuildPaths, lock: RuntimeBuildLock) -> dict[str, Any]: def generate_manifest(paths: RuntimeBuildPaths, lock: RuntimeBuildLock) -> dict[str, Any]:
manifest: dict[str, Any] = { manifest: dict[str, Any] = {
"bundle_version": lock.bundle_version, "bundle_version": lock.bundle_version,
@ -468,6 +760,7 @@ def _build_parser() -> argparse.ArgumentParser: # pragma: no cover - CLI wiring
"build-kernel", "build-kernel",
"build-rootfs", "build-rootfs",
"materialize", "materialize",
"export-environment-oci",
"stage-binaries", "stage-binaries",
"stage-kernel", "stage-kernel",
"stage-rootfs", "stage-rootfs",
@ -483,6 +776,9 @@ def _build_parser() -> argparse.ArgumentParser: # pragma: no cover - CLI wiring
parser.add_argument("--build-dir", default=str(DEFAULT_RUNTIME_BUILD_DIR)) parser.add_argument("--build-dir", default=str(DEFAULT_RUNTIME_BUILD_DIR))
parser.add_argument("--bundle-dir", default=str(DEFAULT_RUNTIME_BUNDLE_DIR)) parser.add_argument("--bundle-dir", default=str(DEFAULT_RUNTIME_BUNDLE_DIR))
parser.add_argument("--materialized-dir", default=str(DEFAULT_RUNTIME_MATERIALIZED_DIR)) parser.add_argument("--materialized-dir", default=str(DEFAULT_RUNTIME_MATERIALIZED_DIR))
parser.add_argument("--environment")
parser.add_argument("--output-dir", default=str(DEFAULT_RUNTIME_OCI_LAYOUT_DIR))
parser.add_argument("--reference")
return parser return parser
@ -508,6 +804,17 @@ def main() -> None: # pragma: no cover - CLI wiring
if args.command == "materialize": if args.command == "materialize":
materialize_sources(paths) materialize_sources(paths)
return return
if args.command == "export-environment-oci":
if not isinstance(args.environment, str) or args.environment == "":
raise RuntimeError("--environment is required for export-environment-oci")
result = export_environment_oci_layout(
paths,
environment=args.environment,
output_dir=Path(args.output_dir),
reference=args.reference,
)
print(json.dumps(result, indent=2, sort_keys=True))
return
if args.command == "bundle": if args.command == "bundle":
build_bundle(paths, sync=True) build_bundle(paths, sync=True)
return return
@ -538,3 +845,7 @@ def main() -> None: # pragma: no cover - CLI wiring
sync_bundle(paths) sync_bundle(paths)
return return
raise RuntimeError(f"unknown command: {args.command}") raise RuntimeError(f"unknown command: {args.command}")
if __name__ == "__main__": # pragma: no cover - CLI entrypoint
main()

View file

@ -1,5 +1,6 @@
from __future__ import annotations from __future__ import annotations
import argparse
import hashlib import hashlib
import json import json
import tarfile import tarfile
@ -11,7 +12,9 @@ from pyro_mcp.runtime_build import (
_build_paths, _build_paths,
_load_lock, _load_lock,
build_bundle, build_bundle,
export_environment_oci_layout,
generate_manifest, generate_manifest,
main,
materialize_binaries, materialize_binaries,
stage_agent, stage_agent,
stage_binaries, stage_binaries,
@ -187,3 +190,127 @@ def test_runtime_build_materializes_firecracker_release(tmp_path: Path) -> None:
assert (paths.materialized_platform_root / "bin/jailer").read_text(encoding="utf-8") == ( assert (paths.materialized_platform_root / "bin/jailer").read_text(encoding="utf-8") == (
"real-jailer\n" "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"