test: replace logic-replay tests for #763 with real-flow coverage

src/apm_cli/install/phases/_redownload.py

@@ -0,0 +1,37 @@
+"""Shared guard for the content-hash skip-redownload fallback (#763, #768).
+
+When a package's ``.git`` directory has been removed, the install pipeline
+cannot compare local HEAD against ``locked_dep.resolved_commit``.  In that
+case it falls back to a content-hash check: if the lockfile recorded a
+``content_hash`` for the package AND the install path is still a directory,
+re-hashing the on-disk content and comparing against the lockfile value is
+enough to confirm the package is intact and skip re-downloading.
+
+This module exposes :func:`_should_skip_redownload` so the three call sites
+(parallel pre-download and two branches of the sequential integrate loop)
+share one auditable definition.  Tests call it directly -- guard mutations
+in this function MUST surface as test failures (mutation-break safety).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+
+def _should_skip_redownload(locked_dep: Any, install_path: Path) -> bool:
+    """Return True when ``install_path`` content matches ``locked_dep.content_hash``.
+
+    Returns False when ``locked_dep`` is missing/has no ``content_hash``, when
+    ``install_path`` is not an existing directory, or when the on-disk content
+    does not hash to the recorded value.  Callers use the True return to skip
+    a redundant re-download after the git-based check fails (#763).
+    """
+    if locked_dep is None or not locked_dep.content_hash:
+        return False
+    if not install_path.is_dir():
+        return False
+
+    from apm_cli.utils.content_hash import verify_package_hash
+
+    return verify_package_hash(install_path, locked_dep.content_hash)

src/apm_cli/install/phases/download.py

@@ -89,11 +89,10 @@ def run(ctx: InstallContext) -> None:
                 # Git check failed (e.g. .git removed after download).
                 # Fall back to content-hash verification so correctly
                 # installed packages are not re-downloaded every run (#763).
-                if _pd_locked_chk.content_hash and _pd_path.is_dir():
-                    from apm_cli.utils.content_hash import verify_package_hash as _pd_verify_hash
+                from apm_cli.install.phases._redownload import _should_skip_redownload
 
-                    if _pd_verify_hash(_pd_path, _pd_locked_chk.content_hash):
-                        continue
+                if _should_skip_redownload(_pd_locked_chk, _pd_path):
+                    continue
         # Build download ref (use locked commit for reproducibility).
         # build_download_ref() uses the manifest ref when ref_changed is True.
         _pd_dlref = build_download_ref(

src/apm_cli/install/phases/integrate.py

@@ -19,6 +19,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple  # noqa: F401, UP035
 
+from apm_cli.install.phases._redownload import _should_skip_redownload
 from apm_cli.install.phases.heal import run_heal_chain
 from apm_cli.install.services import integrate_local_content
 from apm_cli.install.sources import make_dependency_source
@@ -115,12 +116,9 @@ def _resolve_download_strategy(
                         # Git check failed (e.g. .git removed, or virtual
                         # package install_path is not a git repo). Fall back
                         # to content-hash verification (#763).
-                        if locked_dep.content_hash and install_path.is_dir():
-                            from apm_cli.utils.content_hash import verify_package_hash
-
-                            if verify_package_hash(install_path, locked_dep.content_hash):
-                                lockfile_match = True
-                                lockfile_match_via_content_hash_only = True
+                        if _should_skip_redownload(locked_dep, install_path):
+                            lockfile_match = True
+                            lockfile_match_via_content_hash_only = True
             elif not ref_changed:
                 # Normal mode: compare local HEAD with lockfile SHA.
                 try:
@@ -133,12 +131,9 @@ def _resolve_download_strategy(
                     # Git check failed (e.g. .git removed, or virtual package
                     # install_path is not a git repo). Fall back to
                     # content-hash verification (#763).
-                    if locked_dep.content_hash and install_path.is_dir():
-                        from apm_cli.utils.content_hash import verify_package_hash
-
-                        if verify_package_hash(install_path, locked_dep.content_hash):
-                            lockfile_match = True
-                            lockfile_match_via_content_hash_only = True
+                    if _should_skip_redownload(locked_dep, install_path):
+                        lockfile_match = True
+                        lockfile_match_via_content_hash_only = True
 
     # Self-heal pipeline (PR #1158).
     #

src/apm_cli/install/phases/targets.py

@@ -10,10 +10,13 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from collections.abc import Iterable
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from apm_cli.install.context import InstallContext
+    from apm_cli.integration.targets import TargetProfile
 
 
 def _read_yaml_targets(ctx) -> list[str] | None:
@@ -44,6 +47,45 @@ def _read_yaml_targets(ctx) -> list[str] | None:
     return result if result else None
 
 
+def _create_target_dirs(
+    targets: Iterable[TargetProfile],
+    project_root: Path,
+    explicit: str | None,
+    logger: Any = None,
+) -> list[Path]:
+    """Create root_dir for each target when auto_create=True or explicit is set.
+
+    Targets that resolve to an external deploy root (``resolved_deploy_root``)
+    are skipped: their directories live outside the project tree and are
+    created by the integrator's deploy logic, not here.
+
+    Returns the list of directories actually created.
+    """
+    created: list[Path] = []
+    for _t in targets:
+        if not _t.auto_create and not explicit:
+            continue
+        if _t.resolved_deploy_root is not None:
+            continue
+        _root = _t.root_dir
+        _target_dir = project_root / _root
+        if not _target_dir.exists():
+            try:
+                _target_dir.mkdir(parents=True, exist_ok=True)
+            except PermissionError:
+                if logger:
+                    _display_root = f"~/{_root}/"
+                    logger.error(
+                        f"Cannot create {_display_root} -- permission denied. "
+                        f"Check directory permissions or use a different --target."
+                    )
+                raise SystemExit(1) from None
+            created.append(_target_dir)
+            if logger:
+                logger.verbose_detail(f"Created {_root}/ ({_t.name} target)")
+    return created
+
+
 def run(ctx: InstallContext) -> None:
     """Execute the targets phase.
 
@@ -369,26 +411,7 @@ def _fmt_target(t):
                     "deployed. Check 'target:' in apm.yml or use --target."
                 )
 
