Fix issue 93 Windows blob upload (#94)

* Fix issue 93 Windows blob upload

Signed-off-by: Sunny Carter <[email protected]>

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)
+  - patch fix for blob upload Windows, closes issue [93](https://github.com/oras-project/oras-py/issues/93) (0.1.19)
   - patch fix for empty manifest config on Windows, closes issue [90](https://github.com/oras-project/oras-py/issues/90) (0.1.18)
   - patch fix to correct session url pattern, closes issue [78](https://github.com/oras-project/oras-py/issues/78) (0.1.17)
   - add support for tag deletion and retry decorators (0.1.16)

oras/provider.py

@@ -17,6 +17,7 @@
 import oras.schemas
 import oras.utils
 from oras.logger import logger
+from oras.utils.fileio import PathAndOptionalContent
 
 # container type can be string or container
 container_type = Union[str, oras.container.Container]
@@ -164,9 +165,9 @@ def _validate_path(self, path: str) -> bool:
         """
         return os.getcwd() in os.path.abspath(path)
 
-    def _parse_manifest_ref(self, ref: str) -> Union[Tuple[str, str], List[str]]:
+    def _parse_manifest_ref(self, ref: str) -> Tuple[str, str]:
         """
-        Parse an optional manifest config, e.g:
+        Parse an optional manifest config.
 
         Examples
         --------
@@ -176,10 +177,13 @@ def _parse_manifest_ref(self, ref: str) -> Union[Tuple[str, str], List[str]]:
 
         :param ref: the manifest reference to parse (examples above)
         :type ref: str
+        :return - A Tuple of the path and the content-type, using the default unknown
+                  config media type if none found in the reference
         """
-        if ":" not in ref:
-            return ref, oras.defaults.unknown_config_media_type
-        return ref.split(":", 1)
+        path_content: PathAndOptionalContent = oras.utils.split_path_and_content(ref)
+        if not path_content.content:
+            path_content.content = oras.defaults.unknown_config_media_type
+        return path_content.path, path_content.content
 
     def upload_blob(
         self,
@@ -637,8 +641,11 @@ def push(self, *args, **kwargs) -> requests.Response:
         # Upload files as blobs
         for blob in kwargs.get("files", []):
             # You can provide a blob + content type
-            if ":" in str(blob):
-                blob, media_type = str(blob).split(":", 1)
+            path_content: PathAndOptionalContent = oras.utils.split_path_and_content(
+                str(blob)
+            )
+            blob = path_content.path
+            media_type = path_content.content
 
             # Must exist
             if not os.path.exists(blob):

oras/tests/test_provider.py

@@ -8,6 +8,7 @@
 import pytest
 
 import oras.client
+import oras.defaults
 import oras.provider
 import oras.utils
 
@@ -80,3 +81,38 @@ def test_annotated_registry_push(tmp_path):
         res = client.push(
             files=[artifact], target=target, annotation_file=annotation_file
         )
+
+
+def test_parse_manifest():
+    """
+    Test parse manifest function.
+
+    Parse manifest function has additional logic for Windows - this isn't included in
+    these tests as they don't usually run on Windows.
+    """
+    testref = "path/to/config:application/vnd.oci.image.config.v1+json"
+    remote = oras.provider.Registry(hostname=registry, insecure=True)
+    ref, content_type = remote._parse_manifest_ref(testref)
+    assert ref == "path/to/config"
+    assert content_type == "application/vnd.oci.image.config.v1+json"
+
+    testref = "path/to/config:application/vnd.oci.image.config.v1+json:extra"
+    remote = oras.provider.Registry(hostname=registry, insecure=True)
+    ref, content_type = remote._parse_manifest_ref(testref)
+    assert ref == "path/to/config"
+    assert content_type == "application/vnd.oci.image.config.v1+json:extra"
+
+    testref = "/dev/null:application/vnd.oci.image.manifest.v1+json"
+    ref, content_type = remote._parse_manifest_ref(testref)
+    assert ref == "/dev/null"
+    assert content_type == "application/vnd.oci.image.manifest.v1+json"
+
+    testref = "/dev/null"
+    ref, content_type = remote._parse_manifest_ref(testref)
+    assert ref == "/dev/null"
+    assert content_type == oras.defaults.unknown_config_media_type
+
+    testref = "path/to/config.json"
+    ref, content_type = remote._parse_manifest_ref(testref)
+    assert ref == "path/to/config.json"
+    assert content_type == oras.defaults.unknown_config_media_type

oras/tests/test_utils.py

@@ -102,3 +102,31 @@ def test_print_json():
     print("Testing utils.print_json")
     result = utils.print_json({1: 1})
     assert result == '{\n    "1": 1\n}'
+
+
+def test_split_path_and_content():
+    """
+    Test split path and content function.
+
+    Function has additional logic for Windows - this isn't included in these tests as
+    they don't usually run on Windows.
+    """
+    testref = "path/to/config:application/vnd.oci.image.config.v1+json"
+    path_content = utils.split_path_and_content(testref)
+    assert path_content.path == "path/to/config"
+    assert path_content.content == "application/vnd.oci.image.config.v1+json"
+
+    testref = "/dev/null:application/vnd.oci.image.config.v1+json"
+    path_content = utils.split_path_and_content(testref)
+    assert path_content.path == "/dev/null"
+    assert path_content.content == "application/vnd.oci.image.config.v1+json"
+
+    testref = "/dev/null"
+    path_content = utils.split_path_and_content(testref)
+    assert path_content.path == "/dev/null"
+    assert not path_content.content
+
+    testref = "path/to/config.json"
+    path_content = utils.split_path_and_content(testref)
+    assert path_content.path == "path/to/config.json"
+    assert not path_content.content

oras/utils/__init__.py

@@ -14,6 +14,7 @@
     readline,
     recursive_find,
     sanitize_path,
+    split_path_and_content,
     workdir,
     write_file,
     write_json,

oras/utils/fileio.py

@@ -18,6 +18,14 @@
 from typing import Generator, Optional, TextIO, Union
 
 
+class PathAndOptionalContent:
+    """Class for holding a path reference and optional content parsed from a string."""
+
+    def __init__(self, path: str, content: Optional[str] = None):
+        self.path = path
+        self.content = content
+
+
 def make_targz(source_dir: str, dest_name: Optional[str] = None) -> str:
     """
     Make a targz (compressed) archive from a source directory.
@@ -315,3 +323,52 @@ def read_json(filename: str, mode: str = "r") -> dict:
     :type mode: str
     """
     return json.loads(read_file(filename))
+
+
+def split_path_and_content(ref: str) -> PathAndOptionalContent:
+    """
+    Parse a string containing a path and an optional content
+
+    Examples
+    --------
+    <path>:<content-type>
+    path/to/config:application/vnd.oci.image.config.v1+json
+    /dev/null:application/vnd.oci.image.config.v1+json
+    C:\\myconfig:application/vnd.oci.image.config.v1+json
+
+    Or,
+    <path>
+    /dev/null
+    C:\\myconfig
+
+    :param ref: the manifest reference to parse (examples above)
+    :type ref: str
+    : return: A Tuple of the path in the reference, and the content-type if one found,
+              otherwise None.
+    """
+    if ":" not in ref:
+        return PathAndOptionalContent(ref, None)
+
+    if pathlib.Path(ref).drive:
+        # Running on Windows and Path has Windows drive letter in it, it definitely has
+        # one colon and could have two or feasibly more, e.g.
+        # C:\test.tar
+        # C:\test.tar:application/vnd.oci.image.layer.v1.tar
+        # C:\test.tar:application/vnd.oci.image.layer.v1.tar:somethingelse
+        #
+        # This regex matches two colons in the string and returns everything before
+        # the second colon as the "path" group and everything after the second colon
+        # as the "context" group.
+        # i.e.
+        # (C:\test.tar):(application/vnd.oci.image.layer.v1.tar)
+        # (C:\test.tar):(application/vnd.oci.image.layer.v1.tar:somethingelse)
+        # But C:\test.tar along will not match and we just return it as is.
+        path_and_content = re.search(r"(?P<path>.*?:.*?):(?P<content>.*)", ref)
+        if path_and_content:
+            return PathAndOptionalContent(
+                path_and_content.group("path"), path_and_content.group("content")
+            )
+        return PathAndOptionalContent(ref, None)
+    else:
+        path_content_list = ref.split(":", 1)
+        return PathAndOptionalContent(path_content_list[0], path_content_list[1])

oras/version.py

@@ -2,7 +2,7 @@
 __copyright__ = "Copyright The ORAS Authors."
 __license__ = "Apache-2.0"
 
-__version__ = "0.1.18"
+__version__ = "0.1.19"
 AUTHOR = "Vanessa Sochat"
 EMAIL = "[email protected]"
 NAME = "oras"