Make oras-py behave the same way as oras-go for deciding whether to unpack a layer

CHANGELOG.md

@@ -14,6 +14,7 @@ and **Merged pull requests**. Critical items to know are:
 The versions coincide with releases on pip. Only major versions will be released as tags on Github.
 
 ## [0.0.x](https://github.com/oras-project/oras-py/tree/main) (0.0.x)
+ - use same annotation as oras-go for determining whether to unpack a layer or not (0.2.26)
  - check for blob existence before uploading (0.2.26)
    - fix get_tags for ECR when limit is None, closes issue [173](https://github.com/oras-project/oras-py/issues/173)
    - fix empty token for anon tokens to work, closes issue [167](https://github.com/oras-project/oras-py/issues/167)

oras/provider.py

@@ -781,23 +781,24 @@ def push(
                         f"Blob {blob} is not in the present working directory context."
                     )
 
-            # Save directory or blob name before compressing
-            blob_name = os.path.basename(blob)
-
             # If it's a directory, we need to compress
-            cleanup_blob = False
-            if os.path.isdir(blob):
-                blob = oras.utils.make_targz(blob)
-                cleanup_blob = True
+            is_dir_layer = os.path.isdir(blob)
 
+            blob_to_use_for_layer = (
+                oras.utils.make_targz(blob) if is_dir_layer else blob
+            )
             # Create a new layer from the blob
-            layer = oras.oci.NewLayer(blob, is_dir=cleanup_blob, media_type=media_type)
+            layer = oras.oci.NewLayer(
+                blob_to_use_for_layer, is_dir=is_dir_layer, media_type=media_type
+            )
             annotations = annotset.get_annotations(blob)
 
             # Always strip blob_name of path separator
-            layer["annotations"] = {
-                oras.defaults.annotation_title: blob_name.strip(os.sep)
-            }
+            layer["annotations"] = {oras.defaults.annotation_title: blob.strip(os.sep)}
+
+            if is_dir_layer:
+                layer["annotations"][oras.defaults.annotation_unpack] = "True"
+
             if annotations:
                 layer["annotations"].update(annotations)
 
@@ -807,7 +808,7 @@ def push(
 
             # Upload the blob layer
             response = self.upload_blob(
-                blob,
+                blob_to_use_for_layer,
                 container,
                 layer,
                 do_chunked=do_chunked,
@@ -816,8 +817,8 @@ def push(
             self._check_200_response(response)
 
             # Do we need to cleanup a temporary targz?
-            if cleanup_blob and os.path.exists(blob):
-                os.remove(blob)
+            if is_dir_layer and os.path.exists(blob_to_use_for_layer):
+                os.remove(blob_to_use_for_layer)
 
         # Add annotations to the manifest, if provided
         manifest_annots = annotset.get_annotations("$manifest") or {}
@@ -872,22 +873,23 @@ def pull(
         allowed_media_type: Optional[List] = None,
         overwrite: bool = True,
         outdir: Optional[str] = None,
+        skip_unpack: bool = False,
     ) -> List[str]:
         """
         Pull an artifact from a target
 
+        :param target: target location to pull from
+        :type target: str
         :param config_path: path to a config file
         :type config_path: str
         :param allowed_media_type: list of allowed media types
         :type allowed_media_type: list or None
         :param overwrite: if output file exists, overwrite
         :type overwrite: bool
-        :param manifest_config_ref: save manifest config to this file
-        :type manifest_config_ref: str
         :param outdir: output directory path
         :type outdir: str
-        :param target: target location to pull from
-        :type target: str
+        :param skip_unpack: skip unpacking layers
+        :type skip_unpack: bool
         """
         container = self.get_container(target)
         self.auth.load_configs(
@@ -899,9 +901,13 @@ def pull(
 
         files = []
         for layer in manifest.get("layers", []):
-            filename = (layer.get("annotations") or {}).get(
-                oras.defaults.annotation_title
-            )
+            annotations: dict = layer.get("annotations", {})
+            filename = annotations.get(oras.defaults.annotation_title)
+            # A directory will need to be uncompressed and moved
+            unpack_layer = annotations.get(oras.defaults.annotation_unpack, False)
+
+            if unpack_layer and skip_unpack:
+                filename += ".tar.gz"
 
             # If we don't have a filename, default to digest. Hopefully does not happen
             if not filename:
@@ -916,8 +922,7 @@ def pull(
                 )
                 continue
 
-            # A directory will need to be uncompressed and moved
-            if layer["mediaType"] == oras.defaults.default_blob_dir_media_type:
+            if unpack_layer and not skip_unpack:
                 targz = oras.utils.get_tmpfile(suffix=".tar.gz")
                 self.download_blob(container, layer["digest"], targz)
 
@@ -927,7 +932,7 @@ def pull(
             # Anything else just extracted directly
             else:
                 self.download_blob(container, layer["digest"], outfile)
-            logger.info(f"Successfully pulled {outfile}.")
+            logger.info(f"Successfully pulled {outfile}")
             files.append(outfile)
         return files
 

oras/tests/test_oras.py

@@ -161,6 +161,25 @@ def test_directory_push_pull(tmp_path, registry, credentials, target_dir):
     assert "artifact.txt" in os.listdir(files[0])
 
 
+@pytest.mark.with_auth(False)
+def test_directory_push_pull_skip_unpack(tmp_path, registry, credentials, target_dir):
+    """
+    Test push and pull for directory
+    """
+    client = oras.client.OrasClient(hostname=registry, insecure=True)
+
+    # Test upload of a directory
+    upload_dir = os.path.join(here, "upload_data")
+    res = client.push(files=[upload_dir], target=target_dir)
+    assert res.status_code == 201
+    files = client.pull(target=target_dir, outdir=tmp_path, skip_unpack=True)
+
+    assert len(files) == 1
+    assert os.path.basename(files[0]) == "upload_data.tar.gz"
+    assert str(tmp_path) in files[0]
+    assert os.path.exists(files[0])
+
+
 @pytest.mark.with_auth(True)
 def test_directory_push_pull_selfsigned_auth(
     tmp_path, registry, credentials, target_dir