workflow: initial public history

1 files changed, 159 insertions, 0 deletions

diff --git a/tests/test_merge.py b/tests/test_merge.py
new file mode 100644
index 0000000..cb14a20
--- /dev/null
+++ b/tests/test_merge.py
@@ -0,0 +1,159 @@
+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()