"""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()