chore: address comments and improve test coverage (6)

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

Author: jiwangCHEN

oras/copy/graph.py

@@ -347,9 +347,20 @@ def _copy_node(
 
 
 def _verify_digest(data: bytes, expected_digest: str) -> None:
-    """Verify fetched content matches the expected digest."""
-    if not expected_digest or ":" not in expected_digest:
+    """Verify fetched content matches the expected digest.
+
+    An empty digest means there is nothing to verify against and is skipped.
+    A malformed digest or an unsupported algorithm is treated as an error so
+    callers never mistake unverified content for verified content.
+    """
+    if not expected_digest:
         return
+    if ":" not in expected_digest:
+        raise CopyError(
+            "VerifyDigest",
+            CopyErrorOrigin.SOURCE,
+            ValueError(f"invalid digest format: {expected_digest!r}"),
+        )
     algorithm, expected_hash = expected_digest.split(":", 1)
     try:
         h = hashlib.new(algorithm)

oras/tests/conftest.py

@@ -1,68 +1,8 @@
-import io
 import os
-import threading
 from dataclasses import dataclass
-from typing import BinaryIO, Dict, Optional
 
 import pytest
 
-from oras.types import Descriptor
-
-
-class InMemoryTarget:
-    """
-    Thread-safe in-memory Target for copy engine tests.
-
-    Implements the full Target protocol (fetch, exists, push, tag, resolve)
-    using plain dicts backed by a threading.Lock. Raises FileExistsError
-    on duplicate push so copy engine idempotency handling is exercised.
-    """
-
-    def __init__(self):
-        self._content: Dict[str, bytes] = {}
-        self._tags: Dict[str, Descriptor] = {}
-        self._lock = threading.Lock()
-
-    def fetch(self, desc: Descriptor) -> BinaryIO:
-        digest = desc.get("digest", "")
-        with self._lock:
-            data = self._content.get(digest)
-        if data is None:
-            raise FileNotFoundError(f"content not found: {digest}")
-        return io.BytesIO(data)
-
-    def exists(self, desc: Descriptor) -> bool:
-        digest = desc.get("digest", "")
-        with self._lock:
-            return digest in self._content
-
-    def push(self, desc: Descriptor, content: BinaryIO) -> None:
-        data = content.read()
-        digest = desc.get("digest", "")
-        with self._lock:
-            if digest in self._content:
-                raise FileExistsError(f"content already exists: {digest}")
-            self._content[digest] = data
-
-    def tag(self, desc: Descriptor, reference: str) -> None:
-        with self._lock:
-            self._tags[reference] = desc
-
-    def resolve(self, reference: str) -> Descriptor:
-        with self._lock:
-            desc = self._tags.get(reference)
-        if desc is None:
-            raise FileNotFoundError(f"reference not found: {reference}")
-        return desc
-
-    def get_content(self, digest: str) -> bytes:
-        with self._lock:
-            return self._content.get(digest, b"")
-
-    def get_tag(self, reference: str) -> Optional[Descriptor]:
-        with self._lock:
-            return self._tags.get(reference)
-
 
 @dataclass
 class TestCredentials:

oras/tests/helpers.py

@@ -0,0 +1,69 @@
+"""
+Shared test helpers for the oras test suite.
+
+Plain helper module (not a conftest) so it can be safely imported by tests.
+Pytest may load ``conftest.py`` under its own internal module name, and
+importing it as a package module can cause it to execute twice, leading to
+confusing fixture/state duplication. Reusable helpers therefore live here.
+"""
+
+import io
+import threading
+from typing import BinaryIO, Dict, Optional
+
+from oras.types import Descriptor
+
+
+class InMemoryTarget:
+    """
+    Thread-safe in-memory Target for copy engine tests.
+
+    Implements the full Target protocol (fetch, exists, push, tag, resolve)
+    using plain dicts backed by a threading.Lock. Raises FileExistsError
+    on duplicate push so copy engine idempotency handling is exercised.
+    """
+
+    def __init__(self):
+        self._content: Dict[str, bytes] = {}
+        self._tags: Dict[str, Descriptor] = {}
+        self._lock = threading.Lock()
+
+    def fetch(self, desc: Descriptor) -> BinaryIO:
+        digest = desc.get("digest", "")
+        with self._lock:
+            data = self._content.get(digest)
+        if data is None:
+            raise FileNotFoundError(f"content not found: {digest}")
+        return io.BytesIO(data)
+
+    def exists(self, desc: Descriptor) -> bool:
+        digest = desc.get("digest", "")
+        with self._lock:
+            return digest in self._content
+
+    def push(self, desc: Descriptor, content: BinaryIO) -> None:
+        data = content.read()
+        digest = desc.get("digest", "")
+        with self._lock:
+            if digest in self._content:
+                raise FileExistsError(f"content already exists: {digest}")
+            self._content[digest] = data
+
+    def tag(self, desc: Descriptor, reference: str) -> None:
+        with self._lock:
+            self._tags[reference] = desc
+
+    def resolve(self, reference: str) -> Descriptor:
+        with self._lock:
+            desc = self._tags.get(reference)
+        if desc is None:
+            raise FileNotFoundError(f"reference not found: {reference}")
+        return desc
+
+    def get_content(self, digest: str) -> bytes:
+        with self._lock:
+            return self._content.get(digest, b"")
+
+    def get_tag(self, reference: str) -> Optional[Descriptor]:
+        with self._lock:
+            return self._tags.get(reference)

oras/tests/test_copy.py

@@ -31,7 +31,7 @@
 )
 from oras.copy.graph import LimitedRegion, copy_graph, successors
 from oras.copy.tracker import StatusTracker
-from oras.tests.conftest import InMemoryTarget  # shared test fixture
+from oras.tests.helpers import InMemoryTarget  # shared test helper
 from oras.types import (
     descriptor_key,
     descriptors_equal,
@@ -2114,6 +2114,42 @@ def test_correct_content_passes_verification(self):
         assert dst.exists(blob_desc)
         assert dst.get_content(blob_desc["digest"]) == blob_data
 
+    def test_invalid_digest_format_raises_error(self):
+        """A non-empty but malformed digest must be rejected, not skipped."""
+        src = InMemoryTarget()
+        dst = InMemoryTarget()
+
+        blob_data = b"deadbeef"
+        bad_desc = {
+            "mediaType": "application/octet-stream",
+            "digest": "deadbeef",  # missing "algorithm:" prefix
+            "size": len(blob_data),
+        }
+        with src._lock:
+            src._content[bad_desc["digest"]] = blob_data
+
+        with pytest.raises(CopyError, match="invalid digest format"):
+            copy_graph(src, dst, bad_desc)
+        assert not dst.exists(bad_desc)
+
+    def test_unsupported_digest_algorithm_raises_error(self):
+        """A digest with an unsupported algorithm must be rejected, not skipped."""
+        src = InMemoryTarget()
+        dst = InMemoryTarget()
+
+        blob_data = b"payload"
+        bad_desc = {
+            "mediaType": "application/octet-stream",
+            "digest": "fakealgo:abcdef",
+            "size": len(blob_data),
+        }
+        with src._lock:
+            src._content[bad_desc["digest"]] = blob_data
+
+        with pytest.raises(CopyError, match="unsupported digest algorithm"):
+            copy_graph(src, dst, bad_desc)
+        assert not dst.exists(bad_desc)
+
 
 # ---------------------------------------------------------------------------
 # Tests for options mutation safety (C3 fix)