govoplan/tests/test_devkit_maintenance.py

"""Git maintenance runs only in disposable local repositories/bare fixtures."""

from __future__ import annotations

import argparse
from contextlib import contextmanager
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import time
import unittest
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/devkit"))
from govoplan_devkit import maintenance


class MaintenanceTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="govoplan-devkit-git-")
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.repo = self.root / "repo"
        self.repo.mkdir()
        self.git("init", "-b", "main")
        self.git("config", "user.name", "Fixture")
        self.git("config", "user.email", "fixture@example.invalid")
        self.git("config", "commit.gpgsign", "false")
        self.write("selected.txt", "base selected\n")
        self.write("unrelated.txt", "base unrelated\n")
        self.git("add", "--", "selected.txt", "unrelated.txt")
        self.git("commit", "-m", "fixture base")
        self.base = self.git("rev-parse", "HEAD").strip()
        self.project = self.root / "project.json"
        self.project.write_text(
            json.dumps(
                {
                    "schema_version": 1,
                    "name": "fixture",
                    "repositories": [{"name": "repo", "path": "repo"}],
                }
            )
        )
        self.args = argparse.Namespace(
            workspace_root=self.root,
            state_dir=self.root / "state",
            project=self.project,
            repo="repo",
            path=["selected.txt"],
            message="selected change",
            apply=False,
        )

    def git(self, *args):
        return subprocess.run(
            ["git", "-C", str(self.repo), *args],
            check=True,
            capture_output=True,
            text=True,
        ).stdout

    def write(self, name, value):
        (self.repo / name).write_text(value)

    def save_plan(self):
        self.args.apply = True
        result = maintenance.plan(self.args)
        self.args.plan_id = result["plan"]["plan_id"]
        return result

    def test_preview_does_not_write_a_plan_or_stage(self):
        self.write("selected.txt", "changed\n")
        before = self.git("ls-files", "--stage")
        result = maintenance.plan(self.args)
        self.assertEqual(result["state"]["status"], "preview")
        self.assertFalse((self.root / "state").exists())
        self.assertEqual(self.git("ls-files", "--stage"), before)
        self.assertEqual(self.git("rev-parse", "HEAD").strip(), self.base)

    def test_selected_commit_preserves_unrelated_staged_and_dirty_changes(self):
        self.write("selected.txt", "selected final\n")
        self.write("unrelated.txt", "unrelated staged\n")
        self.git("add", "--", "unrelated.txt")
        self.write("unrelated.txt", "unrelated unstaged\n")
        staged = self.git("show", ":unrelated.txt")
        self.save_plan()
        before_plan = (
            maintenance._receipt_path(self.args, self.args.plan_id)
            .joinpath("plan.json")
            .read_bytes()
        )
        self.args.apply = False
        self.assertEqual(maintenance.commit(self.args)["state"]["status"], "preview")
        self.args.apply = True
        result = maintenance.commit(self.args)
        self.assertEqual(result["state"]["status"], "committed")
        self.assertEqual(
            self.git(
                "diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD"
            ).strip(),
            "selected.txt",
        )
        self.assertEqual(self.git("show", ":unrelated.txt"), staged)
        self.assertEqual(
            (self.repo / "unrelated.txt").read_text(), "unrelated unstaged\n"
        )
        self.assertEqual(
            maintenance.commit(self.args)["state"]["commit"], result["state"]["commit"]
        )
        self.assertEqual(
            maintenance._receipt_path(self.args, self.args.plan_id)
            .joinpath("plan.json")
            .read_bytes(),
            before_plan,
        )

    def test_new_selected_file_and_deletion_are_supported(self):
        self.write("new file.txt", "new\n")
        (self.repo / "selected.txt").unlink()
        self.git("add", "--", "selected.txt")
        self.args.path = ["new file.txt", "selected.txt"]
        self.save_plan()
        result = maintenance.commit(self.args)
        self.assertEqual(result["state"]["status"], "committed")
        self.assertEqual(
            set(
                self.git(
                    "diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD"
                ).splitlines()
            ),
            {"new file.txt", "selected.txt"},
        )

    def test_selected_partial_staging_is_not_overwritten(self):
        self.write("selected.txt", "staged\n")
        self.git("add", "--", "selected.txt")
        self.write("selected.txt", "unstaged\n")
        with self.assertRaisesRegex(ValueError, "different staged"):
            self.save_plan()
        self.assertEqual(self.git("show", ":selected.txt"), "staged\n")

    def test_stale_content_index_and_origin_block_commit(self):
        for change in ("content", "index", "origin"):
            with self.subTest(change=change):
                self.write("selected.txt", f"selected {change}\n")
                self.save_plan()
                if change == "content":
                    self.write("selected.txt", "different content\n")
                elif change == "index":
                    self.write("unrelated.txt", "changed index\n")
                    self.git("add", "--", "unrelated.txt")
                else:
                    self.git("remote", "add", "origin", str(self.root / "other.git"))
                with self.assertRaisesRegex(ValueError, "changed; prepare a new plan"):
                    maintenance.commit(self.args)
                self.assertEqual(self.git("rev-parse", "HEAD").strip(), self.base)

    def test_active_hooks_and_filters_are_refused_not_bypassed(self):
        self.write("selected.txt", "changed\n")
        hook = self.repo / ".git/hooks/pre-commit"
        hook.write_text("#!/bin/sh\nexit 99\n")
        hook.chmod(0o700)
        with self.assertRaisesRegex(ValueError, "Active Git hooks"):
            self.save_plan()
        hook.unlink()
        self.git("config", "filter.example.clean", "some-external-command")
        self.write(".gitattributes", "*.txt filter=example\n")
        with self.assertRaisesRegex(ValueError, "Active Git filter"):
            self.save_plan()

    def test_directory_traversal_and_symlinks_are_refused(self):
        self.write("selected.txt", "changed\n")
        (self.repo / "linked.txt").symlink_to(self.repo / "selected.txt")
        for name in (".", "../outside", ".git/config", "linked.txt"):
            with self.subTest(name=name):
                self.args.path = [name]
                with self.assertRaises(ValueError):
                    self.save_plan()

    def test_post_index_change_hook_is_refused_before_it_can_run(self):
        self.write("selected.txt", "changed\n")
        marker = self.root / "hook-executed"
        hook = self.repo / ".git/hooks/post-index-change"
        hook.write_text(f"#!/bin/sh\ntouch '{marker}'\n")
        hook.chmod(0o700)
        with self.assertRaisesRegex(ValueError, "Active Git hooks"):
            self.save_plan()
        self.assertFalse(marker.exists())

    def test_remote_helpers_and_recursive_operations_are_refused(self):
        self.write("selected.txt", "changed\n")
        for setting in (
            "remote.origin.receivepack",
            "remote.origin.uploadpack",
            "remote.origin.vcs",
            "core.gitProxy",
            "core.alternateRefsCommand",
            "push.recurseSubmodules",
            "submodule.recurse",
            "remote.origin.promisor",
        ):
            with self.subTest(setting=setting):
                self.git("config", setting, "true")
                with self.assertRaises(ValueError):
                    self.save_plan()
                self.git("config", "--unset", setting)
        self.assertEqual(self.git("rev-parse", "HEAD").strip(), self.base)

    def test_git_namespace_identity_and_repository_overrides_are_refused(self):
        self.write("selected.txt", "changed\n")
        for key in (
            "GIT_NAMESPACE",
            "GIT_COMMON_DIR",
            "GIT_QUARANTINE_PATH",
            "GIT_SHALLOW_FILE",
            "GIT_REPLACE_REF_BASE",
            "GIT_ALTERNATE_OBJECT_DIRECTORIES",
            "GIT_AUTHOR_NAME",
            "GIT_CONFIG_COUNT",
            "GIT_UNKNOWN_OVERRIDE",
        ):
            with self.subTest(key=key), patch.dict(os.environ, {key: "unexpected"}):
                with self.assertRaisesRegex(ValueError, "environment overrides"):
                    self.save_plan()
        self.assertFalse((self.root / "state").exists())

    def test_noncanonical_paths_are_rejected_before_commit(self):
        (self.repo / "dir").mkdir()
        self.write("dir/file.txt", "new\n")
        self.args.path = ["dir//file.txt"]
        with self.assertRaisesRegex(ValueError, "explicit relative"):
            self.save_plan()
        self.assertEqual(self.git("rev-parse", "HEAD").strip(), self.base)

    def test_replacement_history_is_not_treated_as_the_real_frozen_tree(self):
        self.write("selected.txt", "changed\n")
        tree = self.git("rev-parse", "HEAD^{tree}").strip()
        replacement = self.git("commit-tree", tree, "-m", "replacement fixture").strip()
        self.git("replace", self.base, replacement)
        with self.assertRaisesRegex(ValueError, "replacement/graft"):
            self.save_plan()

    def test_normal_push_is_explicit_and_verified_against_a_local_bare_fixture(self):
        remote = self.root / "origin.git"
        subprocess.run(
            ["git", "init", "--bare", str(remote)], check=True, capture_output=True
        )
        self.git("remote", "add", "origin", str(remote))
        self.write("selected.txt", "selected final\n")
        frozen = self.save_plan()
        self.assertNotIn(str(remote), json.dumps(frozen))
        result = maintenance.commit(self.args)
        self.args.apply = False
        self.assertEqual(maintenance.push(self.args)["state"]["status"], "preview")
        self.assertEqual(
            self.git("ls-remote", "--refs", "origin", "refs/heads/main"), ""
        )
        self.args.apply = True
        pushed = maintenance.push(self.args)
        self.assertEqual(pushed["state"]["status"], "pushed")
        self.assertTrue(
            self.git("ls-remote", "--refs", "origin", "refs/heads/main").startswith(
                result["state"]["commit"]
            )
        )

    def test_interrupted_commit_is_reconciled_without_creating_another_commit(self):
        self.write("selected.txt", "selected final\n")
        self.save_plan()
        original = maintenance._write_state

        def fail_receipt(directory, state, **updates):
            if updates.get("status") == "committed":
                raise OSError("fixture interruption after commit")
            return original(directory, state, **updates)

        with patch.object(maintenance, "_write_state", side_effect=fail_receipt):
            with self.assertRaises(OSError):
                maintenance.commit(self.args)
        head = self.git("rev-parse", "HEAD").strip()
        self.assertNotEqual(head, self.base)
        with self.assertRaisesRegex(ValueError, "reconcile"):
            maintenance.commit(self.args)
        result = maintenance.reconcile(self.args)
        self.assertEqual(result["state"]["status"], "committed")
        self.assertEqual(self.git("rev-parse", "HEAD").strip(), head)

    def test_lost_push_receipt_is_reconciled_without_repeating_push(self):
        remote = self.root / "origin.git"
        subprocess.run(
            ["git", "init", "--bare", str(remote)], check=True, capture_output=True
        )
        self.git("remote", "add", "origin", str(remote))
        self.write("selected.txt", "selected final\n")
        self.save_plan()
        maintenance.commit(self.args)
        original = maintenance._git
        pushes = []

        def lose_response(repo, *argv, **kwargs):
            result = original(repo, *argv, **kwargs)
            if argv[0] == "push":
                pushes.append(argv)
                raise ValueError("fixture response lost after remote update")
            return result

        with patch.object(maintenance, "_git", side_effect=lose_response):
            with self.assertRaises(ValueError):
                maintenance.push(self.args)
        with self.assertRaisesRegex(ValueError, "reconcile"):
            maintenance.push(self.args)
        reconciled = maintenance.reconcile(self.args)
        self.assertEqual(reconciled["state"]["status"], "pushed")
        self.assertEqual(len(pushes), 1)

    def test_reconciliation_rechecks_state_after_acquiring_lock(self):
        self.write("selected.txt", "selected final\n")
        self.save_plan()
        maintenance.commit(self.args)
        payload, state, directory = maintenance._read(self.args, self.args.plan_id)
        maintenance._write_state(
            directory, state, status="needs_reconcile", operation="commit"
        )

        @contextmanager
        def another_reconciliation(_args, _repo):
            maintenance._write_state(directory, state, status="committed")
            yield

        with patch.object(maintenance, "_locked", another_reconciliation):
            result = maintenance.reconcile(self.args)
        self.assertEqual(result["state"]["status"], "committed")
        self.assertTrue(
            any("no replay or downgrade" in line for line in result["summary"])
        )

    def test_worktree_edit_racing_with_commit_remains_uncommitted(self):
        self.write("selected.txt", "approved bytes\n")
        self.save_plan()
        original = maintenance._git

        def race(repo, *argv, **kwargs):
            if argv[0] == "commit-tree":
                self.write("selected.txt", "new editor bytes\n")
            return original(repo, *argv, **kwargs)

        with patch.object(maintenance, "_git", side_effect=race):
            result = maintenance.commit(self.args)
        self.assertEqual(result["state"]["status"], "committed")
        self.assertEqual(self.git("show", "HEAD:selected.txt"), "approved bytes\n")
        self.assertEqual(self.git("show", ":selected.txt"), "approved bytes\n")
        self.assertEqual((self.repo / "selected.txt").read_text(), "new editor bytes\n")

    def test_branch_compare_and_swap_never_overwrites_a_racing_commit(self):
        self.write("selected.txt", "approved bytes\n")
        self.save_plan()
        original = maintenance._git
        raced = []

        def race(repo, *argv, **kwargs):
            if argv[0] == "update-ref" and not raced:
                tree = self.git("rev-parse", "HEAD^{tree}").strip()
                other = self.git(
                    "commit-tree", tree, "-p", self.base, "-m", "other writer"
                ).strip()
                self.git("update-ref", "refs/heads/main", other, self.base)
                raced.append(other)
            return original(repo, *argv, **kwargs)

        with patch.object(maintenance, "_git", side_effect=race):
            with self.assertRaises(ValueError):
                maintenance.commit(self.args)
        self.assertEqual(self.git("rev-parse", "HEAD").strip(), raced[0])
        self.assertEqual(self.git("show", ":selected.txt"), "base selected\n")

    def test_interrupted_index_publication_reconciles_without_another_commit(self):
        self.write("selected.txt", "approved bytes\n")
        self.save_plan()
        with patch.object(
            maintenance, "_publish_index", side_effect=OSError("fixture interruption")
        ):
            with self.assertRaises(OSError):
                maintenance.commit(self.args)
        head = self.git("rev-parse", "HEAD").strip()
        self.assertNotEqual(head, self.base)
        self.assertEqual(self.git("show", ":selected.txt"), "base selected\n")
        result = maintenance.reconcile(self.args)
        self.assertEqual(result["state"]["status"], "committed")
        self.assertEqual(self.git("show", ":selected.txt"), "approved bytes\n")
        self.assertEqual(self.git("rev-parse", "HEAD").strip(), head)

    def test_concurrent_index_changes_are_preserved_instead_of_overwritten(self):
        self.write("selected.txt", "approved bytes\n")
        self.save_plan()
        original = maintenance._publish_index

        def race(repo, locked, prepared, expected):
            # Model an external writer that does not honor Git's index.lock.
            outside = self.root / "outside.index"
            outside.write_bytes((self.repo / ".git/index").read_bytes())
            self.write("unrelated.txt", "new independent staging\n")
            subprocess.run(
                ["git", "-C", str(self.repo), "add", "--", "unrelated.txt"],
                env={**os.environ, "GIT_INDEX_FILE": str(outside)},
                check=True,
                capture_output=True,
            )
            (self.repo / ".git/index").write_bytes(outside.read_bytes())
            return original(repo, locked, prepared, expected)

        with patch.object(maintenance, "_publish_index", side_effect=race):
            with self.assertRaisesRegex(ValueError, "index changed concurrently"):
                maintenance.commit(self.args)
        self.assertEqual(
            self.git("show", ":unrelated.txt"), "new independent staging\n"
        )
        with self.assertRaises(ValueError):
            maintenance.reconcile(self.args)
        self.assertEqual(
            self.git("show", ":unrelated.txt"), "new independent staging\n"
        )

    def test_selected_fifo_swap_cannot_block_capture(self):
        self.write("selected.txt", "approved bytes\n")
        self.save_plan()
        original = maintenance._capture_blob

        def race(repo, item):
            (self.repo / "selected.txt").unlink()
            os.mkfifo(self.repo / "selected.txt")
            return original(repo, item)

        started = time.monotonic()
        with patch.object(maintenance, "_capture_blob", side_effect=race):
            with self.assertRaises(ValueError):
                maintenance.commit(self.args)
        self.assertLess(time.monotonic() - started, 3)
        self.assertEqual(self.git("rev-parse", "HEAD").strip(), self.base)

    def test_reconcile_never_adopts_a_head_that_races_with_index_preparation(self):
        self.write("selected.txt", "approved bytes\n")
        self.save_plan()
        with patch.object(
            maintenance, "_publish_index", side_effect=OSError("fixture interruption")
        ):
            with self.assertRaises(OSError):
                maintenance.commit(self.args)
        candidate = self.git("rev-parse", "HEAD").strip()
        original = maintenance._selected_index
        raced = []

        def race(repo, source, target, files):
            result = original(repo, source, target, files)
            tree = self.git("rev-parse", f"{self.base}^{{tree}}").strip()
            other = self.git(
                "commit-tree", tree, "-p", candidate, "-m", "independent writer"
            ).strip()
            self.git("update-ref", "refs/heads/main", other, candidate)
            raced.append(other)
            return result

        with patch.object(maintenance, "_selected_index", side_effect=race):
            with self.assertRaisesRegex(
                ValueError, "changed during index reconciliation"
            ):
                maintenance.reconcile(self.args)
        recorded = maintenance.status(self.args)["state"]
        self.assertEqual(recorded["status"], "needs_reconcile")
        self.assertEqual(recorded["commit"], candidate)
        self.assertNotEqual(recorded["commit"], raced[0])
        self.assertEqual(self.git("show", ":selected.txt"), "base selected\n")

    def test_final_commit_receipt_cannot_bind_to_a_later_head(self):
        self.write("selected.txt", "approved bytes\n")
        self.save_plan()
        original = maintenance._publish_index
        raced = []

        def race(repo, locked, prepared, expected):
            original(repo, locked, prepared, expected)
            candidate = self.git("rev-parse", "HEAD").strip()
            tree = self.git("rev-parse", f"{self.base}^{{tree}}").strip()
            other = self.git(
                "commit-tree", tree, "-p", candidate, "-m", "independent writer"
            ).strip()
            self.git("update-ref", "refs/heads/main", other, candidate)
            raced.append(other)

        with patch.object(maintenance, "_publish_index", side_effect=race):
            with self.assertRaisesRegex(
                ValueError, "changed before the exact commit result"
            ):
                maintenance.commit(self.args)
        recorded = maintenance.status(self.args)["state"]
        self.assertEqual(recorded["status"], "needs_reconcile")
        self.assertNotEqual(recorded["commit"], raced[0])

    def test_git_output_is_bounded_during_execution(self):
        executable = self.root / "git"
        executable.write_text(f"#!{sys.executable}\nprint('x' * 100000)\n")
        executable.chmod(0o700)
        with (
            patch.dict(
                os.environ, {"PATH": str(self.root) + os.pathsep + os.environ["PATH"]}
            ),
            patch.object(maintenance, "MAX_GIT_OUTPUT_BYTES", 256),
        ):
            with self.assertRaisesRegex(ValueError, "output_limit"):
                maintenance._git(self.repo, "fixture")

    def test_timeout_terminates_git_helper_process_group(self):
        executable = self.root / "git"
        marker = self.root / "helper-ran"
        child = (
            "import signal,time,pathlib; signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(1); pathlib.Path("
            + repr(str(marker))
            + ").write_text('unexpected')"
        )
        executable.write_text(
            f"#!{sys.executable}\nimport subprocess,sys,time\nsubprocess.Popen([sys.executable, '-c', {child!r}])\ntime.sleep(20)\n"
        )
        executable.chmod(0o700)
        with patch.dict(
            os.environ, {"PATH": str(self.root) + os.pathsep + os.environ["PATH"]}
        ):
            with self.assertRaisesRegex(ValueError, "timed_out"):
                maintenance._git(self.repo, "fixture", timeout=0.2)
        time.sleep(1.1)
        self.assertFalse(
            marker.exists(),
            "An orphaned helper must not finish the operation after timeout",
        )


if __name__ == "__main__":
    unittest.main()