import fcntl import subprocess import sys import time import unittest from test_cli import Cli, WF from test_claims import git IDENT = {"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t", "CLAUDE_CODE_MESSAGING_SOCKET": ""} class MergeTest(Cli): def setUp(self): super().setUp() git(self.root, "init", "-q", "-b", "master") (self.root / ".gitignore").write_text(".worktrees/\n.wf/\n") git(self.root, "add", "-A") git(self.root, "commit", "-qm", "init") self.wt = self.root / ".worktrees" / "sonnet" git(self.root, "worktree", "add", "-q", str(self.wt), "-b", "sonnet/t-three") def merge(self, *args): return self.wf("merge", *args, project=False, cwd=self.wt, env=IDENT) def out(self, *args, cwd=None): return subprocess.run(["git", *args], cwd=cwd or self.root, capture_output=True, text=True).stdout def commit_code(self): (self.wt / "code.txt").write_text("x\n") git(self.wt, "add", "code.txt") git(self.wt, "commit", "-qm", "code") def test_merge_ff_commits_bookkeeping_and_detaches(self): self.commit_code() self.wf("done", "t-three", "-m", "ok", project=False, cwd=self.wt, env=IDENT) code, out, err = self.merge("--no-push") self.assertEqual((code, err), (0, ""), out) self.assertEqual(out, "rebased onto master\nfast-forwarded master\n" "committed TASKS.md tasks/archive.md\nmerged sonnet/t-three into master\n") self.assertEqual(self.out("log", "--format=%s", "master"), "t-three done\ncode\ninit\n") self.assertEqual(self.out("branch", "--list", "sonnet/t-three"), "") self.assertEqual(self.out("status", "--porcelain"), "") self.assertEqual((self.root / "code.txt").read_text(), "x\n") def test_merge_commits_notes_and_followups_after_done(self): self.commit_code() self.wf("done", "t-three", "-m", "ok", project=False, cwd=self.wt, env=IDENT) r = self.wf("add", "-p", "2", "-e", "1h", "follow up", project=False, cwd=self.wt, env=IDENT); self.assertEqual(r[0], 0, r) code, out, err = self.merge("--no-push") self.assertEqual((code, err), (0, ""), out) self.assertEqual(self.out("status", "--porcelain"), "") self.assertIn("follow up", self.out("show", "master:TASKS.md")) def test_merge_rebases_onto_moved_master(self): self.commit_code() (self.root / "other.txt").write_text("o\n") git(self.root, "add", "other.txt") git(self.root, "commit", "-qm", "other") code, out, err = self.merge("--no-push") self.assertEqual((code, err), (0, ""), out) self.assertEqual(self.out("log", "--format=%s", "master"), "code\nother\ninit\n") def test_merge_conflict_aborts_and_keeps_branch(self): (self.root / "DESIGN.md").write_text("main side\n") git(self.root, "commit", "-qam", "main edit") (self.wt / "DESIGN.md").write_text("branch side\n") git(self.wt, "commit", "-qam", "branch edit") code, out, err = self.merge("--no-push") self.assertEqual((code, err), (1, "wf: rebase onto master conflicts: git rebase master, resolve, verify, " "then wf merge again\n")) self.assertEqual((self.wt / "DESIGN.md").read_text(), "branch side\n") self.assertEqual(self.out("status", "--porcelain", cwd=self.wt), "") self.assertEqual(self.out("branch", "--show-current", cwd=self.wt), "sonnet/t-three\n") def test_merge_refuses_dirty_worktree(self): (self.wt / "DESIGN.md").write_text("dirty\n") code, out, err = self.merge("--no-push") self.assertEqual((code, err), (1, "wf: worktree has uncommitted changes: commit them first\n")) def test_merge_outside_worktree_fails(self): self.assertEqual(self.fails("merge"), "wf: merge runs inside a linked git worktree (lane worktree)\n") def test_merge_detached_fails(self): git(self.wt, "switch", "-q", "--detach", "master") code, out, err = self.merge("--no-push") self.assertEqual((code, err), (1, "wf: worktree is on a detached HEAD: nothing to merge\n")) def test_merge_detached_commits_leftover_bookkeeping(self): self.commit_code() self.wf("done", "t-three", "-m", "ok", project=False, cwd=self.wt, env=IDENT) self.merge("--no-push") self.wf("add", "-p", "2", "-e", "1h", "late follow up", project=False, cwd=self.wt, env=IDENT) code, out, err = self.merge("--no-push") self.assertEqual((code, err, out), (0, "", "committed TASKS.md tasks/archive.md\n")) self.assertEqual(self.out("status", "--porcelain"), "") self.assertIn("late follow up", self.out("show", "master:TASKS.md")) def test_merge_detached_with_impl_commit_merges_it(self): git(self.wt, "switch", "-q", "--detach", "master") (self.root / "other.txt").write_text("o\n") git(self.root, "add", "other.txt") git(self.root, "commit", "-qm", "other") self.commit_code() r = self.wf("done", "t-three", "-m", "ok", project=False, cwd=self.wt, env=IDENT) self.assertIn("detached HEAD has 1 commit not in master: wf merge merges it", r[1]) code, out, err = self.merge("--no-push") self.assertEqual((code, err), (0, ""), out) self.assertEqual(out, "rebased onto master\nfast-forwarded master\n" "committed TASKS.md tasks/archive.md\nmerged detached HEAD into master\n") self.assertEqual(self.out("log", "--format=%s", "master"), "bookkeeping\ncode\nother\ninit\n") self.assertEqual((self.root / "code.txt").read_text(), "x\n") self.assertEqual(self.out("rev-parse", "HEAD", cwd=self.wt), self.out("rev-parse", "master")) self.assertEqual(self.out("status", "--porcelain"), "") def test_merge_detached_refuses_dirty_worktree(self): git(self.wt, "switch", "-q", "--detach", "master") (self.wt / "code.txt").write_text("x\n") code, out, err = self.merge("--no-push") self.assertEqual((code, err), (1, "wf: worktree has uncommitted changes: commit them first\n")) def test_merge_detached_conflict_aborts_and_keeps_commit(self): git(self.wt, "switch", "-q", "--detach", "master") (self.root / "DESIGN.md").write_text("main side\n") git(self.root, "commit", "-qam", "main edit") (self.wt / "DESIGN.md").write_text("branch side\n") git(self.wt, "commit", "-qam", "branch edit") code, out, err = self.merge("--no-push") self.assertEqual((code, err), (1, "wf: rebase onto master conflicts: git rebase master, resolve, verify, " "then wf merge again\n")) self.assertEqual(self.out("log", "-1", "--format=%s", cwd=self.wt), "branch edit\n") def test_merge_pushes_to_home(self): bare = self.root.parent / "home.git" subprocess.run(["git", "init", "-q", "--bare", str(bare)], check=True) git(self.root, "remote", "add", "home", str(bare)) self.commit_code() code, out, err = self.merge() self.assertEqual((code, err), (0, ""), out) self.assertIn("pushed home\n", out) self.assertEqual(self.out("log", "--format=%s", "master", cwd=bare), "code\ninit\n") def test_merge_waits_for_lock(self): self.commit_code() (self.root / ".wf").mkdir(exist_ok=True) with open(self.root / ".wf" / "lock", "w") as held: fcntl.flock(held, fcntl.LOCK_EX) import os p = subprocess.Popen([sys.executable, str(WF), "merge", "--no-push"], cwd=self.wt, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env={**os.environ, **IDENT}) time.sleep(0.5) self.assertIsNone(p.poll(), "merge ran while the lock was held") out, err = p.communicate(timeout=20) self.assertEqual((p.returncode, err), (0, "")) if __name__ == "__main__": unittest.main()