From 2c3588f52ae1f650b0890d5261999135bee10b94 Mon Sep 17 00:00:00 2001 From: "Ian H. Pittwood" Date: Thu, 19 Feb 2026 14:56:36 -0700 Subject: [PATCH 1/6] Add ORAS CLI integration for multi-platform manifest management Implements an oras-based alternative to `docker buildx imagetools create` for merging multi-platform images. This avoids authentication issues that affect Docker's imagetools when performing cross-registry operations. The workflow: 1. Creates a temporary manifest index from platform-specific sources 2. Copies the index to all target registries/tags 3. Deletes the temporary index Key components: - find_oras_bin() for binary discovery using existing find_bin() pattern - OrasManifestIndexCreate, OrasCopy, OrasManifestDelete command classes - OrasMergeWorkflow orchestrator with from_image_target() factory method - Source validation ensuring all sources are from the same repository Includes 37 unit tests covering command construction, execution, validation, workflow orchestration, and ImageTarget integration. Co-Authored-By: Claude Opus 4.5 --- .../posit_bakery/image/oras/__init__.py | 25 + posit-bakery/posit_bakery/image/oras/oras.py | 351 ++++++++++++ posit-bakery/test/image/oras/__init__.py | 0 posit-bakery/test/image/oras/test_oras.py | 503 ++++++++++++++++++ 4 files changed, 879 insertions(+) create mode 100644 posit-bakery/posit_bakery/image/oras/__init__.py create mode 100644 posit-bakery/posit_bakery/image/oras/oras.py create mode 100644 posit-bakery/test/image/oras/__init__.py create mode 100644 posit-bakery/test/image/oras/test_oras.py diff --git a/posit-bakery/posit_bakery/image/oras/__init__.py b/posit-bakery/posit_bakery/image/oras/__init__.py new file mode 100644 index 00000000..f0f24e44 --- /dev/null +++ b/posit-bakery/posit_bakery/image/oras/__init__.py @@ -0,0 +1,25 @@ +"""ORAS CLI integration for multi-platform manifest management.""" + +from posit_bakery.image.oras.oras import ( + find_oras_bin, + get_repository_from_ref, + parse_image_reference, + OrasCommand, + OrasCopy, + OrasManifestDelete, + OrasManifestIndexCreate, + OrasMergeWorkflow, + OrasMergeWorkflowResult, +) + +__all__ = [ + "find_oras_bin", + "get_repository_from_ref", + "parse_image_reference", + "OrasCommand", + "OrasCopy", + "OrasManifestDelete", + "OrasManifestIndexCreate", + "OrasMergeWorkflow", + "OrasMergeWorkflowResult", +] diff --git a/posit-bakery/posit_bakery/image/oras/oras.py b/posit-bakery/posit_bakery/image/oras/oras.py new file mode 100644 index 00000000..deee8333 --- /dev/null +++ b/posit-bakery/posit_bakery/image/oras/oras.py @@ -0,0 +1,351 @@ +"""ORAS CLI integration for multi-platform manifest management. + +This module provides an alternative to `docker buildx imagetools create` for merging +multi-platform images. It uses the oras CLI to create manifest indexes and copy them +to target registries, avoiding authentication issues that affect Docker's imagetools +when performing cross-registry operations. +""" + +import logging +import subprocess +import uuid +from abc import ABC, abstractmethod +from pathlib import Path +from typing import Annotated, Self +from urllib.parse import urlparse + +from pydantic import BaseModel, Field, model_validator + +from posit_bakery.error import BakeryToolRuntimeError +from posit_bakery.util import find_bin + +log = logging.getLogger(__name__) + + +def find_oras_bin(context: Path) -> str: + """Find the path to the oras binary. + + :param context: The project context to search for the binary in. + :return: The path to the oras binary. + :raises BakeryToolNotFoundError: If the oras binary cannot be found. + """ + return find_bin(context, "oras", "ORAS_PATH") or "oras" + + +def parse_image_reference(ref: str) -> tuple[str, str, str]: + """Parse an image reference into its components. + + :param ref: The image reference to parse (e.g., "registry.io/repo/image@sha256:digest"). + :return: A tuple of (registry, repository, tag_or_digest). + """ + # Handle digest references + if "@" in ref: + name_part, digest = ref.rsplit("@", 1) + tag_or_digest = f"@{digest}" + elif ":" in ref and not ref.rsplit(":", 1)[-1].startswith("sha256"): + # Handle tag references, but be careful with ports + parts = ref.rsplit(":", 1) + # Check if the last part looks like a port (all digits) + if parts[-1].isdigit(): + name_part = ref + tag_or_digest = "" + else: + name_part = parts[0] + tag_or_digest = f":{parts[1]}" + else: + name_part = ref + tag_or_digest = "" + + # Split registry from repository + if "/" in name_part: + first_part = name_part.split("/")[0] + # Check if first part looks like a registry (contains . or :) + if "." in first_part or ":" in first_part: + registry = first_part + repository = "/".join(name_part.split("/")[1:]) + else: + # Default registry + registry = "docker.io" + repository = name_part + else: + registry = "docker.io" + repository = name_part + + return registry, repository, tag_or_digest + + +def get_repository_from_ref(ref: str) -> str: + """Extract the full repository (registry/repo) from an image reference. + + :param ref: The image reference. + :return: The registry and repository portion (without tag or digest). + """ + registry, repository, _ = parse_image_reference(ref) + return f"{registry}/{repository}" + + +class OrasCommand(BaseModel, ABC): + """Base class for oras CLI commands.""" + + oras_bin: Annotated[str, Field(description="Path to the oras binary.")] + + @property + @abstractmethod + def command(self) -> list[str]: + """Return the full command to execute.""" + ... + + def run(self, dry_run: bool = False) -> subprocess.CompletedProcess: + """Execute the oras command. + + :param dry_run: If True, log the command without executing it. + :return: The completed process result. + :raises BakeryToolRuntimeError: If the command fails. + """ + cmd = self.command + log.debug(f"Executing oras command: {' '.join(cmd)}") + + if dry_run: + log.info(f"[DRY RUN] Would execute: {' '.join(cmd)}") + return subprocess.CompletedProcess(args=cmd, returncode=0, stdout=b"", stderr=b"") + + result = subprocess.run(cmd, capture_output=True) + + if result.returncode != 0: + raise BakeryToolRuntimeError( + message=f"oras command failed", + tool_name="oras", + cmd=cmd, + stdout=result.stdout, + stderr=result.stderr, + exit_code=result.returncode, + ) + + return result + + +class OrasManifestIndexCreate(OrasCommand): + """Create a manifest index from multiple source images. + + This command creates a multi-platform manifest index pointing to the provided + source images and pushes it to the destination reference. + """ + + sources: Annotated[list[str], Field(description="List of source image references to include in the index.")] + destination: Annotated[str, Field(description="Destination reference for the created index.")] + annotations: Annotated[dict[str, str], Field(default_factory=dict, description="Annotations to add to the index.")] + + @model_validator(mode="after") + def validate_sources_same_repository(self) -> Self: + """Validate that all sources are from the same repository. + + oras manifest index create requires all sources to be in the same repository + because it creates an index that references existing manifests by digest. + """ + if not self.sources: + raise ValueError("At least one source is required.") + + repositories = set() + for source in self.sources: + repo = get_repository_from_ref(source) + repositories.add(repo) + + if len(repositories) > 1: + raise ValueError(f"All sources must be from the same repository. Found: {', '.join(sorted(repositories))}") + + return self + + @property + def command(self) -> list[str]: + """Build the oras manifest index create command.""" + cmd = [self.oras_bin, "manifest", "index", "create", self.destination] + cmd.extend(self.sources) + + for key, value in self.annotations.items(): + cmd.extend(["--annotation", f"{key}={value}"]) + + return cmd + + +class OrasCopy(OrasCommand): + """Copy an image from source to destination. + + This command copies an image (including manifest indexes) from one location + to another, supporting cross-registry copies. + """ + + source: Annotated[str, Field(description="Source image reference to copy.")] + destination: Annotated[str, Field(description="Destination reference.")] + + @property + def command(self) -> list[str]: + """Build the oras cp command.""" + return [self.oras_bin, "cp", self.source, self.destination] + + +class OrasManifestDelete(OrasCommand): + """Delete a manifest from a registry. + + This command deletes a manifest (image or index) from a registry. + """ + + reference: Annotated[str, Field(description="The manifest reference to delete.")] + + @property + def command(self) -> list[str]: + """Build the oras manifest delete command.""" + return [self.oras_bin, "manifest", "delete", "--force", self.reference] + + +class OrasMergeWorkflowResult(BaseModel): + """Result of an ORAS merge workflow execution.""" + + success: Annotated[bool, Field(description="Whether the workflow completed successfully.")] + temp_index_ref: Annotated[str | None, Field(default=None, description="Reference to the temporary index created.")] + destinations: Annotated[list[str], Field(default_factory=list, description="List of destination references.")] + error: Annotated[str | None, Field(default=None, description="Error message if the workflow failed.")] + + +class OrasMergeWorkflow(BaseModel): + """Orchestrates the multi-platform merge workflow using oras. + + This workflow: + 1. Creates a temporary manifest index from platform-specific source images + 2. Copies the index to all target registries/tags + 3. Deletes the temporary index + """ + + oras_bin: Annotated[str, Field(description="Path to the oras binary.")] + sources: Annotated[list[str], Field(description="List of source image references (one per platform).")] + temp_registry: Annotated[str, Field(description="Registry to use for temporary index storage.")] + image_name: Annotated[str, Field(description="Name of the image (used for temp tag).")] + tag_suffixes: Annotated[list[str], Field(description="Tag suffixes to apply to the final image.")] + target_registries: Annotated[list[str], Field(description="Target registries to push to.")] + annotations: Annotated[dict[str, str], Field(default_factory=dict, description="Annotations for the index.")] + + @model_validator(mode="after") + def validate_sources(self) -> Self: + """Validate that sources are provided.""" + if not self.sources: + raise ValueError("At least one source is required.") + return self + + @property + def temp_index_tag(self) -> str: + """Generate a unique temporary index tag.""" + uid = str(uuid.uuid4())[:8] + return f"{self.temp_registry}/{self.image_name}/tmp:{uid}" + + def _get_destinations(self) -> list[str]: + """Generate all destination references.""" + destinations = [] + for registry in self.target_registries: + for suffix in self.tag_suffixes: + # Check if registry already includes a repository path + if "/" in registry.split(":", 1)[0].split("/", 1)[-1]: + # Registry includes repository (e.g., "ghcr.io/org/repo") + destinations.append(f"{registry}:{suffix}") + else: + # Registry is just the host (e.g., "ghcr.io") + destinations.append(f"{registry}/{self.image_name}:{suffix}") + return destinations + + def execute(self, dry_run: bool = False) -> OrasMergeWorkflowResult: + """Execute the merge workflow. + + :param dry_run: If True, log commands without executing them. + :return: Result of the workflow execution. + """ + temp_ref = self.temp_index_tag + destinations = self._get_destinations() + + log.info(f"Starting ORAS merge workflow for {self.image_name}") + log.debug(f"Sources: {self.sources}") + log.debug(f"Temporary index: {temp_ref}") + log.debug(f"Destinations: {destinations}") + + try: + # Step 1: Create the manifest index + log.info(f"Creating manifest index at {temp_ref}") + create_cmd = OrasManifestIndexCreate( + oras_bin=self.oras_bin, + sources=self.sources, + destination=temp_ref, + annotations=self.annotations, + ) + create_cmd.run(dry_run=dry_run) + + # Step 2: Copy to all destinations + for dest in destinations: + log.info(f"Copying index to {dest}") + copy_cmd = OrasCopy( + oras_bin=self.oras_bin, + source=temp_ref, + destination=dest, + ) + copy_cmd.run(dry_run=dry_run) + + # Step 3: Delete the temporary index + log.info(f"Cleaning up temporary index {temp_ref}") + delete_cmd = OrasManifestDelete( + oras_bin=self.oras_bin, + reference=temp_ref, + ) + delete_cmd.run(dry_run=dry_run) + + log.info(f"ORAS merge workflow completed successfully") + return OrasMergeWorkflowResult( + success=True, + temp_index_ref=temp_ref, + destinations=destinations, + ) + + except BakeryToolRuntimeError as e: + log.error(f"ORAS merge workflow failed: {e}") + return OrasMergeWorkflowResult( + success=False, + temp_index_ref=temp_ref, + destinations=destinations, + error=str(e), + ) + + @classmethod + def from_image_target(cls, target: "ImageTarget", oras_bin: str | None = None) -> "OrasMergeWorkflow": + """Create an OrasMergeWorkflow from an ImageTarget. + + :param target: The ImageTarget to merge. + :param oras_bin: Path to the oras binary. If not provided, will be discovered. + :return: A configured OrasMergeWorkflow instance. + :raises ValueError: If the target is missing required settings. + """ + # Import here to avoid circular imports + from posit_bakery.image.image_target import ImageTarget + + if not target.settings.temp_registry: + raise ValueError("ImageTarget must have temp_registry set in settings for ORAS merge workflow.") + + if oras_bin is None: + oras_bin = find_oras_bin(target.context.base_path) + + sources = target._get_merge_sources() + + # Convert labels to annotations + annotations = {k: v for k, v in target.labels.items()} + + # Get target registries - extract base URLs from registries + target_registries = [] + for registry in target.image_version.all_registries: + if hasattr(registry, "repository"): + target_registries.append(f"{registry.base_url}/{registry.repository}") + else: + target_registries.append(f"{registry.base_url}/{target.image_name}") + + return cls( + oras_bin=oras_bin, + sources=sources, + temp_registry=target.settings.temp_registry, + image_name=target.image_name, + tag_suffixes=target.tag_suffixes, + target_registries=target_registries, + annotations=annotations, + ) diff --git a/posit-bakery/test/image/oras/__init__.py b/posit-bakery/test/image/oras/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/posit-bakery/test/image/oras/test_oras.py b/posit-bakery/test/image/oras/test_oras.py new file mode 100644 index 00000000..93dbfa81 --- /dev/null +++ b/posit-bakery/test/image/oras/test_oras.py @@ -0,0 +1,503 @@ +"""Tests for the ORAS CLI integration module.""" + +import subprocess +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from pydantic import ValidationError + +from posit_bakery.error import BakeryToolRuntimeError +from posit_bakery.image.oras import ( + find_oras_bin, + get_repository_from_ref, + parse_image_reference, + OrasCopy, + OrasManifestDelete, + OrasManifestIndexCreate, + OrasMergeWorkflow, + OrasMergeWorkflowResult, +) + +pytestmark = [ + pytest.mark.unit, +] + + +class TestParseImageReference: + """Tests for the parse_image_reference function.""" + + @pytest.mark.parametrize( + "ref,expected", + [ + # Standard registry with digest + ( + "ghcr.io/posit-dev/test/tmp@sha256:abc123", + ("ghcr.io", "posit-dev/test/tmp", "@sha256:abc123"), + ), + # Standard registry with tag + ( + "ghcr.io/posit-dev/test:latest", + ("ghcr.io", "posit-dev/test", ":latest"), + ), + # Registry with port + ( + "localhost:5000/repo/image:tag", + ("localhost:5000", "repo/image", ":tag"), + ), + # Docker Hub implicit registry + ( + "library/ubuntu:22.04", + ("docker.io", "library/ubuntu", ":22.04"), + ), + # Simple image name (Docker Hub) + ( + "ubuntu:22.04", + ("docker.io", "ubuntu", ":22.04"), + ), + # Azure Container Registry + ( + "myregistry.azurecr.io/repo/image@sha256:def456", + ("myregistry.azurecr.io", "repo/image", "@sha256:def456"), + ), + # No tag or digest + ( + "ghcr.io/posit-dev/image", + ("ghcr.io", "posit-dev/image", ""), + ), + ], + ) + def test_parse_image_reference(self, ref, expected): + """Test parsing various image reference formats.""" + result = parse_image_reference(ref) + assert result == expected + + @pytest.mark.parametrize( + "ref,expected_repo", + [ + ("ghcr.io/posit-dev/test/tmp@sha256:abc123", "ghcr.io/posit-dev/test/tmp"), + ("ghcr.io/posit-dev/test:latest", "ghcr.io/posit-dev/test"), + ("localhost:5000/repo/image:tag", "localhost:5000/repo/image"), + ("docker.io/library/ubuntu:22.04", "docker.io/library/ubuntu"), + ], + ) + def test_get_repository_from_ref(self, ref, expected_repo): + """Test extracting repository from image reference.""" + result = get_repository_from_ref(ref) + assert result == expected_repo + + +class TestOrasManifestIndexCreate: + """Tests for the OrasManifestIndexCreate command.""" + + def test_command_construction(self): + """Test that the command is constructed correctly.""" + cmd = OrasManifestIndexCreate( + oras_bin="oras", + sources=[ + "ghcr.io/posit/test/tmp@sha256:amd64digest", + "ghcr.io/posit/test/tmp@sha256:arm64digest", + ], + destination="ghcr.io/posit/test/tmp:merged", + annotations={"org.opencontainers.image.title": "Test Image"}, + ) + + expected = [ + "oras", + "manifest", + "index", + "create", + "ghcr.io/posit/test/tmp:merged", + "ghcr.io/posit/test/tmp@sha256:amd64digest", + "ghcr.io/posit/test/tmp@sha256:arm64digest", + "--annotation", + "org.opencontainers.image.title=Test Image", + ] + assert cmd.command == expected + + def test_command_without_annotations(self): + """Test command construction without annotations.""" + cmd = OrasManifestIndexCreate( + oras_bin="oras", + sources=["ghcr.io/posit/test/tmp@sha256:digest"], + destination="ghcr.io/posit/test/tmp:tag", + ) + + expected = [ + "oras", + "manifest", + "index", + "create", + "ghcr.io/posit/test/tmp:tag", + "ghcr.io/posit/test/tmp@sha256:digest", + ] + assert cmd.command == expected + + def test_validates_sources_same_repository(self): + """Test that validation fails when sources are from different repositories.""" + with pytest.raises(ValidationError) as exc_info: + OrasManifestIndexCreate( + oras_bin="oras", + sources=[ + "ghcr.io/posit/image1/tmp@sha256:digest1", + "ghcr.io/posit/image2/tmp@sha256:digest2", + ], + destination="ghcr.io/posit/test/tmp:tag", + ) + + assert "same repository" in str(exc_info.value).lower() + + def test_validates_sources_required(self): + """Test that validation fails when no sources are provided.""" + with pytest.raises(ValidationError) as exc_info: + OrasManifestIndexCreate( + oras_bin="oras", + sources=[], + destination="ghcr.io/posit/test/tmp:tag", + ) + + assert "at least one source" in str(exc_info.value).lower() + + def test_run_success(self): + """Test successful command execution.""" + cmd = OrasManifestIndexCreate( + oras_bin="oras", + sources=["ghcr.io/posit/test/tmp@sha256:digest"], + destination="ghcr.io/posit/test/tmp:tag", + ) + + with patch("subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess(args=cmd.command, returncode=0, stdout=b"", stderr=b"") + result = cmd.run() + + mock_run.assert_called_once_with(cmd.command, capture_output=True) + assert result.returncode == 0 + + def test_run_failure(self): + """Test command execution failure.""" + cmd = OrasManifestIndexCreate( + oras_bin="oras", + sources=["ghcr.io/posit/test/tmp@sha256:digest"], + destination="ghcr.io/posit/test/tmp:tag", + ) + + with patch("subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess( + args=cmd.command, returncode=1, stdout=b"", stderr=b"error message" + ) + with pytest.raises(BakeryToolRuntimeError) as exc_info: + cmd.run() + + assert exc_info.value.tool_name == "oras" + assert exc_info.value.exit_code == 1 + + def test_dry_run(self): + """Test dry run mode doesn't execute command.""" + cmd = OrasManifestIndexCreate( + oras_bin="oras", + sources=["ghcr.io/posit/test/tmp@sha256:digest"], + destination="ghcr.io/posit/test/tmp:tag", + ) + + with patch("subprocess.run") as mock_run: + result = cmd.run(dry_run=True) + + mock_run.assert_not_called() + assert result.returncode == 0 + + +class TestOrasCopy: + """Tests for the OrasCopy command.""" + + def test_command_construction(self): + """Test that the command is constructed correctly.""" + cmd = OrasCopy( + oras_bin="oras", + source="ghcr.io/posit/test/tmp:source", + destination="docker.io/posit/test:dest", + ) + + expected = ["oras", "cp", "ghcr.io/posit/test/tmp:source", "docker.io/posit/test:dest"] + assert cmd.command == expected + + def test_run_success(self): + """Test successful copy execution.""" + cmd = OrasCopy( + oras_bin="oras", + source="ghcr.io/posit/test:source", + destination="docker.io/posit/test:dest", + ) + + with patch("subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess(args=cmd.command, returncode=0, stdout=b"", stderr=b"") + result = cmd.run() + + mock_run.assert_called_once_with(cmd.command, capture_output=True) + assert result.returncode == 0 + + +class TestOrasManifestDelete: + """Tests for the OrasManifestDelete command.""" + + def test_command_construction(self): + """Test that the command is constructed correctly.""" + cmd = OrasManifestDelete( + oras_bin="oras", + reference="ghcr.io/posit/test/tmp:tag", + ) + + expected = ["oras", "manifest", "delete", "--force", "ghcr.io/posit/test/tmp:tag"] + assert cmd.command == expected + + def test_run_success(self): + """Test successful delete execution.""" + cmd = OrasManifestDelete( + oras_bin="oras", + reference="ghcr.io/posit/test/tmp:tag", + ) + + with patch("subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess(args=cmd.command, returncode=0, stdout=b"", stderr=b"") + result = cmd.run() + + mock_run.assert_called_once_with(cmd.command, capture_output=True) + assert result.returncode == 0 + + +class TestOrasMergeWorkflow: + """Tests for the OrasMergeWorkflow orchestrator.""" + + @pytest.fixture + def basic_workflow(self): + """Return a basic workflow configuration.""" + return OrasMergeWorkflow( + oras_bin="oras", + sources=[ + "ghcr.io/posit/test/tmp@sha256:amd64digest", + "ghcr.io/posit/test/tmp@sha256:arm64digest", + ], + temp_registry="ghcr.io/posit-dev", + image_name="test-image", + tag_suffixes=["1.0.0", "latest"], + target_registries=["ghcr.io/posit-dev/test-image", "docker.io/posit/test-image"], + annotations={"org.opencontainers.image.title": "Test Image"}, + ) + + def test_temp_index_tag_generation(self, basic_workflow): + """Test that temporary index tag is generated with unique ID.""" + tag = basic_workflow.temp_index_tag + assert tag.startswith("ghcr.io/posit-dev/test-image/tmp:") + # Should have an 8-character UUID suffix + suffix = tag.split(":")[-1] + assert len(suffix) == 8 + + def test_destinations_generation(self, basic_workflow): + """Test that destinations are generated correctly.""" + destinations = basic_workflow._get_destinations() + + expected = [ + "ghcr.io/posit-dev/test-image:1.0.0", + "ghcr.io/posit-dev/test-image:latest", + "docker.io/posit/test-image:1.0.0", + "docker.io/posit/test-image:latest", + ] + assert destinations == expected + + def test_execute_success(self, basic_workflow): + """Test successful workflow execution.""" + with patch("subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess(args=[], returncode=0, stdout=b"", stderr=b"") + result = basic_workflow.execute() + + assert result.success is True + assert result.error is None + assert len(result.destinations) == 4 + assert result.temp_index_ref is not None + + # Should have called: + # 1 create + 4 copy + 1 delete = 6 calls + assert mock_run.call_count == 6 + + def test_execute_dry_run(self, basic_workflow): + """Test dry run mode.""" + with patch("subprocess.run") as mock_run: + result = basic_workflow.execute(dry_run=True) + + mock_run.assert_not_called() + assert result.success is True + assert len(result.destinations) == 4 + + def test_execute_failure_on_create(self, basic_workflow): + """Test workflow handles failure during index creation.""" + with patch("subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess( + args=[], returncode=1, stdout=b"", stderr=b"failed to create index" + ) + result = basic_workflow.execute() + + assert result.success is False + assert result.error is not None + # Should fail on first call (create) + assert mock_run.call_count == 1 + + def test_execute_failure_on_copy(self, basic_workflow): + """Test workflow handles failure during copy.""" + call_count = 0 + + def side_effect(*args, **kwargs): + nonlocal call_count + call_count += 1 + # Fail on second call (first copy) + if call_count == 2: + return subprocess.CompletedProcess(args=[], returncode=1, stdout=b"", stderr=b"copy failed") + return subprocess.CompletedProcess(args=[], returncode=0, stdout=b"", stderr=b"") + + with patch("subprocess.run", side_effect=side_effect): + result = basic_workflow.execute() + + assert result.success is False + assert result.error is not None + + def test_validates_sources_required(self): + """Test that validation fails when no sources are provided.""" + with pytest.raises(ValidationError) as exc_info: + OrasMergeWorkflow( + oras_bin="oras", + sources=[], + temp_registry="ghcr.io/posit-dev", + image_name="test-image", + tag_suffixes=["latest"], + target_registries=["ghcr.io/posit-dev/test-image"], + ) + + assert "at least one source" in str(exc_info.value).lower() + + +class TestOrasMergeWorkflowFromImageTarget: + """Tests for creating OrasMergeWorkflow from ImageTarget.""" + + @pytest.fixture + def mock_image_target(self): + """Create a mock ImageTarget for testing.""" + mock_target = MagicMock() + mock_target.image_name = "test-image" + mock_target.context.base_path = Path("/project") + mock_target.settings.temp_registry = "ghcr.io/posit-dev" + mock_target._get_merge_sources.return_value = [ + "ghcr.io/posit-dev/test/tmp@sha256:amd64", + "ghcr.io/posit-dev/test/tmp@sha256:arm64", + ] + mock_target.labels = { + "org.opencontainers.image.title": "Test Image", + "org.opencontainers.image.version": "1.0.0", + } + mock_target.tag_suffixes = ["1.0.0", "latest"] + + # Mock registries + mock_registry1 = MagicMock() + mock_registry1.base_url = "ghcr.io" + mock_registry1.repository = "posit-dev/test-image" + + mock_registry2 = MagicMock() + mock_registry2.base_url = "docker.io" + del mock_registry2.repository # Simulate registry without repository + + mock_target.image_version.all_registries = [mock_registry1, mock_registry2] + + return mock_target + + def test_from_image_target(self, mock_image_target): + """Test creating workflow from ImageTarget.""" + with patch("posit_bakery.image.oras.oras.find_oras_bin", return_value="oras"): + workflow = OrasMergeWorkflow.from_image_target(mock_image_target) + + assert workflow.oras_bin == "oras" + assert workflow.image_name == "test-image" + assert workflow.temp_registry == "ghcr.io/posit-dev" + assert len(workflow.sources) == 2 + assert workflow.tag_suffixes == ["1.0.0", "latest"] + assert len(workflow.target_registries) == 2 + assert "org.opencontainers.image.title" in workflow.annotations + + def test_from_image_target_missing_temp_registry(self, mock_image_target): + """Test that ValueError is raised when temp_registry is not set.""" + mock_image_target.settings.temp_registry = None + + with pytest.raises(ValueError) as exc_info: + OrasMergeWorkflow.from_image_target(mock_image_target) + + assert "temp_registry" in str(exc_info.value).lower() + + def test_from_image_target_with_custom_oras_bin(self, mock_image_target): + """Test creating workflow with custom oras binary path.""" + workflow = OrasMergeWorkflow.from_image_target(mock_image_target, oras_bin="/custom/path/oras") + + assert workflow.oras_bin == "/custom/path/oras" + + +class TestFindOrasBin: + """Tests for the find_oras_bin function.""" + + def test_find_from_env_var(self, tmp_path): + """Test finding oras from environment variable.""" + with patch.dict("os.environ", {"ORAS_PATH": "/custom/oras"}): + result = find_oras_bin(tmp_path) + assert result == "/custom/oras" + + def test_find_from_path(self, tmp_path): + """Test finding oras from PATH.""" + with patch.dict("os.environ", {}, clear=False): + # Remove ORAS_PATH if it exists + import os + + os.environ.pop("ORAS_PATH", None) + + with patch("shutil.which", return_value="/usr/bin/oras"): + result = find_oras_bin(tmp_path) + assert result == "oras" + + def test_find_from_tools_dir(self, tmp_path): + """Test finding oras from project tools directory.""" + tools_dir = tmp_path / "tools" + tools_dir.mkdir() + oras_bin = tools_dir / "oras" + oras_bin.touch() + + with patch.dict("os.environ", {}, clear=False): + import os + + os.environ.pop("ORAS_PATH", None) + + # Patch 'which' in the module where it's imported + with patch("posit_bakery.util.which", return_value=None): + result = find_oras_bin(tmp_path) + + assert result == str(oras_bin) + + +class TestOrasMergeWorkflowResult: + """Tests for the OrasMergeWorkflowResult model.""" + + def test_success_result(self): + """Test creating a successful result.""" + result = OrasMergeWorkflowResult( + success=True, + temp_index_ref="ghcr.io/test/tmp:abc123", + destinations=["ghcr.io/test:1.0.0", "ghcr.io/test:latest"], + ) + + assert result.success is True + assert result.error is None + assert len(result.destinations) == 2 + + def test_failure_result(self): + """Test creating a failure result.""" + result = OrasMergeWorkflowResult( + success=False, + temp_index_ref="ghcr.io/test/tmp:abc123", + destinations=[], + error="Command failed with exit code 1", + ) + + assert result.success is False + assert result.error is not None From 0acb3a98659e8ebf31d23f2154de4a5dafe5ed4f Mon Sep 17 00:00:00 2001 From: "Ian H. Pittwood" Date: Mon, 23 Feb 2026 09:28:26 -0700 Subject: [PATCH 2/6] Add ORAS installer step to CI --- .github/workflows/ci.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 36a57e31..8676e71a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -62,6 +62,8 @@ jobs: - name: Setup docker buildx uses: docker/setup-buildx-action@v3 + - uses: oras-project/setup-oras@v1 + - name: Install poetry working-directory: ./posit-bakery run: | From e0a329df3186972461568a39986b1280812e45f1 Mon Sep 17 00:00:00 2001 From: "Ian H. Pittwood" Date: Mon, 23 Feb 2026 09:31:51 -0700 Subject: [PATCH 3/6] Remove header docstring and unused import --- posit-bakery/posit_bakery/image/oras/oras.py | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/posit-bakery/posit_bakery/image/oras/oras.py b/posit-bakery/posit_bakery/image/oras/oras.py index deee8333..a9a719d1 100644 --- a/posit-bakery/posit_bakery/image/oras/oras.py +++ b/posit-bakery/posit_bakery/image/oras/oras.py @@ -1,18 +1,9 @@ -"""ORAS CLI integration for multi-platform manifest management. - -This module provides an alternative to `docker buildx imagetools create` for merging -multi-platform images. It uses the oras CLI to create manifest indexes and copy them -to target registries, avoiding authentication issues that affect Docker's imagetools -when performing cross-registry operations. -""" - import logging import subprocess import uuid from abc import ABC, abstractmethod from pathlib import Path from typing import Annotated, Self -from urllib.parse import urlparse from pydantic import BaseModel, Field, model_validator @@ -319,7 +310,6 @@ def from_image_target(cls, target: "ImageTarget", oras_bin: str | None = None) - :raises ValueError: If the target is missing required settings. """ # Import here to avoid circular imports - from posit_bakery.image.image_target import ImageTarget if not target.settings.temp_registry: raise ValueError("ImageTarget must have temp_registry set in settings for ORAS merge workflow.") From 618b319201d5bad414f33aa65ab6a9336c584bcf Mon Sep 17 00:00:00 2001 From: "Ian H. Pittwood" Date: Mon, 23 Feb 2026 12:25:01 -0700 Subject: [PATCH 4/6] Skip unnecessary annotation copy --- posit-bakery/posit_bakery/image/oras/oras.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/posit-bakery/posit_bakery/image/oras/oras.py b/posit-bakery/posit_bakery/image/oras/oras.py index a9a719d1..f7a6a56b 100644 --- a/posit-bakery/posit_bakery/image/oras/oras.py +++ b/posit-bakery/posit_bakery/image/oras/oras.py @@ -319,9 +319,6 @@ def from_image_target(cls, target: "ImageTarget", oras_bin: str | None = None) - sources = target._get_merge_sources() - # Convert labels to annotations - annotations = {k: v for k, v in target.labels.items()} - # Get target registries - extract base URLs from registries target_registries = [] for registry in target.image_version.all_registries: @@ -337,5 +334,5 @@ def from_image_target(cls, target: "ImageTarget", oras_bin: str | None = None) - image_name=target.image_name, tag_suffixes=target.tag_suffixes, target_registries=target_registries, - annotations=annotations, + annotations=target.labels.items(), ) From 0b50896a13b355902163270abaa0a89730de165d Mon Sep 17 00:00:00 2001 From: "Ian H. Pittwood" Date: Mon, 23 Feb 2026 12:46:10 -0700 Subject: [PATCH 5/6] Rework OrasMergeWorkflow to utilize `image_target` directly --- posit-bakery/posit_bakery/image/oras/oras.py | 82 ++++++-------------- posit-bakery/test/image/oras/test_oras.py | 8 +- 2 files changed, 29 insertions(+), 61 deletions(-) diff --git a/posit-bakery/posit_bakery/image/oras/oras.py b/posit-bakery/posit_bakery/image/oras/oras.py index f7a6a56b..0d389285 100644 --- a/posit-bakery/posit_bakery/image/oras/oras.py +++ b/posit-bakery/posit_bakery/image/oras/oras.py @@ -1,6 +1,6 @@ +import hashlib import logging import subprocess -import uuid from abc import ABC, abstractmethod from pathlib import Path from typing import Annotated, Self @@ -207,11 +207,7 @@ class OrasMergeWorkflow(BaseModel): """ oras_bin: Annotated[str, Field(description="Path to the oras binary.")] - sources: Annotated[list[str], Field(description="List of source image references (one per platform).")] - temp_registry: Annotated[str, Field(description="Registry to use for temporary index storage.")] - image_name: Annotated[str, Field(description="Name of the image (used for temp tag).")] - tag_suffixes: Annotated[list[str], Field(description="Tag suffixes to apply to the final image.")] - target_registries: Annotated[list[str], Field(description="Target registries to push to.")] + image_target: Annotated["ImageTarget", Field(description="The image target of the sources.")] annotations: Annotated[dict[str, str], Field(default_factory=dict, description="Annotations for the index.")] @model_validator(mode="after") @@ -224,79 +220,65 @@ def validate_sources(self) -> Self: @property def temp_index_tag(self) -> str: """Generate a unique temporary index tag.""" - uid = str(uuid.uuid4())[:8] - return f"{self.temp_registry}/{self.image_name}/tmp:{uid}" - - def _get_destinations(self) -> list[str]: - """Generate all destination references.""" - destinations = [] - for registry in self.target_registries: - for suffix in self.tag_suffixes: - # Check if registry already includes a repository path - if "/" in registry.split(":", 1)[0].split("/", 1)[-1]: - # Registry includes repository (e.g., "ghcr.io/org/repo") - destinations.append(f"{registry}:{suffix}") - else: - # Registry is just the host (e.g., "ghcr.io") - destinations.append(f"{registry}/{self.image_name}:{suffix}") - return destinations - - def execute(self, dry_run: bool = False) -> OrasMergeWorkflowResult: - """Execute the merge workflow. + source_hash = hashlib.sha256("".join(self.image_target._get_merge_sources()).encode("UTF-8")).hexdigest()[:10] + return ( + f"{self.image_target.temp_registry}/{self.image_target.image_name}/tmp:{self.image_target.uid}{source_hash}" + ) + + def run(self, dry_run: bool = False) -> OrasMergeWorkflowResult: + """Run the merge workflow. :param dry_run: If True, log commands without executing them. :return: Result of the workflow execution. """ - temp_ref = self.temp_index_tag - destinations = self._get_destinations() - log.info(f"Starting ORAS merge workflow for {self.image_name}") - log.debug(f"Sources: {self.sources}") - log.debug(f"Temporary index: {temp_ref}") - log.debug(f"Destinations: {destinations}") + log.info(f"Starting ORAS merge workflow for {self.image_target.image_name}") + log.debug(f"Sources: {self.image_target.sources}") + log.debug(f"Temporary index: {self.temp_index_tag}") + log.debug(f"Destinations: {self.image_target.tags}") try: # Step 1: Create the manifest index - log.info(f"Creating manifest index at {temp_ref}") + log.info(f"Creating manifest index at {self.temp_index_tag}") create_cmd = OrasManifestIndexCreate( oras_bin=self.oras_bin, - sources=self.sources, - destination=temp_ref, + sources=self.image_target._get_merge_sources(), + destination=self.temp_index_tag, annotations=self.annotations, ) create_cmd.run(dry_run=dry_run) # Step 2: Copy to all destinations - for dest in destinations: + for dest in self.image_target.tags: log.info(f"Copying index to {dest}") copy_cmd = OrasCopy( oras_bin=self.oras_bin, - source=temp_ref, + source=self.temp_index_tag, destination=dest, ) copy_cmd.run(dry_run=dry_run) # Step 3: Delete the temporary index - log.info(f"Cleaning up temporary index {temp_ref}") + log.info(f"Cleaning up temporary index {self.temp_index_tag}") delete_cmd = OrasManifestDelete( oras_bin=self.oras_bin, - reference=temp_ref, + reference=self.temp_index_tag, ) delete_cmd.run(dry_run=dry_run) log.info(f"ORAS merge workflow completed successfully") return OrasMergeWorkflowResult( success=True, - temp_index_ref=temp_ref, - destinations=destinations, + temp_index_ref=self.temp_index_tag, + destinations=self.image_target.tags, ) except BakeryToolRuntimeError as e: log.error(f"ORAS merge workflow failed: {e}") return OrasMergeWorkflowResult( success=False, - temp_index_ref=temp_ref, - destinations=destinations, + temp_index_ref=self.temp_index_tag, + destinations=self.image_target.tags, error=str(e), ) @@ -317,22 +299,8 @@ def from_image_target(cls, target: "ImageTarget", oras_bin: str | None = None) - if oras_bin is None: oras_bin = find_oras_bin(target.context.base_path) - sources = target._get_merge_sources() - - # Get target registries - extract base URLs from registries - target_registries = [] - for registry in target.image_version.all_registries: - if hasattr(registry, "repository"): - target_registries.append(f"{registry.base_url}/{registry.repository}") - else: - target_registries.append(f"{registry.base_url}/{target.image_name}") - return cls( oras_bin=oras_bin, - sources=sources, - temp_registry=target.settings.temp_registry, - image_name=target.image_name, - tag_suffixes=target.tag_suffixes, - target_registries=target_registries, + image_target=target, annotations=target.labels.items(), ) diff --git a/posit-bakery/test/image/oras/test_oras.py b/posit-bakery/test/image/oras/test_oras.py index 93dbfa81..ad465f4e 100644 --- a/posit-bakery/test/image/oras/test_oras.py +++ b/posit-bakery/test/image/oras/test_oras.py @@ -307,7 +307,7 @@ def test_execute_success(self, basic_workflow): """Test successful workflow execution.""" with patch("subprocess.run") as mock_run: mock_run.return_value = subprocess.CompletedProcess(args=[], returncode=0, stdout=b"", stderr=b"") - result = basic_workflow.execute() + result = basic_workflow.run() assert result.success is True assert result.error is None @@ -321,7 +321,7 @@ def test_execute_success(self, basic_workflow): def test_execute_dry_run(self, basic_workflow): """Test dry run mode.""" with patch("subprocess.run") as mock_run: - result = basic_workflow.execute(dry_run=True) + result = basic_workflow.run(dry_run=True) mock_run.assert_not_called() assert result.success is True @@ -333,7 +333,7 @@ def test_execute_failure_on_create(self, basic_workflow): mock_run.return_value = subprocess.CompletedProcess( args=[], returncode=1, stdout=b"", stderr=b"failed to create index" ) - result = basic_workflow.execute() + result = basic_workflow.run() assert result.success is False assert result.error is not None @@ -353,7 +353,7 @@ def side_effect(*args, **kwargs): return subprocess.CompletedProcess(args=[], returncode=0, stdout=b"", stderr=b"") with patch("subprocess.run", side_effect=side_effect): - result = basic_workflow.execute() + result = basic_workflow.run() assert result.success is False assert result.error is not None From dcac285c719e7c7dbc4ad104377dfef9fae2b05a Mon Sep 17 00:00:00 2001 From: "Ian H. Pittwood" Date: Tue, 24 Feb 2026 15:00:43 -0700 Subject: [PATCH 6/6] Fix OrasMergeWorkflow bugs and update tests for new interface - Fix validate_sources validator to use image_target._get_merge_sources() - Add sources property to expose merge sources from image target - Fix from_image_target to pass target.labels instead of target.labels.items() - Fix log statement to use self.sources instead of self.image_target.sources - Add destination property to Tag model for grouping by registry/repository - Group copy operations by destination for more efficient tag pushing - Update tests to use mock ImageTarget fixtures Co-Authored-By: Claude Opus 4.5 --- .../posit_bakery/image/image_target.py | 12 ++ posit-bakery/posit_bakery/image/oras/oras.py | 34 ++++-- posit-bakery/test/image/oras/test_oras.py | 114 +++++++++++------- 3 files changed, 106 insertions(+), 54 deletions(-) diff --git a/posit-bakery/posit_bakery/image/image_target.py b/posit-bakery/posit_bakery/image/image_target.py index 6ea39769..b95a3411 100644 --- a/posit-bakery/posit_bakery/image/image_target.py +++ b/posit-bakery/posit_bakery/image/image_target.py @@ -65,6 +65,18 @@ def validate(self): raise ValueError("At least one of registry, repository, or suffix must be provided for a valid tag.") return self + @property + def destination(self): + """Return the destination portion of the tag (registry and repository) without the suffix or digest.""" + destination = "" + if self.registry: + destination += self.registry.base_url + if self.repository: + if len(destination) > 0 and not destination.endswith("/"): + destination += "/" + destination += self.repository + return destination + def __hash__(self): return hash((self.registry.base_url if self.registry else None, self.repository, self.suffix, self.digest)) diff --git a/posit-bakery/posit_bakery/image/oras/oras.py b/posit-bakery/posit_bakery/image/oras/oras.py index 0d389285..0888c586 100644 --- a/posit-bakery/posit_bakery/image/oras/oras.py +++ b/posit-bakery/posit_bakery/image/oras/oras.py @@ -1,11 +1,15 @@ import hashlib +import itertools import logging import subprocess from abc import ABC, abstractmethod from pathlib import Path -from typing import Annotated, Self +from typing import Annotated, Any, Self, TYPE_CHECKING -from pydantic import BaseModel, Field, model_validator +if TYPE_CHECKING: + from posit_bakery.image.image_target import ImageTarget + +from pydantic import BaseModel, ConfigDict, Field, model_validator from posit_bakery.error import BakeryToolRuntimeError from posit_bakery.util import find_bin @@ -206,17 +210,24 @@ class OrasMergeWorkflow(BaseModel): 3. Deletes the temporary index """ + model_config = ConfigDict(arbitrary_types_allowed=True) + oras_bin: Annotated[str, Field(description="Path to the oras binary.")] - image_target: Annotated["ImageTarget", Field(description="The image target of the sources.")] + image_target: Annotated[Any, Field(description="The image target of the sources.")] annotations: Annotated[dict[str, str], Field(default_factory=dict, description="Annotations for the index.")] @model_validator(mode="after") def validate_sources(self) -> Self: """Validate that sources are provided.""" - if not self.sources: + if not self.image_target._get_merge_sources(): raise ValueError("At least one source is required.") return self + @property + def sources(self) -> list[str]: + """Get the list of source image references from the image target.""" + return self.image_target._get_merge_sources() + @property def temp_index_tag(self) -> str: """Generate a unique temporary index tag.""" @@ -233,7 +244,7 @@ def run(self, dry_run: bool = False) -> OrasMergeWorkflowResult: """ log.info(f"Starting ORAS merge workflow for {self.image_target.image_name}") - log.debug(f"Sources: {self.image_target.sources}") + log.debug(f"Sources: {self.sources}") log.debug(f"Temporary index: {self.temp_index_tag}") log.debug(f"Destinations: {self.image_target.tags}") @@ -249,12 +260,13 @@ def run(self, dry_run: bool = False) -> OrasMergeWorkflowResult: create_cmd.run(dry_run=dry_run) # Step 2: Copy to all destinations - for dest in self.image_target.tags: - log.info(f"Copying index to {dest}") + for destination, tags in itertools.groupby(self.image_target.tags, lambda x: x.destination): + log.info(f"Copying index to {destination}") + combine_tag_str = destination + ":" + ",".join(tag.suffix for tag in tags) copy_cmd = OrasCopy( oras_bin=self.oras_bin, source=self.temp_index_tag, - destination=dest, + destination=combine_tag_str, ) copy_cmd.run(dry_run=dry_run) @@ -270,7 +282,7 @@ def run(self, dry_run: bool = False) -> OrasMergeWorkflowResult: return OrasMergeWorkflowResult( success=True, temp_index_ref=self.temp_index_tag, - destinations=self.image_target.tags, + destinations=[str(tag) for tag in self.image_target.tags], ) except BakeryToolRuntimeError as e: @@ -278,7 +290,7 @@ def run(self, dry_run: bool = False) -> OrasMergeWorkflowResult: return OrasMergeWorkflowResult( success=False, temp_index_ref=self.temp_index_tag, - destinations=self.image_target.tags, + destinations=[str(tag) for tag in self.image_target.tags], error=str(e), ) @@ -302,5 +314,5 @@ def from_image_target(cls, target: "ImageTarget", oras_bin: str | None = None) - return cls( oras_bin=oras_bin, image_target=target, - annotations=target.labels.items(), + annotations=target.labels, ) diff --git a/posit-bakery/test/image/oras/test_oras.py b/posit-bakery/test/image/oras/test_oras.py index ad465f4e..7efa1469 100644 --- a/posit-bakery/test/image/oras/test_oras.py +++ b/posit-bakery/test/image/oras/test_oras.py @@ -268,40 +268,69 @@ class TestOrasMergeWorkflow: """Tests for the OrasMergeWorkflow orchestrator.""" @pytest.fixture - def basic_workflow(self): + def mock_image_target(self): + """Create a mock ImageTarget for testing.""" + mock_target = MagicMock() + mock_target.image_name = "test-image" + mock_target.uid = "test-image-1-0-0" + mock_target.temp_registry = "ghcr.io/posit-dev" + mock_target._get_merge_sources.return_value = [ + "ghcr.io/posit-dev/test/tmp@sha256:amd64digest", + "ghcr.io/posit-dev/test/tmp@sha256:arm64digest", + ] + mock_target.labels = { + "org.opencontainers.image.title": "Test Image", + } + + # Create mock tags + mock_tag1 = MagicMock() + mock_tag1.destination = "ghcr.io/posit-dev/test-image" + mock_tag1.suffix = "1.0.0" + mock_tag1.__str__ = lambda self: "ghcr.io/posit-dev/test-image:1.0.0" + + mock_tag2 = MagicMock() + mock_tag2.destination = "ghcr.io/posit-dev/test-image" + mock_tag2.suffix = "latest" + mock_tag2.__str__ = lambda self: "ghcr.io/posit-dev/test-image:latest" + + mock_tag3 = MagicMock() + mock_tag3.destination = "docker.io/posit/test-image" + mock_tag3.suffix = "1.0.0" + mock_tag3.__str__ = lambda self: "docker.io/posit/test-image:1.0.0" + + mock_tag4 = MagicMock() + mock_tag4.destination = "docker.io/posit/test-image" + mock_tag4.suffix = "latest" + mock_tag4.__str__ = lambda self: "docker.io/posit/test-image:latest" + + mock_target.tags = [mock_tag1, mock_tag2, mock_tag3, mock_tag4] + + return mock_target + + @pytest.fixture + def basic_workflow(self, mock_image_target): """Return a basic workflow configuration.""" return OrasMergeWorkflow( oras_bin="oras", - sources=[ - "ghcr.io/posit/test/tmp@sha256:amd64digest", - "ghcr.io/posit/test/tmp@sha256:arm64digest", - ], - temp_registry="ghcr.io/posit-dev", - image_name="test-image", - tag_suffixes=["1.0.0", "latest"], - target_registries=["ghcr.io/posit-dev/test-image", "docker.io/posit/test-image"], + image_target=mock_image_target, annotations={"org.opencontainers.image.title": "Test Image"}, ) def test_temp_index_tag_generation(self, basic_workflow): """Test that temporary index tag is generated with unique ID.""" tag = basic_workflow.temp_index_tag - assert tag.startswith("ghcr.io/posit-dev/test-image/tmp:") - # Should have an 8-character UUID suffix + assert tag.startswith("ghcr.io/posit-dev/test-image/tmp:test-image-1-0-0") + # Should have a 10-character hash suffix after the uid suffix = tag.split(":")[-1] - assert len(suffix) == 8 - - def test_destinations_generation(self, basic_workflow): - """Test that destinations are generated correctly.""" - destinations = basic_workflow._get_destinations() + assert suffix.startswith("test-image-1-0-0") + assert len(suffix) == len("test-image-1-0-0") + 10 - expected = [ - "ghcr.io/posit-dev/test-image:1.0.0", - "ghcr.io/posit-dev/test-image:latest", - "docker.io/posit/test-image:1.0.0", - "docker.io/posit/test-image:latest", - ] - assert destinations == expected + def test_sources_property(self, basic_workflow): + """Test that sources property returns image target's merge sources.""" + sources = basic_workflow.sources + assert len(sources) == 2 + assert "ghcr.io/posit-dev/test/tmp@sha256:amd64digest" in sources + assert "ghcr.io/posit-dev/test/tmp@sha256:arm64digest" in sources def test_execute_success(self, basic_workflow): """Test successful workflow execution.""" @@ -315,8 +344,8 @@ def test_execute_success(self, basic_workflow): assert result.temp_index_ref is not None # Should have called: - # 1 create + 4 copy + 1 delete = 6 calls - assert mock_run.call_count == 6 + # 1 create + 2 copy (grouped by destination) + 1 delete = 4 calls + assert mock_run.call_count == 4 def test_execute_dry_run(self, basic_workflow): """Test dry run mode.""" @@ -360,14 +389,13 @@ def side_effect(*args, **kwargs): def test_validates_sources_required(self): """Test that validation fails when no sources are provided.""" + mock_target = MagicMock() + mock_target._get_merge_sources.return_value = [] + with pytest.raises(ValidationError) as exc_info: OrasMergeWorkflow( oras_bin="oras", - sources=[], - temp_registry="ghcr.io/posit-dev", - image_name="test-image", - tag_suffixes=["latest"], - target_registries=["ghcr.io/posit-dev/test-image"], + image_target=mock_target, ) assert "at least one source" in str(exc_info.value).lower() @@ -381,8 +409,10 @@ def mock_image_target(self): """Create a mock ImageTarget for testing.""" mock_target = MagicMock() mock_target.image_name = "test-image" + mock_target.uid = "test-image-1-0-0" mock_target.context.base_path = Path("/project") mock_target.settings.temp_registry = "ghcr.io/posit-dev" + mock_target.temp_registry = "ghcr.io/posit-dev" mock_target._get_merge_sources.return_value = [ "ghcr.io/posit-dev/test/tmp@sha256:amd64", "ghcr.io/posit-dev/test/tmp@sha256:arm64", @@ -391,18 +421,19 @@ def mock_image_target(self): "org.opencontainers.image.title": "Test Image", "org.opencontainers.image.version": "1.0.0", } - mock_target.tag_suffixes = ["1.0.0", "latest"] - # Mock registries - mock_registry1 = MagicMock() - mock_registry1.base_url = "ghcr.io" - mock_registry1.repository = "posit-dev/test-image" + # Create mock tags + mock_tag1 = MagicMock() + mock_tag1.destination = "ghcr.io/posit-dev/test-image" + mock_tag1.suffix = "1.0.0" + mock_tag1.__str__ = lambda self: "ghcr.io/posit-dev/test-image:1.0.0" - mock_registry2 = MagicMock() - mock_registry2.base_url = "docker.io" - del mock_registry2.repository # Simulate registry without repository + mock_tag2 = MagicMock() + mock_tag2.destination = "ghcr.io/posit-dev/test-image" + mock_tag2.suffix = "latest" + mock_tag2.__str__ = lambda self: "ghcr.io/posit-dev/test-image:latest" - mock_target.image_version.all_registries = [mock_registry1, mock_registry2] + mock_target.tags = [mock_tag1, mock_tag2] return mock_target @@ -412,11 +443,8 @@ def test_from_image_target(self, mock_image_target): workflow = OrasMergeWorkflow.from_image_target(mock_image_target) assert workflow.oras_bin == "oras" - assert workflow.image_name == "test-image" - assert workflow.temp_registry == "ghcr.io/posit-dev" + assert workflow.image_target is mock_image_target assert len(workflow.sources) == 2 - assert workflow.tag_suffixes == ["1.0.0", "latest"] - assert len(workflow.target_registries) == 2 assert "org.opencontainers.image.title" in workflow.annotations def test_from_image_target_missing_temp_registry(self, mock_image_target):