diff --git a/tests/unit/integration/test_base_integrator_utils.py b/tests/unit/integration/test_base_integrator_utils.py new file mode 100644 index 00000000..8fc4a68e --- /dev/null +++ b/tests/unit/integration/test_base_integrator_utils.py @@ -0,0 +1,396 @@ +"""Tests for untested BaseIntegrator utility methods. + +Covers: +- normalize_managed_files() +- cleanup_empty_parents() +- sync_remove_files() +- find_files_by_glob() +""" + +import pytest +from pathlib import Path + +from apm_cli.integration.base_integrator import BaseIntegrator + + +# --------------------------------------------------------------------------- +# normalize_managed_files +# --------------------------------------------------------------------------- + + +class TestNormalizeManagedFiles: + """Tests for BaseIntegrator.normalize_managed_files().""" + + def test_returns_none_for_none_input(self): + result = BaseIntegrator.normalize_managed_files(None) + assert result is None + + def test_returns_empty_set_for_empty_input(self): + result = BaseIntegrator.normalize_managed_files(set()) + assert result == set() + + def test_forward_slashes_unchanged(self): + files = {".github/prompts/foo.md", ".claude/commands/bar.md"} + result = BaseIntegrator.normalize_managed_files(files) + assert result == files + + def test_backslashes_converted_to_forward_slashes(self): + files = {".github\\prompts\\foo.md", ".claude\\commands\\bar.md"} + result = BaseIntegrator.normalize_managed_files(files) + assert result == {".github/prompts/foo.md", ".claude/commands/bar.md"} + + def test_mixed_separators_normalized(self): + files = {".github/prompts\\foo.md"} + result = BaseIntegrator.normalize_managed_files(files) + assert ".github/prompts/foo.md" in result + + def test_returns_new_set_not_in_place(self): + original = {".github\\prompts\\foo.md"} + result = BaseIntegrator.normalize_managed_files(original) + # Original should be unchanged + assert ".github\\prompts\\foo.md" in original + assert ".github/prompts/foo.md" in result + + def test_deduplication_after_normalization(self): + # Both backslash and forward-slash versions point to same path + files = {".github\\prompts\\foo.md", ".github/prompts/foo.md"} + result = BaseIntegrator.normalize_managed_files(files) + assert len(result) == 1 + assert ".github/prompts/foo.md" in result + + def test_large_set_normalized(self): + files = {f".github\\prompts\\file_{i}.md" for i in range(100)} + result = BaseIntegrator.normalize_managed_files(files) + assert len(result) == 100 + assert all("/" in p and "\\" not in p for p in result) + + +# --------------------------------------------------------------------------- +# cleanup_empty_parents +# --------------------------------------------------------------------------- + + +class TestCleanupEmptyParents: + """Tests for BaseIntegrator.cleanup_empty_parents().""" + + def test_no_op_for_empty_list(self, tmp_path): + # Should not raise + BaseIntegrator.cleanup_empty_parents([], tmp_path) + + def test_removes_empty_intermediate_dirs(self, tmp_path): + nested = tmp_path / "a" / "b" / "c" + nested.mkdir(parents=True) + file_path = nested / "file.txt" + file_path.write_text("content") + file_path.unlink() + + BaseIntegrator.cleanup_empty_parents([file_path], tmp_path) + + assert not (tmp_path / "a").exists() + + def test_does_not_remove_stop_at_dir(self, tmp_path): + stop_dir = tmp_path / "base" + stop_dir.mkdir() + nested = stop_dir / "sub" + nested.mkdir() + file_path = nested / "file.txt" + file_path.write_text("content") + file_path.unlink() + + BaseIntegrator.cleanup_empty_parents([file_path], stop_dir) + + # stop_dir should still exist + assert stop_dir.exists() + # sub should be removed (empty) + assert not nested.exists() + + def test_does_not_remove_non_empty_dirs(self, tmp_path): + nested = tmp_path / "a" / "b" + nested.mkdir(parents=True) + # Put another file in 'a' so it's not empty + (tmp_path / "a" / "sibling.txt").write_text("keep me") + deleted = nested / "file.txt" + deleted.write_text("gone") + deleted.unlink() + nested.rmdir() + + BaseIntegrator.cleanup_empty_parents([deleted], tmp_path) + + # 'a' still has sibling.txt, so should remain + assert (tmp_path / "a").exists() + + def test_multiple_deleted_files_same_dir(self, tmp_path): + parent = tmp_path / "parent" + parent.mkdir() + f1 = parent / "file1.txt" + f2 = parent / "file2.txt" + f1.write_text("a") + f2.write_text("b") + f1.unlink() + f2.unlink() + + BaseIntegrator.cleanup_empty_parents([f1, f2], tmp_path) + + assert not parent.exists() + + def test_deeply_nested_cleanup(self, tmp_path): + deep = tmp_path / "a" / "b" / "c" / "d" / "e" + deep.mkdir(parents=True) + f = deep / "file.txt" + f.write_text("x") + f.unlink() + + BaseIntegrator.cleanup_empty_parents([f], tmp_path) + + assert not (tmp_path / "a").exists() + + def test_cleanup_stops_at_ancestor_boundary(self, tmp_path): + # stop_at is a direct child of tmp_path + stop = tmp_path / "stop_here" + stop.mkdir() + nested = stop / "inner" / "deep" + nested.mkdir(parents=True) + f = nested / "file.txt" + f.write_text("x") + f.unlink() + + BaseIntegrator.cleanup_empty_parents([f], stop) + + # stop_here should survive + assert stop.exists() + # inner and deep should be cleaned up + assert not (stop / "inner").exists() + + +# --------------------------------------------------------------------------- +# sync_remove_files +# --------------------------------------------------------------------------- + + +class TestSyncRemoveFiles: + """Tests for BaseIntegrator.sync_remove_files().""" + + def _make_files(self, root, paths): + """Create files at the given relative paths under root.""" + for p in paths: + full = root / p + full.parent.mkdir(parents=True, exist_ok=True) + full.write_text("content") + + def test_removes_matching_files(self, tmp_path): + self._make_files(tmp_path, [".github/prompts/foo.md"]) + managed = {".github/prompts/foo.md"} + + stats = BaseIntegrator.sync_remove_files( + tmp_path, + managed, + prefix=".github/prompts/", + ) + + assert stats["files_removed"] == 1 + assert stats["errors"] == 0 + assert not (tmp_path / ".github/prompts/foo.md").exists() + + def test_skips_files_not_matching_prefix(self, tmp_path): + self._make_files(tmp_path, [".claude/commands/foo.md"]) + managed = {".claude/commands/foo.md"} + + stats = BaseIntegrator.sync_remove_files( + tmp_path, + managed, + prefix=".github/prompts/", + ) + + assert stats["files_removed"] == 0 + # File should still exist + assert (tmp_path / ".claude/commands/foo.md").exists() + + def test_skips_nonexistent_files(self, tmp_path): + managed = {".github/prompts/nonexistent.md"} + + stats = BaseIntegrator.sync_remove_files( + tmp_path, + managed, + prefix=".github/prompts/", + ) + + assert stats["files_removed"] == 0 + assert stats["errors"] == 0 + + def test_uses_legacy_glob_when_managed_files_none(self, tmp_path): + glob_dir = tmp_path / ".github" / "prompts" + glob_dir.mkdir(parents=True) + (glob_dir / "foo-apm.prompt.md").write_text("x") + (glob_dir / "unrelated.md").write_text("y") + + stats = BaseIntegrator.sync_remove_files( + tmp_path, + None, # managed_files is None -> use legacy glob + prefix=".github/prompts/", + legacy_glob_dir=glob_dir, + legacy_glob_pattern="*-apm.prompt.md", + ) + + assert stats["files_removed"] == 1 + assert not (glob_dir / "foo-apm.prompt.md").exists() + assert (glob_dir / "unrelated.md").exists() + + def test_legacy_glob_skipped_when_dir_missing(self, tmp_path): + missing_dir = tmp_path / "nonexistent" + + stats = BaseIntegrator.sync_remove_files( + tmp_path, + None, + prefix=".github/prompts/", + legacy_glob_dir=missing_dir, + legacy_glob_pattern="*.md", + ) + + assert stats["files_removed"] == 0 + assert stats["errors"] == 0 + + def test_removes_multiple_files(self, tmp_path): + paths = [ + ".github/prompts/a.md", + ".github/prompts/b.md", + ".github/prompts/c.md", + ] + self._make_files(tmp_path, paths) + managed = set(paths) + + stats = BaseIntegrator.sync_remove_files( + tmp_path, + managed, + prefix=".github/prompts/", + ) + + assert stats["files_removed"] == 3 + for p in paths: + assert not (tmp_path / p).exists() + + def test_rejects_traversal_paths(self, tmp_path): + """validate_deploy_path prevents removal of paths with '..'.""" + managed = {"../etc/passwd"} + + stats = BaseIntegrator.sync_remove_files( + tmp_path, + managed, + prefix="../", + ) + + # The traversal path should be rejected by validate_deploy_path + assert stats["files_removed"] == 0 + + def test_skips_files_not_in_allowed_prefix(self, tmp_path): + # A file under /tmp/evil/ is NOT an APM-managed prefix + managed = {"evil/file.md"} + + stats = BaseIntegrator.sync_remove_files( + tmp_path, + managed, + prefix="evil/", + ) + + assert stats["files_removed"] == 0 + + def test_legacy_glob_no_pattern_does_nothing(self, tmp_path): + """Without both dir and pattern, legacy fallback is skipped.""" + stats = BaseIntegrator.sync_remove_files( + tmp_path, + None, + prefix=".github/prompts/", + legacy_glob_dir=tmp_path, + legacy_glob_pattern=None, + ) + assert stats["files_removed"] == 0 + + +# --------------------------------------------------------------------------- +# find_files_by_glob +# --------------------------------------------------------------------------- + + +class TestFindFilesByGlob: + """Tests for BaseIntegrator.find_files_by_glob().""" + + def test_finds_matching_files(self, tmp_path): + (tmp_path / "a.prompt.md").write_text("x") + (tmp_path / "b.prompt.md").write_text("x") + (tmp_path / "other.md").write_text("x") + + results = BaseIntegrator.find_files_by_glob(tmp_path, "*.prompt.md") + + assert len(results) == 2 + names = {f.name for f in results} + assert names == {"a.prompt.md", "b.prompt.md"} + + def test_returns_empty_list_when_no_matches(self, tmp_path): + (tmp_path / "file.txt").write_text("x") + + results = BaseIntegrator.find_files_by_glob(tmp_path, "*.prompt.md") + + assert results == [] + + def test_returns_empty_list_for_nonexistent_dir(self, tmp_path): + missing = tmp_path / "nonexistent" + + results = BaseIntegrator.find_files_by_glob(missing, "*.md") + + assert results == [] + + def test_searches_subdirs(self, tmp_path): + subdir = tmp_path / ".apm" / "prompts" + subdir.mkdir(parents=True) + (subdir / "skill.prompt.md").write_text("x") + + results = BaseIntegrator.find_files_by_glob( + tmp_path, "*.prompt.md", subdirs=[".apm/prompts"] + ) + + assert len(results) == 1 + assert results[0].name == "skill.prompt.md" + + def test_deduplicates_across_dirs(self, tmp_path): + """If the same file appears via multiple dirs it's included once.""" + f = tmp_path / "skill.prompt.md" + f.write_text("x") + + # Pass tmp_path as both root and subdir to force duplicate discovery + results = BaseIntegrator.find_files_by_glob( + tmp_path, "*.prompt.md", subdirs=["."] + ) + + names = [r.name for r in results] + assert names.count("skill.prompt.md") == 1 + + def test_combines_root_and_subdir_results(self, tmp_path): + (tmp_path / "root.prompt.md").write_text("x") + subdir = tmp_path / ".apm" / "prompts" + subdir.mkdir(parents=True) + (subdir / "sub.prompt.md").write_text("x") + + results = BaseIntegrator.find_files_by_glob( + tmp_path, "*.prompt.md", subdirs=[".apm/prompts"] + ) + + names = {f.name for f in results} + assert names == {"root.prompt.md", "sub.prompt.md"} + + def test_skips_missing_subdir(self, tmp_path): + (tmp_path / "root.prompt.md").write_text("x") + + results = BaseIntegrator.find_files_by_glob( + tmp_path, "*.prompt.md", subdirs=["missing_subdir"] + ) + + assert len(results) == 1 + assert results[0].name == "root.prompt.md" + + def test_returns_sorted_results_per_dir(self, tmp_path): + for name in ["c.prompt.md", "a.prompt.md", "b.prompt.md"]: + (tmp_path / name).write_text("x") + + results = BaseIntegrator.find_files_by_glob(tmp_path, "*.prompt.md") + + names = [f.name for f in results] + assert names == sorted(names)