chore: address comments and improve test coverage(6)

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

Author: jiwangCHEN

oras/copy/copy.md

@@ -0,0 +1,95 @@
+1. The algorithm is a post-order concurrent DAG walk
+Children are always fully committed before their parent is pushed. This is not just an ordering preference — it is a correctness invariant. A registry will reject a manifest whose layers do not yet exist. The semaphore-release trick (region.end() before recursing into children, region.start() before copying self) is what makes this safe at any concurrency level, including 1.
+2. Deduplication is a first-class primitive, not an afterthought
+StatusTracker is the first gate every node passes through. Without it, shared blobs referenced by multiple manifests in an index would be fetched and pushed N times concurrently and race on dst.push(). The tracker collapses N workers into 1 doer + N-1 waiters.
+3. The cache proxy decouples resolution from traversal
+CacheProxy absorbs manifest and index bytes on first fetch (during _resolve_root / successors). When _process_node later copies the root, it reads from memory instead of re-hitting the network. The stop_caching flag exists because map_root may re-fetch different content (platform selection) that should not pollute the cache.
+4. Hooks are the extension seam, not subclassing
+pre_copy, post_copy, on_copy_skipped, on_mounted are injected callables. The tagging mechanism (ensuring dst_ref ends up pointing at the root) is implemented entirely through these hooks, not through a special code path in the traversal. SkipNode raised from pre_copy short-circuits _do_copy_node — this is how ReferencePusher atomically pushes + tags the root in a single PUT.
+5. Two destination capabilities drive two tagging strategies
+- Plain Target (e.g., disk layout, in-memory): tag is a separate call; the hook fires after the copy
+- ReferencePusher (e.g., OCI registry): tag is embedded in the push URL; the hook fires before the copy and skips the normal push..
+6. Mount is an opportunistic cross-repo blob copy
+_mount_or_copy_node only applies to non-manifest blobs when dst is a Mounter. It posts a server-side copy hint; the registry may honor it (201 → done) or refuse (202 → fall back to full upload). The fallback is unified with normal upload via the get_content closure, so mount and copy share the same post/pre-copy hook lifecycle.
+7. Foreign layer filtering is a pre-traversal filter, not a skip hook
+remove_foreign_layers() strips non-distributable layers from the successors list. These layers are never fetched, never pushed, and never dedup-checked. If you need to transfer them, you must override find_successors.
+
+```
+### COPY Overview
+
+ACTION copy(src, src_ref, dst, dst_ref, opts):
+    state ← ValidatingInputs
+    opts  ← opts.clone() or default Options   -- mutation safety
+
+    if dst_ref == "": dst_ref ← src_ref
+
+    -- ValidatingInputs
+    assert src ≠ null  else raise CopyError(SOURCE, "Resolve")
+    assert dst ≠ null  else raise CopyError(DEST,   "Resolve")
+    state ← CreatingProxy
+
+    -- CreatingProxy
+    proxy ← CacheProxy(base=src, max_bytes=opts.max_metadata_bytes)
+    state ← ResolvingRoot
+
+    -- ResolvingRoot
+    root ← resolve_root(src, src_ref, proxy)    -- see below
+    state ← MappingRoot
+
+    -- MappingRoot
+    if opts.map_root ≠ null:
+        proxy.stop_caching ← true
+        root ← opts.map_root(proxy, root)
+        proxy.stop_caching ← false
+    state ← InstallingHooks
+
+    -- InstallingHooks
+    install_tagging_hooks(dst, dst_ref, proxy, root, opts)  -- see below
+    state ← RunningGraph
+
+    -- RunningGraph
+    copy_graph(src=src, dst=dst, root=root, proxy=proxy, opts=opts)
+    state ← Done(root)
+    return root
+```
+
+```
+### Concurrency
+
+─────────────────────────────────────────────────────────────────────
+WORKER LAUNCHER (with semaphore gating)
+─────────────────────────────────────────────────────────────────────
+
+ACTION launch_workers(sem, fn, descriptors, cancel):
+    for each desc in descriptors:
+        thread ← new daemon Thread:
+            if cancel.is_set(): return
+            region ← LimitedRegion(sem)
+            region.start()              -- acquire slot (blocks if full)
+            if cancel.is_set():
+                region.end(); return
+            try:
+                fn(region, desc)
+            catch e:
+                first_error ← e         -- capture first error only
+                cancel.set()            -- signal all workers to stop
+            finally:
+                region.end()            -- always release slot
+        thread.start()
+
+
+─────────────────────────────────────────────────────────────────────
+COPY NODE
+─────────────────────────────────────────────────────────────────────
+
+Per-node:      Claimed → CheckingExistence
+               → [Skipped | FindingSuccessors]
+               → [YieldingSlot → WaitingForChildren → ReacquiringSlot]?
+               → Copying → NodeDone
+
+Single copy:   PreCopy → [Skipped_ByHook | Transferring → PostCopy → NodeCopied]
+
+States:         Fetching → Buffering → VerifyingDigest → VerifyingSize
+               → Pushing → Transferred
+
+```

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)