Implement manifest command (#378)

Co-authored-by: Gabriel de Marmiesse <[email protected]>

python_on_whales/components/manifest/cli_wrapper.py

@@ -1,19 +1,155 @@
-from python_on_whales.client_config import DockerCLICaller
+from typing import Any, Dict, List, Union
+
+from python_on_whales.client_config import (
+    ClientConfig,
+    DockerCLICaller,
+    ReloadableObjectFromJson,
+)
+from python_on_whales.components.buildx.imagetools.models import ImageVariantManifest
+from python_on_whales.components.manifest.models import ManifestListInspectResult
+from python_on_whales.utils import run, to_list
+
+
+class ManifestList(ReloadableObjectFromJson):
+    def __init__(
+        self, client_config: ClientConfig, reference: str, is_immutable_id=False
+    ):
+        self.reference = reference
+        super().__init__(client_config, "name", reference, is_immutable_id)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.remove()
+
+    def _fetch_inspect_result_json(self, reference):
+        return f'[{run(self.docker_cmd + ["manifest", "inspect", reference])}]'
+
+    def _parse_json_object(
+        self, json_object: Dict[str, Any]
+    ) -> ManifestListInspectResult:
+        json_object["name"] = self.reference
+        return ManifestListInspectResult.parse_obj(json_object)
+
+    def _get_inspect_result(self) -> ManifestListInspectResult:
+        """Only there to allow tools to know the return type"""
+        return super()._get_inspect_result()
+
+    @property
+    def name(self) -> str:
+        return self._get_inspect_result().name
+
+    @property
+    def schema_version(self) -> str:
+        return self._get_inspect_result().schema_version
+
+    @property
+    def media_type(self) -> str:
+        return self._get_inspect_result().media_type
+
+    @property
+    def manifests(self) -> List[ImageVariantManifest]:
+        return self._get_inspect_result().manifests
+
+    def __repr__(self):
+        return f"python_on_whales.ManifestList(name='{self.name}')"
+
+    def remove(self) -> None:
+        """Removes this Docker manifest list.
+
+        Rather than removing it manually, you can use a context manager to
+        make sure the manifest list is deleted even if an exception is raised.
+        """
+        ManifestCLI(self.client_config).remove(self)
+
+
+ValidManifestList = Union[ManifestList, str]
 
 
 class ManifestCLI(DockerCLICaller):
-    def annotate(self):
-        """Not yet implemented"""
-        raise NotImplementedError
+    def annotate(
+        self,
+        name: str,
+        manifest: str,
+        arch: str = None,
+        os: str = None,
+        os_features: List[str] = None,
+        os_version: str = None,
+        variant: str = None,
+    ) -> ManifestList:
+        """Annotates a Docker manifest list.
+
+        # Arguments
+            name: The name of the manifest list
+            manifest: The individual manifest to annotate
+            arch: The manifest's architecture
+            os: The manifest's operating system
+            os_features: The manifest's operating system features
+            os_version: The manifest's operating system version
+            variant: The manifest's architecture variant
+        """
+        full_cmd = self.docker_cmd + ["manifest", "annotate"]
+        full_cmd.add_simple_arg("--arch", arch)
+        full_cmd.add_simple_arg("--os", os)
+        full_cmd.add_simple_arg("--os-features", os_features)
+        full_cmd.add_simple_arg("--os-version", os_version)
+        full_cmd.add_simple_arg("--variant", variant)
+        full_cmd.append(name)
+        full_cmd.append(manifest)
+        run(full_cmd)
+
+    def create(
+        self,
+        name: str,
+        manifests: List[str],
+        ammend: bool = False,
+        insecure: bool = False,
+    ) -> ManifestList:
+        """Creates a Docker manifest list.
+
+        # Arguments
+            name: The name of the manifest list
+            manifests: The list of manifests to add to the manifest list
+
+        # Returns
+            A `python_on_whales.ManifestList`.
+        """
+        full_cmd = self.docker_cmd + ["manifest", "create"]
+        full_cmd.add_flag("--amend", ammend)
+        full_cmd.add_flag("--insecure", insecure)
+        full_cmd.append(name)
+        full_cmd += to_list(manifests)
+        return ManifestList(
+            self.client_config, run(full_cmd)[22:], is_immutable_id=True
+        )
+
+    def inspect(self, x: str) -> ManifestList:
+        """Returns a Docker manifest list object."""
+        return ManifestList(self.client_config, x)
+
+    def push(self, x: str, purge: bool = False, quiet: bool = False):
+        """Push a manifest list to a repository.
+
+        # Options
+            purge: Remove the local manifest list after push
+        """
+        # this is just to raise a correct exception if the manifest list doesn't exist
+        self.inspect(x)
 
-    def create(self):
-        """Not yet implemented"""
-        raise NotImplementedError
+        full_cmd = self.docker_cmd + ["manifest", "push"]
+        full_cmd.add_flag("--purge", purge)
+        full_cmd.append(x)
+        run(full_cmd, capture_stdout=quiet, capture_stderr=quiet)
 
-    def inspect(self):
-        """Not yet implemented"""
-        raise NotImplementedError
+    def remove(self, manifest_lists: Union[ValidManifestList, List[ValidManifestList]]):
+        """Removes a Docker manifest list or lists.
 
-    def push(self):
-        """Not yet implemented"""
-        raise NotImplementedError
+        # Arguments
+            manifest_lists: One or more manifest lists.
+        """
+        if manifest_lists == []:
+            return
+        full_cmd = self.docker_cmd + ["manifest", "rm"]
+        full_cmd += to_list(manifest_lists)
+        run(full_cmd)

python_on_whales/components/manifest/models.py

@@ -0,0 +1,12 @@
+from typing import List
+
+from python_on_whales.components.buildx.imagetools.models import ImageVariantManifest
+from python_on_whales.utils import DockerCamelModel, all_fields_optional
+
+
+@all_fields_optional
+class ManifestListInspectResult(DockerCamelModel):
+    name: str
+    schema_version: int
+    media_type: str
+    manifests: List[ImageVariantManifest]

tests/python_on_whales/components/test_manifest.py

@@ -0,0 +1,38 @@
+import pytest
+
+from python_on_whales import docker
+from python_on_whales.components.manifest.cli_wrapper import (
+    ManifestList,
+    ManifestListInspectResult,
+)
+from python_on_whales.test_utils import get_all_jsons, random_name
+
+
+@pytest.fixture
+def with_manifest():
+    manifest_name = random_name()
+    # utilizing old, pre-manifest-list images
+    images = ["busybox:1.26", "busybox:1.27.1"]
+    docker.image.pull(images)
+    with docker.manifest.create(manifest_name, images) as my_manifest:
+        yield my_manifest
+    docker.image.remove(images)
+
+
+@pytest.mark.parametrize("json_file", get_all_jsons("manifests"))
+def test_load_json(json_file):
+    json_as_txt = json_file.read_text()
+    ManifestListInspectResult.parse_raw(json_as_txt)
+    # we could do more checks here if needed
+
+
+def test_manifest_create_remove(with_manifest):
+    assert isinstance(with_manifest, ManifestList)
+
+
+def test_manifest_annotate(with_manifest):
+    docker.manifest.annotate(
+        with_manifest.name, "busybox:1.26", os="linux", arch="arm64"
+    )
+    assert with_manifest.manifests[0].platform.os == "linux"
+    assert with_manifest.manifests[0].platform.architecture == "arm64"