Add support for pushing manifest lists to the registry

closes pulp#469

CHANGES/469.feature

@@ -0,0 +1 @@
+Added support for pushing manifest lists via the Registry API.

pulp_container/app/registry_api.py

@@ -764,6 +764,7 @@ def put(self, request, path, pk=None):
         """
         Responds with the actual manifest
         """
+        # TODO: when uploading an empty manifest list, a new repository has to be created in advance
         _, repository = self.get_dr_push(request, path)
         # iterate over all the layers and create
         chunk = request.META["wsgi.input"]
@@ -777,9 +778,89 @@ def put(self, request, path, pk=None):
         if request.content_type not in (
             models.MEDIA_TYPE.MANIFEST_V2,
             models.MEDIA_TYPE.MANIFEST_OCI,
+            models.MEDIA_TYPE.MANIFEST_LIST,
+            models.MEDIA_TYPE.INDEX_OCI,
         ):
-            # we suport only v2 docker/oci schema upload
+            # we support only v2 docker/oci schema upload
             raise ManifestInvalid(digest=manifest_digest)
+
+        if request.content_type in (
+            models.MEDIA_TYPE.MANIFEST_LIST,
+            models.MEDIA_TYPE.INDEX_OCI,
+        ):
+            manifest_list = models.Manifest(
+                digest=manifest_digest,
+                schema_version=content_data["schemaVersion"],
+                media_type=content_data.get("mediaType", models.MEDIA_TYPE.INDEX_OCI),
+            )
+            try:
+                manifest_list.save()
+            except IntegrityError:
+                manifest = models.Manifest.objects.get(digest=manifest_list.digest)
+                manifest.touch()
+
+            ca = ContentArtifact(
+                artifact=artifact, content=manifest_list, relative_path=manifest_list.digest
+            )
+            try:
+                ca.save()
+            except IntegrityError:
+                pass
+
+            manifests_to_list = []
+            for manifest_metadata in content_data.get("manifests"):
+                manifest = models.Manifest.objects.get(digest=manifest_metadata["digest"])
+
+                platform = manifest_metadata["platform"]
+                manifest_to_list = models.ManifestListManifest(
+                    manifest_list=manifest,
+                    image_manifest=manifest_list,
+                    architecture=platform["architecture"],
+                    os=platform["os"],
+                    features=platform.get("features", ""),
+                    variant=platform.get("variant", ""),
+                    os_version=platform.get("os.version", ""),
+                    os_features=platform.get("os.features", ""),
+                )
+                manifests_to_list.append(manifest_to_list)
+
+            models.ManifestListManifest.objects.bulk_create(manifests_to_list)
+
+            tag = models.Tag(name=pk, tagged_manifest=manifest_list)
+            try:
+                tag.save()
+            except IntegrityError:
+                tag = models.Tag.objects.get(name=tag.name, tagged_manifest=manifest_list)
+                tag.touch()
+
+            tags_to_remove = models.Tag.objects.filter(
+                pk__in=repository.latest_version().content.all(), name=tag
+            ).exclude(tagged_manifest=manifest_list)
+            dispatched_task = dispatch(
+                add_and_remove,
+                exclusive_resources=[repository],
+                kwargs={
+                    "repository_pk": str(repository.pk),
+                    "add_content_units": [str(tag.pk), str(manifest_list.pk)],
+                    "remove_content_units": [str(pk) for pk in tags_to_remove.values_list("pk")],
+                },
+            )
+
+            # Wait a small amount of time
+            for dummy in range(3):
+                time.sleep(1)
+                task = Task.objects.get(pk=dispatched_task.pk)
+                if task.state == "completed":
+                    task.delete()
+                    return ManifestResponse(manifest, path, request, status=201)
+                elif task.state in ["waiting", "running"]:
+                    continue
+                else:
+                    error = task.error
+                    task.delete()
+                    raise Exception(str(error))
+            raise Throttled()
+
         # both docker/oci format should contain config, digest, mediaType, size
         config_layer = content_data.get("config")
         try:

pulp_container/tests/functional/api/test_push_content.py

@@ -10,6 +10,9 @@
     monitor_task,
     PulpTestCase,
 )
+
+from pulp_container.constants import MEDIA_TYPE
+
 from pulp_container.tests.functional.api import rbac_base
 from pulp_container.tests.functional.constants import REGISTRY_V2_REPO_PULP
 from pulp_container.tests.functional.utils import (
@@ -21,6 +24,9 @@
 )
 
 from pulpcore.client.pulp_container import (
+    ContentManifestsApi,
+    ContentTagsApi,
+    DistributionsContainerApi,
     PulpContainerNamespacesApi,
     RepositoriesContainerPushApi,
 )
@@ -310,3 +316,129 @@ def test_matching_username(self):
         # cleanup, namespace removal also removes related distributions
         namespace = self.namespace_api.list(name=namespace_name).results[0]
         self.addCleanup(self.namespace_api.delete, namespace.pulp_href)