-        for _t in _targets:
-            if not _t.auto_create and not _explicit:
-                continue
-            if _t.resolved_deploy_root is not None:
-                continue
-            _root = _t.root_dir
-            _target_dir = ctx.project_root / _root
-            if not _target_dir.exists():
-                try:
-                    _target_dir.mkdir(parents=True, exist_ok=True)
-                except PermissionError:
-                    if ctx.logger:
-                        _display_root = f"~/{_root}/"
-                        ctx.logger.error(
-                            f"Cannot create {_display_root} -- permission denied. "
-                            f"Check directory permissions or use a different --target."
-                        )
-                    raise SystemExit(1) from None
-                if ctx.logger:
-                    ctx.logger.verbose_detail(f"Created {_root}/ ({_t.name} target)")
+        _create_target_dirs(_targets, ctx.project_root, _explicit, ctx.logger)
 
     # Legacy detect_target call -- return values are not consumed by any
     # downstream code but the call is preserved for behaviour parity with

tests/unit/test_install_command.py

@@ -1083,7 +1083,7 @@ def test_ghes_host_skips_ssh_attempt(self, mock_run):
 
 
 class TestExplicitTargetDirCreation:
-    """Verify --target creates root_dir even when auto_create=False (GH bug fix)."""
+    """Verify _create_target_dirs creates explicit root_dir for auto_create=False (#763)."""
 
     def setup_method(self):
         self._tmpdir = tempfile.mkdtemp()
@@ -1095,44 +1095,34 @@ def teardown_method(self):
         shutil.rmtree(self._tmpdir, ignore_errors=True)
 
     def test_explicit_target_creates_dir_for_auto_create_false(self):
-        """When _explicit is set, target dirs are created even if auto_create=False."""
+        """When explicit is set, target dirs are created even if auto_create=False."""
+        from apm_cli.install.phases.targets import _create_target_dirs
         from apm_cli.integration.targets import KNOWN_TARGETS
 
         claude = KNOWN_TARGETS["claude"]
         assert claude.auto_create is False
 
-        # Simulate the fixed loop logic: create dir when _explicit is set
-        _explicit = "claude"
-        _targets = [claude]
-        for _t in _targets:
-            if not _t.auto_create and not _explicit:
-                continue
-            _target_dir = self.project_root / _t.root_dir
-            if not _target_dir.exists():
-                _target_dir.mkdir(parents=True, exist_ok=True)
+        created = _create_target_dirs([claude], self.project_root, explicit="claude")
 
         assert (self.project_root / ".claude").is_dir()
+        assert (self.project_root / ".claude") in created
 
     def test_auto_detect_skips_dir_for_auto_create_false(self):
-        """Without _explicit, auto_create=False targets don't get dirs created."""
+        """Without explicit, auto_create=False targets don't get dirs created."""
+        from apm_cli.install.phases.targets import _create_target_dirs
         from apm_cli.integration.targets import KNOWN_TARGETS
 
         claude = KNOWN_TARGETS["claude"]
         assert claude.auto_create is False
 
-        _explicit = None
-        _targets = [claude]
-        for _t in _targets:
-            if not _t.auto_create and not _explicit:
-                continue
-            _target_dir = self.project_root / _t.root_dir
-            if not _target_dir.exists():
-                _target_dir.mkdir(parents=True, exist_ok=True)
+        created = _create_target_dirs([claude], self.project_root, explicit=None)
 
         assert not (self.project_root / ".claude").exists()
+        assert created == []
 
     def test_auto_create_true_always_creates_dir(self):
-        """auto_create=True targets create dir regardless of _explicit."""
+        """auto_create=True targets create dir regardless of explicit."""
+        from apm_cli.install.phases.targets import _create_target_dirs
         from apm_cli.integration.targets import KNOWN_TARGETS
 
         copilot = KNOWN_TARGETS["copilot"]
@@ -1143,24 +1133,24 @@ def test_auto_create_true_always_creates_dir(self):
 
             shutil.rmtree(self.project_root / copilot.root_dir, ignore_errors=True)
 
