thaloco/pyro-mcp
1
Fork 0
Code Pull requests Releases Packages Activity Actions

Switch official environment publishing to Docker Hub

Browse source
This commit is contained in:
Thales Maciel 2026-03-09 17:39:18 -03:00
parent 0c4ac17b82
commit 6988d85f7d
11 changed files with 590 additions and 73 deletions
Download patch file Download diff file Expand all files Collapse all files

410
tests/test_runtime_build.py
View file

@ -13,9 +13,14 @@ 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,
@ -362,13 +367,77 @@ def test_runtime_build_main_dispatches_export(
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://ghcr.io/token?"):
if url.startswith("https://auth.docker.io/token?"):
requested_token_urls.append(url)
authorization = (
_request_header(request, "Authorization")
if isinstance(request, urllib.request.Request)
@ -389,9 +458,9 @@ def test_request_registry_retries_with_bearer_token(monkeypatch: pytest.MonkeyPa
_http_headers(
{
"WWW-Authenticate": (
'Bearer realm="https://ghcr.io/token",'
'service="ghcr.io",'
'scope="repository:thaloco/pyro-environments/debian-12-base:pull,push"'
'Bearer realm="https://auth.docker.io/token",'
'service="registry.docker.io",'
'scope="repository:thalesmaciel/pyro-environment-debian-12-base:pull,push"'
)
}
),
@ -402,9 +471,9 @@ def test_request_registry_retries_with_bearer_token(monkeypatch: pytest.MonkeyPa
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
status, payload, headers = _request_registry(
"https://ghcr.io/v2/thaloco/pyro-environments/debian-12-base/test",
"https://registry-1.docker.io/v2/thalesmaciel/pyro-environment-debian-12-base/test",
method="GET",
repository="thaloco/pyro-environments/debian-12-base",
repository="thalesmaciel/pyro-environment-debian-12-base",
scope_actions="pull,push",
username="user",
password="token",
@ -416,9 +485,45 @@ def test_request_registry_retries_with_bearer_token(monkeypatch: pytest.MonkeyPa
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)
@ -434,18 +539,20 @@ def test_request_registry_requires_auth_challenge(monkeypatch: pytest.MonkeyPatc
with pytest.raises(RuntimeError, match="denied access without an auth challenge"):
_request_registry(
"https://ghcr.io/v2/thaloco/pyro-environments/debian-12-base/test",
"https://registry-1.docker.io/v2/thalesmaciel/pyro-environment-debian-12-base/test",
method="GET",
repository="thaloco/pyro-environments/debian-12-base",
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://ghcr.io/token?"):
if url.startswith("https://auth.docker.io/token?"):
return FakeResponse(b'{"token":"secret-token"}')
authorization = (
_request_header(request, "Authorization")
@ -460,9 +567,9 @@ def test_request_registry_returns_allowed_retry_status(monkeypatch: pytest.Monke
_http_headers(
{
"WWW-Authenticate": (
'Bearer realm="https://ghcr.io/token",'
'service="ghcr.io",'
'scope="repository:thaloco/pyro-environments/debian-12-base:pull,push"'
'Bearer realm="https://auth.docker.io/token",'
'service="registry.docker.io",'
'scope="repository:thalesmaciel/pyro-environment-debian-12-base:pull,push"'
)
}
),
@ -479,9 +586,9 @@ def test_request_registry_returns_allowed_retry_status(monkeypatch: pytest.Monke
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
status, payload, headers = _request_registry(
"https://ghcr.io/v2/thaloco/pyro-environments/debian-12-base/test",
"https://registry-1.docker.io/v2/thalesmaciel/pyro-environment-debian-12-base/test",
method="GET",
repository="thaloco/pyro-environments/debian-12-base",
repository="thalesmaciel/pyro-environment-debian-12-base",
scope_actions="pull,push",
allow_statuses=(404,),
)
@ -491,6 +598,78 @@ def test_request_registry_returns_allowed_retry_status(monkeypatch: pytest.Monke
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"),
[
@ -523,8 +702,8 @@ def test_upload_blob_skips_existing_blob(monkeypatch: pytest.MonkeyPatch, tmp_pa
monkeypatch.setattr("pyro_mcp.runtime_build._blob_exists", lambda **_: True)
result = _upload_blob(
registry="ghcr.io",
repository="thaloco/pyro-environments/debian-12-base",
registry="registry-1.docker.io",
repository="thalesmaciel/pyro-environment-debian-12-base",
digest="sha256:abc123",
blob_path=blob_path,
username=None,
@ -549,8 +728,132 @@ def test_upload_blob_requires_location(monkeypatch: pytest.MonkeyPatch, tmp_path
with pytest.raises(RuntimeError, match="did not return a blob upload location"):
_upload_blob(
registry="ghcr.io",
repository="thaloco/pyro-environments/debian-12-base",
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,
@ -602,9 +905,13 @@ def test_publish_environment_oci_layout_uploads_blobs_and_manifest(
output_dir=tmp_path / "oci_layouts",
)
uploads: list[tuple[str, int]] = []
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
@ -627,16 +934,20 @@ def test_publish_environment_oci_layout_uploads_blobs_and_manifest(
status=202,
headers={"Location": f"/upload/session-{upload_counter}"},
)
if method == "PUT" and "/upload/session-" in url:
digest = url.split("digest=", 1)[1]
if method == "PATCH" and "/upload/session-" in url:
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)))
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
@ -660,20 +971,65 @@ def test_publish_environment_oci_layout_uploads_blobs_and_manifest(
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",
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(uploads) == 4
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(

35
tests/test_vm_environments.py
View file

@ -246,15 +246,15 @@ def test_fetch_oci_manifest_resolves_linux_amd64_index_with_bearer_auth(
runtime_paths = _fake_runtime_paths(tmp_path)
store = EnvironmentStore(runtime_paths=runtime_paths, cache_dir=tmp_path / "cache")
spec = VmEnvironment(
name="debian:12-ghcr",
name="debian:12-registry",
version="1.0.0",
description="OCI-backed environment",
default_packages=("bash", "git"),
distribution="debian",
distribution_version="12",
source_profile="missing-profile",
oci_registry="ghcr.io",
oci_repository="thaloco/pyro-environments/debian-12",
oci_registry="registry-1.docker.io",
oci_repository="thalesmaciel/pyro-environment-debian-12",
oci_reference="1.0.0",
)
@ -288,7 +288,7 @@ def test_fetch_oci_manifest_resolves_linux_amd64_index_with_bearer_auth(
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?"):
if url.startswith("https://auth.docker.io/token?"):
return FakeResponse(b'{"token":"secret-token"}')
authorization = _authorization_header(request)
if url.endswith("/manifests/1.0.0"):
@ -300,9 +300,9 @@ def test_fetch_oci_manifest_resolves_linux_amd64_index_with_bearer_auth(
_http_headers(
{
"WWW-Authenticate": (
'Bearer realm="https://ghcr.io/token",'
'service="ghcr.io",'
'scope="repository:thaloco/pyro-environments/debian-12:pull"'
'Bearer realm="https://auth.docker.io/token",'
'service="registry.docker.io",'
'scope="repository:thalesmaciel/pyro-environment-debian-12:pull"'
)
}
),
@ -322,9 +322,9 @@ def test_fetch_oci_manifest_resolves_linux_amd64_index_with_bearer_auth(
_http_headers(
{
"WWW-Authenticate": (
'Bearer realm="https://ghcr.io/token",'
'service="ghcr.io",'
'scope="repository:thaloco/pyro-environments/debian-12:pull"'
'Bearer realm="https://auth.docker.io/token",'
'service="registry.docker.io",'
'scope="repository:thalesmaciel/pyro-environment-debian-12:pull"'
)
}
),
@ -345,8 +345,11 @@ def test_fetch_oci_manifest_resolves_linux_amd64_index_with_bearer_auth(
assert manifest == child_manifest
assert resolved_digest == child_digest
assert authorized_urls == [
"https://ghcr.io/v2/thaloco/pyro-environments/debian-12/manifests/1.0.0",
f"https://ghcr.io/v2/thaloco/pyro-environments/debian-12/manifests/{child_digest}",
"https://registry-1.docker.io/v2/thalesmaciel/pyro-environment-debian-12/manifests/1.0.0",
(
"https://registry-1.docker.io/v2/thalesmaciel/pyro-environment-debian-12/"
f"manifests/{child_digest}"
),
]
@ -377,15 +380,15 @@ def test_environment_store_installs_from_oci_when_runtime_source_missing(
manifest_payload = json.dumps(manifest).encode("utf-8")
manifest_digest = _sha256_digest(manifest_payload)
environment = VmEnvironment(
name="debian:12-ghcr",
name="debian:12-registry",
version="1.0.0",
description="OCI-backed environment",
default_packages=("bash", "git"),
distribution="debian",
distribution_version="12",
source_profile="missing-profile",
oci_registry="ghcr.io",
oci_repository="thaloco/pyro-environments/debian-12",
oci_registry="registry-1.docker.io",
oci_repository="thalesmaciel/pyro-environment-debian-12",
oci_reference="1.0.0",
)
@ -415,7 +418,7 @@ def test_environment_store_installs_from_oci_when_runtime_source_missing(
assert installed.kernel_image.read_text(encoding="utf-8") == "kernel\n"
assert installed.rootfs_image.read_text(encoding="utf-8") == "rootfs\n"
assert installed.source == (
"oci://ghcr.io/thaloco/pyro-environments/debian-12"
"oci://registry-1.docker.io/thalesmaciel/pyro-environment-debian-12"
f"@{manifest_digest}"
)
metadata = json.loads((installed.install_dir / "environment.json").read_text(encoding="utf-8"))