diff options
Diffstat (limited to 'tests/test_sessions_field.py')
| -rw-r--r-- | tests/test_sessions_field.py | 197 |
1 files changed, 197 insertions, 0 deletions
diff --git a/tests/test_sessions_field.py b/tests/test_sessions_field.py new file mode 100644 index 0000000..af1cb62 --- /dev/null +++ b/tests/test_sessions_field.py @@ -0,0 +1,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() |