feat: add ability to get layer digests

Add method for fetching layer digests of a specified image.

coregio/registry_api.py

@@ -401,3 +401,30 @@ def get_tags(self, repository: str, page_size: int = 100, limit: int = 2000) ->
         return self.get_paginated_response(
             uri, list_name="tags", page_size=page_size, limit=limit
         )
+
+    def get_layers(
+        self,
+        repository: str,
+        reference: str,
+    ) -> List[str]:
+        """
+        Get hashes of layers in a repository by a reference (manifest digest or tag).
+
+        Args:
+            repository (str): Repository name
+            reference (str): Manifest digest or tag
+
+        Returns:
+            list: Hashes of the layers ordered from the base image
+                (the base image is at index 0)
+        """
+        manifest = self.get_manifest(repository, reference)
+        if manifest["schemaVersion"] == 1:
+            return [layer["blobSum"] for layer in manifest["fsLayers"]][::-1]
+        if manifest["mediaType"] in (
+            ACCEPT_HEADERS["docker_manifest_v2"],
+            ACCEPT_HEADERS["oci_manifest"],
+        ):
+            return [layer["digest"] for layer in manifest["layers"]]
+        LOGGER.warning("Failed to extract layers for %s:%s", repository, reference)
+        return []

tests/test_registry_api.py

@@ -235,3 +235,52 @@ def test_get_tags(mock_get: MagicMock) -> None:
     mock_get.assert_called_once_with(
         "v2/repo/tags/list", list_name="tags", page_size=100, limit=2000
     )
+
+
+@pytest.mark.parametrize(
+    ["manifest", "layers"],
+    [
+        (
+            {
+                "schemaVersion": 1,
+                "fsLayers": [{"blobSum": "2"}, {"blobSum": "1"}],
+            },
+            ["1", "2"],
+        ),
+        (
+            {
+                "schemaVersion": 2,
+                "mediaType": "application/vnd.docker.distribution.manifest.v2+json",
+                "layers": [{"digest": "1"}, {"digest": "2"}],
+            },
+            ["1", "2"],
+        ),
+        (
+            {
+                "schemaVersion": 2,
+                "mediaType": "application/vnd.oci.image.manifest.v1+json",
+                "layers": [{"digest": "1"}, {"digest": "2"}],
+            },
+            ["1", "2"],
+        ),
+        (
+            {
+                "schemaVersion": 2,
+                "mediaType": "application/vnd.docker.distribution.manifest.list.v2+json",
+            },
+            [],
+        ),
+    ],
+)
+@patch("coregio.registry_api.ContainerRegistry.get_manifest")
+def test_get_layers(
+    mock_get_manifest: MagicMock,
+    manifest: Any,
+    layers: Any,
+) -> None:
+    mock_get_manifest.return_value = manifest
+    registry_api = ContainerRegistry(url="registry")
+    result = registry_api.get_layers("repo", "ref")
+
+    assert result == layers
+    mock_get_manifest.assert_called_once_with("repo", "ref")