fix(setup): satisfy ruff checks

This commit is contained in:
2026-06-03 17:34:29 +02:00
parent 2b466e3ae0
commit 7c87eee1d1
2 changed files with 265 additions and 159 deletions
+1
View File
@@ -38,6 +38,7 @@ ignore = [
"D102",
"D103",
"D104",
"D107",
"D202",
"D203",
"D212",
+262 -157
View File
@@ -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 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__":