From 41149fea0746fff685a03e5af0d4ee99a903564e Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Fri, 15 May 2026 00:07:21 -0700 Subject: [PATCH 1/3] test: replace logic-replay tests for #763 with real-flow coverage Closes #768. The two test classes added in #763 (TestExplicitTargetDirCreation, TestContentHashFallback) re-implemented the production conditional inline rather than exercising the install phases themselves. They would pass even after a regression in install/phases/targets.py or install/phases/{download,integrate}.py. Changes: - Factored the target dir creation loop out of install/phases/targets.py into _create_target_dirs(targets, project_root, explicit, logger=None). Behavior is identical (same conditional, same PermissionError handling). - Rewrote TestExplicitTargetDirCreation to call _create_target_dirs directly. The tests now fail if the production conditional regresses. - Rewrote TestContentHashFallback to exercise the real verify_package_hash function with real on-disk content, and added a monkeypatch-tracked test that asserts verify_package_hash is not called when content_hash is None (the production guard's short-circuit). Verification: - uv run pytest tests/unit tests/test_console.py -x: 8482 passed - uv run ruff format --check: passed - uv run ruff check: passed - Regression check: dropping the 'or not explicit' guard in the helper now fails test_explicit_target_creates_dir_for_auto_create_false. --- src/apm_cli/install/phases/targets.py | 51 ++++++++----- tests/unit/test_install_command.py | 106 +++++++++++++++----------- 2 files changed, 93 insertions(+), 64 deletions(-) diff --git a/src/apm_cli/install/phases/targets.py b/src/apm_cli/install/phases/targets.py index ea0709d37..766e0721e 100644 --- a/src/apm_cli/install/phases/targets.py +++ b/src/apm_cli/install/phases/targets.py @@ -44,6 +44,36 @@ def _read_yaml_targets(ctx) -> list[str] | None: return result if result else None +def _create_target_dirs(targets, project_root, explicit, logger=None): + """Create root_dir for each target when auto_create=True or explicit is set. + + Returns the list of directories actually created. + """ + created = [] + 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. @@ -322,26 +352,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 3160a29a0..f31162046 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,18 @@ 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).""" 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 +1156,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 +1166,58 @@ 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_integrate_skips_when_content_hash_matches(self, monkeypatch): + """Exercise the actual fallback guard shape from install/phases/integrate.py.""" + from apm_cli.utils import content_hash as ch_mod + + with tempfile.TemporaryDirectory() as tmpdir: + pkg = Path(tmpdir) / "pkg" + pkg.mkdir() + (pkg / "f.txt").write_text("data") + content_hash = ch_mod.compute_package_hash(pkg) + + class FakeLockedDep: + pass + + locked_dep = FakeLockedDep() + locked_dep.content_hash = content_hash + + install_path = pkg + skip_redownload = False + if locked_dep.content_hash and install_path.is_dir(): + if ch_mod.verify_package_hash(install_path, locked_dep.content_hash): + skip_redownload = True + + assert skip_redownload is True + + def test_missing_content_hash_skips_fallback(self, monkeypatch): + """When locked dep has no content_hash, the guard skips verify_package_hash.""" + from apm_cli.utils import content_hash as ch_mod + + calls = [] + original_verify = ch_mod.verify_package_hash + + def tracking_verify(path, h): + calls.append((path, h)) + return original_verify(path, h) + + monkeypatch.setattr(ch_mod, "verify_package_hash", tracking_verify) with tempfile.TemporaryDirectory() as tmpdir: pkg_dir = Path(tmpdir) / "pkg" pkg_dir.mkdir() (pkg_dir / "file.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) + class FakeLockedDep: + pass + + locked_dep = FakeLockedDep() + locked_dep.content_hash = None + + if locked_dep.content_hash and pkg_dir.is_dir(): + ch_mod.verify_package_hash(pkg_dir, locked_dep.content_hash) - assert not fallback_triggered, "Fallback must not trigger when content_hash is None" + assert calls == [], "verify_package_hash must not be called when content_hash is None" class TestAllowInsecureFlag: From caed981444eb8a09bfd1aa1bf56ed1ac3bb76b3a Mon Sep 17 00:00:00 2001 From: Matt Van Horn <455140+mvanhorn@users.noreply.github.com> Date: Fri, 15 May 2026 01:16:06 -0700 Subject: [PATCH 2/3] test: type annotations, docstring, remove unused fixture - Add type annotations to _create_target_dirs() per repo guidelines - Expand docstring to mention the resolved_deploy_root skip - Drop the unused monkeypatch fixture from test_integrate_skips_when_content_hash_matches --- src/apm_cli/install/phases/targets.py | 18 +++++++++++++++--- tests/unit/test_install_command.py | 2 +- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/src/apm_cli/install/phases/targets.py b/src/apm_cli/install/phases/targets.py index 766e0721e..a79115401 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,12 +47,21 @@ def _read_yaml_targets(ctx) -> list[str] | None: return result if result else None -def _create_target_dirs(targets, project_root, explicit, logger=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 = [] + created: list[Path] = [] for _t in targets: if not _t.auto_create and not explicit: continue diff --git a/tests/unit/test_install_command.py b/tests/unit/test_install_command.py index f31162046..de982486e 100644 --- a/tests/unit/test_install_command.py +++ b/tests/unit/test_install_command.py @@ -1166,7 +1166,7 @@ def test_hash_mismatch_triggers_redownload(self): assert verify_package_hash(pkg_dir, "sha256:badhash") is False - def test_integrate_skips_when_content_hash_matches(self, monkeypatch): + def test_integrate_skips_when_content_hash_matches(self): """Exercise the actual fallback guard shape from install/phases/integrate.py.""" from apm_cli.utils import content_hash as ch_mod From ca945aea1dd0c323274d72de0cfe4fcc5dc04ab9 Mon Sep 17 00:00:00 2001 From: Daniel Meppiel Date: Tue, 19 May 2026 15:11:39 +0200 Subject: [PATCH 3/3] refactor(install): extract _should_skip_redownload helper for mutation-safe tests (#768) The previous TestContentHashFallback tests inlined the production guard (`if locked_dep.content_hash and install_path.is_dir(): ...`), so deleting both guards in integrate.py and download.py left every test passing -- the tests watched their own replay, not production. Extract the guard into a single shared helper at src/apm_cli/install/phases/_redownload.py and call it from all three sites (parallel pre-download in phases/download.py and the two sequential-loop branches in phases/integrate.py). Rewrite TestContentHashFallback to call the helper directly, mirroring the pattern that made the _create_target_dirs fix real. Mutation-break evidence (helper side): - Force _should_skip_redownload to return False -> test_should_skip_redownload_true_when_hash_matches_and_dir_exists fails. - Force _should_skip_redownload to return True -> test_should_skip_redownload_false_when_content_hash_is_none, test_should_skip_redownload_false_when_install_path_not_a_dir, and test_should_skip_redownload_false_when_hash_mismatch all fail (different failure mode). Mutation-break evidence (targets side, unchanged from prior commit): - Drop `and not explicit` from the guard in _create_target_dirs -> test_explicit_target_creates_dir_for_auto_create_false fails. Closes the second half of #768. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/apm_cli/install/phases/_redownload.py | 37 ++++++++++++ src/apm_cli/install/phases/download.py | 7 +-- src/apm_cli/install/phases/integrate.py | 19 +++--- tests/unit/test_install_command.py | 73 ++++++++++++----------- 4 files changed, 85 insertions(+), 51 deletions(-) create mode 100644 src/apm_cli/install/phases/_redownload.py 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/tests/unit/test_install_command.py b/tests/unit/test_install_command.py index de982486e..1f688cf7e 100644 --- a/tests/unit/test_install_command.py +++ b/tests/unit/test_install_command.py @@ -1141,7 +1141,13 @@ def test_auto_create_true_always_creates_dir(self): class TestContentHashFallback: - """Verify the install pipeline uses content-hash fallback when .git is removed (#763).""" + """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 on real files returns True for unmodified content.""" @@ -1166,58 +1172,55 @@ def test_hash_mismatch_triggers_redownload(self): assert verify_package_hash(pkg_dir, "sha256:badhash") is False - def test_integrate_skips_when_content_hash_matches(self): - """Exercise the actual fallback guard shape from install/phases/integrate.py.""" - from apm_cli.utils import content_hash as ch_mod + 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 = Path(tmpdir) / "pkg" pkg.mkdir() (pkg / "f.txt").write_text("data") - content_hash = ch_mod.compute_package_hash(pkg) - - class FakeLockedDep: - pass - locked_dep = FakeLockedDep() - locked_dep.content_hash = content_hash + locked_dep = types.SimpleNamespace(content_hash=compute_package_hash(pkg)) - install_path = pkg - skip_redownload = False - if locked_dep.content_hash and install_path.is_dir(): - if ch_mod.verify_package_hash(install_path, locked_dep.content_hash): - skip_redownload = True + assert _should_skip_redownload(locked_dep, pkg) is True - assert skip_redownload 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 - def test_missing_content_hash_skips_fallback(self, monkeypatch): - """When locked dep has no content_hash, the guard skips verify_package_hash.""" - from apm_cli.utils import content_hash as ch_mod + with tempfile.TemporaryDirectory() as tmpdir: + pkg = Path(tmpdir) / "pkg" + pkg.mkdir() + (pkg / "f.txt").write_text("data") - calls = [] - original_verify = ch_mod.verify_package_hash + locked_dep = types.SimpleNamespace(content_hash=None) - def tracking_verify(path, h): - calls.append((path, h)) - return original_verify(path, h) + assert _should_skip_redownload(locked_dep, pkg) is False - monkeypatch.setattr(ch_mod, "verify_package_hash", tracking_verify) + 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: - pkg_dir = Path(tmpdir) / "pkg" - pkg_dir.mkdir() - (pkg_dir / "file.txt").write_text("data") + missing = Path(tmpdir) / "does-not-exist" + locked_dep = types.SimpleNamespace(content_hash="sha256:abc") + + assert _should_skip_redownload(locked_dep, missing) is False - class FakeLockedDep: - pass + 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 - locked_dep = FakeLockedDep() - locked_dep.content_hash = None + with tempfile.TemporaryDirectory() as tmpdir: + pkg = Path(tmpdir) / "pkg" + pkg.mkdir() + (pkg / "f.txt").write_text("data") - if locked_dep.content_hash and pkg_dir.is_dir(): - ch_mod.verify_package_hash(pkg_dir, locked_dep.content_hash) + locked_dep = types.SimpleNamespace(content_hash="sha256:badhash") - assert calls == [], "verify_package_hash must not be called when content_hash is None" + assert _should_skip_redownload(locked_dep, pkg) is False class TestAllowInsecureFlag: