diff options
Diffstat (limited to 'tests/test_cli.py')
| -rw-r--r-- | tests/test_cli.py | 752 |
1 files changed, 752 insertions, 0 deletions
diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..43135c9 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,752 @@ +import datetime +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +HERE = Path(__file__).resolve().parent.parent +WF = HERE / "wf.py" +sys.path.insert(0, str(HERE)) + +TOML = ('format = 1\ntasks = "TASKS.md"\narchive = "tasks/archive.md"\ndocs = ["DESIGN.md", "docs/"]\n' + 'verify = ["make test"]\ndone = ["update CATALOG"]\nledgers = ".superpowers/sdd"\n') +TASKS = """\ +# Tasks — demo + +## Awaiting your decision + +- **a-key**: Key needed. Which one? + +## Pending + +- **t-one** [P1] (1h) (in progress: master): First thing. Do it. + - Steps: a + Ref: DESIGN.md#terrain, docs/plan.md + +- **t-two** [P2] (5h) (blocked: [[a-key]]): Second thing. + +- **t-three** [P2] (<1h): Third thing. Later. + - After: [[t-one]] + +## Needs human + +- **t-play** [P1] (<1h): Play-test feel. + - [ ] keyboard works + +## Deferred + +- **t-far** [P3] (100h): Far future. +""" +ARCHIVE = "# Archive (newest first)\n\n- 2026-09-01 **t-done** Done thing — ok\n- 2026-08-01 **t-older** Older thing — fine\n" +TODAY = datetime.date.today().isoformat() + + +class Cli(unittest.TestCase): + tasks_text = TASKS + toml = TOML + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.root = Path(self.tmp.name).resolve() / "demo" + (self.root / "tasks").mkdir(parents=True) + (self.root / "docs").mkdir() + (self.root / "workflow.toml").write_text(self.toml) + (self.root / "TASKS.md").write_bytes(self.tasks_text.encode()) + (self.root / "tasks" / "archive.md").write_text(ARCHIVE) + (self.root / "DESIGN.md").write_text("# Design\n\n## terrain\n\nHeightmap.\n") + (self.root / "docs" / "plan.md").write_text("# The Plan\n\n**Goal:** Ship it.\n") + self.inbox = self.root.parent / "inbox.md" + + def tearDown(self): + self.tmp.cleanup() + + def wf(self, *args, stdin="", cwd=None, project=True, env=None, hints=False): + cmd = [sys.executable, str(WF), *(["--project", str(self.root)] if project else []), *args] + full = {**os.environ, "WF_INBOX": str(self.inbox), "WF_ROOT": str(self.root.parent), + "WF_TOOL_ROOT": str(self.root.parent / "no-tool"), "CLAUDE_CONFIG_DIR": str(self.root.parent / "no-claude"), **(env or {})} + r = subprocess.run(cmd, input=stdin, capture_output=True, text=True, cwd=cwd or self.root, env=full, timeout=30) + err = r.stderr if hints else "".join(l for l in r.stderr.splitlines(True) if not l.startswith("hint:")) + return r.returncode, r.stdout, err + + def ok(self, *args, **kw): + code, out, err = self.wf(*args, **kw) + self.assertEqual((code, err), (0, ""), out + err) + return out + + def fails(self, *args, code=1, **kw): + got, out, err = self.wf(*args, **kw) + self.assertEqual(got, code, out + err) + self.assertNotIn("Traceback", err) + return err + + def tasks(self): + return (self.root / "TASKS.md").read_text() + + def archive(self): + return (self.root / "tasks" / "archive.md").read_text() + + def item(self, id): + return self.ok("show", id) + + +class ReadTest(Cli): + def test_list_default_is_pending(self): + self.assertEqual(self.ok("list"), + "t-one P1 1h prog opus slow First thing\n" + "t-two P2 5h blkd opus fast Second thing\n" + "t-three P2 <1h - opus fast Third thing\n" + "pending 3 · human 1 · awaiting 1 · deferred 1\n") + + def test_list_all(self): + self.assertEqual(self.ok("list", "-s", "all"), + "== Awaiting your decision\n" + "a-key -- - - - - Key needed\n" + "== Pending\n" + "t-one P1 1h prog opus slow First thing\n" + "t-two P2 5h blkd opus fast Second thing\n" + "t-three P2 <1h - opus fast Third thing\n" + "== Needs human\n" + "t-play P1 <1h - opus fast Play-test feel\n" + "== Deferred\n" + "t-far P3 100h - opus fast Far future\n" + "pending 3 · human 1 · awaiting 1 · deferred 1\n") + + def first_column(self, *args): + return [l.split()[0] for l in self.ok("list", *args).splitlines()[:-1]] + + def test_list_filters(self): + self.assertEqual(self.first_column("-p", "1"), ["t-one"]) + self.assertEqual(self.first_column("--ready"), ["t-one"]) + self.assertEqual(self.first_column("--blocked"), ["t-two"]) + self.assertEqual(self.first_column("--progress"), ["t-one"]) + self.assertEqual(self.first_column("--ref", "DESIGN.md"), ["t-one"]) + self.assertEqual(self.first_column("--ref", "DESIGN.md#terrain"), ["t-one"]) + self.assertEqual(self.first_column("--ref", "DESIGN.md#other"), []) + self.assertEqual(self.first_column("-n", "2"), ["t-one", "t-two"]) + self.assertEqual(self.first_column("-s", "deferred"), ["t-far"]) + + def test_list_cuts_long_titles_to_100_columns(self): + self.ok("set", "t-far", "--title", "x" * 150) + line = self.ok("list", "-s", "deferred").splitlines()[0] + self.assertEqual(len(line), 100) + self.assertTrue(line.endswith("xxx…")) + + def test_show(self): + self.assertEqual(self.ok("show", "t-three", "a-key"), + "- **t-three** [P2] (<1h): Third thing. Later.\n - After: [[t-one]]\n\n" + "- **a-key**: Key needed. Which one?\n") + + def test_show_by_unique_prefix(self): + self.assertEqual(self.ok("show", "t-thr"), "- **t-three** [P2] (<1h): Third thing. Later.\n - After: [[t-one]]\n") + + def test_show_ambiguous_prefix(self): + self.assertEqual(self.fails("show", "t-t"), "wf: 't-t' matches t-three, t-two\n") + + def test_show_unknown(self): + self.assertRegex(self.fails("show", "t-zzz"), r"^wf: unknown id 't-zzz' \(nearest: .*\)\n$") + + def test_next(self): + self.assertEqual(self.ok("next", "--as", "opus"), + "===== Awaiting your decision (mention, don't block) =====\n" + "- a-key: Key needed. Which one?\n\n" + "===== Needs human (not picked) =====\n" + "- t-play: Play-test feel\n\n" + "===== Lanes =====\n" + "fast: 0 pickable · 2 waiting · no session\n" + "slow: 1 pickable · 0 waiting · no session → orchestrator or owner: start one (wf next --lane slow)\n\n" + "===== Next task =====\n" + "- **t-one** [P1] (1h) (in progress: master): First thing. Do it.\n" + " - Steps: a\n" + " Ref: DESIGN.md#terrain, docs/plan.md\n\n" + "===== DESIGN.md#terrain (line 3) =====\n## terrain\n\nHeightmap.\n\n" + "docs/plan.md: The Plan — Ship it.\n") + + def test_next_brief(self): + self.assertEqual(self.ok("next", "--as", "opus", "--brief"), + "- **t-one** [P1] (1h) (in progress: master): First thing. Do it.\n" + " - Steps: a\n Ref: DESIGN.md#terrain, docs/plan.md\n") + + def test_next_lists_what_it_skipped(self): + self.ok("move", "t-one", "deferred") + self.ok("prio", "t-far", "3") + code, out, err = self.wf("next", "--as", "opus") + self.assertEqual(code, 1) + self.assertIn("===== Skipped =====\n- t-two: blocked: a-key\n- t-three: after: t-one\n", out) + self.assertEqual(err, "wf: nothing pickable for all lanes (opus) in Pending\n") + + def test_next_shows_ledgers_and_inbox(self): + sdd = self.root / ".superpowers" / "sdd" / "p" + sdd.mkdir(parents=True) + (self.root / "docs" / "p.md").write_text("# P\n**Goal:** g\n### Task 1: A\n### Task 2: B\n### Task 3: C\n") + (sdd / "progress.md").write_text("# SDD ledger — plan: docs/p.md\nTask 1: complete (x)\nTask 2: Ruling: y\nTask 2: complete (z)\n") + self.inbox.write_text("- 2026-09-29 demo bug: x @abc\n- 2026-09-29 demo idea: y @abc\n") + out = self.ok("next", "--as", "opus") + self.assertIn("===== In-flight plans =====\n- docs/p.md: done 1, 2; resume at Task 3 (task-start PLAN 3) " + "(1 rulings; ledger .superpowers/sdd/p/progress.md)\n\n", out) + self.assertTrue(out.endswith("\nworkflow inbox: 2 reports (triage: workflow session)\n"), out) + + def test_ctx_item(self): + self.assertEqual(self.ok("ctx", "t-one"), + "- **t-one** [P1] (1h) (in progress: master): First thing. Do it.\n" + " - Steps: a\n" + " Ref: DESIGN.md#terrain, docs/plan.md\n\n" + "Section: Pending\n" + "Needed by: t-three\n" + "Verify (run before wf finish): make test\n\n" + "===== DESIGN.md#terrain (line 3) =====\n## terrain\n\nHeightmap.\n\n" + "docs/plan.md: The Plan — Ship it.\n") + + def test_ctx_dependencies_and_links(self): + out = self.ok("ctx", "t-three") + self.assertIn("Section: Pending\nAfter: t-one (open, Pending)\n", out) + self.assertEqual(self.ok("ctx", "a-key").split("\n\n", 1)[1], "Section: Awaiting your decision\nBlocks: t-two\n") + + def test_ctx_archived_id(self): + self.assertEqual(self.ok("ctx", "t-done"), "done: - 2026-09-01 **t-done** Done thing — ok\n") + + def test_ctx_anchor(self): + self.assertEqual(self.ok("ctx", "DESIGN.md#terrain"), + "===== DESIGN.md#terrain (line 3) =====\n## terrain\n\nHeightmap.\n\n" + "Tasks: t-one\n") + + def test_search(self): + self.assertEqual(self.ok("search", "thing", "later"), + "t-three Third thing. Later.\n" + "t-one First thing. Do it.\n" + "t-two Second thing.\n" + "tasks/archive.md:3 t-done Done thing — ok\n" + "tasks/archive.md:4 t-older Older thing — fine\n") + + def test_search_docs(self): + self.assertEqual(self.ok("search", "--docs", "height"), "DESIGN.md:3 terrain — Heightmap.\n") + + def test_search_nothing(self): + self.assertEqual(self.fails("search", "zzz"), "wf: no hits\n") + + def test_log(self): + self.assertEqual(self.ok("log", "-n", "1"), "- 2026-09-01 **t-done** Done thing — ok\n") + self.assertEqual(self.ok("log", "older"), "- 2026-08-01 **t-older** Older thing — fine\n") + + def test_check_clean(self): + self.assertEqual(self.ok("check"), "OK: 0 errors · 0 warnings\n") + + def test_check_errors(self): + (self.root / "TASKS.md").write_text(TASKS.replace("[[t-one]]", "[[t-gone]]")) + code, out, err = self.wf("check") + self.assertEqual((code, err), (1, "")) + self.assertEqual(out, "ERROR: TASKS.md:16: [[t-gone]]: no such id in TASKS or archive\n1 errors · 0 warnings\n") + + def test_projects(self): + other = self.root.parent / "group" / "other" + (other / "tasks").mkdir(parents=True) + (other / "workflow.toml").write_text('format = 1\ntasks = "TASKS.md"\narchive = "tasks/archive.md"\n') + (other / "TASKS.md").write_text("## Awaiting your decision\n\n## Pending\n\n- **t-x** [P9] (1h): X.\n\n## Needs human\n\n## Deferred\n") + (other / "tasks" / "archive.md").write_text("# Archive\n") + hidden = self.root.parent / ".gate" + hidden.mkdir() + (hidden / "workflow.toml").write_text(TOML) + tool = self.root.parent / "workflow" # a wf checkout: its templates/ is no project + (tool / "wflib").mkdir(parents=True) + (tool / "templates").mkdir() + (tool / "wf.py").write_text("") + (tool / "templates" / "workflow.toml").write_text(TOML) + self.assertEqual(self.ok("projects", project=False), + "demo pending 3 human 1 awaiting 1 errors 0 next: t-one First thing\n" + "group/other pending 1 human 0 awaiting 0 errors 1 next: t-x X\n") + + def test_outside_a_project(self): + err = self.fails("list", project=False, cwd=self.root.parent) + self.assertRegex(err, r"^wf: no workflow.toml in .* or above: not a wf project \(wf init makes one\)\n$") + + def test_broken_config(self): + (self.root / "workflow.toml").write_text(TOML + "verfy = []\n") + self.assertEqual(self.fails("list"), "wf: workflow.toml: unknown key 'verfy'\n") + + def test_finds_project_from_subfolder(self): + self.assertIn("t-one", self.ok("list", project=False, cwd=self.root / "docs")) + + def test_no_command_is_usage_error(self): + self.fails(code=2) + + +class AddTest(Cli): + def test_add_places_by_priority_and_prints_header(self): + out = self.ok("add", "Cave seams. Close the slit.", "-p", "1", "-e", "<1h") + self.assertEqual(out, "- **t-cave-seams** [P1] (<1h): Cave seams. Close the slit.\n") + self.assertEqual(self.first_ids(), ["t-one", "t-cave-seams", "t-two", "t-three"]) + + def first_ids(self, section="pending"): + return [l.split()[0] for l in self.ok("list", "-s", section).splitlines()[:-1]] + + def test_add_with_options(self): + self.ok("add", "Docs pass", "-p", "2", "-e", "5h", "--sessions", "owner", "--id", "t-docs", + "--after", "t-one,t-done", "--ref", "docs/plan.md,DESIGN.md#terrain") + self.assertEqual(self.item("t-docs"), + "- **t-docs** [P2] (5h): Docs pass.\n" + " Sessions: owner\n" + " - After: [[t-one]], [[t-done]]\n" + " Ref: docs/plan.md, DESIGN.md#terrain\n") + + def test_add_cloud(self): + self.ok("add", "Code fix", "-p", "1", "-e", "1h", "--done", "tests pass", "--model", "sonnet", "--cloud", "yes", + "--id", "t-cf") + self.assertEqual(self.item("t-cf"), "- **t-cf** [P1] (1h): Code fix.\n Done: tests pass\n" + " Model: sonnet\n Cloud: yes\n") + self.ok("add", "Gui fix", "-p", "1", "-e", "1h", "--cloud", "no", "--id", "t-gf") + self.assertIn(" Cloud: no\n", self.item("t-gf")) + + def test_add_body_from_stdin(self): + self.ok("add", "With body", "-p", "3", "-e", "1h", "-b", stdin="- Steps: a\n - sub\n- Done: b\n") + self.assertEqual(self.item("t-with-body"), + "- **t-with-body** [P3] (1h): With body.\n - Steps: a\n - sub\n - Done: b\n") + + def test_add_to_other_sections(self): + self.assertEqual(self.ok("add", "Which colour. Red or blue?", "-s", "awaiting"), + "- **a-which-colour** : Which colour. Red or blue?\n".replace(" :", ":")) + self.ok("add", "Feel check", "-p", "2", "-e", "<1h", "-s", "human") + self.assertEqual(self.first_ids("human"), ["t-play", "t-feel-check"]) + + def test_add_slice(self): + self.ok("add", "Map scaffold", "-e", "1h", "--parent", "t-far") + self.ok("add", "Combat rows", "-e", "1h", "--parent", "t-far") + self.assertEqual(self.ok("show", "t-far", "t-far-1", "t-far-2"), + "- **t-far** [P3] (100h): Far future.\n - Slices: [[t-far-1]], [[t-far-2]]\n\n" + "- **t-far-1** [P3] (1h): Map scaffold.\n\n" + "- **t-far-2** [P3] (1h): Combat rows.\n - After: [[t-far-1]]\n") + self.assertEqual(self.first_ids("deferred"), ["t-far", "t-far-1", "t-far-2"]) + + def test_add_slice_skips_deferred_previous_slice(self): + self.ok("add", "Map scaffold", "-e", "1h", "--parent", "t-far") + self.ok("add", "Combat rows", "-e", "1h", "--parent", "t-far", "-s", "pending") + self.ok("add", "Loot rows", "-e", "1h", "--parent", "t-far", "-s", "pending") + self.assertEqual(self.ok("show", "t-far-2", "t-far-3"), + "- **t-far-2** [P3] (1h): Combat rows.\n\n" + "- **t-far-3** [P3] (1h): Loot rows.\n - After: [[t-far-2]]\n") + + def test_add_named_slice_chains_after_previous(self): + self.ok("add", "Map scaffold", "-e", "1h", "--parent", "t-far", "--id", "t-map") + self.ok("add", "Combat rows", "-e", "1h", "--parent", "t-far", "--id", "t-combat") + self.assertEqual(self.ok("show", "t-far", "t-combat"), + "- **t-far** [P3] (100h): Far future.\n - Slices: [[t-map]], [[t-combat]]\n\n" + "- **t-combat** [P3] (1h): Combat rows.\n - After: [[t-map]]\n") + self.assertEqual(self.fails("done", "t-far", "-m", "x"), "wf: 't-far' has open slices: t-map, t-combat\n") + self.ok("done", "t-map", "-m", "x") + self.assertIn("last slice of t-far: finish it", self.ok("done", "t-combat", "-m", "x")) + + def test_add_leading_id_in_text(self): + self.assertEqual(self.ok("add", "a-colour: Which colour? Red or blue?", "-s", "awaiting"), + "- **a-colour**: Which colour? Red or blue?\n") + self.assertEqual(self.ok("add", "t-seams: Cave seams. Close them", "-p", "1", "-e", "1h"), + "- **t-seams** [P1] (1h): Cave seams. Close them.\n") + + def test_add_help_says_id_works_with_parent(self): + self.assertIn("--parent", self.ok("add", "-h").split("--id ID", 2)[2].split("--after")[0]) + + def test_add_block(self): + self.ok("add", "-", stdin="- **t-blk** [P0] (1h): Block. Goal.\n - Steps: a\n Done: x\n") + self.assertEqual(self.item("t-blk"), "- **t-blk** [P0] (1h): Block. Goal.\n - Steps: a\n Done: x\n") + self.assertEqual(self.first_ids()[0], "t-blk") + + def test_add_block_in_old_format_is_refused(self): + self.assertEqual(self.fails("add", "-", stdin="3. **[P2] Old** (Effort: 1h) — goal.\n"), + "wf: old numbered format: write '- **id** [Pn] (effort): Title. Goal.'\n") + + def test_add_needs_priority_and_effort(self): + self.assertIn("-p", self.fails("add", "No prio", "-e", "1h", code=2)) + self.assertIn("-e", self.fails("add", "No effort", "-p", "1", code=2)) + self.assertEqual(self.tasks(), TASKS) + + def test_add_title_without_letters(self): + self.assertEqual(self.fails("add", "???", "-p", "1", "-e", "1h"), + "wf: no id can be made from title '???': give one with --id\n") + + def test_add_same_title_twice(self): + self.ok("add", "Cave seams", "-p", "1", "-e", "1h") + self.assertEqual(self.ok("add", "Cave seams", "-p", "1", "-e", "1h"), "- **t-cave-seams-2** [P1] (1h): Cave seams.\n") + + def test_add_refused_when_result_fails_check(self): + self.assertEqual(self.fails("add", "Bad ref", "-p", "1", "-e", "1h", "--ref", "docs/none.md"), + "wf: refused: nothing written, the change adds problems:\n TASKS.md:14: t-bad-ref: Ref 'docs/none.md' does not exist\n") + self.assertEqual(self.tasks(), TASKS) + + def test_dry_run(self): + out = self.ok("add", "Cave seams", "-p", "1", "-e", "1h", "--dry-run") + self.assertIn("+- **t-cave-seams** [P1] (1h): Cave seams.\n", out) + self.assertIn("--- TASKS.md\n+++ TASKS.md (new)\n", out) + self.assertEqual(self.tasks(), TASKS) + + +class DoneTest(Cli): + def test_done_builtin_checklist_empty_cfg(self): + self.toml = TOML.replace('verify = ["make test"]\ndone = ["update CATALOG"]\n', '') + self.setUp() + out = self.ok("done", "t-one", "-m", "x") + self.assertEqual(out, "done: t-one → tasks/archive.md\nfast work now pickable: t-three (no session: tell the owner)\nchecklist:\n - re-learned anything (>3 greps to find)? one anchor line → that area's code map\n") + + def test_done(self): + out = self.ok("done", "t-one", "-m", "shipped\nwith care") + self.assertEqual(out, "done: t-one → tasks/archive.md\nfast work now pickable: t-three (no session: tell the owner)\nverify:\n make test\nchecklist:\n - re-learned anything (>3 greps to find)? one anchor line → that area's code map\n - update CATALOG\n") + self.assertNotIn("t-one**", self.tasks()) + self.assertEqual(self.archive().splitlines()[2], f"- {TODAY} **t-one** First thing — shipped with care") + self.assertEqual(self.ok("check"), "OK: 0 errors · 0 warnings\n") + self.assertEqual(self.ok("next", "--as", "opus", "--brief").splitlines()[0], "- **t-three** [P2] (<1h): Third thing. Later.") + + def test_done_needs_entry(self): + self.assertEqual(self.fails("done", "t-one", code=2), "wf: done: -m \"<entry>\" is required for a single task\n") + + def test_done_several_uses_goals(self): + self.ok("done", "t-one", "t-three") + self.assertEqual(self.archive().splitlines()[2:4], + [f"- {TODAY} **t-three** Third thing — Later.", f"- {TODAY} **t-one** First thing — Do it."]) + + def test_done_refuses_parent_with_open_slices(self): + self.ok("add", "Map scaffold", "-e", "1h", "--parent", "t-far") + self.assertEqual(self.fails("done", "t-far", "-m", "x"), "wf: 't-far' has open slices: t-far-1\n") + + def test_done_awaiting_item_unblocks(self): + self.assertEqual(self.ok("done", "a-key"), "removed: a-key\nunblocked: t-two\n") + self.assertEqual(self.item("t-two"), "- **t-two** [P2] (5h): Second thing.\n") + self.assertEqual(self.archive(), ARCHIVE) + + def test_done_unknown(self): + self.assertRegex(self.fails("done", "t-on", "-m", "x"), r"^wf: unknown id 't-on' \(nearest: t-one") + self.assertEqual(self.archive(), ARCHIVE) + + +class EditTest(Cli): + def order(self): + return [l.split()[0] for l in self.ok("list").splitlines()[:-1]] + + def test_prio(self): + self.assertEqual(self.ok("prio", "t-three", "0"), "- **t-three** [P0] (<1h): Third thing. Later.\n") + + def test_prio_refused_when_it_jumps_a_dependency(self): + # t-three is After: t-one, the insert rule keeps it behind t-one + self.ok("prio", "t-three", "0") + self.assertEqual(self.order(), ["t-one", "t-three", "t-two"]) + + def test_move_section(self): + self.assertEqual(self.ok("move", "t-two", "deferred"), "- **t-two** [P2] (5h) (blocked: [[a-key]]): Second thing.\n") + self.assertEqual(self.order(), ["t-one", "t-three"]) + + def test_move_relative(self): + self.ok("move", "t-three", "--before", "t-two") + self.assertEqual(self.order(), ["t-one", "t-three", "t-two"]) + self.assertEqual(self.fails("move", "t-two", "--before", "t-one"), + "wf: moving 't-two' there breaks priority order (wf prio, or --force)\n") + self.ok("move", "t-two", "--before", "t-one", "--force") + self.assertEqual(self.order(), ["t-two", "t-one", "t-three"]) + + def test_move_needs_one_target(self): + self.fails("move", "t-two", code=2) + self.fails("move", "t-two", "deferred", "--before", "t-one", code=2) + + def test_status(self): + self.assertEqual(self.ok("status", "t-three", "progress", "feature/x"), + "- **t-three** [P2] (<1h) (in progress: feature/x): Third thing. Later.\n") + self.assertEqual(self.ok("status", "t-three", "blocked", "a-key"), + "- **t-three** [P2] (<1h) (blocked: [[a-key]]): Third thing. Later.\n") + self.assertEqual(self.ok("status", "t-three", "clear"), "- **t-three** [P2] (<1h): Third thing. Later.\n") + self.assertEqual(self.fails("status", "t-three", "blocked", "a-none"), "wf: 'a-none' is not an open Awaiting item\n") + self.fails("status", "t-three", "progress", code=2) + + def test_set(self): + self.ok("set", "t-two", "--title", "Second: renamed", "--effort", "10h", "--sessions", "owner", + "--after", "t-one", "--ref", "docs/plan.md (goal)") + self.assertEqual(self.item("t-two"), + "- **t-two** [P2] (10h) (blocked: [[a-key]]): Second: renamed.\n" + " Sessions: owner\n - After: [[t-one]]\n Ref: docs/plan.md (goal)\n") + self.ok("set", "t-two", "--after", "", "--ref", "", "--sessions", "") + self.assertEqual(self.item("t-two"), "- **t-two** [P2] (10h) (blocked: [[a-key]]): Second: renamed.\n") + + def test_set_done(self): + self.ok("set", "t-two", "--done", "two works") + self.assertEqual(self.item("t-two"), "- **t-two** [P2] (5h) (blocked: [[a-key]]): Second thing.\n Done: two works\n") + self.ok("set", "t-two", "--done", "") + self.assertEqual(self.item("t-two"), "- **t-two** [P2] (5h) (blocked: [[a-key]]): Second thing.\n") + + def test_rename(self): + self.assertEqual(self.ok("rename", "a-key", "a-which-key"), "renamed: a-key → a-which-key (1 link)\n") + self.assertEqual(self.item("t-two"), "- **t-two** [P2] (5h) (blocked: [[a-which-key]]): Second thing.\n") + self.assertEqual(self.fails("rename", "t-one", "t-done"), "wf: id 't-done' already used in the archive (ids are never reused)\n") + + def test_set_refused_on_bad_ref(self): + self.assertIn("Ref 'missing.md' does not exist", self.fails("set", "t-two", "--ref", "missing.md")) + self.assertEqual(self.tasks(), TASKS) + + def test_set_without_fields(self): + self.fails("set", "t-two", code=2) + + def test_note(self): + self.ok("note", "t-one", "cause found") + self.assertEqual(self.item("t-one"), + "- **t-one** [P1] (1h) (in progress: master): First thing. Do it.\n" + " - Steps: a\n - cause found\n Ref: DESIGN.md#terrain, docs/plan.md\n") + + def test_body(self): + self.ok("body", "t-one", stdin="- Steps: b\n- Done: c\n") + self.assertEqual(self.item("t-one"), + "- **t-one** [P1] (1h) (in progress: master): First thing. Do it.\n" + " - Steps: b\n - Done: c\n Ref: DESIGN.md#terrain, docs/plan.md\n") + + def test_body_refusal_says_refused_and_writes_nothing(self): + before = self.tasks() + for _ in range(2): + err = self.fails("body", "t-one", stdin="- After: [[t-three]]\n") + self.assertTrue(err.startswith("wf: refused: nothing written"), err) + self.assertIn("placed before 't-three'", err) + self.assertEqual(self.tasks(), before) + + def test_body_positional_text_hints_stdin(self): + self.assertIn("stdin", self.fails("body", "t-one", "some", "text")) + self.assertIn("stdin", self.ok("body", "-h")) + + def test_tick(self): + self.assertEqual(self.ok("tick", "t-play", "keyboard"), "ticked: keyboard works\n") + self.assertIn(" - [x] keyboard works\n", self.tasks()) + self.assertEqual(self.fails("tick", "t-play", "1"), "wf: box already ticked: keyboard works\n") + + def test_write_needs_exact_id(self): + self.assertRegex(self.fails("prio", "t-thr", "1"), r"^wf: unknown id 't-thr' \(nearest: t-three") + + def test_untouched_text_survives_a_write(self): + self.ok("note", "t-far", "x") + self.assertEqual(self.tasks(), TASKS.replace("Far future.\n", "Far future.\n - x\n")) + + +class OldProblemsTest(Cli): + tasks_text = TASKS + "\n## Notes\n\nSee [[t-lost]].\n" + + def test_existing_problems_do_not_block_other_edits(self): + self.ok("note", "t-far", "x") + self.assertIn(" - x\n", self.tasks()) + + +class CrlfTest(Cli): + tasks_text = TASKS.replace("\n", "\r\n") + + def test_write_keeps_crlf(self): + self.ok("add", "Cave seams", "-p", "1", "-e", "1h") + raw = (self.root / "TASKS.md").read_bytes() + self.assertIn(b"- **t-cave-seams** [P1] (1h): Cave seams.\r\n", raw) + self.assertEqual(raw.count(b"\n"), raw.count(b"\r\n")) + self.assertTrue(raw.endswith(b"Far future.\r\n")) + + def test_read_commands_work(self): + self.assertEqual(self.ok("show", "a-key"), "- **a-key**: Key needed. Which one?\n") + + +class FormatGateTest(Cli): + toml = TOML.replace("format = 1", "format = 0") + + def test_write_refused(self): + self.assertEqual(self.fails("note", "t-one", "x"), + "wf: TASKS format 0, wf needs 1: run wf migrate --write (idle project, one commit)\n") + self.assertEqual(self.tasks(), TASKS) + + def test_read_works(self): + self.assertIn("t-one", self.ok("list")) + + +class NewerFormatTest(Cli): + toml = TOML.replace("format = 1", "format = 2") + + def test_refused(self): + self.assertEqual(self.fails("list"), "wf: project format 2 is newer than this wf (1): update /projects/public/workflow\n") + + +class ReportTest(Cli): + def test_report_line(self): + self.assertEqual(self.ok("report", "prio change\nneeds two commands", "--kind", "friction", "--cmd", "wf prio x 1"), + "reported (workflow inbox); carry on\n") + self.assertRegex(self.inbox.read_text(), + rf"^- {TODAY} demo friction: prio change needs two commands \(cmd: wf prio x 1\) @[0-9a-f]{{7}}\n$") + + def test_default_kind_and_outside_project(self): + self.ok("report", "something", project=False, cwd=self.root.parent) + self.assertRegex(self.inbox.read_text(), rf"^- {TODAY} {self.root.parent.name} bug: something @") + + def test_parallel_reports_stay_whole(self): + env = {**os.environ, "WF_INBOX": str(self.inbox)} + procs = [subprocess.Popen([sys.executable, str(WF), "--project", str(self.root), "report", f"report {n} " + "x" * 3000], + env=env, stdout=subprocess.DEVNULL) for n in range(8)] + for p in procs: + self.assertEqual(p.wait(timeout=30), 0) + lines = self.inbox.read_text().splitlines() + self.assertEqual(len(lines), 8) + for line in lines: + self.assertRegex(line, rf"^- {TODAY} demo bug: report \d x{{3000}} @[0-9a-f]{{7}}$") + + def test_empty_report(self): + self.fails("report", " ", code=2) + + +class InitTest(Cli): + def test_init(self): + new = self.root.parent / "fresh" + new.mkdir() + self.assertEqual(self.ok("init", project=False, cwd=new), + "wrote workflow.toml\nwrote TASKS.md\nwrote tasks/archive.md\nwrote CLAUDE.md\n") + self.assertTrue((new / "TASKS.md").read_text().startswith("# Tasks — fresh\n")) + self.assertTrue((new / "CLAUDE.md").read_text().startswith("# CLAUDE.md — fresh\n")) + self.assertEqual(self.ok("check", project=False, cwd=new), "OK: 0 errors · 0 warnings\n") + self.assertEqual(self.ok("add", "First", "-p", "1", "-e", "1h", project=False, cwd=new), "- **t-first** [P1] (1h): First.\n") + + def test_init_twice(self): + self.assertEqual(self.fails("init"), f"wf: {self.root}/workflow.toml exists already\n") + + def test_init_keeps_existing_tasks_file(self): + new = self.root.parent / "old" + new.mkdir() + (new / "TASKS.md").write_text("mine\n") + self.assertEqual(self.ok("init", project=False, cwd=new), + "wrote workflow.toml\nkept TASKS.md (exists; old format → wf migrate)\nwrote tasks/archive.md\nwrote CLAUDE.md\n") + self.assertEqual((new / "TASKS.md").read_text(), "mine\n") + + def test_init_keeps_existing_claude_md(self): + new = self.root.parent / "oldc" + new.mkdir() + (new / "CLAUDE.md").write_text("mine\n") + out = self.ok("init", project=False, cwd=new) + self.assertIn("kept CLAUDE.md (exists)\n", out) + self.assertEqual((new / "CLAUDE.md").read_text(), "mine\n") + + def test_init_inside_a_project_subfolder_makes_a_new_project_there(self): + self.assertEqual(self.ok("init", project=False, cwd=self.root / "docs").splitlines()[0], "wrote workflow.toml") + self.assertTrue((self.root / "docs" / "workflow.toml").is_file()) + + +class ProjectsRootTest(unittest.TestCase): + def setUp(self): + import wf + self.wf = wf + self.tmp = tempfile.TemporaryDirectory() + self.top = Path(self.tmp.name) + self.tool = self.top / "public" / "workflow" + self.tool.mkdir(parents=True) + + def tearDown(self): + self.tmp.cleanup() + + def project(self, rel): + (self.top / rel).mkdir(parents=True) + (self.top / rel / "workflow.toml").write_text("") + + def test_parent_with_projects_wins(self): + self.project("public/a") + self.project("games/b") + self.assertEqual(self.wf.default_root(self.tool), self.top / "public") + + def test_parent_without_projects_falls_back_to_grandparent(self): + self.project("games/b") + self.assertEqual(self.wf.default_root(self.tool), self.top) + + def test_no_projects_anywhere_keeps_parent(self): + self.assertEqual(self.wf.default_root(self.tool), self.top / "public") + + +class WriteSafetyTest(unittest.TestCase): + def setUp(self): + import wf + self.wf = wf + self.tmp = tempfile.TemporaryDirectory() + self.path = Path(self.tmp.name) / "TASKS.md" + self.path.write_text("one\n") + + def tearDown(self): + self.tmp.cleanup() + + def test_write(self): + stamp = self.wf.stamp(self.path) + self.wf.write_if_unchanged(self.path, "two\n", stamp) + self.assertEqual(self.path.read_text(), "two\n") + self.assertEqual(sorted(p.name for p in self.path.parent.iterdir()), ["TASKS.md"]) + + def test_stops_when_the_file_changed_since_read(self): + stamp = self.wf.stamp(self.path) + self.path.write_text("someone else wrote this\n") + with self.assertRaisesRegex(self.wf.Failure, "TASKS.md changed on disk since it was read: nothing written, run again"): + self.wf.write_if_unchanged(self.path, "two\n", stamp) + self.assertEqual(self.path.read_text(), "someone else wrote this\n") + + def test_keeps_file_mode(self): + self.path.chmod(0o640) + self.wf.write_if_unchanged(self.path, "two\n", self.wf.stamp(self.path)) + self.assertEqual(self.path.stat().st_mode & 0o777, 0o640) + + +if __name__ == "__main__": + unittest.main() + + +OLD_TASKS = """\ +# Tasks — demo + +## Pending + +1. **[P1] First thing** (Effort: 1h) — do it. + - Steps: a + Reference: [DESIGN.md#terrain](DESIGN.md#terrain) + +2. **[P2] Second thing** (Effort: 5h) — later. + - After: First thing + +## Needs human + +## Awaiting your decision + +- Which key to use. +""" +MIGRATED = """\ +# Tasks — demo + +## Pending + +- **t-first-thing** [P1] (1h): First thing. do it. + - Steps: a + Ref: DESIGN.md#terrain + +- **t-second-thing** [P2] (5h): Second thing. later. + - After: [[t-first-thing]] + +## Needs human + +## Awaiting your decision + +- **a-which-key-to-use**: Which key to use. + +## Deferred +""" + + +class MigrateCliTest(Cli): + tasks_text = OLD_TASKS + toml = TOML.replace("format = 1", "format = 0") + + def test_dry_run_prints_diff_and_map_and_writes_nothing(self): + out = self.ok("migrate") + self.assertIn("+- **t-first-thing** [P1] (1h): First thing. do it.\n", out) + self.assertIn("\nids:\n t-first-thing First thing\n t-second-thing Second thing\n a-which-key-to-use Which key to use\n", out) + self.assertTrue(out.endswith("check after migrate: 0 errors\ndry run: nothing written (wf migrate --write)\n"), out) + self.assertEqual(self.tasks(), OLD_TASKS) + self.assertIn("format = 0", (self.root / "workflow.toml").read_text()) + + def test_write(self): + out = self.ok("migrate", "--write") + self.assertTrue(out.endswith("check after migrate: 0 errors\nwrote TASKS.md, workflow.toml (format = 1)\n"), out) + self.assertEqual(self.tasks(), MIGRATED) + self.assertEqual((self.root / "workflow.toml").read_text(), TOML) + self.assertEqual(self.ok("check"), "OK: 0 errors · 0 warnings\n") + self.assertEqual(self.ok("next", "--as", "opus", "--brief").splitlines()[0], "- **t-first-thing** [P1] (1h): First thing. do it.") + + def test_second_run_changes_nothing(self): + self.ok("migrate", "--write") + self.assertEqual(self.ok("migrate", "--write"), "nothing to migrate\n") + self.assertEqual(self.tasks(), MIGRATED) + + def test_ids_avoid_archived_ones(self): + (self.root / "tasks" / "archive.md").write_text("# A\n\n- 2026-09-01 **t-first-thing** First thing — old\n") + self.ok("migrate", "--write") + self.assertIn("- **t-first-thing-2** [P1] (1h): First thing. do it.\n", self.tasks()) |