+
+
+class PushManifestListTestCase(PulpTestCase, rbac_base.BaseRegistryTest):
+
+    @classmethod
+    def setUpClass(cls):
+        cfg = config.get_config()
+        cls.registry = cli.RegistryClient(cfg)
+        cls.registry.raise_if_unsupported(unittest.SkipTest, "Tests require podman/docker")
+        cls.registry_name = urlparse(cfg.get_base_url()).netloc
+
+        admin_user, admin_password = cfg.pulp_auth
+        cls.user_admin = {"username": admin_user, "password": admin_password}
+
+        api_client = gen_container_client()
+        api_client.configuration.username = cls.user_admin["username"]
+        api_client.configuration.password = cls.user_admin["password"]
+        cls.pushrepository_api = RepositoriesContainerPushApi(api_client)
+        cls.distributions_api = DistributionsContainerApi(api_client)
+        cls.manifests_api = ContentManifestsApi(api_client)
+        cls.tags_api = ContentTagsApi(api_client)
+
+        image_base_path = f"{REGISTRY_V2_REPO_PULP}:manifest_"
+        cls._pull(image_base_path + "a")
+        cls._pull(image_base_path + "b")
+        cls._pull(image_base_path + "c")
+
+        manifest_a_digest = cls.registry.inspect(image_base_path + "a")[0]["Digest"]
+        manifest_b_digest = cls.registry.inspect(image_base_path + "b")[0]["Digest"]
+        manifest_c_digest = cls.registry.inspect(image_base_path + "c")[0]["Digest"]
+        cls.manifests_digests = sorted([manifest_a_digest, manifest_b_digest, manifest_c_digest])
+
+        cls.image_tag = "manifest_list"
+        cls.image_path = f"{REGISTRY_V2_REPO_PULP}:{cls.image_tag}"
+        cls.local_url = f"{cls.registry_name}/foo:{cls.image_tag}"
+        # create a new manifest list composed of three manifest images
+        cls.registry._dispatch_command("manifest", "create", cls.image_path)
+        cls.registry._dispatch_command("manifest", "add", cls.image_path, image_base_path + "a")
+        cls.registry._dispatch_command("manifest", "add", cls.image_path, image_base_path + "b")
+        cls.registry._dispatch_command("manifest", "add", cls.image_path, image_base_path + "c")
+
+        cls.empty_image_tag = "empty_manifest_list"
+        cls.empty_image_path = f"{REGISTRY_V2_REPO_PULP}:{cls.empty_image_tag}"
+        cls.empty_image_local_url = f"{cls.registry_name}/foo:{cls.empty_image_tag}"
+        # create an empty manifest list
+        cls.registry._dispatch_command("manifest", "create", cls.empty_image_path)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.registry._dispatch_command("manifest", "rm", cls.image_path)
+        cls.registry._dispatch_command("manifest", "rm", cls.empty_image_path)
+        delete_orphans()
+
+    def test_push_manifest_list_v2s2(self):
+        self.registry.login(
+            "-u", self.user_admin["username"], "-p", self.user_admin["password"], self.registry_name
+        )
+        self.registry._dispatch_command(
+            "manifest", "push", self.image_path, self.local_url, "--all", "--format", "v2s2"
+        )
+
+        distribution = self.distributions_api.list(name="foo").results[0]
+        self.addCleanup(self.distributions_api.delete, distribution.pulp_href)
+
+        repo_version = self.pushrepository_api.read(distribution.repository).latest_version_href
+        latest_tag = self.tags_api.list(repository_version_added=repo_version).results[0]
+        assert latest_tag.name == self.image_tag
+
+        manifest_list = self.manifests_api.read(latest_tag.tagged_manifest)
+        assert manifest_list.media_type == MEDIA_TYPE.MANIFEST_LIST
+        assert manifest_list.schema_version == 2
+
+        referenced_manifests_digests = sorted(
+            [
+                self.manifests_api.read(manifest_href).digest
+                for manifest_href in manifest_list.listed_manifests
+             ]
+        )
+        assert referenced_manifests_digests == self.manifests_digests
+
+    def test_push_manifest_list_oci(self):
+        self.registry.login(
+            "-u", self.user_admin["username"], "-p", self.user_admin["password"], self.registry_name
+        )
+        self.registry._dispatch_command(
+            "manifest", "push", self.image_path, self.local_url, "--all", "--format", "oci"
+        )
+
+        distribution = self.distributions_api.list(name="foo").results[0]
+        self.addCleanup(self.distributions_api.delete, distribution.pulp_href)
+
+        repo_version = self.pushrepository_api.read(distribution.repository).latest_version_href
+        latest_tag = self.tags_api.list(repository_version_added=repo_version).results[0]
+        assert latest_tag.name == self.image_tag
+
+        manifest_list = self.manifests_api.read(latest_tag.tagged_manifest)
+        assert manifest_list.media_type == MEDIA_TYPE.INDEX_OCI
+        assert manifest_list.schema_version == 2
+
+        referenced_manifests_digests = sorted(
+            [
+                self.manifests_api.read(manifest_href).digest
+                for manifest_href in manifest_list.listed_manifests
+            ]
+        )
+        assert referenced_manifests_digests == self.manifests_digests
+
+    def test_push_empty_manifest_list(self):
+        self.registry.login(
+            "-u", self.user_admin["username"], "-p", self.user_admin["password"], self.registry_name
+        )
+        self.registry._dispatch_command(
+            "manifest", "push", self.empty_image_path, self.empty_image_local_url
+        )
+
+        distribution = self.distributions_api.list(name="foo").results[0]
+        self.addCleanup(self.distributions_api.delete, distribution.pulp_href)
+
+        repo_version = self.pushrepository_api.read(distribution.repository).latest_version_href
+        latest_tag = self.tags_api.list(repository_version_added=repo_version).results[0]
+        assert latest_tag.name == self.empty_image_tag
+
+        manifest_list = self.manifests_api.read(latest_tag.tagged_manifest)
+        assert manifest_list.media_type == MEDIA_TYPE.INDEX_OCI
+        assert manifest_list.schema_version == 2
+        assert manifest_list.listed_manifests == []