"""Integration tests for git cleanup, using disposable repositories only. Run with Python 3.11+. Set GIT_CLEANUP_SCRIPT to test another checkout's script. """ import json import os from pathlib import Path import re import shutil import subprocess import sys import tempfile import unittest SCRIPT = Path(os.environ.get( "GIT_CLEANUP_SCRIPT", Path(__file__).resolve().parents[1] / "git-cleanup" )).resolve() class CleanupTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix="git cleanup fixtures ") self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name).resolve() self.home = self.root / "home" self.home.mkdir() # A pre-push hook or parent Git command must not redirect fixtures into # the real checkout or share its config, index, objects, or worktrees. self.env = { key: value for key, value in os.environ.items() if not key.startswith(("GIT_", "GH_")) } self.env.update({ "HOME": str(self.home), "XDG_CONFIG_HOME": str(self.home / ".config"), "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": os.devnull, "GIT_AUTHOR_NAME": "Cleanup Tests", "GIT_AUTHOR_EMAIL": "cleanup@example.invalid", "GIT_COMMITTER_NAME": "Cleanup Tests", "GIT_COMMITTER_EMAIL": "cleanup@example.invalid", "GIT_TERMINAL_PROMPT": "0", "LC_ALL": "C", }) self.repo = self.root / "repo" self.repo.mkdir() self.git("init", "--quiet", "--initial-branch=main") self.write("tracked.txt", "initial\n") self.write(".gitignore", "ignored/\n.notes/\n") self.commit("Initial files") def git(self, *args, cwd=None, check=True, env=None): result = subprocess.run( ["git", *args], cwd=cwd or self.repo, env=env or self.env, capture_output=True, text=True, ) if check and result.returncode: self.fail(f"git {args!r} failed ({result.returncode}):\n{result.stderr}") return result.stdout.strip() if check else result def write(self, name, content, cwd=None): path = (cwd or self.repo) / name path.parent.mkdir(parents=True, exist_ok=True) if isinstance(content, bytes): path.write_bytes(content) else: path.write_text(content) return path def commit(self, message="Fixture change", cwd=None): self.git("add", "--all", cwd=cwd) self.git("commit", "--quiet", "-m", message, cwd=cwd) return self.git("rev-parse", "HEAD", cwd=cwd) def topic(self, name="topic", filename="feature.txt", content="feature\n"): self.git("switch", "--quiet", "-c", name) self.write(filename, content) oid = self.commit() self.git("switch", "--quiet", "main") return oid def worktree(self, name="topic", suffix=None): self.git("branch", name) path = self.root / (suffix or f"worktree {name}") self.git("worktree", "add", "--quiet", str(path), name) return path def cleanup(self, *args, cwd=None, check=True, env=None): result = subprocess.run( [sys.executable, str(SCRIPT), *args], cwd=cwd or self.repo, env=env or self.env, capture_output=True, text=True, ) if check and result.returncode: self.fail( f"cleanup {args!r} failed ({result.returncode}):\n" f"stdout: {result.stdout}\nstderr: {result.stderr}" ) return result def plan(self, *args, cwd=None): result = self.cleanup(*args, "--json", cwd=cwd) try: return json.loads(result.stdout) except json.JSONDecodeError as exc: self.fail(f"JSON output was invalid: {exc}\n{result.stdout}") def entry(self, plan, ref): entries = [entry for entry in plan["entries"] if entry["ref"] == ref] self.assertEqual(len(entries), 1, f"Missing/duplicate {ref}: {plan}") return entries[0] def exists(self, ref, cwd=None): result = self.git("show-ref", "--verify", "--quiet", ref, cwd=cwd, check=False) self.assertIn(result.returncode, (0, 1), result.stderr) return result.returncode == 0 def backup_oids(self): return self.git("for-each-ref", "--format=%(objectname)", "refs/cleanup/").splitlines() def remote(self): path = self.root / "upstream.git" self.git("init", "--quiet", "--bare", "--initial-branch=main", str(path)) self.git("remote", "add", "origin", str(path)) self.git("push", "--quiet", "--set-upstream", "origin", "main") self.git("remote", "set-head", "origin", "main") return path def git_shim(self, body): """Install a fixture-only Git wrapper to inject a precise failure/race.""" real_git = shutil.which("git", path=self.env["PATH"]) self.assertIsNotNone(real_git) binary_dir = self.root / "bin" binary_dir.mkdir(exist_ok=True) shim = binary_dir / "git" shim.write_text( f"#!{sys.executable}\n" "import os, pathlib, subprocess, sys\n" f"real_git = {real_git!r}\n" "args = sys.argv[1:]\n" f"{body}\n" "os.execv(real_git, [real_git, *args])\n" ) shim.chmod(0o755) self.env["PATH"] = f"{binary_dir}{os.pathsep}{self.env['PATH']}" def git_transaction_shim(self, body): """Inject a race while forwarding Git's interactive transaction protocol.""" self.git_shim( "if 'update-ref' in args and '--stdin' in args:\n" " import threading\n" " child = subprocess.Popen([real_git, *args], stdin=subprocess.PIPE)\n" " threading.Thread(target=lambda: os._exit(child.wait()), daemon=True).start()\n" " for line in sys.stdin.buffer:\n" + "\n".join(" " + line for line in body.splitlines()) + "\n" " child.stdin.write(line)\n" " child.stdin.flush()\n" " child.stdin.close()\n" " sys.exit(child.wait())" ) def github_fixture(self, prs, failure=False): """Serve synthetic PR responses without invoking a real forge client.""" self.git("remote", "set-url", "origin", "https://github.com/fixture/cleanup.git") binary_dir = self.root / "bin" binary_dir.mkdir(exist_ok=True) fake = binary_dir / "gh" fake.write_text( f"#!{sys.executable}\n" "import json, sys\n" f"prs = {prs!r}\n" f"failure = {failure!r}\n" "if failure:\n" " print('fixture: GitHub unavailable', file=sys.stderr)\n" " sys.exit(1)\n" "state = sys.argv[sys.argv.index('--state') + 1].upper()\n" "assert sys.argv[sys.argv.index('--repo') + 1] == 'github.com/fixture/cleanup'\n" "print(json.dumps([p for p in prs if state == 'ALL' or p['state'] == state]))\n" ) fake.chmod(0o755) self.env["PATH"] = f"{binary_dir}{os.pathsep}{self.env['PATH']}" @staticmethod def pr(head, state="MERGED", landing=None, cross_repository=False): return { "number": 17, "state": state, "headRefName": "topic", "headRefOid": head, "isCrossRepository": cross_repository, "mergeCommit": {"oid": landing} if landing else None, } def test_help_outside_a_repository(self): result = self.cleanup("--help", cwd=self.home) self.assertIn("--apply", result.stdout) self.assertIn("--obsolete", result.stdout) def test_color_modes_preserve_plain_plan_text(self): self.git("branch", "candidate") plain = self.cleanup("--local", "--no-color").stdout self.assertNotIn("\033[", plain) self.assertEqual(plain, self.cleanup("--local", "--color=never").stdout) self.assertEqual(plain, self.cleanup("--local", "--color=always", "--no-color").stdout) for flags in (("--color=always",), ("--color",), ("--no-color", "--color")): with self.subTest(flags=flags): colored = self.cleanup("--local", *flags).stdout self.assertIn("\033[31mdelete", colored) self.assertIn("\033[32mkeep", colored) self.assertEqual(re.sub(r"\033\[[0-9;]*m", "", colored), plain) def test_auto_color_keeps_redirected_output_plain(self): env = dict(self.env, TERM="xterm-256color") env.pop("NO_COLOR", None) for flags in ((), ("--color=auto",)): with self.subTest(flags=flags): result = self.cleanup("--local", *flags, env=env) self.assertNotIn("\033[", result.stdout) def test_json_stays_plain_even_when_color_is_forced(self): self.git("branch", "candidate") result = self.cleanup("--local", "--color=always", "--json") self.assertNotIn("\033", result.stdout) plan = json.loads(result.stdout) self.assertEqual(self.entry(plan, "refs/heads/candidate")["action"], "delete") @unittest.skipUnless(os.name == "posix", "pseudo-terminal check requires POSIX") def test_auto_color_respects_terminal_and_environment(self): import errno import pty cases = [ ((), {}, True), (("--color=auto",), {"NO_COLOR": "1"}, False), ((), {"TERM": "dumb"}, False), (("--no-color",), {}, False), (("--color=always",), {"NO_COLOR": "1", "TERM": "dumb"}, True), ((), {"NO_COLOR": ""}, True), ] for flags, overrides, colored in cases: with self.subTest(flags=flags, overrides=overrides): env = dict(self.env, TERM="xterm-256color") env.pop("NO_COLOR", None) env.update(overrides) master, slave = pty.openpty() process = None try: process = subprocess.Popen( [sys.executable, str(SCRIPT), "--local", *flags], cwd=self.repo, env=env, stdout=slave, stderr=subprocess.PIPE, ) os.close(slave) slave = None chunks = [] while True: try: chunk = os.read(master, 65536) except OSError as error: if error.errno == errno.EIO: break raise if not chunk: break chunks.append(chunk) _, stderr = process.communicate(timeout=10) self.assertEqual(process.returncode, 0, stderr) self.assertEqual(b"\033[" in b"".join(chunks), colored) finally: os.close(master) if slave is not None: os.close(slave) if process is not None and process.poll() is None: process.kill() process.communicate() def test_preview_changes_no_refs_worktrees_or_files(self): path = self.worktree() refs = self.git("show-ref") trees = self.git("worktree", "list", "--porcelain") status = self.git("status", "--porcelain=v1", "--untracked-files=all") plan = self.plan("--local") self.assertEqual(plan["mode"], "preview") self.assertEqual(self.entry(plan, "refs/heads/topic")["action"], "delete") self.assertEqual(self.git("show-ref"), refs) self.assertEqual(self.git("worktree", "list", "--porcelain"), trees) self.assertEqual(self.git("status", "--porcelain=v1", "--untracked-files=all"), status) self.assertTrue(path.exists()) self.assertEqual(self.backup_oids(), []) common = Path(self.git("rev-parse", "--path-format=absolute", "--git-common-dir")) self.assertFalse((common / "git-cleanup").exists()) def test_apply_removes_merged_branch_and_clean_worktree_with_recovery_ref(self): oid = self.topic() self.git("merge", "--quiet", "--ff-only", "topic") path = self.root / "merged worktree" self.git("worktree", "add", "--quiet", str(path), "topic") self.cleanup("--local", "--apply") self.assertFalse(self.exists("refs/heads/topic")) self.assertFalse(path.exists()) self.assertIn(oid, self.backup_oids()) self.assertTrue(self.exists("refs/heads/main")) common = Path(self.git("rev-parse", "--path-format=absolute", "--git-common-dir")) journals = list((common / "git-cleanup").glob("*.json")) self.assertTrue(journals, "Apply should retain a recovery journal") for path in journals: json.loads(path.read_text()) def test_unique_work_is_retained(self): self.topic() plan = self.plan("--local") self.assertEqual(self.entry(plan, "refs/heads/topic")["action"], "keep") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic")) def test_integration_current_and_protected_branches_are_retained(self): self.git("branch", "master") self.git("branch", "release/stable") self.git("branch", "expendable") self.cleanup("--local", "--protect", "release/*", "--apply") for name in ("main", "master", "release/stable"): self.assertTrue(self.exists(f"refs/heads/{name}")) self.assertFalse(self.exists("refs/heads/expendable")) def test_current_linked_and_primary_worktrees_are_retained(self): path = self.worktree("current") self.git("switch", "--quiet", "-c", "primary") self.cleanup("--local", "--apply", cwd=path) self.assertTrue(path.exists()) self.assertTrue(self.exists("refs/heads/current")) self.assertTrue(self.exists("refs/heads/primary")) def test_dirty_untracked_ignored_and_staged_worktrees_are_retained(self): paths = {} for name in ("dirty", "untracked", "ignored", "staged", "notes"): paths[name] = self.worktree(name) self.write("tracked.txt", "unsaved\n", cwd=paths["dirty"]) self.write("new.txt", "untracked\n", cwd=paths["untracked"]) self.write("ignored/evidence.txt", "keep me\n", cwd=paths["ignored"]) self.write("tracked.txt", "staged\n", cwd=paths["staged"]) self.git("add", "tracked.txt", cwd=paths["staged"]) self.write(".notes/review.md", "review evidence\n", cwd=paths["notes"]) self.cleanup("--local", "--apply") for name, path in paths.items(): with self.subTest(name=name): self.assertTrue(path.exists()) self.assertTrue(self.exists(f"refs/heads/{name}")) def test_locked_worktree_is_retained(self): path = self.worktree() self.git("worktree", "lock", "--reason", "mounted elsewhere", str(path)) self.cleanup("--local", "--apply") self.assertTrue(path.exists()) self.assertTrue(self.exists("refs/heads/topic")) def test_missing_worktree_registration_is_retained(self): path = self.worktree() path.rename(self.root / "temporarily moved") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic")) self.assertIn(str(path), self.git("worktree", "list", "--porcelain")) def test_operation_in_progress_is_retained(self): path = self.worktree() operation = Path(self.git("rev-parse", "--path-format=absolute", "--git-path", "rebase-merge", cwd=path)) operation.mkdir() self.cleanup("--local", "--apply") self.assertTrue(path.exists()) self.assertTrue(self.exists("refs/heads/topic")) def test_detached_worktree_is_retained(self): path = self.root / "detached" self.git("worktree", "add", "--quiet", "--detach", str(path), "main") self.cleanup("--local", "--apply") self.assertTrue(path.exists()) self.assertIn(str(path), self.git("worktree", "list", "--porcelain")) def test_worktree_path_with_newline_is_handled(self): path = self.worktree(suffix="worktree with\nnewline") self.cleanup("--local", "--apply") self.assertFalse(path.exists()) self.assertFalse(self.exists("refs/heads/topic")) def test_squashed_changes_are_recognized_with_unrelated_base_changes(self): oid = self.topic(filename="name with\nnewline.txt", content=b"\0binary\xff") self.write("unrelated.txt", "main only\n") self.commit() self.git("merge", "--squash", "topic") self.commit("Squash topic") self.assertNotEqual(self.git("rev-parse", "main"), oid) self.cleanup("--local", "--apply") self.assertFalse(self.exists("refs/heads/topic")) self.assertIn(oid, self.backup_oids()) def test_renamed_deleted_and_executable_paths_in_squash(self): self.write("old.txt", "rename me\n") self.write("delete.txt", "delete me\n") self.write("script", "#!/bin/sh\nexit 0\n") self.commit() self.git("switch", "--quiet", "-c", "topic") self.git("mv", "old.txt", "new.txt") (self.repo / "delete.txt").unlink() (self.repo / "script").chmod(0o755) self.commit() self.git("switch", "--quiet", "main") self.git("merge", "--squash", "topic") self.commit("Squash topic") self.cleanup("--local", "--apply") self.assertFalse(self.exists("refs/heads/topic")) def test_partial_incorporation_is_not_enough(self): self.git("switch", "--quiet", "-c", "topic") self.write("one.txt", "one\n") self.write("two.txt", "two\n") self.commit() self.git("switch", "--quiet", "main") self.write("one.txt", "one\n") self.commit() self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic")) def test_work_added_after_squash_is_retained(self): self.topic() self.git("merge", "--squash", "topic") self.commit("Squash topic") self.git("switch", "--quiet", "topic") self.write("later.txt", "new work\n") self.commit() self.git("switch", "--quiet", "main") self.cleanup("--local", "--scan-history", "100", "--apply") self.assertTrue(self.exists("refs/heads/topic")) def test_empty_commit_in_unmerged_history_is_retained(self): self.topic() self.git("switch", "--quiet", "topic") self.git("commit", "--quiet", "--allow-empty", "-m", "Meaningful marker") self.git("switch", "--quiet", "main") self.git("merge", "--squash", "topic") self.commit("Squash topic") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic")) def test_reverted_work_in_unmerged_history_is_retained(self): self.topic() self.git("switch", "--quiet", "topic") self.write("temporary.txt", "temporary\n") temporary = self.commit() self.git("revert", "--no-edit", temporary) self.git("switch", "--quiet", "main") self.git("merge", "--squash", "topic") self.commit("Squash topic") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic")) def test_cancelled_path_without_repeated_tree_is_retained(self): self.topic(filename="one.txt", content="one\n") self.git("switch", "--quiet", "topic") self.write("temporary.txt", "temporary\n") self.commit() self.write("one.txt", "two\n") self.commit() (self.repo / "temporary.txt").unlink() self.commit() self.git("switch", "--quiet", "main") self.git("merge", "--squash", "topic") self.commit("Squash topic") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic")) def test_unmerged_merge_history_is_retained(self): self.topic("topic", "topic.txt") self.topic("auxiliary", "auxiliary.txt") self.git("switch", "--quiet", "topic") self.git("merge", "--quiet", "--no-ff", "-m", "Merge auxiliary", "auxiliary") self.git("switch", "--quiet", "main") self.git("merge", "--squash", "topic") self.commit("Squash topic") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic")) def test_historical_squash_matching_requires_scan_opt_in(self): self.topic() self.git("merge", "--squash", "topic") self.commit("Squash topic") self.write("feature.txt", "later main change\n") self.commit() self.assertEqual(self.entry(self.plan("--local"), "refs/heads/topic")["action"], "keep") self.cleanup("--local", "--scan-history", "100", "--apply") self.assertFalse(self.exists("refs/heads/topic")) def test_worker_ancestry_in_retained_parent(self): self.topic("worker") self.git("switch", "--quiet", "-c", "parent", "worker") self.write("parent.txt", "parent work\n") parent_oid = self.commit() self.git("switch", "--quiet", "main") self.cleanup("--local", "--apply") self.assertFalse(self.exists("refs/heads/worker")) self.assertEqual(self.git("rev-parse", "parent"), parent_oid) def test_squashed_worker_into_explicit_parent(self): self.topic("worker") self.git("switch", "--quiet", "-c", "parent") self.write("parent.txt", "parent work\n") self.commit() self.git("merge", "--squash", "worker") parent_oid = self.commit() self.git("switch", "--quiet", "main") self.cleanup("--local", "--into", "parent", "--apply") self.assertFalse(self.exists("refs/heads/worker")) self.assertEqual(self.git("rev-parse", "parent"), parent_oid) def test_identical_unmerged_peers_keep_a_surviving_branch(self): oid = self.topic("peer-a") self.git("branch", "peer-b", "peer-a") self.cleanup("--local", "--apply") surviving = [name for name in ("peer-a", "peer-b") if self.exists(f"refs/heads/{name}")] self.assertTrue(surviving, "Duplicate peers must not delete each other") for name in surviving: self.assertEqual(self.git("rev-parse", name), oid) def test_obsolete_selection_is_explicit_and_recoverable(self): oid = self.topic("abandoned") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/abandoned")) self.cleanup("--local", "--obsolete", "abandoned", "--apply") self.assertFalse(self.exists("refs/heads/abandoned")) self.assertIn(oid, self.backup_oids()) def test_obsolete_does_not_override_worktree_protection(self): path = self.worktree() self.write("new.txt", "untracked\n", cwd=path) self.cleanup("--local", "--obsolete", "topic", "--apply") self.assertTrue(self.exists("refs/heads/topic")) self.assertTrue(path.exists()) def test_invalid_explicit_base_fails_without_changes(self): self.git("branch", "candidate") before = self.git("show-ref") result = self.cleanup("--local", "--base", "nonexistent", "--apply", check=False) self.assertNotEqual(result.returncode, 0) self.assertEqual(self.git("show-ref"), before) def test_unrelated_config_does_not_authorize_cleanup(self): self.topic() self.git("config", "inspection.base", "topic") plan = self.plan("--local") self.assertEqual(plan["base"]["oid"], self.git("rev-parse", "main")) self.assertEqual(self.entry(plan, "refs/heads/topic")["action"], "keep") def test_configured_main_branch_is_preserved(self): self.git("branch", "-m", "main", "trunk") self.git("config", "cleanup.mainBranch", "trunk") self.git("branch", "candidate") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/trunk")) self.assertFalse(self.exists("refs/heads/candidate")) def test_remote_merged_branch_is_removed_and_backed_up(self): remote = self.remote() oid = self.topic() self.git("push", "--quiet", "origin", "topic") self.git("merge", "--quiet", "--ff-only", "topic") self.git("push", "--quiet", "origin", "main") self.cleanup("--apply") self.assertFalse(self.exists("refs/heads/topic", cwd=remote)) self.assertFalse(self.exists("refs/heads/topic")) self.assertIn(oid, self.backup_oids()) self.assertTrue(self.exists("refs/heads/main", cwd=remote)) def test_matching_remote_copy_does_not_prove_incorporation(self): remote = self.remote() self.topic() self.git("push", "--quiet", "--set-upstream", "origin", "topic") self.cleanup("--apply") self.assertTrue(self.exists("refs/heads/topic")) self.assertTrue(self.exists("refs/heads/topic", cwd=remote)) def test_unpublished_local_main_cannot_delete_remote_branch(self): remote = self.remote() self.topic() self.git("push", "--quiet", "origin", "topic") self.git("merge", "--quiet", "--ff-only", "topic") plan = self.plan() self.assertEqual(plan["base"]["oid"], self.git("rev-parse", "origin/main")) self.cleanup("--apply") self.assertTrue(self.exists("refs/heads/topic", cwd=remote)) def test_local_mode_never_deletes_remote_branches(self): remote = self.remote() self.topic() self.git("push", "--quiet", "origin", "topic") self.git("merge", "--quiet", "--ff-only", "topic") self.git("push", "--quiet", "origin", "main") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic", cwd=remote)) self.assertFalse(self.exists("refs/heads/topic")) def test_gone_upstream_does_not_delete_unique_local_work(self): remote = self.remote() self.topic() self.git("push", "--quiet", "--set-upstream", "origin", "topic") self.git("update-ref", "-d", "refs/heads/topic", cwd=remote) self.cleanup("--apply") self.assertTrue(self.exists("refs/heads/topic")) def test_split_fetch_push_destination_is_refused(self): remote = self.remote() self.topic() self.git("push", "--quiet", "origin", "topic") self.git("merge", "--quiet", "--ff-only", "topic") self.git("push", "--quiet", "origin", "main") other = self.root / "other.git" self.git("clone", "--quiet", "--bare", str(remote), str(other)) self.git("config", "remote.origin.pushurl", str(other)) self.cleanup("--apply", check=False) self.assertTrue(self.exists("refs/heads/topic", cwd=remote)) self.assertTrue(self.exists("refs/heads/topic", cwd=other)) def test_status_failure_cannot_look_like_a_clean_worktree(self): path = self.worktree() marker = self.root / "status failure injected" self.git_shim( f"if 'status' in args and os.getcwd() == {str(path)!r}:\n" f" pathlib.Path({str(marker)!r}).write_text('injected')\n" " print('fixture: status failed', file=sys.stderr)\n" " sys.exit(128)" ) self.cleanup("--local", "--apply", check=False) self.assertTrue(marker.exists(), "Fixture did not reach status validation") self.assertTrue(path.exists()) self.assertTrue(self.exists("refs/heads/topic")) def test_local_ref_changed_at_deletion_is_not_removed(self): self.git("branch", "topic") replacement = self.topic("new-work") marker = self.root / "local race injected" self.git_transaction_shim( f"if line.startswith(b'delete refs/heads/topic ') and not pathlib.Path({str(marker)!r}).exists():\n" f" pathlib.Path({str(marker)!r}).write_text('injected')\n" f" subprocess.run([real_git, '-C', {str(self.repo)!r}, 'update-ref', 'refs/heads/topic', {replacement!r}], check=True)" ) self.cleanup("--local", "--apply", check=False) self.assertTrue(marker.exists(), "Fixture did not reach guarded deletion") self.assertEqual(self.git("rev-parse", "topic"), replacement) def test_remote_ref_changed_at_push_is_not_removed(self): remote = self.remote() self.git("branch", "topic") self.git("push", "--quiet", "origin", "topic") replacement = self.topic("remote-new-work") self.git("push", "--quiet", "origin", "remote-new-work") self.git("update-ref", "-d", "refs/heads/remote-new-work", cwd=remote) marker = self.root / "remote race injected" self.git_shim( f"if 'push' in args and any(':refs/heads/topic' in arg for arg in args) and not pathlib.Path({str(marker)!r}).exists():\n" f" pathlib.Path({str(marker)!r}).write_text('injected')\n" f" subprocess.run([real_git, '-C', {str(remote)!r}, 'update-ref', 'refs/heads/topic', {replacement!r}], check=True)" ) self.cleanup("--apply", check=False) self.assertTrue(marker.exists(), "Fixture did not reach remote deletion") self.assertEqual(self.git("rev-parse", "refs/heads/topic", cwd=remote), replacement) def test_local_into_does_not_authorize_remote_deletion(self): remote = self.remote() self.topic("worker") self.git("push", "--quiet", "origin", "worker") self.git("switch", "--quiet", "-c", "parent") self.write("parent.txt", "parent work\n") self.commit() self.git("merge", "--squash", "worker") self.commit("Squash worker") self.git("switch", "--quiet", "main") self.cleanup("--into", "parent", "--apply") self.assertTrue(self.exists("refs/heads/worker", cwd=remote)) def test_explicit_remote_obsolete_is_recoverable(self): remote = self.remote() oid = self.topic("abandoned") self.git("push", "--quiet", "origin", "abandoned") self.cleanup("--obsolete", "origin/abandoned", "--apply") self.assertFalse(self.exists("refs/heads/abandoned", cwd=remote)) self.assertTrue(self.exists("refs/heads/abandoned")) self.assertIn(oid, self.backup_oids()) def test_explicit_base_and_recipient_refs_are_retained(self): self.topic("base") self.git("branch", "recipient", "base") self.git("branch", "candidate", "base") self.cleanup("--local", "--base", "base", "--into", "recipient", "--apply") self.assertTrue(self.exists("refs/heads/base")) self.assertTrue(self.exists("refs/heads/recipient")) self.assertFalse(self.exists("refs/heads/candidate")) def test_unknown_obsolete_selection_does_not_delete_other_candidates(self): self.git("branch", "candidate") before = self.git("show-ref") result = self.cleanup("--local", "--obsolete", "misspelled", "--apply", check=False) self.assertNotEqual(result.returncode, 0) self.assertEqual(self.git("show-ref"), before) def test_shallow_repository_is_refused(self): remote = self.remote() clone = self.root / "shallow" self.git("clone", "--quiet", "--depth=1", remote.as_uri(), str(clone)) self.git("branch", "candidate", cwd=clone) before = self.git("show-ref", cwd=clone) result = self.cleanup("--local", "--apply", cwd=clone, check=False) self.assertNotEqual(result.returncode, 0) self.assertEqual(self.git("show-ref", cwd=clone), before) def test_replacement_objects_cannot_supply_false_content_proof(self): topic = self.topic() self.write("main-only.txt", "main work\n") main = self.commit() self.git("replace", main, topic) self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic")) def test_legacy_grafts_are_refused(self): self.git("branch", "candidate") before = self.git("show-ref") grafts = Path(self.git("rev-parse", "--path-format=absolute", "--git-path", "info/grafts")) grafts.write_text(self.git("rev-parse", "HEAD") + "\n") result = self.cleanup("--local", "--apply", check=False) self.assertNotEqual(result.returncode, 0) self.assertEqual(self.git("show-ref"), before) def test_assume_unchanged_cannot_hide_uncommitted_work(self): path = self.worktree() self.git("update-index", "--assume-unchanged", "tracked.txt", cwd=path) self.write("tracked.txt", "hidden uncommitted data\n", cwd=path) self.assertEqual(self.git("status", "--porcelain", cwd=path), "") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic")) self.assertEqual((path / "tracked.txt").read_text(), "hidden uncommitted data\n") def test_skip_worktree_cannot_hide_uncommitted_work(self): path = self.worktree() self.git("update-index", "--skip-worktree", "tracked.txt", cwd=path) self.write("tracked.txt", "hidden uncommitted data\n", cwd=path) self.assertEqual(self.git("status", "--porcelain", cwd=path), "") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic")) self.assertEqual((path / "tracked.txt").read_text(), "hidden uncommitted data\n") def test_github_exact_merged_head_proves_historical_landing(self): self.remote() head = self.topic() self.git("push", "--quiet", "origin", "topic") self.git("merge", "--squash", "topic") landing = self.commit("Squash topic") self.write("feature.txt", "later main work\n") self.commit() self.git("push", "--quiet", "origin", "main") self.github_fixture([self.pr(head, landing=landing)]) self.env["GH_HOST"] = "different.example.invalid" self.env["GH_REPO"] = "unrelated/repository" plan = self.plan("--github") for ref in ("refs/heads/topic", "refs/remotes/origin/topic"): entry = self.entry(plan, ref) self.assertEqual(entry["action"], "delete") self.assertIn("PR #17", entry["reason"]) def test_github_closed_unmerged_pr_is_not_deletion_proof(self): self.remote() head = self.topic() self.git("push", "--quiet", "origin", "topic") self.github_fixture([self.pr(head, state="CLOSED")]) plan = self.plan("--github") for ref in ("refs/heads/topic", "refs/remotes/origin/topic"): self.assertEqual(self.entry(plan, ref)["action"], "keep") def test_github_open_pr_preserves_an_otherwise_merged_branch(self): self.remote() head = self.git("rev-parse", "main") self.git("branch", "topic") self.git("push", "--quiet", "origin", "topic") self.github_fixture([self.pr(head, state="OPEN")]) plan = self.plan("--github") for ref in ("refs/heads/topic", "refs/remotes/origin/topic"): self.assertEqual(self.entry(plan, ref)["action"], "keep") def test_github_merged_pr_does_not_cover_later_source_commits(self): self.remote() old_head = self.topic() self.git("merge", "--squash", "topic") landing = self.commit("Squash topic") self.git("push", "--quiet", "origin", "main") self.git("switch", "--quiet", "topic") self.write("later.txt", "work after merged PR\n") self.commit() self.git("push", "--quiet", "origin", "topic") self.git("switch", "--quiet", "main") self.github_fixture([self.pr(old_head, landing=landing)]) plan = self.plan("--github") for ref in ("refs/heads/topic", "refs/remotes/origin/topic"): self.assertEqual(self.entry(plan, ref)["action"], "keep") def test_github_cross_repository_pr_cannot_supply_proof(self): self.remote() head = self.topic() self.git("push", "--quiet", "origin", "topic") self.github_fixture([self.pr(head, landing=self.git("rev-parse", "main"), cross_repository=True)]) plan = self.plan("--github") for ref in ("refs/heads/topic", "refs/remotes/origin/topic"): self.assertEqual(self.entry(plan, ref)["action"], "keep") def test_github_query_failure_aborts_without_mutations(self): self.remote() self.git("branch", "topic") self.github_fixture([], failure=True) before = self.git("show-ref") result = self.cleanup("--github", check=False) self.assertNotEqual(result.returncode, 0) self.assertEqual(self.git("show-ref"), before) self.assertEqual(self.backup_oids(), []) def test_annotated_tag_base_uses_raw_ref_value_for_transaction(self): self.git("branch", "candidate") self.git("tag", "--annotate", "published", "-m", "Published baseline") tag_oid = self.git("rev-parse", "refs/tags/published") self.assertNotEqual(tag_oid, self.git("rev-parse", "published^{commit}")) self.cleanup("--local", "--base", "published", "--apply") self.assertFalse(self.exists("refs/heads/candidate")) self.assertEqual(self.git("rev-parse", "refs/tags/published"), tag_oid) def test_filemode_config_cannot_hide_executable_bit_change(self): path = self.worktree() self.git("config", "core.filemode", "false", cwd=path) file = path / "tracked.txt" file.chmod(0o755) self.assertEqual(self.git("status", "--porcelain", cwd=path), "") self.cleanup("--local", "--apply") self.assertTrue(self.exists("refs/heads/topic")) self.assertTrue(file.stat().st_mode & 0o111) def test_repository_path_trailing_whitespace_cannot_select_a_sibling(self): self.git("branch", "original-must-survive") original = self.git("show-ref") for suffix in (" ", "\n"): with self.subTest(suffix=repr(suffix)): clone = self.root / (self.repo.name + suffix) self.git("clone", "--quiet", str(self.repo), str(clone)) self.git("branch", "candidate", cwd=clone) self.cleanup("--local", "--apply", cwd=clone) self.assertEqual(self.git("show-ref"), original, "Cleanup changed the sibling repository") self.assertFalse(self.exists("refs/heads/candidate", cwd=clone)) def test_protection_patterns_cover_local_and_remote_branch_names(self): remote = self.remote() for name in ("release/by-flag", "stable/by-config", "candidate"): self.git("branch", name) self.git("push", "--quiet", "origin", name) self.git("config", "--add", "cleanup.protect", "stable/*") self.cleanup("--protect", "release/*", "--apply") for name in ("release/by-flag", "stable/by-config"): with self.subTest(name=name): self.assertTrue(self.exists(f"refs/heads/{name}")) self.assertTrue(self.exists(f"refs/heads/{name}", cwd=remote)) self.assertFalse(self.exists("refs/heads/candidate")) self.assertFalse(self.exists("refs/heads/candidate", cwd=remote)) def test_git_command_config_overrides_preserve_protected_branches(self): remote = self.remote() for name in ("topic", "candidate"): self.git("branch", name) self.git("push", "--quiet", "origin", name) # Git transports `git -c cleanup.protect=topic cleanup` to external # subcommands using these environment variables. self.env.update({ "GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "cleanup.protect", "GIT_CONFIG_VALUE_0": "topic", }) self.cleanup("--apply") self.assertTrue(self.exists("refs/heads/topic")) self.assertTrue(self.exists("refs/heads/topic", cwd=remote)) self.assertFalse(self.exists("refs/heads/candidate")) self.assertFalse(self.exists("refs/heads/candidate", cwd=remote)) def test_cleanup_push_does_not_publish_unrelated_local_tags(self): remote = self.remote() self.git("branch", "topic") self.git("push", "--quiet", "origin", "topic") self.git("tag", "--annotate", "local-only", "-m", "Keep unpublished") self.git("config", "push.followTags", "true") self.assertFalse(self.exists("refs/tags/local-only", cwd=remote)) self.cleanup("--apply") self.assertFalse(self.exists("refs/heads/topic", cwd=remote)) self.assertFalse(self.exists("refs/tags/local-only", cwd=remote)) self.assertTrue(self.exists("refs/tags/local-only")) def test_cleanup_fetch_does_not_prune_unpublished_local_tags(self): self.remote() self.git("tag", "--annotate", "local-only", "-m", "Keep unpublished") self.git("config", "fetch.pruneTags", "true") oid = self.git("rev-parse", "refs/tags/local-only") self.cleanup("--fetch") self.assertEqual(self.git("rev-parse", "refs/tags/local-only"), oid) def assert_chained_url_destination(self, rewrite_kind): original = self.remote() self.git("branch", "topic") self.git("push", "--quiet", "origin", "topic") expected = self.root / "expected=endpoint.git" unintended = self.root / "unintended=endpoint.git" self.git("clone", "--quiet", "--bare", str(original), str(expected)) self.git("clone", "--quiet", "--bare", str(original), str(unintended)) self.git("config", f"url.{expected}.insteadOf", str(original)) self.git("config", f"url.{unintended}.{rewrite_kind}", str(expected)) self.assertEqual(self.git("remote", "get-url", "origin"), str(expected)) self.assertEqual(self.git("remote", "get-url", "--push", "origin"), str(expected)) self.cleanup("--apply") self.assertFalse(self.exists("refs/heads/topic", cwd=expected)) self.assertTrue(self.exists("refs/heads/topic", cwd=unintended)) self.assertTrue(self.exists("refs/heads/topic", cwd=original)) def test_chained_url_rewrites_cannot_redirect_remote_deletion(self): self.assert_chained_url_destination("pushInsteadOf") def test_chained_fetch_rewrites_cannot_change_the_inspected_endpoint(self): self.assert_chained_url_destination("insteadOf") def test_existing_symbolic_recipient_alias_preserves_its_target(self): topic = self.topic() self.git("branch", "recipient", "topic") self.git("symbolic-ref", "refs/heads/alias", "refs/heads/recipient") self.cleanup("--local", "--base", "alias", "--apply") self.assertFalse(self.exists("refs/heads/topic")) self.assertEqual(self.git("rev-parse", "recipient"), topic) self.assertEqual(self.git("rev-parse", "alias"), topic) self.assertEqual(self.git("symbolic-ref", "refs/heads/alias"), "refs/heads/recipient") def test_recipient_snapshot_cannot_mix_proof_and_verified_values(self): main = self.git("rev-parse", "main") topic = self.topic() self.git("branch", "recipient", "topic") marker = self.root / "recipient snapshot race injected" self.git_shim( f"if '--symbolic-full-name' in args and args[-1] == 'recipient' and not pathlib.Path({str(marker)!r}).exists():\n" f" pathlib.Path({str(marker)!r}).write_text('injected')\n" f" subprocess.run([real_git, '-C', {str(self.repo)!r}, 'update-ref', 'refs/heads/recipient', {main!r}], check=True)" ) self.cleanup("--local", "--base", "recipient", "--apply", check=False) self.assertTrue(marker.exists(), "Fixture did not reach recipient capture") self.assertEqual(self.git("rev-parse", "recipient"), main) self.assertEqual(self.git("rev-parse", "topic"), topic) def test_recipient_changed_to_symbolic_ref_cannot_become_dangling(self): topic = self.topic() self.git("branch", "recipient", "topic") marker = self.root / "symbolic recipient race injected" self.git_transaction_shim( f"if line.startswith(b'delete refs/heads/topic ') and not pathlib.Path({str(marker)!r}).exists():\n" f" pathlib.Path({str(marker)!r}).write_text('injected')\n" f" subprocess.run([real_git, '-C', {str(self.repo)!r}, 'symbolic-ref', 'refs/heads/recipient', 'refs/heads/topic'], check=True)" ) result = self.cleanup("--local", "--base", "recipient", "--apply", check=False) self.assertTrue(marker.exists(), "Fixture did not reach guarded deletion") self.assertNotEqual(result.returncode, 0) self.assertEqual(self.git("rev-parse", "topic"), topic) self.assertEqual(self.git("rev-parse", "recipient"), topic) self.assertIn(topic, self.backup_oids()) def test_candidate_changed_to_symbolic_ref_cannot_delete_protected_main(self): main = self.git("rev-parse", "main") self.git("branch", "topic") self.git("branch", "recipient") marker = self.root / "symbolic candidate race injected" self.git_transaction_shim( f"if line.startswith(b'delete refs/heads/topic ') and not pathlib.Path({str(marker)!r}).exists():\n" f" pathlib.Path({str(marker)!r}).write_text('injected')\n" f" subprocess.run([real_git, '-C', {str(self.repo)!r}, 'symbolic-ref', 'refs/heads/topic', 'refs/heads/main'], check=True)" ) result = self.cleanup("--local", "--base", "recipient", "--apply", check=False) self.assertTrue(marker.exists(), "Fixture did not reach guarded deletion") self.assertNotEqual(result.returncode, 0) self.assertEqual(self.git("rev-parse", "main"), main) self.assertEqual(self.git("rev-parse", "topic"), main) self.assertEqual(self.git("symbolic-ref", "refs/heads/topic"), "refs/heads/main") self.assertIn(main, self.backup_oids()) def test_partial_clone_local_preview_never_fetches_missing_objects(self): remote = self.remote() self.topic() self.git("push", "--quiet", "origin", "topic") self.write("main-only.txt", "main work\n") self.commit() self.git("push", "--quiet", "origin", "main") self.git("config", "uploadpack.allowFilter", "true", cwd=remote) clone = self.root / "partial" self.git("clone", "--quiet", "--filter=tree:0", "--no-checkout", remote.as_uri(), str(clone)) self.git("branch", "topic", "origin/topic", cwd=clone) packs = clone / ".git/objects/pack" before_packs = {path.name: path.read_bytes() for path in packs.iterdir()} before_refs = self.git("show-ref", cwd=clone) trace = self.root / "partial clone trace" self.env["GIT_TRACE"] = str(trace) result = self.cleanup("--local", "--json", cwd=clone, check=False) self.assertNotEqual(result.returncode, 0) self.assertIn("partial", result.stderr.lower()) self.assertEqual({path.name: path.read_bytes() for path in packs.iterdir()}, before_packs) self.assertEqual(self.git("show-ref", cwd=clone), before_refs) self.assertTrue(trace.exists(), "Fixture did not trace the cleanup process") self.assertNotIn("built-in: git fetch", trace.read_text()) def test_empty_explicit_selectors_fail_before_fetch_or_delete(self): remote = self.remote() self.git("branch", "candidate") self.git("push", "--quiet", "origin", "candidate") before_local = self.git("show-ref") before_remote = self.git("show-ref", cwd=remote) marker = self.root / "unexpected fetch" self.git_shim( f"if 'fetch' in args:\n" f" pathlib.Path({str(marker)!r}).write_text('fetch happened')" ) for selector in ("--base", "--remote", "--into", "--obsolete"): with self.subTest(selector=selector): result = self.cleanup(selector, "", "--apply", check=False) self.assertNotEqual(result.returncode, 0) self.assertFalse(marker.exists(), "Invalid selection fetched before validation") self.assertEqual(self.git("show-ref"), before_local) self.assertEqual(self.git("show-ref", cwd=remote), before_remote) def test_empty_configuration_selectors_fail_before_fetch_or_delete(self): remote = self.remote() self.git("branch", "candidate") self.git("push", "--quiet", "origin", "candidate") before_local = self.git("show-ref") before_remote = self.git("show-ref", cwd=remote) marker = self.root / "unexpected configured fetch" self.git_shim( f"if 'fetch' in args:\n" f" pathlib.Path({str(marker)!r}).write_text('fetch happened')" ) for key in ("cleanup.base", "cleanup.mainBranch"): with self.subTest(key=key): self.git("config", key, "") try: result = self.cleanup("--apply", check=False) self.assertNotEqual(result.returncode, 0) self.assertFalse(marker.exists(), "Invalid configuration fetched before validation") self.assertEqual(self.git("show-ref"), before_local) self.assertEqual(self.git("show-ref", cwd=remote), before_remote) finally: self.git("config", "--unset", key) if __name__ == "__main__": unittest.main()