aboutsummaryrefslogtreecommitdiffziptar.gz
path: root/tests/test_merge.py
blob: cb14a20988503fe0c3bb451f66e4cb9ea6bdddcb (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
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()