-            _targets = [copilot]
-            for _t in _targets:
-                if not _t.auto_create and not _explicit:
-                    continue
-                _target_dir = self.project_root / _t.root_dir
-                if not _target_dir.exists():
-                    _target_dir.mkdir(parents=True, exist_ok=True)
-
+            created = _create_target_dirs([copilot], self.project_root, explicit=_explicit)
             assert (self.project_root / ".github").is_dir(), (
-                f"auto_create=True should create dir when _explicit={_explicit!r}"
+                f"auto_create=True should create dir when explicit={_explicit!r}"
             )
+            assert (self.project_root / ".github") in created
 
 
 class TestContentHashFallback:
-    """Verify content-hash fallback when .git is removed from installed packages."""
+    """Verify the install pipeline uses content-hash fallback when .git is removed (#763).
+
+    These tests exercise the production helper ``_should_skip_redownload`` directly
+    (the same one called from both ``phases/integrate.py`` and ``phases/download.py``).
+    Mutation-break: deleting the guard contents in ``_should_skip_redownload``
+    (forcing ``return False`` or ``return True``) MUST fail these tests.
+    """
 
     def test_hash_match_skips_redownload(self):
-        """Content hash verification allows skipping re-download."""
+        """Content hash verification on real files returns True for unmodified content."""
         from apm_cli.utils.content_hash import compute_package_hash, verify_package_hash
 
         with tempfile.TemporaryDirectory() as tmpdir:
@@ -1172,7 +1162,7 @@ def test_hash_match_skips_redownload(self):
             assert verify_package_hash(pkg_dir, content_hash) is True
 
     def test_hash_mismatch_triggers_redownload(self):
-        """Mismatched content hash means re-download should proceed."""
+        """Mismatched content hash on real files returns False."""
         from apm_cli.utils.content_hash import verify_package_hash
 
         with tempfile.TemporaryDirectory() as tmpdir:
@@ -1182,24 +1172,55 @@ def test_hash_mismatch_triggers_redownload(self):
 
             assert verify_package_hash(pkg_dir, "sha256:badhash") is False
 
-    def test_missing_content_hash_skips_fallback(self):
-        """When locked dep has no content_hash, the fallback guard prevents
-        verify_package_hash from being called."""
-        from apm_cli.utils.content_hash import verify_package_hash
+    def test_should_skip_redownload_true_when_hash_matches_and_dir_exists(self):
+        """Production helper returns True when content_hash matches an existing dir."""
+        from apm_cli.install.phases._redownload import _should_skip_redownload
+        from apm_cli.utils.content_hash import compute_package_hash
 
         with tempfile.TemporaryDirectory() as tmpdir:
-            pkg_dir = Path(tmpdir) / "pkg"
-            pkg_dir.mkdir()
-            (pkg_dir / "file.txt").write_text("data")
+            pkg = Path(tmpdir) / "pkg"
+            pkg.mkdir()
+            (pkg / "f.txt").write_text("data")
+
+            locked_dep = types.SimpleNamespace(content_hash=compute_package_hash(pkg))
+
+            assert _should_skip_redownload(locked_dep, pkg) is True
+
+    def test_should_skip_redownload_false_when_content_hash_is_none(self):
+        """Production helper returns False when locked dep has no content_hash."""
+        from apm_cli.install.phases._redownload import _should_skip_redownload
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            pkg = Path(tmpdir) / "pkg"
+            pkg.mkdir()
+            (pkg / "f.txt").write_text("data")
+
+            locked_dep = types.SimpleNamespace(content_hash=None)
+
+            assert _should_skip_redownload(locked_dep, pkg) is False
+
+    def test_should_skip_redownload_false_when_install_path_not_a_dir(self):
+        """Production helper returns False when install_path is missing/not a dir."""
+        from apm_cli.install.phases._redownload import _should_skip_redownload
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            missing = Path(tmpdir) / "does-not-exist"
+            locked_dep = types.SimpleNamespace(content_hash="sha256:abc")
+
+            assert _should_skip_redownload(locked_dep, missing) is False
+
+    def test_should_skip_redownload_false_when_hash_mismatch(self):
+        """Production helper returns False when content_hash does not match on-disk."""
+        from apm_cli.install.phases._redownload import _should_skip_redownload
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            pkg = Path(tmpdir) / "pkg"
+            pkg.mkdir()
+            (pkg / "f.txt").write_text("data")
 
-            # Simulate the guard logic from install.py:
-            # if _pd_locked_chk.content_hash and _pd_path.is_dir():
-            content_hash = None  # no content_hash recorded in lockfile
-            fallback_triggered = False
-            if content_hash and pkg_dir.is_dir():
-                fallback_triggered = verify_package_hash(pkg_dir, content_hash)
+            locked_dep = types.SimpleNamespace(content_hash="sha256:badhash")
 
-            assert not fallback_triggered, "Fallback must not trigger when content_hash is None"
+            assert _should_skip_redownload(locked_dep, pkg) is False
 
 
 class TestAllowInsecureFlag: