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
|
import os
import subprocess
import unittest
from pathlib import Path
from test_cli import TOML, Cli
from test_claims import git
from test_merge import IDENT
class SplitCli(Cli):
"""Private project (self.root, git, workflow.toml with code_root) + sibling code repo (self.code)."""
toml = TOML + 'code_root = "../code"\n'
def setUp(self):
super().setUp()
git(self.root, "init", "-q", "-b", "master")
(self.root / ".gitignore").write_text(".wf/\nout/\n")
git(self.root, "add", "-A")
git(self.root, "commit", "-qm", "init")
self.code = self.root.parent / "code"
self.code.mkdir()
git(self.code, "init", "-q", "-b", "master")
(self.code / ".gitignore").write_text(".worktrees/\n.wf-home\n")
(self.code / "lib.txt").write_text("a\n")
git(self.code, "add", "-A")
git(self.code, "commit", "-qm", "init")
def add_wt(self, lane="sonnet", id="t-three"):
"""Code worktree on branch <lane>/<id> with .wf-home (what wf start makes)."""
wt = self.code / ".worktrees" / lane
git(self.code, "worktree", "add", "-q", str(wt), "-b", f"{lane}/{id}")
(wt / ".wf-home").write_text(f"{self.root}\n")
return wt
def log(self, repo, rev="master"):
return subprocess.run(["git", "log", "--format=%s", rev], cwd=repo, capture_output=True, text=True).stdout
def sha(self, repo, rev="master"):
return subprocess.run(["git", "rev-parse", "--short", rev], cwd=repo, capture_output=True, text=True).stdout.strip()
class DiscoveryTest(SplitCli):
def test_wf_in_code_worktree_finds_private_project(self):
wt = self.add_wt()
code, out, err = self.wf("show", "t-three", project=False, cwd=wt)
self.assertEqual(code, 0, err)
self.assertIn("**t-three**", out)
def test_dangling_wf_home_is_config_error(self):
wt = self.add_wt()
(wt / ".wf-home").write_text(f"{self.root.parent / 'gone'}\n")
code, out, err = self.wf("show", "t-three", project=False, cwd=wt)
self.assertNotEqual(code, 0)
self.assertIn(".wf-home points to", err)
self.assertNotIn("Traceback", err)
def test_split_and_code_main(self):
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from wflib import config
cfg = config.load(self.root)
self.assertTrue(cfg.split)
self.assertEqual(config.code_main(cfg), self.code)
self.assertEqual(cfg.push, [])
def test_code_root_inside_same_repo_not_split(self):
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from wflib import config
(self.root / "workflow.toml").write_text(TOML + 'code_root = "."\npush = ["true"]\n')
cfg = config.load(self.root)
self.assertFalse(cfg.split)
self.assertIsNone(config.code_main(cfg))
self.assertEqual(cfg.push, ["true"])
class StartSplitTest(SplitCli):
toml = SplitCli.toml + 'worktree_setup = ["pwd > where.txt"]\n'
def test_start_makes_code_worktree_with_wf_home(self):
wt = self.code / ".worktrees" / "slow"
code, out, err = self.wf("start", "t-three", "--worktree", str(wt), "--branch", "slow/t-three",
project=False, env=IDENT)
self.assertEqual(code, 0, out + err)
self.assertEqual((wt / ".wf-home").read_text(), f"{self.root}\n")
self.assertEqual(subprocess.run(["git", "branch", "--show-current"], cwd=wt, capture_output=True,
text=True).stdout.strip(), "slow/t-three")
self.assertEqual((wt / "where.txt").read_text().strip(), str(wt)) # setup cwd = code worktree
self.assertIn("(in progress: slow/t-three)", (self.root / "TASKS.md").read_text())
self.assertIn(f"cd {wt} && (make test) && ", out)
self.assertFalse((self.root / ".worktrees").exists()) # nothing in the private repo
def test_setup_in_code_worktree(self):
wt = self.add_wt()
code, out, err = self.wf("setup", project=False, cwd=wt)
self.assertEqual(code, 0, err)
self.assertEqual((wt / "where.txt").read_text().strip(), str(wt))
def env(self, name, pid): # copy of tests/test_claims.py ClaimTest.env
sock = self.root / f"{name}.sock"
sock.write_text("")
return {"CLAUDE_CODE_MESSAGING_SOCKET": str(sock), "CLAUDE_PID": str(pid), "CLAUDE_CODE_SESSION_ID": name}
def test_next_advice_names_wf_start_in_code_repo(self):
# pattern: tests/test_claims.py::test_next_says_worktree_mode_when_two_sessions_live
son, me = self.env("s", os.getppid()), self.env("o", os.getpid())
self.wf("next", "--lane", "slow", "--as", "sonnet", "--brief", env=son)
code, out, err = self.wf("next", "--lane", "fast", "--as", "opus", env=me)
self.assertIn(f" wf start <task> --worktree {self.code / '.worktrees' / 'fast'} --branch fast/<task> "
f"(run in {self.root}; wf there writes this TASKS.md)\n", out)
|
