diff --git a/pyproject.toml b/pyproject.toml index 2ebad53..2a4341c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,7 @@ ignore = [ "D102", "D103", "D104", + "D107", "D202", "D203", "D212", diff --git a/setup.py b/setup.py index 7d8f97e..bba1a73 100755 --- a/setup.py +++ b/setup.py @@ -34,6 +34,7 @@ from dataclasses import dataclass from enum import Enum, auto from pathlib import Path, PurePath from typing import override +from unittest.mock import patch SCRIPT_DIR = Path(__file__).resolve().parent HOME = Path.home() @@ -55,6 +56,7 @@ class PackageManager(ABC): [*self.check_cmd, name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + check=False, ).returncode == 0 ) @@ -88,7 +90,12 @@ class Apt(PackageManager): @override def is_installed(self, name: str) -> bool: - r = subprocess.run([*self.check_cmd, name], capture_output=True, text=True) + r = subprocess.run( + [*self.check_cmd, name], + capture_output=True, + text=True, + check=False, + ) return r.returncode == 0 and r.stdout.strip() == "install ok installed" @@ -231,7 +238,9 @@ class Resolved: items: list[Item] -def apply_filter(ctx: Context, resolved: Resolved, filters: list[str]) -> Resolved: +def apply_filter( + ctx: Context, resolved: Resolved, filters: list[str] +) -> Resolved: exclusions = copy_exclusions(resolved.items) items: list[Item] = [] matched: set[str] = set() @@ -249,9 +258,13 @@ def apply_filter(ctx: Context, resolved: Resolved, filters: list[str]) -> Resolv return Resolved(pkgs=[], items=items) -def items_for_filter(items: list[Item], path: str, exclusions: set[str]) -> list[Item]: +def items_for_filter( + items: list[Item], path: str, exclusions: set[str] +) -> list[Item]: matches = [ - item for item in items if item.action != Action.COPY and item.src == path + item + for item in items + if item.action != Action.COPY and item.src == path ] copy_matches = [ item @@ -275,7 +288,9 @@ def items_for_filter(items: list[Item], path: str, exclusions: set[str]) -> list def filter_copy_item(item: Item, path: str) -> Item: rel = relative_path(path, item.src) if rel is None: - raise ValueError(f"copy filter path is outside registered item: {path!r}") + raise ValueError( + f"copy filter path is outside registered item: {path!r}" + ) target = join_path(item.target, rel) return Item( @@ -287,7 +302,9 @@ def filter_copy_item(item: Item, path: str) -> Item: def copy_target_excluded(item: Item, path: str, exclusions: set[str]) -> bool: rel = relative_path(path, item.src) - return rel is None or path_is_excluded(join_path(item.target, rel), exclusions) + return rel is None or path_is_excluded( + join_path(item.target, rel), exclusions + ) def copy_exclusions(items: list[Item]) -> set[str]: @@ -310,13 +327,20 @@ def join_path(base: str, rel: PurePath) -> str: def path_is_excluded(path: str, exclusions: set[str]) -> bool: - return any(path_matches_registered(path, excluded) for excluded in exclusions) + return any( + path_matches_registered(path, excluded) for excluded in exclusions + ) def path_matches_registered(path: str, registered: str) -> bool: p = PurePath(path) base = PurePath(registered) - if p.is_absolute() or base.is_absolute() or ".." in p.parts or ".." in base.parts: + if ( + p.is_absolute() + or base.is_absolute() + or ".." in p.parts + or ".." in base.parts + ): return False try: @@ -367,9 +391,9 @@ def prompt_yes_no(question: str, default: bool = False) -> bool: answer = input(question + suffix).strip().lower() except EOFError: return default - if answer in ("y", "yes"): + if answer in {"y", "yes"}: return True - if answer in ("n", "no"): + if answer in {"n", "no"}: return False return default @@ -382,7 +406,9 @@ def check_packages_installed(ctx: Context, pkgs: list[Pkg]) -> None: ctx.error(f"don't know how to manage packages on distro {DISTRO!r}") return - missing = [name for pkg in pkgs if not PM.is_installed(name := pkg.name(DISTRO))] + missing = [ + name for pkg in pkgs if not PM.is_installed(name := pkg.name(DISTRO)) + ] if not missing: return @@ -405,9 +431,12 @@ def check_terminfo(ctx: Context) -> None: result = subprocess.run( ["infocmp", "tmux-256color"], stdout=subprocess.DEVNULL, + check=False, ) if result.returncode != 0: - ctx.error("Missing terminfo for tmux-256color. Try installing ncurses-term.") + ctx.error( + "Missing terminfo for tmux-256color. Try installing ncurses-term." + ) def _describe_kind(p: Path) -> str: @@ -483,7 +512,9 @@ def remove_path(ctx: Context, path: Path) -> None: def same_item(src: Path, dst: Path) -> bool: if src.is_symlink() or dst.is_symlink(): return ( - src.is_symlink() and dst.is_symlink() and src.readlink() == dst.readlink() + src.is_symlink() + and dst.is_symlink() + and src.readlink() == dst.readlink() ) if src.is_file() and dst.is_file(): @@ -496,6 +527,107 @@ def same_item(src: Path, dst: Path) -> bool: return False +def run_copy_diff(ctx: Context, src: Path, dst: Path) -> None: + if not ctx.diff: + return + + diff_args = ["-urN"] if src.is_dir() and dst.is_dir() else ["-u"] + subprocess.run( + [ + "diff", + "--color=always", + *diff_args, + str(dst), + str(src), + ], + check=False, + ) + + +def prepare_copy_target(ctx: Context, src: Path, dst: Path) -> bool: + if not dst.exists() and not dst.is_symlink(): + return True + + if same_item(src, dst): + return False + + run_copy_diff(ctx, src, dst) + + if ctx.ignore_existing: + return False + + if not ctx.force: + ctx.error(f"path already exists and differs: {dst}") + return False + + if src.is_dir() or src.is_symlink() or dst.is_dir() or dst.is_symlink(): + remove_path(ctx, dst) + + return True + + +def copy_children( + ctx: Context, + src: Path, + dst: Path, + rel_src: str, + rel_dst: str, + exclusions: set[str], +) -> None: + for child in sorted(src.iterdir()): + child_dst = dst / child.name + child_rel_src = str(Path(rel_src) / child.name) + child_rel_dst = str(Path(rel_dst) / child.name) + copy_path( + ctx, + child, + child_dst, + child_rel_src, + child_rel_dst, + exclusions, + ) + + +def copy_dir( + ctx: Context, + src: Path, + dst: Path, + rel_src: str, + rel_dst: str, + exclusions: set[str], +) -> None: + if dst.is_dir() and not dst.is_symlink(): + copy_children(ctx, src, dst, rel_src, rel_dst, exclusions) + return + + if not prepare_copy_target(ctx, src, dst): + return + + print(f"Creating directory: {dst}") + if not ctx.dry_run: + dst.mkdir() + + copy_children(ctx, src, dst, rel_src, rel_dst, exclusions) + + +def copy_symlink(ctx: Context, src: Path, dst: Path, rel_src: str) -> None: + if not prepare_copy_target(ctx, src, dst): + return + + print(f"Copying item: from {rel_src} to {dst.parent}/") + if not ctx.dry_run: + dst.symlink_to(src.readlink()) + + +def copy_file(ctx: Context, src: Path, dst: Path, rel_src: str) -> None: + if not prepare_copy_target(ctx, src, dst): + return + + print(f"Copying item: from {rel_src} to {dst.parent}/") + if not ctx.dry_run: + shutil.copy2(src, dst) + + def copy_path( ctx: Context, src: Path, @@ -507,67 +639,23 @@ def copy_path( if path_is_excluded(rel_dst, exclusions): return - removed_dst = False - if dst.exists() or dst.is_symlink(): - if same_item(src, dst): - return - - if src.is_dir() and dst.is_dir() and not dst.is_symlink(): - for child in sorted(src.iterdir()): - child_dst = dst / child.name - child_rel_src = str(Path(rel_src) / child.name) - child_rel_dst = str(Path(rel_dst) / child.name) - copy_path( - ctx, - child, - child_dst, - child_rel_src, - child_rel_dst, - exclusions, - ) - return - - if ctx.ignore_existing: - return - - if not ctx.force: - ctx.error(f"path already exists and differs: {dst}") - return - - if src.is_dir() or src.is_symlink() or dst.is_dir() or dst.is_symlink(): - remove_path(ctx, dst) - removed_dst = True + if same_item(src, dst): + return if src.is_symlink(): - print(f"Copying item: from {rel_src} to {dst.parent}/") - if not ctx.dry_run: - dst.symlink_to(src.readlink()) + copy_symlink(ctx, src, dst, rel_src) return if src.is_dir(): - if removed_dst or not dst.exists(): - print(f"Creating directory: {dst}") - if not ctx.dry_run: - dst.mkdir() - elif dst.is_symlink() or not dst.is_dir(): - remove_path(ctx, dst) - print(f"Creating directory: {dst}") - if not ctx.dry_run: - dst.mkdir() - - for child in sorted(src.iterdir()): - child_dst = dst / child.name - child_rel_src = str(Path(rel_src) / child.name) - child_rel_dst = str(Path(rel_dst) / child.name) - copy_path(ctx, child, child_dst, child_rel_src, child_rel_dst, exclusions) + copy_dir(ctx, src, dst, rel_src, rel_dst, exclusions) return - print(f"Copying item: from {rel_src} to {dst.parent}/") - if not ctx.dry_run: - shutil.copy2(src, dst) + copy_file(ctx, src, dst, rel_src) -def copy_item(ctx: Context, item: Item, exclusions: set[str] | None = None) -> None: +def copy_item( + ctx: Context, item: Item, exclusions: set[str] | None = None +) -> None: excluded = exclusions or set() rel_src = item.src rel_dst = item.target @@ -590,33 +678,19 @@ def copy_item(ctx: Context, item: Item, exclusions: set[str] | None = None) -> N if not ctx.dry_run: dst.parent.mkdir(parents=True, exist_ok=True) - if dst.exists() or dst.is_symlink(): - if same_item(src, dst): - return - - if ctx.diff: - diff_args = ["-urN"] if src.is_dir() and dst.is_dir() else ["-u"] - subprocess.run(["diff", "--color=always", *diff_args, str(dst), str(src)]) - - if ctx.force and not src.is_dir(): - remove_path(ctx, dst) - elif ctx.ignore_existing: - return - elif not ctx.force and not ( - src.is_dir() and dst.is_dir() and not dst.is_symlink() - ): - ctx.error(f"path already exists and differs: {dst}") - return - copy_path(ctx, src, dst, rel_src, rel_dst, excluded) def _sudo_test(*args: str) -> bool: - return subprocess.run(["sudo", "test", *args]).returncode == 0 + return subprocess.run(["sudo", "test", *args], check=False).returncode == 0 def _sudo_cmp(a: Path, b: Path) -> bool: - return subprocess.run(["sudo", "cmp", "-s", str(a), str(b)]).returncode == 0 + result = subprocess.run( + ["sudo", "cmp", "-s", str(a), str(b)], + check=False, + ) + return result.returncode == 0 def install_system_file(ctx: Context, item: Item) -> None: @@ -628,7 +702,9 @@ def install_system_file(ctx: Context, item: Item) -> None: return parent_exists = ( - dst.parent.is_dir() if ctx.dry_run else _sudo_test("-d", str(dst.parent)) + dst.parent.is_dir() + if ctx.dry_run + else _sudo_test("-d", str(dst.parent)) ) dst_exists = dst.exists() if ctx.dry_run else _sudo_test("-e", str(dst)) @@ -639,14 +715,25 @@ def install_system_file(ctx: Context, item: Item) -> None: if dst_exists: matches = ( - filecmp.cmp(src, dst, shallow=False) if ctx.dry_run else _sudo_cmp(src, dst) + filecmp.cmp(src, dst, shallow=False) + if ctx.dry_run + else _sudo_cmp(src, dst) ) if matches: return if ctx.diff: diff_cmd = ["diff"] if ctx.dry_run else ["sudo", "diff"] - subprocess.run([*diff_cmd, "--color=always", "-u", str(dst), str(src)]) + subprocess.run( + [ + *diff_cmd, + "--color=always", + "-u", + str(dst), + str(src), + ], + check=False, + ) if ctx.force: print(f"Overwriting: {dst}") @@ -671,14 +758,18 @@ def create_all_symlinks(ctx: Context, items: list[Item]) -> None: create_symlink(ctx, item.src, item.target) -def copy_all_items(ctx: Context, items: list[Item], exclusions: set[str]) -> None: +def copy_all_items( + ctx: Context, items: list[Item], exclusions: set[str] +) -> None: for item in items: if item.action == Action.COPY: copy_item(ctx, item, exclusions) def install_all_system_files(ctx: Context, items: list[Item]) -> None: - system_installs = [item for item in items if item.action == Action.SYSTEM_INSTALL] + system_installs = [ + item for item in items if item.action == Action.SYSTEM_INSTALL + ] if not system_installs: return if not ctx.dry_run: @@ -719,14 +810,20 @@ def parse_args() -> argparse.Namespace: "-n", "--dry-run", action="store_true", - help="Show what would be changed without writing files or installing packages", + help=( + "Show what would be changed without writing files or installing " + "packages" + ), ) extras = sorted(PROFILES.keys() - {"base"}) parser.add_argument( "profiles", nargs="*", metavar="PROFILE", - help=f"Extra profiles to apply on top of base (available: {', '.join(extras)})", + help=( + "Extra profiles to apply on top of base " + f"(available: {', '.join(extras)})" + ), ) parser.add_argument( "-f", @@ -799,19 +896,23 @@ class SetupTests(unittest.TestCase): globals()["HOME"] = self.old_home self.tmp.cleanup() + @staticmethod def capture_copy( - self, ctx: Context, item: Item, exclusions: set[str] | None = None, ) -> str: output = io.StringIO() errors = io.StringIO() - with contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors): + with ( + contextlib.redirect_stdout(output), + contextlib.redirect_stderr(errors), + ): copy_item(ctx, item, exclusions) return output.getvalue() - def test_filter_allows_descendant_of_registered_copy_path(self) -> None: + @staticmethod + def test_filter_allows_descendant_of_registered_copy_path() -> None: ctx = _test_ctx() resolved = Resolved( pkgs=[], @@ -820,10 +921,11 @@ class SetupTests(unittest.TestCase): filtered = apply_filter(ctx, resolved, [".codex/skills"]) - self.assertEqual(filtered.items, [cp(".codex/skills")]) - self.assertEqual(ctx.errors, []) + assert filtered.items == [cp(".codex/skills")] + assert ctx.errors == [] - def test_filter_maps_descendant_of_registered_copy_path(self) -> None: + @staticmethod + def test_filter_maps_descendant_of_registered_copy_path() -> None: ctx = _test_ctx() resolved = Resolved( pkgs=[], @@ -832,13 +934,13 @@ class SetupTests(unittest.TestCase): filtered = apply_filter(ctx, resolved, ["codex/skills/check"]) - self.assertEqual( - filtered.items, - [cp("codex/skills/check", ".codex/skills/check")], - ) - self.assertEqual(ctx.errors, []) + assert filtered.items == [ + cp("codex/skills/check", ".codex/skills/check") + ] + assert ctx.errors == [] - def test_filter_rejects_sibling_of_registered_copy_path(self) -> None: + @staticmethod + def test_filter_rejects_sibling_of_registered_copy_path() -> None: ctx = _test_ctx() resolved = Resolved( pkgs=[], @@ -849,13 +951,13 @@ class SetupTests(unittest.TestCase): with contextlib.redirect_stderr(errors): filtered = apply_filter(ctx, resolved, [".codex-nope"]) - self.assertEqual(filtered.items, []) - self.assertEqual( - ctx.errors, - ["--filter path not in selected profiles: '.codex-nope'"], - ) + assert filtered.items == [] + assert ctx.errors == [ + "--filter path not in selected profiles: '.codex-nope'" + ] - def test_filter_rejects_descendant_of_symlink_owned_copy_path(self) -> None: + @staticmethod + def test_filter_rejects_descendant_of_symlink_owned_copy_path() -> None: ctx = _test_ctx() resolved = Resolved( pkgs=[], @@ -869,13 +971,14 @@ class SetupTests(unittest.TestCase): with contextlib.redirect_stderr(errors): filtered = apply_filter(ctx, resolved, [".claude/skills/check"]) - self.assertEqual(filtered.items, []) - self.assertEqual( - ctx.errors, - ["--filter path not in selected profiles: '.claude/skills/check'"], - ) + assert filtered.items == [] + assert ctx.errors == [ + "--filter path not in selected profiles: '.claude/skills/check'" + ] - def test_dry_run_recursive_copy_prints_children_without_writing(self) -> None: + def test_dry_run_recursive_copy_prints_children_without_writing( + self, + ) -> None: src = SCRIPT_DIR / "cfg" src.mkdir() (src / "a.txt").write_text("a\n") @@ -885,11 +988,11 @@ class SetupTests(unittest.TestCase): output = self.capture_copy(ctx, cp("cfg")) - self.assertEqual(ctx.errors, []) - self.assertFalse((HOME / "cfg").exists()) - self.assertIn(f"Creating directory: {HOME / 'cfg'}", output) - self.assertIn("Copying item: from cfg/a.txt", output) - self.assertIn("Copying item: from cfg/sub/b.txt", output) + assert ctx.errors == [] + assert not (HOME / "cfg").exists() + assert f"Creating directory: {HOME / 'cfg'}" in output + assert "Copying item: from cfg/a.txt" in output + assert "Copying item: from cfg/sub/b.txt" in output def test_copy_item_can_map_destination(self) -> None: src = SCRIPT_DIR / "cfg" @@ -899,12 +1002,14 @@ class SetupTests(unittest.TestCase): output = self.capture_copy(ctx, cp("cfg", "mapped/cfg")) - self.assertEqual(ctx.errors, []) - self.assertEqual((HOME / "mapped" / "cfg" / "a.txt").read_text(), "a\n") - self.assertFalse((HOME / "cfg").exists()) - self.assertIn("Copying item: from cfg/a.txt", output) + assert ctx.errors == [] + assert (HOME / "mapped" / "cfg" / "a.txt").read_text() == "a\n" + assert not (HOME / "cfg").exists() + assert "Copying item: from cfg/a.txt" in output - def test_force_directory_copy_merges_without_removing_extra_files(self) -> None: + def test_force_directory_copy_merges_without_removing_extra_files( + self, + ) -> None: src = SCRIPT_DIR / "cfg" dst = HOME / "cfg" src.mkdir() @@ -916,11 +1021,11 @@ class SetupTests(unittest.TestCase): output = self.capture_copy(ctx, cp("cfg")) - self.assertEqual(ctx.errors, []) - self.assertEqual((dst / "a.txt").read_text(), "source\n") - self.assertEqual((dst / "keep.txt").read_text(), "keep\n") - self.assertNotIn(f"Trashing item: {dst}", output) - self.assertIn("Copying item: from cfg/a.txt", output) + assert ctx.errors == [] + assert (dst / "a.txt").read_text() == "source\n" + assert (dst / "keep.txt").read_text() == "keep\n" + assert f"Trashing item: {dst}" not in output + assert "Copying item: from cfg/a.txt" in output def test_force_directory_copy_replaces_destination_symlink(self) -> None: src = SCRIPT_DIR / "cfg" @@ -935,11 +1040,11 @@ class SetupTests(unittest.TestCase): output = self.capture_copy(ctx, cp("cfg")) - self.assertEqual(ctx.errors, []) - self.assertIn(f"Trashing item: {dst}", output) - self.assertIn(f"Creating directory: {dst}", output) - self.assertIn("Copying item: from cfg/a.txt", output) - self.assertEqual((target / "keep.txt").read_text(), "keep\n") + assert ctx.errors == [] + assert f"Trashing item: {dst}" in output + assert f"Creating directory: {dst}" in output + assert "Copying item: from cfg/a.txt" in output + assert (target / "keep.txt").read_text() == "keep\n" def test_directory_copy_skips_symlink_owned_children(self) -> None: src = SCRIPT_DIR / "cfg" @@ -952,12 +1057,14 @@ class SetupTests(unittest.TestCase): (dst / "linked.txt").symlink_to(SCRIPT_DIR / "elsewhere") ctx = _test_ctx(force=True, dry_run=True) - output = self.capture_copy(ctx, cp("cfg"), exclusions={"cfg/linked.txt"}) + output = self.capture_copy( + ctx, cp("cfg"), exclusions={"cfg/linked.txt"} + ) - self.assertEqual(ctx.errors, []) - self.assertIn("Copying item: from cfg/managed.txt", output) - self.assertNotIn("cfg/linked.txt", output) - self.assertTrue((dst / "linked.txt").is_symlink()) + assert ctx.errors == [] + assert "Copying item: from cfg/managed.txt" in output + assert "cfg/linked.txt" not in output + assert (dst / "linked.txt").is_symlink() def test_directory_copy_without_force_errors_when_managed_file_differs( self, @@ -972,39 +1079,37 @@ class SetupTests(unittest.TestCase): self.capture_copy(ctx, cp("cfg")) - self.assertEqual( - ctx.errors, - [f"path already exists and differs: {dst / 'a.txt'}"], - ) - self.assertEqual((dst / "a.txt").read_text(), "dest\n") + assert ctx.errors == [ + f"path already exists and differs: {dst / 'a.txt'}" + ] + assert (dst / "a.txt").read_text() == "dest\n" - def test_install_uses_item_mode(self) -> None: + @staticmethod + def test_install_uses_item_mode() -> None: src = SCRIPT_DIR / "bin" / "tool" src.parent.mkdir() src.write_text("tool\n") ctx = _test_ctx() calls: list[list[str]] = [] - def fake_run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess: - calls.append(cmd) - if cmd[0:4] == ["sudo", "test", "-e", "/usr/local/bin/tool"]: - return subprocess.CompletedProcess(cmd, 1) - return subprocess.CompletedProcess(cmd, 0) + def fake_run( + args: list[str], **kwargs: object + ) -> subprocess.CompletedProcess[str]: + calls.append(args) + if args[0:4] == ["sudo", "test", "-e", "/usr/local/bin/tool"]: + return subprocess.CompletedProcess(args, 1, "", "") + return subprocess.CompletedProcess(args, 0, "", "") - old_run = subprocess.run - subprocess.run = fake_run - try: + with patch("subprocess.run", new=fake_run): output = io.StringIO() with contextlib.redirect_stdout(output): install_system_file( ctx, install("bin/tool", "usr/local/bin/tool", mode=0o755), ) - finally: - subprocess.run = old_run - self.assertEqual(ctx.errors, []) - self.assertEqual(calls[-1][0:4], ["sudo", "install", "-m", "755"]) + assert ctx.errors == [] + assert calls[-1][0:4] == ["sudo", "install", "-m", "755"] if __name__ == "__main__":