diff --git a/src/apm_cli/install/phases/_redownload.py b/src/apm_cli/install/phases/_redownload.py new file mode 100644 index 000000000..4ef97801a --- /dev/null +++ b/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) diff --git a/src/apm_cli/install/phases/download.py b/src/apm_cli/install/phases/download.py index dc17de69f..2cf108960 100644 --- a/src/apm_cli/install/phases/download.py +++ b/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( diff --git a/src/apm_cli/install/phases/integrate.py b/src/apm_cli/install/phases/integrate.py index ecbcb4393..cb8f83635 100644 --- a/src/apm_cli/install/phases/integrate.py +++ b/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). # diff --git a/src/apm_cli/install/phases/targets.py b/src/apm_cli/install/phases/targets.py index 9c681a7b3..7a9d82c26 100644 --- a/src/apm_cli/install/phases/targets.py +++ b/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 diff --git a/tests/unit/test_install_command.py b/tests/unit/test_install_command.py index f2267251f..c119b1d03 100644 --- a/tests/unit/test_install_command.py +++ b/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: