raw · 9113 bytes
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 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | import json import os 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_check import Base, CLEAN SESS = """\ # Tasks — demo ## Awaiting your decision ## Pending - **t-solo** [P1] (1h): Solo. - Sessions: solo — lib bump Model: sonnet - **t-own** [P1] (1h, interactive): Own. - **t-owner** [P2] (1h): Owner. Sessions: owner - **t-par** [P2] (<1h): Par. Sessions: parallel - **t-plain** [P3] (<1h): Plain. ## Needs human ## Deferred """ class SessionsLineTest(unittest.TestCase): def setUp(self): self.doc = T.parse(SESS) def test_values(self): self.assertEqual([(i.id, i.sessions) for i in self.doc.section("pending").items], [("t-solo", "solo"), ("t-own", "owner"), ("t-owner", "owner"), ("t-par", "parallel"), ("t-plain", "parallel")]) def test_set_replaces_in_place_and_keeps_tail_order(self): T.set_fields(self.doc, "t-solo", sessions="owner") self.assertEqual(self.doc.item("t-solo").body, [" Sessions: owner", " Model: sonnet"]) def test_set_adds_before_model(self): T.set_fields(self.doc, "t-plain", sessions="solo") self.assertEqual(self.doc.item("t-plain").body, [" Sessions: solo"]) def test_set_empty_removes(self): T.set_fields(self.doc, "t-owner", sessions="") self.assertEqual(self.doc.item("t-owner").body, []) def test_set_drops_interactive_flag(self): T.set_fields(self.doc, "t-own", sessions="owner") item = self.doc.item("t-own") self.assertEqual(item.lines(), ["- **t-own** [P1] (1h): Own.", " Sessions: owner"]) def test_set_bad_value(self): with self.assertRaises(T.TaskError): T.set_fields(self.doc, "t-plain", sessions="many") def test_note_goes_before_sessions_line(self): T.add_note(self.doc, "t-owner", "x") self.assertEqual(self.doc.item("t-owner").body, [" - x", " Sessions: owner"]) class SessionsPickTest(unittest.TestCase): def setUp(self): self.doc = T.parse(SESS) def pick(self, lane, model, **kw): return L.pick(self.doc, set(), L.DEFAULT_LANES, "1h", lane, model, **kw) def test_alone_solo_picked_owner_skipped(self): item, skipped = self.pick(None, "sonnet") self.assertEqual(item.id, "t-solo") item, skipped = self.pick("slow", "opus", others=1) self.assertEqual((item.id, [(i.id, why) for i, why in skipped]), ("t-par", [("t-solo", "solo: 1 other live session"), ("t-own", "owner: needs the owner (wf next --owner)"), ("t-owner", "owner: needs the owner (wf next --owner)")])) def test_owner_present(self): item, _ = self.pick("slow", "opus", others=1, owner=True) self.assertEqual(item.id, "t-own") def test_solo_skipped_with_other_live_sessions(self): item, skipped = self.pick(None, "sonnet", others=1) self.assertEqual((item, [(i.id, why) for i, why in skipped]), (None, [("t-solo", "solo: 1 other live session")])) _, skipped = self.pick(None, "sonnet", others=2) self.assertEqual(skipped[0][1], "solo: 2 other live sessions") def test_solo_running(self): self.assertEqual(T.solo_running(self.doc, {"t-solo": "sonnet session uds:/a"}), ("t-solo", "sonnet session uds:/a")) self.assertIsNone(T.solo_running(self.doc, {"t-par": "x"})) self.assertIsNone(T.solo_running(self.doc, {})) def test_solo_done_block(self): sessions = {"opus": {"socket": "/o", "alive": True, "pid": 1}, "haiku": {"socket": "/h", "alive": False, "pid": 2}, "sonnet": {"socket": "/s", "alive": True, "pid": 3}} self.assertEqual(L.solo_done_block(["t-solo"], sessions, "3"), ["notify opus uds:/o: solo t-solo done, run wf next"]) self.assertEqual(L.solo_done_block([], sessions, "3"), []) class SessionsCheckTest(Base): def test_bad_word_error_interactive_warning(self): text = CLEAN.replace("- **t-one** [P1] (1h)", "- **t-one** [P1] (1h, interactive)").replace( "## Needs human", "- **t-x** [P3] (1h): X.\n Sessions: lots\n\n## Needs human") self.tasks(text) errors, warnings = self.run_check() self.assertTrue(any(e.endswith("t-x: Sessions 'lots' (want parallel, solo, owner)") for e in errors), errors) self.assertTrue(any(w.endswith("t-one: 'interactive' flag: write 'Sessions: owner' " "(wf set t-one --sessions owner)") for w in warnings), warnings) class SessionsCliTest(Cli): tasks_text = SESS 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 register(self, lane, model, sock, pid): d = self.root / ".wf" / "sessions" d.mkdir(parents=True, exist_ok=True) (d / f"{lane}.json").write_text(json.dumps({"lane": lane, "model": model, "socket": str(sock), "pid": pid, "session": "s", "at": "2026-10-04T10:00"})) def test_list_marks(self): self.assertEqual(self.ok("list"), "t-solo P1 1h - sonnet slow [solo] Solo\n" "t-own P1 1h - opus slow [owner] Own\n" "t-owner P2 1h - opus slow [owner] Owner\n" "t-par P2 <1h - opus fast Par\n" "t-plain P3 <1h - opus fast Plain\n" "pending 5 · human 0 · awaiting 0 · deferred 0\n") def test_add_and_set(self): self.ok("add", "Four.", "-p", "3", "-e", "1h", "--sessions", "solo", "--model", "haiku") self.assertEqual(self.item("t-four"), "- **t-four** [P3] (1h): Four.\n Sessions: solo\n Model: haiku\n") self.ok("set", "t-four", "--sessions", "") self.assertEqual(self.item("t-four"), "- **t-four** [P3] (1h): Four.\n Model: haiku\n") self.fails("set", "t-four", "--sessions", "many", code=2) self.fails("add", "Five.", "-p", "3", "-e", "1h", "--sessions", "many", code=2) def test_interactive_option_is_old_spelling(self): code, out, err = self.wf("add", "Four.", "-p", "3", "-e", "1h", "--interactive") self.assertEqual((code, err), (0, "wf: --interactive is now --sessions owner\n")) self.assertEqual(self.item("t-four"), "- **t-four** [P3] (1h): Four.\n Sessions: owner\n") code, out, err = self.wf("set", "t-plain", "--interactive", "yes") self.assertEqual((code, err), (0, "wf: --interactive is now --sessions owner\n")) self.assertEqual(self.item("t-plain"), "- **t-plain** [P3] (<1h): Plain.\n Sessions: owner\n") def test_next_owner(self): self.ok("done", "t-solo", "-m", "ok") out = self.ok("next", "--lane", "slow", "--as", "opus", env={"CLAUDE_PID": "", "CLAUDE_CODE_MESSAGING_SOCKET": ""}) self.assertIn("- t-own: owner: needs the owner (wf next --owner)\n", out) self.assertIn("===== Next task =====\n- **t-par**", out) # slow empty → fallback fast self.assertTrue(self.ok("next", "--lane", "slow", "--as", "opus", "--owner", "--brief").startswith("- **t-own**")) def test_next_solo_with_other_live_session(self): sock = self.root / "o.sock" sock.write_text("") self.register("fast", "opus", sock, os.getppid()) err = self.fails("next", "--as", "sonnet", env=self.env("s", os.getpid())) self.assertEqual(err, "wf: nothing pickable for all lanes (sonnet) in Pending\n") out = self.wf("next", "--as", "sonnet", env=self.env("s", os.getpid()))[1] self.assertIn("- t-solo: solo: 1 other live session\n", out) def test_solo_in_progress_blocks_others_and_done_notifies(self): s, o = self.env("s", os.getpid()), self.env("o", os.getppid()) self.ok("next", "--lane", "slow", "--as", "sonnet", "--brief", env=s) self.ok("status", "t-solo", "progress", "x", env=s) code, out, err = self.wf("next", "--lane", "fast", "--as", "opus", env=o) self.assertEqual((code, err), (1, f"wf: solo t-solo in progress by sonnet session uds:{self.root / 's.sock'}: " "wait (its wf done notifies you)\n")) self.assertNotIn("Next task", out) out = self.ok("done", "t-solo", "-m", "ok", env=s) self.assertIn(f"notify fast uds:{self.root / 'o.sock'}: solo t-solo done, run wf next\n", out) self.assertTrue(self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=o).startswith("- **t-par**")) if __name__ == "__main__": unittest.main() |