From 81d4e80fd5aabe4e80f58e960affa795cf7d34ec Mon Sep 17 00:00:00 2001 From: godosa Date: Wed, 7 Oct 2026 07:27:17 +0200 Subject: workflow: initial public history --- tests/test_claims.py | 182 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 182 insertions(+) create mode 100644 tests/test_claims.py (limited to 'tests/test_claims.py') diff --git a/tests/test_claims.py b/tests/test_claims.py new file mode 100644 index 0000000..e6b5c5d --- /dev/null +++ b/tests/test_claims.py @@ -0,0 +1,182 @@ +import json +import os +import subprocess +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from wflib import tasks as T +from wflib import lanes as L +from test_cli import Cli +from test_model import LANES + +FOUR = LANES.replace("## Needs human", "- **t-four** [P3] (1h): Four.\n\n## Needs human") + + +class HeldPickTest(unittest.TestCase): + def test_pick_skips_held(self): + item, skipped = L.pick(T.parse(FOUR), set(), L.DEFAULT_LANES, "1h", None, "opus", + held={"t-one": "opus session uds:/a"}) + self.assertEqual((item.id, [(i.id, why) for i, why in skipped]), + ("t-two", [("t-one", "in progress by opus session uds:/a")])) + + +class ClaimCliTest(Cli): + tasks_text = FOUR + + def env(self, name, pid): + 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 dead_pid(self): + p = subprocess.Popen(["true"]) + p.wait() + return p.pid + + def claim(self, id): + return json.loads((self.root / ".wf" / "claims" / f"{id}.json").read_text()) + + def test_progress_claims_and_other_session_skips(self): + a, b = self.env("a", os.getpid()), self.env("b", os.getppid()) + self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a) + self.ok("status", "t-three", "progress", "x", env=a) + c = self.claim("t-three") + self.assertEqual({k: c[k] for k in ("pid", "socket", "lane", "model")}, + {"pid": os.getpid(), "socket": str(self.root / "a.sock"), "lane": "fast", "model": "opus"}) + self.assertEqual(self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a).splitlines()[0], + "- **t-three** [P2] (<1h) (in progress: x): Three.") + code, out, err = self.wf("next", "--lane", "fast", "--as", "opus", env=b) + self.assertEqual((code, err), (1, "wf: nothing pickable for fast (opus) in Pending\n")) + self.assertIn(f"===== Skipped =====\n- t-three: in progress by opus session uds:{self.root / 'a.sock'}\n", out) + out = self.ok("next", "--lane", "slow", "--as", "opus", env=b) + self.assertIn("===== Next task =====\n- **t-one**", out) + + def test_dead_claim_ignored(self): + self.ok("status", "t-three", "progress", "x", env=self.env("a", self.dead_pid())) + self.assertEqual(self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=self.env("b", os.getpid())).splitlines()[0], + "- **t-three** [P2] (<1h) (in progress: x): Three.") + + def test_claim_without_in_progress_status_ignored(self): + self.ok("status", "t-three", "progress", "x", env=self.env("a", os.getppid())) + self.ok("status", "t-three", "clear", env=self.env("b", os.getpid())) + self.assertFalse((self.root / ".wf" / "claims" / "t-three.json").exists()) + self.assertEqual(self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=self.env("b", os.getpid())).splitlines()[0], + "- **t-three** [P2] (<1h): Three.") + + def test_done_and_blocked_release(self): + a = self.env("a", os.getpid()) + self.ok("status", "t-three", "progress", "x", env=a) + self.ok("done", "t-three", "-m", "ok", env=a) + self.assertFalse((self.root / ".wf" / "claims" / "t-three.json").exists()) + self.ok("status", "t-four", "progress", "x", env=a) + self.ok("add", "-s", "awaiting", "Q?", env=a) + self.ok("status", "t-four", "blocked", "a-q", env=a) + self.assertFalse((self.root / ".wf" / "claims" / "t-four.json").exists()) + + def test_no_env_no_claim(self): + self.ok("status", "t-three", "progress", "x", env={"CLAUDE_PID": "", "CLAUDE_CODE_MESSAGING_SOCKET": ""}) + self.assertFalse((self.root / ".wf" / "claims").exists()) + self.assertFalse((self.root / ".wf" / "sessions").exists()) + + def test_same_lane_live_session_warns_and_keeps_registry(self): + a, b = self.env("a", os.getppid()), self.env("b", os.getpid()) + self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a) + out = self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=b) + self.assertEqual(out.splitlines()[0], + f"another live fast session holds this lane: uds:{self.root / 'a.sock'} " + "(claims keep tasks apart; tell the owner if unintended)") + reg = json.loads((self.root / ".wf" / "sessions" / "fast.json").read_text()) + self.assertEqual(reg["socket"], str(self.root / "a.sock")) + self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a) # own re-register: no warning + self.assertNotIn("another live", self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a)) + + def test_lanes_unregister_drops_own_record(self): + a, b = self.env("a", os.getpid()), self.env("b", os.getppid()) + self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=a) + self.ok("next", "--lane", "slow", "--as", "sonnet", "--brief", env=b) + out = self.ok("lanes", "--unregister", env=a) + self.assertFalse((self.root / ".wf" / "sessions" / "fast.json").exists()) + self.assertTrue((self.root / ".wf" / "sessions" / "slow.json").exists()) + self.assertNotIn(str(self.root / "a.sock"), out) + + +def git(cwd, *args): + subprocess.run(["git", "-C", str(cwd), *args], check=True, capture_output=True, + env={**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", "GIT_COMMITTER_NAME": "t", + "GIT_COMMITTER_EMAIL": "t@t"}) + + +class WorktreeTest(Cli): + tasks_text = FOUR + + def setUp(self): + super().setUp() + git(self.root, "init", "-q", "-b", "master") + (self.root / ".gitignore").write_text(".worktrees/\n") + git(self.root, "add", "-A") + git(self.root, "commit", "-qm", "init") + self.wt = self.root / ".worktrees" / "fast" + git(self.root, "worktree", "add", "-q", str(self.wt), "-b", "fast/t-three") + + def test_worktree_writes_main_tree(self): + none = {"CLAUDE_CODE_MESSAGING_SOCKET": ""} + self.wf("status", "t-three", "progress", "x", project=False, cwd=self.wt, env=none) + self.assertIn("(in progress: x)", (self.root / "TASKS.md").read_text()) + self.assertNotIn("in progress", (self.wt / "TASKS.md").read_text()) + code, out, err = self.wf("done", "t-four", "-m", "ok", project=False, cwd=self.wt / "docs", env=none) + self.assertEqual(code, 0, err) + self.assertIn("**t-four**", self.archive()) + self.assertNotIn("t-four", (self.wt / "tasks" / "archive.md").read_text()) + + def test_done_in_worktree_prints_merge_steps(self): + code, out, err = self.wf("done", "t-four", "-m", "ok", project=False, cwd=self.wt, + env={"CLAUDE_CODE_MESSAGING_SOCKET": ""}) + self.assertEqual(code, 0, err) + self.assertTrue(out.endswith( + "worktree mode (branch fast/t-three), after verify: commit your code here (explicit paths), then:\n" + " wf merge (rebase, ff-merge into master, commit TASKS.md tasks/archive.md, push home; " + "conflict → git rebase master, resolve, verify, wf merge again)\n"), out) + self.assertNotIn("worktree mode", self.ok("done", "t-three", "-m", "ok")) + + def env(self, name, pid): + 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_says_worktree_mode_when_two_sessions_live(self): + son, me = self.env("s", os.getppid()), self.env("o", os.getpid()) + self.assertNotIn("Multi-session", self.ok("next", "--lane", "fast", "--as", "opus", env=me)) + self.ok("next", "--lane", "slow", "--as", "sonnet", "--brief", env=son) + out = self.ok("next", "--lane", "fast", "--as", "opus", env=me) + self.assertIn("===== Multi-session =====\n" + "2 live sessions here: work in your lane's worktree, never on master:\n" + " cd .worktrees/fast && git switch -c fast/ master (wf there writes this TASKS.md)\n\n", out) + self.ok("next", "--lane", "slow", "--as", "sonnet", "--brief", env=son) + out = self.ok("next", "--lane", "slow", "--as", "sonnet", env=son) + self.assertIn(" git worktree add .worktrees/slow -b slow/ master (wf there writes this TASKS.md)\n", out) + code, out, err = self.wf("next", "--lane", "fast", "--as", "opus", project=False, cwd=self.wt, env=me) + self.assertIn("===== Multi-session =====\n2 live sessions here: you are in worktree .worktrees/fast " + "(branch fast/t-three); wf writes the main tree's TASKS.md\n", out) + + +if __name__ == "__main__": + unittest.main() + + +class StaleTest(ClaimCliTest): + def test_stale_list_and_clear(self): + self.ok("status", "t-three", "progress", "x", env=self.env("a", self.dead_pid())) + self.ok("status", "t-four", "progress", "y", env=self.env("b", os.getpid())) + self.assertEqual(self.ok("list", "--stale").splitlines()[0].split()[0], "t-three") + self.assertNotIn("t-four", self.ok("list", "--stale")) + self.assertIn("cleared t-three", self.ok("status", "--clear-stale")) + self.assertEqual(self.ok("list", "--progress").count("prog"), 1) + self.assertIn("t-four", self.ok("list", "--progress")) + self.assertFalse((self.root / ".wf" / "claims" / "t-three.json").exists()) + + def test_no_claim_listed(self): + self.ok("status", "t-three", "progress", "x") + (self.root / ".wf" / "claims" / "t-three.json").unlink(missing_ok=True) + self.assertIn("t-three", self.ok("list", "--stale")) -- cgit