chore: address comments and improve test coverage(6)

Signed-off-by: jiwangCHEN <hyper1char@gmail.com>

Author: jiwangCHEN

oras/content/registry.py

@@ -154,6 +154,10 @@ def push(self, desc: Descriptor, content: BinaryIO) -> None:
                 self._registry._check_200_response(response)
             finally:
                 if tmp is not None:
+                    try:
+                        tmp.close()
+                    except Exception:
+                        pass
                     try:
                         os.unlink(tmp.name)
                     except OSError:

oras/content/storage.py

@@ -27,7 +27,12 @@
 
 import io
 import threading
-from typing import BinaryIO, Callable, Protocol, Tuple, runtime_checkable
+from typing import BinaryIO, Callable, Tuple
+
+try:
+    from typing import Protocol, runtime_checkable  # Python 3.8+
+except ImportError:  # Python 3.7
+    from typing_extensions import Protocol, runtime_checkable
 
 from oras.types import Descriptor
 

oras/copy/copy.py

@@ -119,6 +119,8 @@ def _resolve_root(
     The fetched content is fed through ``successors`` to ensure
     it gets cached in the proxy for later use during graph traversal.
 
+    Currently referenceFetcher is not implemented for registry targets.
+
     Matches oras-go's resolveRoot.
     """
     if isinstance(src, storage.ReferenceFetcher):
@@ -137,8 +139,6 @@ def _resolve_root(
             if hasattr(rc, "close"):
                 rc.close()
 
-        # Cache the root content by feeding it through successors
-        # This ensures the proxy has the manifest/index cached
         def fetch_root(desc):
             if descriptors_equal(desc, root):
                 return io.BytesIO(data)

oras/layout/layout.py

@@ -6,7 +6,7 @@
 
 import json
 import pathlib
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, List, Optional
 
 import jsonschema
 import requests
@@ -147,7 +147,7 @@ def is_oci_layout(path: str) -> bool:
         except (FileNotFoundError, ValueError, OSError):
             return False
 
-    def get_ordered_blobs(self, tag: str = "latest") -> list[str]:
+    def get_ordered_blobs(self, tag: str = "latest") -> List[str]:
         """
         Traverse an OCI layout and collect blob digests in dependency order for pushing.
 
@@ -170,7 +170,7 @@ def get_ordered_blobs(self, tag: str = "latest") -> list[str]:
             raise ValueError(f"Tag '{tag}' not found in index")
 
         # Collect blobs in dependency order
-        collected: list[str] = []
+        collected: List[str] = []
         self._process_manifest(manifest_entry["digest"], collected)
         return collected
 
@@ -196,7 +196,7 @@ def find_index_entry(self, reference: str) -> Optional[dict]:
                 return manifest_entry
         return None
 
-    def _process_manifest(self, digest: str, collected: list[str]) -> None:
+    def _process_manifest(self, digest: str, collected: List[str]) -> None:
         """
         Recursively process a manifest blob and collect dependencies.
 

oras/provider.py

@@ -632,11 +632,10 @@ def chunked_upload(
                 # Important to update with auth token if acquired
                 # TODO call to auth here
                 start = end + 1
-                self._check_200_response(
-                    r := self.do_request(
-                        session_url, "PATCH", data=chunk, headers=headers
-                    )
+                r = self.do_request(
+                    session_url, "PATCH", data=chunk, headers=headers
                 )
+                self._check_200_response(r)
                 session_url = self._get_location(r, container)
                 if not session_url:
                     raise ValueError(f"Issue retrieving session url: {r.json()}")
@@ -732,8 +731,8 @@ def copy(
         self.auth.load_configs(src_container, configs=configs)
         self.auth.load_configs(dst_container, configs=configs)
 
-        src_target = RegistryTarget(self, src_container)
-        dst_target = RegistryTarget(self, dst_container)
+        src_target = RegistryTarget(self, src_container, opts)
+        dst_target = RegistryTarget(self, dst_container, opts)
 
         src_ref = src_container.digest or src_container.tag
         dst_ref = dst_container.digest or dst_container.tag