import json import shutil import subprocess import sys import tempfile import tomllib import unittest from datetime import date from pathlib import Path SCRIPT = Path(__file__).resolve().parents[1] / "tickets.py" CONFIG_UTILS = SCRIPT.parents[2] / "bmad" / "scripts" / "config_utils.py" def ticket( status, ticket_id=None, after="[]", hitl="false", tracker_id='""', assignee='""', blocked_at=None, blocked_reason=None, kind="story", refined="false", tracker_status=None, ): lines = ["---"] if ticket_id is not None: lines.append(f"id: {ticket_id}") lines += [ f"tracker_id: {tracker_id}", 'remote: ""', f"type: {kind}", 'title: "x"', "parent: epic-cart", "covers: [R1]", f"after: {after}", f"assignee: {assignee}", ] if status: lines.append(f"status: {status}") if tracker_status: lines.append(f"tracker_status: {tracker_status}") lines += [ f"refined: {refined}", f"hitl: {hitl}", "risk: low", ] if blocked_at: lines.append(f'blocked_at: "{blocked_at}"') if blocked_reason: lines.append(f'blocked_reason: "{blocked_reason}"') lines += ["---", "", "# x", ""] return "\n".join(lines) def plan(ticket_id, status=None, assignee=None, blocked_at=None, blocked_reason=None): """A plan as the build's template writes it: quoted values, trailing comments, block lists.""" lines = [ "---", "title: 'x'", "type: 'feature' # feature | bugfix | refactor | chore", f"ticket: '{ticket_id}' # the entry id from the ticket tree, or the story file's stem when the entry has no id; empty outside it", ] if status: lines.append(f"status: '{status}' # draft | ready-for-dev | in-progress | in-review | built | done") for key, value in (("assignee", assignee), ("blocked_at", blocked_at), ("blocked_reason", blocked_reason)): if value: lines.append(f"{key}: '{value}'") lines += ["lenses_ran: []", "deferred:", " - summary: a", " evidence: b", "---", "", "# x", ""] return "\n".join(lines) def run(*args, cwd=None, stdin=None): # tickets.py writes UTF-8; text=True alone would decode with the locale's code page on Windows. return subprocess.run( [sys.executable, str(SCRIPT), *args], encoding="utf-8", capture_output=True, check=False, cwd=cwd, input=stdin ) class TreeCase(unittest.TestCase): """A temp project with `_bmad/custom/` and the epic `out/initiative-checkout/epic-cart`.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) (self.root / "_bmad" / "custom").mkdir(parents=True) self.initiative = self.root / "out" / "initiative-checkout" self.epic = self.add_epic("epic-cart") self.write_store("repo") def tearDown(self): self.tmp.cleanup() def write_store(self, name="repo", root=None): line = f'root = "{root}"\n' if root is not None else "" (self.root / "_bmad" / "custom" / "ticketing-store-config.toml").write_text( f'[tickets]\nstore = "{name}"\n{line}' ) def add_epic(self, slug, status="", after="[]"): folder = self.initiative / slug folder.mkdir(parents=True) line = f"status: {status}\n" if status else "" (folder / f"{slug}.md").write_text(f"---\ntype: epic\n{line}after: {after}\n---\n# {slug}\n") return folder def add_backlog(self): backlog = self.root / "out" / "backlog" backlog.mkdir() return backlog def status_rows(self, folder): return json.loads(run("status", str(folder)).stdout)["tickets"] def add(self, name, text, folder=None): ((folder or self.epic) / name).write_text(text, encoding="utf-8") def seed(self, s1="", s2="", s3=""): self.add("story-scaffold.md", ticket(s1, 1, hitl="true")) self.add("story-ui-shell.md", ticket(s2, 2)) self.add("story-tracer.md", ticket(s3, 3, after="[1, 2]")) self.add("story-codes.md", ticket("", 4, after="[story-tracer]")) self.add("spike-tax.md", ticket("", 5, after="[3]", kind="spike")) def pricing(self): pricing = self.add_epic("epic-pricing") (pricing / "tickets.toml").write_text( '[[entry]]\nid = 1\ntype = "story"\ntitle = "Pricing contract"\n\n' '[[entry]]\nid = 2\ntype = "story"\ntitle = "Pricing rules"\nafter = [1]\n' ) (self.initiative / "tickets.toml").write_text( '[[epic]]\nid = 1\nslug = "epic-pricing"\n\n[[epic]]\nid = 2\nslug = "epic-cart"\n' ) return pricing class TicketsTests(TreeCase): def next(self, *extra): r = run("next", str(self.epic), *extra) self.assertEqual(r.returncode, 0, r.stderr) return json.loads(r.stdout) def files(self, rows): return [r["file"] for r in rows] def test_unstarted_tickets_with_no_prerequisites_are_ready_to_refine(self): self.seed() out = self.next() self.assertEqual(self.files(out["ready_to_refine"]), ["story-scaffold.md", "story-ui-shell.md"]) self.assertEqual(out["ready_to_start"], []) self.assertEqual(self.files(out["blocked"]), ["story-tracer.md", "story-codes.md", "spike-tax.md"]) self.assertTrue(out["ready_to_refine"][0]["hitl"]) def test_unrefined_ticket_is_never_ready_to_start(self): self.seed(s1="done", s2="draft") out = self.next() self.assertEqual(out["ready_to_start"], []) self.assertIn("story-ui-shell.md", self.files(out["ready_to_refine"])) def test_refined_is_ready_to_start_and_ready_set_moves_when_prerequisites_done(self): self.seed(s1="done") self.add("story-ui-shell.md", ticket("ready-for-dev", 2, refined="true")) out = self.next() self.assertEqual(self.files(out["ready_to_start"]), ["story-ui-shell.md"]) self.assertIn("story-tracer.md", self.files(out["blocked"])) self.seed(s1="done", s2="done") out = self.next() self.assertEqual(self.files(out["ready_to_refine"]), ["story-tracer.md"]) def test_prerequisites_resolve_by_id_stem_and_tracker_id(self): self.seed(s1="done", s2="done", s3="done") self.add("story-by-id.md", ticket("", 6, after="[CART-4]")) self.add("story-codes.md", ticket("done", 4, after="[story-tracer]", tracker_id='"CART-4"')) out = self.next() self.assertIn("story-by-id.md", self.files(out["ready_to_refine"])) self.assertIn("spike-tax.md", self.files(out["ready_to_refine"])) def test_in_progress_review_blocked_build_and_blocked_at(self): self.seed(s1="in-progress", s2="in-review") self.add("story-ui-shell.md", ticket("", 2, blocked_at="2026-09-05")) self.add("story-codes.md", ticket("blocked", 4)) out = self.next() self.assertEqual(self.files(out["in_progress"]), ["story-scaffold.md"]) self.assertEqual( self.files(out["blocked"]), ["story-ui-shell.md", "story-tracer.md", "story-codes.md", "spike-tax.md"] ) self.assertEqual([r["state"] for r in out["blocked"]], ["backlog", "backlog", "in-progress", "backlog"]) def test_tracker_status_wins_over_the_build_status_for_state(self): self.seed(s1="in-progress", s2="ready-for-dev") self.add("story-scaffold.md", ticket("in-progress", 1, tracker_status="done")) self.add("story-ui-shell.md", ticket("", 2, tracker_status="in-progress")) out = self.next() self.assertEqual(self.files(out["in_progress"]), ["story-ui-shell.md"]) self.assertEqual(out["in_progress"][0]["status"], "") self.assertEqual(self.files(out["blocked"]), ["story-tracer.md", "story-codes.md", "spike-tax.md"]) self.add("story-ui-shell.md", ticket("", 2, tracker_status="done")) self.assertEqual(self.files(self.next()["ready_to_refine"]), ["story-tracer.md"]) self.add("story-ui-shell.md", ticket("", 2, tracker_status="todo")) self.assertIn("tracker_status", run("next", str(self.epic)).stderr) def test_status_counts_order_and_chain(self): self.seed(s1="done") r = run("status", str(self.epic)) self.assertEqual(r.returncode, 0, r.stderr) out = json.loads(r.stdout) self.assertEqual([t["id"] for t in out["tickets"]], [1, 2, 3, 4, 5]) self.assertEqual(out["counts"], {"total": 5, "done": 1, "backlog": 4}) self.assertEqual(out["longest_remaining_chain"], ["epic-cart/2", "epic-cart/3", "epic-cart/4"]) self.assertEqual(out["tickets"][2]["blocks"], [4, 5]) BREAKDOWN = """ [[entry]] id = 1 type = "story" title = "Scaffold" covers = ["R1"] [[entry]] id = 2 type = "story" title = "UI shell" after = [1] covers = ["R1"] [[entry]] id = 3 type = "spike" title = "Tax engine?" after = [1] covers = ["R4"] hitl = true [[entry]] id = 4 type = "story" title = "Codes" after = [2, 3] covers = ["R2", "R3"] """ def breakdown_epic(self, text=None): (self.epic / "tickets.toml").write_text(text or self.BREAKDOWN, encoding="utf-8") def test_text_outside_ascii_is_read_and_written_as_utf8(self): self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Café ✓ menu"')) r = subprocess.run([sys.executable, str(SCRIPT), "next", str(self.epic)], capture_output=True, check=False) self.assertEqual(r.returncode, 0, r.stderr) out = json.loads(r.stdout.decode("utf-8")) self.assertEqual(out["ready_to_start"][0]["title"], "Café ✓ menu") r = subprocess.run([sys.executable, str(SCRIPT), "pull", str(self.epic), "1"], capture_output=True, check=False) self.assertEqual(r.returncode, 0, r.stderr) self.assertIn("# Café ✓ menu", (self.epic / "story-cafe-menu.md").read_text(encoding="utf-8")) def test_titles_that_slug_alike_get_their_own_file_and_plan(self): self.breakdown_epic( '[[entry]]\nid = 1\ntype = "story"\ntitle = "Café menu"\n\n' '[[entry]]\nid = 2\ntype = "story"\ntitle = "Cafe menu"\n' ) plans = [Path(json.loads(run("find", str(self.epic), n).stdout)["plan"]).name for n in ("1", "2")] self.assertEqual(plans, ["story-cafe-menu-plan.md", "story-cafe-menu-2-plan.md"]) self.assertEqual(json.loads(run("pull", str(self.epic), "2").stdout)["file"], "story-cafe-menu-2.md") self.assertEqual(json.loads(run("pull", str(self.epic), "1").stdout)["file"], "story-cafe-menu.md") plans = [Path(json.loads(run("find", str(self.epic), n).stdout)["plan"]).name for n in ("1", "2")] self.assertEqual(plans, ["story-cafe-menu-plan.md", "story-cafe-menu-2-plan.md"]) def test_a_leaf_file_with_a_byte_order_mark_is_read(self): (self.epic / "story-scaffold.md").write_bytes(b"\xef\xbb\xbf" + ticket("done", 1).encode("utf-8")) rows = self.status_rows(self.epic) self.assertEqual([(r["file"], r["state"]) for r in rows], [("story-scaffold.md", "done")]) def test_mark_keeps_a_plans_crlf_line_endings_and_byte_order_mark(self): self.breakdown_epic() path = self.epic / "story-scaffold-plan.md" text = plan(1, "in-progress").replace("\n", "\r\n") path.write_bytes(b"\xef\xbb\xbf" + text.encode("utf-8")) self.mark(str(self.epic), "1", "done") data = path.read_bytes() self.assertTrue(data.startswith(b"\xef\xbb\xbf")) self.assertNotIn(b"\n", data.replace(b"\r\n", b"")) self.assertIn(b"\r\nstatus: done\r\n", data) self.assertEqual(self.status_rows(self.epic)[0]["status"], "done") def test_planned_entries_surface_when_unblocked(self): self.breakdown_epic() self.add("story-scaffold.md", ticket("done", 1)) out = self.next() self.assertEqual( [(e["id"], e["type"], e["title"]) for e in out["ready_to_start"]], [(2, "story", "UI shell"), (3, "spike", "Tax engine?")], ) self.assertEqual(out["ready_to_start"][1]["covers"], ["R4"]) self.assertEqual(out["ready_to_start"][1]["after"], [1]) self.assertTrue(out["ready_to_start"][1]["hitl"]) self.assertEqual([e["id"] for e in out["blocked"]], [4]) self.add("story-ui-shell.md", ticket("", 2, after="[1]")) out = self.next() self.assertEqual(self.files(out["ready_to_start"]), ["story-ui-shell.md", None]) status = json.loads(run("status", str(self.epic)).stdout) self.assertEqual(status["counts"], {"total": 4, "done": 1, "backlog": 1, "planned": 2}) self.assertEqual(status["tickets"][0]["blocks"], [2, 3]) def test_table_order_is_build_order_not_id(self): self.breakdown_epic( '[[entry]]\nid = 2\ntype = "story"\ntitle = "Second first"\n\n' '[[entry]]\nid = 1\ntype = "story"\ntitle = "First second"\n\n' '[[entry]]\nid = 3\ntype = "story"\ntitle = "Third"\nafter = [1, 2]\n' ) self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], [2, 1]) self.add("story-first-second.md", ticket("draft", 1)) rows = self.status_rows(self.epic) self.assertEqual([r["id"] for r in rows], [2, 1, 3]) def test_entry_waiting_on_unwritten_entry_stays_blocked(self): self.breakdown_epic() out = self.next() self.assertEqual([e["id"] for e in out["ready_to_start"]], [1]) self.assertEqual([e["id"] for e in out["blocked"]], [2, 3, 4]) def test_a_prerequisite_in_review_is_met_and_one_in_progress_is_not(self): self.breakdown_epic() self.add("story-scaffold-plan.md", plan(1, "in-review")) out = self.next() self.assertEqual([e["id"] for e in out["ready_to_start"]], [2, 3]) self.assertEqual([e["id"] for e in out["in_progress"]], [1]) self.add("story-scaffold-plan.md", plan(1, "in-progress")) out = self.next() self.assertEqual(out["ready_to_start"], []) self.assertEqual([e["id"] for e in out["blocked"]], [2, 3, 4]) def test_an_epic_gate_waits_for_the_epic_to_be_done_even_with_its_tickets_in_review(self): pricing = self.pricing() self.breakdown_epic('[[entry]]\nid = 1\ntype = "story"\ntitle = "Cart"\nafter = ["1.2"]\n') self.add("story-pricing-contract-plan.md", plan(1, "in-review"), pricing) self.add("story-pricing-rules-plan.md", plan(2, "in-review"), pricing) self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], [1]) (self.epic / "epic-cart.md").write_text("---\ntype: epic\nafter: [epic-pricing]\n---\n# epic-cart\n") out = self.next() self.assertEqual(out["ready_to_start"], []) self.assertEqual(out["blocked"][0]["gated_by"], ["epic-pricing"]) (pricing / "epic-pricing.md").write_text("---\ntype: epic\nstatus: done\n---\n# epic-pricing\n") self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], [1]) def test_every_row_carries_a_ref_that_find_resolves_to_the_same_row(self): self.pricing() self.breakdown_epic() backlog = self.add_backlog() self.add("bug-x.md", ticket("draft", kind="bug"), backlog) cases = ( (self.initiative, ["1.1", "1.2", "2.1", "2.2", "2.3", "2.4"]), (self.epic, ["2.1", "2.2", "2.3", "2.4"]), (backlog, ["bug-x.md"]), ) for folder, refs in cases: rows = self.status_rows(folder) self.assertEqual([r["ref"] for r in rows], refs, folder.name) for row in rows: r = run("find", str(folder), row["ref"]) self.assertEqual(r.returncode, 0, r.stderr) found = json.loads(r.stdout) self.assertEqual( [found[k] for k in ("epic", "id", "file", "ref")], [row[k] for k in ("epic", "id", "file", "ref")] ) for group in ("ready_to_start", "blocked"): for row in json.loads(run("next", str(folder)).stdout)[group]: self.assertIn(row["ref"], refs) def test_a_ref_is_the_bare_id_in_an_epic_the_initiative_does_not_number(self): self.breakdown_epic() self.assertEqual([r["ref"] for r in self.next()["ready_to_start"]], ["1"]) self.assertEqual(json.loads(run("find", str(self.epic), "1").stdout)["title"], "Scaffold") def test_file_prerequisites_win_over_the_entry_and_status_flags_the_drift(self): self.breakdown_epic() self.add("story-scaffold.md", ticket("draft", 1)) self.add("story-ui-shell.md", ticket("draft", 2, after="[]")) self.assertIn("story-ui-shell.md", self.files(self.next()["ready_to_start"])) rows = self.status_rows(self.epic) self.assertTrue(rows[1]["drift"]) self.assertNotIn("drift", rows[0]) def test_a_leaf_file_without_after_or_hitl_reads_them_as_empty_and_flags_the_drift(self): self.breakdown_epic() self.add("story-ui-shell.md", '---\nid: 2\ntype: story\ntitle: "UI shell"\n---\n# UI shell\n') self.add("spike-tax-engine.md", '---\nid: 3\ntype: spike\ntitle: "Tax engine?"\nafter: [1]\n---\n# Tax\n') rows = {r["id"]: r for r in self.status_rows(self.epic)} self.assertEqual((rows[2]["after"], rows[2]["drift"]), ([], {"after": {"file": [], "entry": [1]}})) self.assertEqual((rows[3]["hitl"], rows[3]["drift"]), (False, {"hitl": {"file": False, "entry": True}})) self.assertNotIn("drift", rows[1]) def test_a_blocked_row_names_the_prerequisites_it_waits_on(self): self.pricing() self.breakdown_epic(self.BREAKDOWN.replace("after = [2, 3]", 'after = [2, 3, "1.2"]')) self.add("story-scaffold-plan.md", plan(1, "in-progress")) self.add("spike-tax-engine-plan.md", plan(3, "in-review")) self.add("story-ui-shell-plan.md", plan(2, "blocked", blocked_reason="legal")) out = self.next() blocked = {e["id"]: e for e in out["blocked"]} self.assertEqual(blocked[4]["waiting_on"], [2, "1.2"]) self.assertEqual(blocked[2]["waiting_on"], [1]) # 3 is in review with 1 still open: a ticket already under way is not waiting. self.assertEqual([e["id"] for e in out["in_progress"]], [1, 3]) self.assertFalse(any("waiting_on" in e for e in out["in_progress"])) self.add("story-scaffold-plan.md", plan(1, "done")) self.assertNotIn("waiting_on", {e["id"]: e for e in self.next()["blocked"]}[2]) def test_status_gives_the_id_a_new_ticket_takes(self): self.pricing() self.breakdown_epic() self.assertEqual(json.loads(run("status", str(self.epic)).stdout)["next_id"], 5) self.breakdown_epic(self.BREAKDOWN + '\n[[entry]]\nid = "6a"\ntype = "story"\ntitle = "Lettered"\n') self.assertEqual(json.loads(run("status", str(self.epic)).stdout)["next_id"], 7) self.add("story-unlisted.md", ticket("", 9)) out = json.loads(run("status", str(self.epic)).stdout) self.assertEqual(out["next_id"], 10) out = json.loads(run("status", str(self.initiative)).stdout) self.assertNotIn("next_id", out) self.assertEqual({e["slug"]: e["next_id"] for e in out["epics"]}, {"epic-pricing": 3, "epic-cart": 10}) backlog = self.add_backlog() self.assertEqual(json.loads(run("status", str(backlog)).stdout)["next_id"], 1) self.add("bug-x.md", ticket("draft", 7, kind="bug"), backlog) self.assertEqual(json.loads(run("status", str(backlog)).stdout)["next_id"], 8) def test_a_folders_own_next_id_keeps_a_moved_entrys_id_from_coming_back(self): self.breakdown_epic("next_id = 9\n" + self.BREAKDOWN) self.assertEqual(json.loads(run("status", str(self.epic)).stdout)["next_id"], 9) self.breakdown_epic("next_id = 3\n" + self.BREAKDOWN) self.assertEqual(json.loads(run("status", str(self.epic)).stdout)["next_id"], 5) self.breakdown_epic('next_id = "9"\n' + self.BREAKDOWN) r = run("status", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("`next_id` must be a whole number", r.stderr) def test_next_and_find_carry_the_drift_too(self): self.breakdown_epic() self.add("story-scaffold.md", ticket("draft", 1)) self.add("story-ui-shell.md", ticket("draft", 2, after="[]")) drift = {"after": {"file": [], "entry": [1]}} ready = {r["id"]: r for r in self.next()["ready_to_start"]} self.assertEqual(ready[2]["drift"], drift) self.assertNotIn("drift", ready[1]) self.assertEqual(json.loads(run("find", str(self.epic), "2").stdout)["drift"], drift) def test_an_unknown_holds_a_ticket_until_it_is_settled(self): self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold"\nunknown = "Which host?"')) out = self.next() self.assertEqual(out["ready_to_start"], []) row = {e["id"]: e for e in out["blocked"]}[1] self.assertEqual(row["unknown"], "Which host?") self.assertNotIn("waiting_on", row) # Pulled, the file's Notes hold it. self.assertEqual(run("pull", str(self.epic), "1").returncode, 0) self.assertEqual({e["id"]: e for e in self.next()["blocked"]}[1]["unknown"], "Which host?") leaf = self.epic / "story-scaffold.md" text = leaf.read_text(encoding="utf-8") leaf.write_text(text.replace("- Unknown: Which host?", "- Decision: the EU host."), encoding="utf-8") out = self.next() self.assertEqual([e["id"] for e in out["ready_to_start"]], [1]) self.assertNotIn("unknown", out["ready_to_start"][0]) def test_an_unknown_does_not_keep_a_ticket_from_refining(self): self.breakdown_epic( self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold"\nrefine = true\nunknown = "Which host?"') ) self.assertEqual([e["id"] for e in self.next()["ready_to_refine"]], [1]) # Refined, and the only `Unknown:` left is a template's example in a comment. self.add("story-scaffold.md", ticket("", 1, refined="true") + "\n") self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], [1]) def test_rows_carry_the_checkpoints_an_entry_sets(self): self.breakdown_epic( self.BREAKDOWN.replace( 'title = "Scaffold"', 'title = "Scaffold"\nplan_checkpoint = true\ndone_checkpoint = true' ) ) first, second = self.next()["ready_to_start"][0], self.status_rows(self.epic)[1] self.assertEqual((first["plan_checkpoint"], first["done_checkpoint"]), (True, True)) self.assertNotIn("plan_checkpoint", second) self.assertTrue(json.loads(run("find", str(self.epic), "1").stdout)["done_checkpoint"]) def test_every_row_carries_the_tickets_risk_and_the_leaf_files_wins(self): self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold"\nrisk = "medium"')) rows = {r["id"]: r for r in self.status_rows(self.epic)} self.assertEqual((rows[1]["risk"], rows[2]["risk"]), ("medium", "")) self.assertEqual(self.next()["ready_to_start"][0]["risk"], "medium") self.add("story-scaffold.md", ticket("", 1).replace("risk: low", "risk: high")) self.assertEqual(json.loads(run("find", str(self.epic), "1").stdout)["risk"], "high") def test_a_plan_without_an_assignee_keeps_the_leafs_and_one_with_it_wins(self): self.breakdown_epic() self.add("story-scaffold.md", ticket("", 1, assignee='"ann"')) self.add("story-scaffold-plan.md", plan(1, "in-progress")) self.assertEqual(self.status_rows(self.epic)[0]["assignee"], "ann") self.add("story-scaffold-plan.md", plan(1, "in-progress", assignee="bob")) self.assertEqual(self.status_rows(self.epic)[0]["assignee"], "bob") def mirror(self, items, folder=None): return run("mirror", str(folder or self.epic), stdin=json.dumps(items)) def test_mirror_refuses_the_repo_store(self): self.breakdown_epic() r = self.mirror([{"ref": "1", "tracker_status": "done"}]) self.assertEqual(r.returncode, 2) self.assertIn("no tracker", r.stderr) self.assertEqual(list(self.epic.glob("story-*.md")), []) def test_mirror_pulls_a_leaf_and_writes_only_the_keys_given(self): self.write_store("jira") self.breakdown_epic() r = self.mirror( [ { "ref": 2, "tracker_id": "42", "remote": "https://x.test/42", "tracker_status": "in-progress", "assignee": "ann", "after": [1, "41"], }, {"ref": "scaffold", "tracker_id": "41", "tracker_status": "done"}, ] ) self.assertEqual(r.returncode, 0, r.stderr) out = json.loads(r.stdout) self.assertEqual( [(m["ref"], m["file"], m["pulled"]) for m in out["mirrored"]], [("2", "story-ui-shell.md", True), ("1", "story-scaffold.md", True)], ) self.assertEqual(out["unmatched"], []) text = (self.epic / "story-ui-shell.md").read_text(encoding="utf-8") for line in ( 'tracker_id: "42"', 'remote: "https://x.test/42"', "tracker_status: in-progress", 'assignee: "ann"', 'after: [1, "41"]', ): self.assertIn(f"\n{line}\n", text) self.assertNotIn("\nstatus:", text) rows = {t["id"]: t for t in self.status_rows(self.epic)} self.assertEqual((rows[2]["state"], rows[2]["assignee"], rows[2]["after"]), ("in-progress", "ann", [1])) self.assertEqual(rows[1]["state"], "done") self.assertNotIn("drift", rows[2]) self.assertEqual([e["id"] for e in self.next("--synced")["ready_to_start"]], [3]) def test_mirror_finds_a_ticket_by_its_tracker_id_and_an_empty_value_removes_the_line(self): self.write_store("github") self.breakdown_epic() self.add("story-scaffold.md", ticket("", 1, tracker_id='"41"', assignee='"ann"', tracker_status="backlog")) r = self.mirror([{"tracker_id": "41", "tracker_status": "review", "assignee": ""}]) self.assertEqual(r.returncode, 0, r.stderr) self.assertFalse(json.loads(r.stdout)["mirrored"][0]["pulled"]) text = (self.epic / "story-scaffold.md").read_text(encoding="utf-8") self.assertIn("\ntracker_status: review\n", text) self.assertNotIn("assignee", text) self.assertIn('\ntracker_id: "41"\n', text) def test_a_tracker_id_alone_matches_only_the_ticket_that_carries_it(self): self.write_store("github") self.breakdown_epic(self.BREAKDOWN.replace('title = "Codes"', 'title = "Port SHOP-12 codes"')) self.add("story-scaffold.md", ticket("", 1, tracker_id='"3"')) r = self.mirror([{"tracker_id": 3, "tracker_status": "done"}, {"tracker_id": "SHOP-12", "assignee": "ann"}]) self.assertEqual(r.returncode, 0, r.stderr) out = json.loads(r.stdout) self.assertEqual([m["file"] for m in out["mirrored"]], ["story-scaffold.md"]) self.assertEqual([u["ref"] for u in out["unmatched"]], ["SHOP-12"]) self.assertEqual([p.name for p in self.epic.glob("s*.md")], ["story-scaffold.md"]) self.assertEqual(self.status_rows(self.epic)[0]["state"], "done") self.add("story-ui-shell.md", ticket("", 2, tracker_id='"3"')) r = self.mirror([{"tracker_id": "3", "tracker_status": "backlog"}]) self.assertEqual(r.returncode, 1) self.assertIn("more than one", json.loads(r.stderr)["error"]) def test_mirror_lists_a_ticket_the_tree_does_not_hold_and_writes_the_rest(self): self.write_store("linear") self.breakdown_epic() r = self.mirror([{"ref": "SHOP-99", "tracker_status": "done"}, {"ref": 1, "tracker_status": "done"}]) self.assertEqual(r.returncode, 0, r.stderr) out = json.loads(r.stdout) self.assertEqual([u["ref"] for u in out["unmatched"]], ["SHOP-99"]) self.assertEqual(self.status_rows(self.epic)[0]["state"], "done") def test_a_failed_mirror_still_lists_the_unmatched_tickets(self): self.write_store("linear") self.breakdown_epic() self.add("story-scaffold.md", ticket("", 1, tracker_id='"SHOP-1"')) before = (self.epic / "story-scaffold.md").read_bytes() # A known ticket now waits on a tracker item the tree does not hold yet. r = self.mirror( [{"tracker_id": "SHOP-1", "after": ["SHOP-2"]}, {"tracker_id": "SHOP-2", "tracker_status": "backlog"}] ) self.assertEqual(r.returncode, 1) err = json.loads(r.stderr) self.assertIn("nothing was mirrored", err["error"]) self.assertEqual([u["ref"] for u in err["unmatched"]], ["SHOP-2"]) self.assertEqual((self.epic / "story-scaffold.md").read_bytes(), before) def test_a_tracker_id_on_two_tickets_is_refused_and_a_mirror_that_makes_one_rolls_back(self): self.write_store("linear") self.breakdown_epic() self.add("story-scaffold.md", ticket("", 1, tracker_id='"SHOP-1"')) before = (self.epic / "story-scaffold.md").read_bytes() # The same id again, in another case, would make `after = ["SHOP-1"]` point at whichever was read last. r = self.mirror([{"ref": 2, "tracker_id": "shop-1"}]) self.assertEqual(r.returncode, 1) self.assertIn("more than one ticket", json.loads(r.stderr)["error"]) self.assertEqual((self.epic / "story-scaffold.md").read_bytes(), before) self.assertFalse((self.epic / "story-ui-shell.md").exists()) self.add("story-ui-shell.md", ticket("", 2, after="[1]", tracker_id='"SHOP-1"')) r = run("status", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("tracker_id 'shop-1' is on more than one ticket", json.loads(r.stderr)["error"]) def test_mirror_writes_nothing_when_a_value_is_refused_or_breaks_the_tree(self): self.write_store("jira") self.breakdown_epic() self.add("story-scaffold.md", ticket("", 1)) before = (self.epic / "story-scaffold.md").read_text(encoding="utf-8") for items, message in ( ([{"ref": 1, "tracker_status": "done"}, {"ref": 2, "status": "done"}], "never mirrored"), ([{"ref": 1, "tracker_status": "closed"}], "tracker_status"), ([{"ref": 1, "tracker_status": "done"}, {"ref": 2, "after": ["NOPE-7"]}], "nothing was mirrored"), ([{"ref": 1, "tracker_status": "done"}, {"ref": "1", "assignee": "ann"}], "named twice"), ({"ref": 1}, "JSON array"), ): r = self.mirror(items) self.assertEqual(r.returncode, 1, items) self.assertIn(message, json.loads(r.stderr)["error"]) self.assertEqual((self.epic / "story-scaffold.md").read_text(encoding="utf-8"), before) self.assertFalse((self.epic / "story-ui-shell.md").exists()) r = run("mirror", str(self.epic), stdin="not json") self.assertEqual(r.returncode, 1) def test_mirror_keeps_a_leafs_crlf_line_endings(self): self.write_store("jira") self.breakdown_epic() (self.epic / "story-scaffold.md").write_bytes(ticket("", 1).replace("\n", "\r\n").encode("utf-8")) self.assertEqual(self.mirror([{"ref": 1, "tracker_status": "done"}]).returncode, 0) raw = (self.epic / "story-scaffold.md").read_bytes() self.assertIn(b"\r\ntracker_status: done\r\n", raw) self.assertNotIn(b"\n", raw.replace(b"\r\n", b"")) def test_a_pulled_leaf_matches_its_entry_until_the_entry_changes(self): self.breakdown_epic() for n in ("2", "3"): self.assertEqual(run("pull", str(self.epic), n).returncode, 0) self.assertFalse(any("drift" in r for r in self.status_rows(self.epic))) self.breakdown_epic(self.BREAKDOWN.replace("hitl = true", "hitl = false")) rows = {r["id"]: r for r in self.status_rows(self.epic)} self.assertTrue(rows[3]["drift"]) self.assertTrue(rows[3]["hitl"]) def test_pull_writes_the_leaf_and_only_a_refine_entry_waits_for_refinement(self): self.breakdown_epic( self.BREAKDOWN.replace( 'title = "Scaffold"', 'title = "Scaffold: the cart!"\nverify = "It runs."\nrefine = true' ) ) r = run("pull", str(self.epic), "1") self.assertEqual(r.returncode, 0, r.stderr) self.assertEqual(json.loads(r.stdout), {"file": "story-scaffold-the-cart.md", "refine": True}) text = (self.epic / "story-scaffold-the-cart.md").read_text(encoding="utf-8") self.assertIn("Verify: It runs.", text) self.assertIn("covers: [R1]", text) self.assertEqual(self.files(self.next()["ready_to_refine"]), ["story-scaffold-the-cart.md"]) self.assertEqual(run("pull", str(self.epic), "1").returncode, 1) run("mark", str(self.epic), "1", "done") run("pull", str(self.epic), "2") out = self.next() self.assertEqual(self.files(out["ready_to_start"]), ["story-ui-shell.md", None]) self.assertEqual(out["ready_to_start"][0]["after"], [1]) def test_a_bug_waits_for_refinement_without_refine_set(self): self.breakdown_epic('[[entry]]\nid = 1\ntype = "bug"\ntitle = "Missing criteria"\n') out = self.next() self.assertEqual([t["id"] for t in out["ready_to_refine"]], [1]) self.assertEqual(out["ready_to_start"], []) r = run("pull", str(self.epic), "1") self.assertEqual(r.returncode, 0, r.stderr) self.assertEqual(json.loads(r.stdout), {"file": "bug-missing-criteria.md", "refine": True}) self.assertEqual(self.files(self.next()["ready_to_refine"]), ["bug-missing-criteria.md"]) def test_a_leaf_file_that_says_bug_waits_for_refinement(self): self.breakdown_epic('[[entry]]\nid = 1\ntype = "story"\ntitle = "Was a story"\n') self.add("bug-was-a-story.md", ticket("ready-for-dev", 1, kind="bug")) self.assertEqual(self.files(self.next()["ready_to_refine"]), ["bug-was-a-story.md"]) def test_a_refined_bug_is_ready_to_start(self): self.breakdown_epic('[[entry]]\nid = 1\ntype = "bug"\ntitle = "Refined bug"\n') self.add("bug-refined-bug.md", ticket("ready-for-dev", 1, kind="bug", refined="true")) out = self.next() self.assertEqual(self.files(out["ready_to_start"]), ["bug-refined-bug.md"]) self.assertEqual(out["ready_to_refine"], []) def test_pull_leaves_out_empty_fields_but_always_writes_after_and_hitl(self): self.breakdown_epic() run("pull", str(self.epic), "1") head = (self.epic / "story-scaffold.md").read_text(encoding="utf-8").split("---")[1] self.assertEqual( head, '\nid: 1\ntype: story\ntitle: "Scaffold"\nparent: epic-cart\ncovers: [R1]\nafter: []\nhitl: false\n' ) self.assertEqual(self.next()["ready_to_start"][0]["state"], "backlog") run("pull", str(self.epic), "3") head = (self.epic / "spike-tax-engine.md").read_text(encoding="utf-8").split("---")[1] self.assertIn("\nafter: [1]\n", head) self.assertIn("\nhitl: true\n", head) self.assertNotIn("refined", head) def test_pull_ends_after_the_references(self): self.breakdown_epic() run("pull", str(self.epic), "1") text = (self.epic / "story-scaffold.md").read_text(encoding="utf-8") self.assertTrue( text.endswith("## References\n\n- parent — out/initiative-checkout/epic-cart/epic-cart.md\n"), text ) def test_find_by_cross_ref_id_file_tracker_id_and_title(self): pricing = self.pricing() self.breakdown_epic() self.add("story-contract.md", ticket("done", 1, tracker_id='"PRICE-1"'), pricing) run("pull", str(self.epic), "1") def find(folder, ref): r = run("find", str(folder), ref) self.assertEqual(r.returncode, 0, r.stderr) return json.loads(r.stdout) hit = find(self.initiative, "2.1") self.assertEqual((hit["folder"], hit["id"], hit["file"]), ("epic-cart", 1, "story-scaffold.md")) self.assertEqual(hit["story_file"], str((self.epic / "story-scaffold.md").resolve())) self.assertNotIn("path", hit) self.assertEqual(find(self.epic, "3")["title"], "Tax engine?") self.assertIsNone(find(self.epic, "3")["story_file"]) self.assertEqual(find(self.epic, "1.1")["file"], "story-contract.md") self.assertEqual(find(self.epic, "price-1")["file"], "story-contract.md") self.assertEqual(find(self.initiative, "story-scaffold")["folder"], "epic-cart") self.assertEqual(find(self.initiative, "ui shell")["id"], 2) self.assertEqual(find(self.epic, "codes")["id"], 4) r = run("find", str(self.initiative), "in") self.assertEqual(r.returncode, 1) self.assertIn("more than one", r.stderr) r = run("find", str(self.initiative), "9.1") self.assertEqual(r.returncode, 1) self.assertIn("no ticket matches", r.stderr) def test_pull_refuses_a_file_name_already_taken(self): self.breakdown_epic() self.add("story-ui-shell.md", "notes, not a ticket\n") r = run("pull", str(self.epic), "2") self.assertEqual(r.returncode, 1) self.assertIn("exists already", r.stderr) def test_pull_writes_references_and_notes(self): self.breakdown_epic( self.BREAKDOWN.replace( 'title = "Scaffold"', 'title = "Scaffold"\nunknown = "Which host?"\nreferences = ["SPINE.md#ad-8"]\nnotes = ["Reuse the mailer."]', ) ) self.assertEqual(run("pull", str(self.epic), "1").returncode, 0) text = (self.epic / "story-scaffold.md").read_text(encoding="utf-8") parent = (self.epic / "epic-cart.md").resolve().relative_to(self.root.resolve()).as_posix() self.assertIn(f"## References\n\n- parent — {parent}\n- SPINE.md#ad-8\n", text) self.assertIn("## Notes\n\n- Unknown: Which host?\n- Reuse the mailer.\n", text) def test_a_quoted_title_survives_the_pull(self): self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', "title = 'Say \"hi\"'")) self.assertEqual(run("pull", str(self.epic), "1").returncode, 0) self.assertEqual(self.next()["ready_to_start"][0]["title"], 'Say "hi"') def test_references_must_be_a_list(self): self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold"\nreferences = "SPINE.md"')) r = run("status", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("`references` must be a list", r.stderr) def test_dropped_prerequisite_still_blocks(self): self.breakdown_epic() self.add("story-scaffold.md", ticket("dropped", 1)) self.add("story-ui-shell.md", ticket("", 2, after="[1]")) out = self.next() self.assertEqual(self.files(out["blocked"])[0], "story-ui-shell.md") self.assertEqual(out["ready_to_refine"], []) self.assertEqual(out["ready_to_start"], []) def test_malformed_breakdown_errors(self): for text, message in ( (self.BREAKDOWN.replace("id = 3", "id = 2"), "two entries with id 2"), (self.BREAKDOWN.replace("after = [2, 3]", "after = [2, 9]"), "names no entry in epic-cart"), (self.BREAKDOWN.replace('type = "spike"', 'type = "task"'), "is not one of"), (self.BREAKDOWN.replace("id = 4", 'id = "3"'), "two entries with id 3"), (self.BREAKDOWN.replace("id = 4", 'id = "4-b"'), "needs an `id`"), (self.BREAKDOWN.replace("id = 4", 'id = "true"'), "needs an `id`"), (self.BREAKDOWN.replace("id = 4\n", ""), "needs an `id`"), ("[[entry]\nid = ", "tickets.toml"), ('[entry]\nid = 1\ntype = "story"\n', "[[entry]]"), (self.BREAKDOWN.replace('covers = ["R4"]', "covers = 1"), "must be a list"), ): self.breakdown_epic(text) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1, text) self.assertIn(message, json.loads(r.stderr)["error"]) def test_ticket_waits_on_an_entry_in_another_epic(self): pricing = self.pricing() self.breakdown_epic( self.BREAKDOWN.replace('title = "UI shell"\nafter = [1]', 'title = "UI shell"\nafter = [1, "1.1"]') ) self.add("story-scaffold.md", ticket("done", 1)) out = self.next() self.assertEqual([e["id"] for e in out["ready_to_start"]], [3]) self.assertEqual(out["blocked"][0]["after"], [1, "1.1"]) self.add("story-contract.md", ticket("done", 1), pricing) self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], [2, 3]) def test_whole_epic_prerequisite_and_epic_gate(self): pricing = self.pricing() self.add("story-scaffold.md", ticket("draft", 1, after="[epic-pricing]")) self.assertEqual(self.files(self.next()["blocked"]), ["story-scaffold.md"]) (pricing / "epic-pricing.md").write_text("---\ntype: epic\nstatus: done\n---\n") self.assertEqual(self.files(self.next()["ready_to_refine"]), ["story-scaffold.md"]) gated = self.add_epic("epic-tax", after="[epic-cart]") self.add("story-rates.md", ticket("draft", 1), gated) out = json.loads(run("next", str(gated)).stdout) self.assertEqual(out["blocked"][0]["after"], []) self.assertEqual(out["blocked"][0]["gated_by"], ["epic-cart"]) def test_gate_deadlock_with_a_ticketless_epic_is_a_cycle(self): self.add_epic("epic-tax", after="[epic-cart]") self.add("story-scaffold.md", ticket("draft", 1, after="[epic-tax]")) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("cycle", r.stderr) def test_tracker_ids_resolve_across_epics_and_duplicates_collapse(self): pricing = self.pricing() self.add("story-contract.md", ticket("done", 1, tracker_id='"PRICE-1"'), pricing) self.add("story-scaffold.md", ticket("draft", 1, after="[PRICE-1, 1.1, 1.01]")) out = self.next() self.assertEqual(out["ready_to_refine"][0]["after"], ["1.1"]) def test_malformed_files_name_their_folder(self): pricing = self.pricing() self.seed() cases = ( (ticket("todo", 1), "epic-pricing/story-contract.md"), (ticket("draft", 1).replace("after: []", "after:\n - 1"), "must be inline"), (ticket("draft", 1).replace("---\n\n# x", "\n# x"), "does not close"), ) for text, message in cases: self.add("story-contract.md", text, pricing) self.assertIn(message, json.loads(run("next", str(self.epic)).stderr)["error"]) self.add("story-contract.md", ticket("draft", 1), pricing) (pricing / "epic-pricing.md").write_text("---\ntype: epic\nstatus: Finished\n---\n") self.assertIn("epic-pricing/epic-pricing.md", json.loads(run("next", str(self.epic)).stderr)["error"]) def test_cross_epic_unknown_target_and_cycle_error(self): pricing = self.pricing() for ref, message in (("9.1", "names no epic id"), ("1.9", "names no entry"), ("epic-missing", "names no epic")): self.add("story-scaffold.md", ticket("draft", 1, after=f"[{ref}]")) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn(message, json.loads(r.stderr)["error"]) self.add("story-scaffold.md", ticket("draft", 1, after="[1.2]")) self.add("story-contract.md", ticket("draft", 1, after="[2.1]"), pricing) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("cycle", r.stderr) def test_initiative_view_spans_epics_in_file_order_and_reports_unpinned_after(self): self.pricing() self.breakdown_epic() (self.initiative / "tickets.toml").write_text( '[[epic]]\nid = 1\nslug = "epic-pricing"\n\n' '[[epic]]\nid = 2\nslug = "epic-cart"\nafter = [{ epic = 1, needs = "the pricing contract" }]\n' ) out = json.loads(run("next", str(self.initiative)).stdout) self.assertEqual([(e["epic"], e["id"]) for e in out["ready_to_start"]], [("epic-pricing", 1), ("epic-cart", 1)]) self.assertEqual( out["unpinned_after"], [{"epic": "epic-cart", "after": "epic-pricing", "needs": "the pricing contract"}] ) self.breakdown_epic( self.BREAKDOWN.replace('title = "UI shell"\nafter = [1]', 'title = "UI shell"\nafter = [1, "1.1"]') ) status = json.loads(run("status", str(self.initiative)).stdout) self.assertEqual(status["unpinned_after"], []) self.assertEqual(status["counts"], {"total": 6, "planned": 6}) self.assertEqual([(e["slug"], e["id"]) for e in status["epics"]], [("epic-pricing", 1), ("epic-cart", 2)]) self.assertEqual(status["tickets"][0]["blocks"], [2, "2.2"]) self.assertEqual(status["longest_remaining_chain"][1:], ["2.2", "2.4"]) (self.initiative / "tickets.toml").write_text( '[[epic]]\nid = 1\nslug = "epic-pricing"\n\n[[epic]]\nid = 2\nslug = "epic-cart"\nafter = [{ epic = "epic-x" }]\n' ) r = run("status", str(self.initiative)) self.assertEqual(r.returncode, 1) self.assertIn("no epic listed", r.stderr) (self.initiative / "tickets.toml").write_text( '[[epic]]\nid = 2\nslug = "epic-cart"\n\n[[epic]]\nid = 2\nslug = "epic-pricing"\n' ) self.assertIn("two epics with id 2", run("status", str(self.initiative)).stderr) def test_status_shows_the_declared_epic_graph_before_and_after_inception(self): self.pricing() (self.initiative / "tickets.toml").write_text( '[[epic]]\nid = 1\nslug = "epic-pricing"\n\n' '[[epic]]\nid = 2\nslug = "epic-cart"\nafter = [{ epic = 1, needs = "the pricing contract" }]\n' ) for _ in range(2): status = json.loads(run("status", str(self.initiative)).stdout) self.assertEqual( [(e["slug"], e["after"], e["gated_by"]) for e in status["epics"]], [ ("epic-pricing", [], []), ("epic-cart", [{"epic": "epic-pricing", "needs": "the pricing contract"}], []), ], ) self.breakdown_epic() def test_an_undeclared_or_backward_cross_epic_after_is_reported(self): self.pricing() self.breakdown_epic() (self.initiative / "tickets.toml").write_text( '[[epic]]\nid = 1\nslug = "epic-pricing"\n\n[[epic]]\nid = 2\nslug = "epic-cart"\n' ) (self.initiative / "epic-pricing" / "tickets.toml").write_text( '[[entry]]\nid = 1\ntype = "story"\ntitle = "Pricing contract"\nafter = ["2.1"]\n' ) for command, folder in (("status", self.initiative), ("next", self.initiative), ("status", self.epic)): out = json.loads(run(command, str(folder)).stdout) if folder == self.epic: self.assertEqual((out["undeclared_after"], out["order_conflict"]), ([], [])) continue self.assertEqual( out["undeclared_after"], [{"epic": "epic-pricing", "after": "epic-cart", "ref": "1.1", "names": "2.1"}] ) self.assertEqual(out["order_conflict"], [{"epic": "epic-pricing", "after": "epic-cart"}]) (self.initiative / "tickets.toml").write_text( '[[epic]]\nid = 1\nslug = "epic-pricing"\nafter = [{ epic = 2, needs = "the cart" }]\n\n' '[[epic]]\nid = 2\nslug = "epic-cart"\n' ) out = json.loads(run("status", str(self.initiative)).stdout) self.assertEqual(out["undeclared_after"], []) self.assertEqual(out["order_conflict"], [{"epic": "epic-pricing", "after": "epic-cart"}]) def test_after_naming_a_backlog_file_says_to_move_it_into_the_epic(self): self.add("story-thing.md", ticket("", kind="story"), self.add_backlog()) self.add("story-scaffold.md", ticket("draft", 1, after="[story-thing]")) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("move a backlog ticket into the epic as an entry", json.loads(r.stderr)["error"]) self.breakdown_epic() self.add("story-scaffold.md", ticket("draft", 1, after="[story-ui-shell]")) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("use the entry's id", json.loads(r.stderr)["error"]) def test_an_epic_gate_counts_once_and_its_order_before_inception(self): self.pricing() (self.initiative / "tickets.toml").write_text( '[[epic]]\nid = 1\nslug = "epic-cart"\n\n[[epic]]\nid = 2\nslug = "epic-pricing"\n' ) (self.epic / "epic-cart.md").write_text("---\ntype: epic\nafter: [epic-pricing]\n---\n# epic-cart\n") out = json.loads(run("status", str(self.initiative)).stdout) self.assertEqual( (out["undeclared_after"], out["order_conflict"]), ([], [{"epic": "epic-cart", "after": "epic-pricing"}]) ) self.breakdown_epic('[[entry]]\nid = 1\ntype = "story"\ntitle = "Scaffold"\nafter = ["epic-pricing"]\n') out = json.loads(run("status", str(self.initiative)).stdout) self.assertEqual( out["undeclared_after"], [{"epic": "epic-cart", "after": "epic-pricing", "ref": "1.1", "names": "epic-pricing"}], ) self.assertEqual(out["order_conflict"], [{"epic": "epic-cart", "after": "epic-pricing"}]) def test_epics_need_an_id_a_slug_and_a_valid_after(self): self.pricing() toml = self.initiative / "tickets.toml" good = toml.read_text(encoding="utf-8") for bad, message in ( (good.replace("id = 2\n", ""), "needs an `id`"), (good.replace('slug = "epic-cart"', 'slug = "epic-pricing"'), "two epics with slug"), (good.replace('slug = "epic-cart"\n', ""), "needs a `slug`"), (good + "after = [{ epic = true }]\n", "epic = "), ): toml.write_text(bad, encoding="utf-8") r = run("status", str(self.initiative)) self.assertEqual(r.returncode, 1, r.stdout) self.assertIn(message, r.stderr) def test_find_prefers_the_folder_asked_about_for_a_file_name(self): pricing = self.pricing() self.add("story-scaffold.md", ticket("", 1), pricing) self.breakdown_epic() run("pull", str(self.epic), "1") r = run("find", str(pricing), "story-scaffold") self.assertEqual(r.returncode, 0, r.stderr) self.assertEqual(json.loads(r.stdout)["folder"], "epic-pricing") r = run("find", str(self.initiative), "story-scaffold") self.assertEqual(r.returncode, 1) self.assertIn("more than one", r.stderr) def test_a_numeric_tracker_id_is_never_a_sibling_id(self): self.seed(s1="done") self.add("story-ui-shell.md", ticket("done", 2, tracker_id='"47"')) self.add("story-tracer.md", ticket("", 3, after='["47"]')) self.assertIn("story-tracer.md", self.files(self.next()["ready_to_refine"])) def test_a_quoted_number_is_a_tracker_id_and_a_bare_one_a_sibling(self): self.add("story-scaffold.md", ticket("done", 1, tracker_id="101")) self.add("story-ui-shell.md", ticket("draft", 2, after='["101"]')) self.assertEqual(self.files(self.next()["ready_to_refine"]), ["story-ui-shell.md"]) self.add("story-ui-shell.md", ticket("draft", 2, after="[101]")) self.assertIn("names no entry", run("next", str(self.epic)).stderr) def test_an_id_takes_letters_and_digits_wherever_a_number_goes(self): pricing = self.add_epic("epic-pricing") (pricing / "tickets.toml").write_text( '[[entry]]\nid = "6a"\ntype = "story"\ntitle = "Pricing contract"\n\n' '[[entry]]\nid = 7\ntype = "story"\ntitle = "Pricing rules"\nafter = ["6a"]\n' ) (self.initiative / "tickets.toml").write_text( '[[epic]]\nid = "1b"\nslug = "epic-pricing"\n\n' '[[epic]]\nid = 2\nslug = "epic-cart"\nafter = [{ epic = "1b", needs = "prices" }]\n' ) self.breakdown_epic('[[entry]]\nid = "2a"\ntype = "story"\ntitle = "Totals for 6a"\nafter = ["1b.6a"]\n') status = json.loads(run("status", str(self.initiative)).stdout) rows = {r["ref"]: r for r in status["tickets"]} self.assertEqual(list(rows), ["1b.6a", "1b.7", "2.2a"]) self.assertEqual((rows["1b.7"]["after"], rows["2.2a"]["after"]), (["6a"], ["1b.6a"])) self.assertEqual(status["unpinned_after"] + status["undeclared_after"], []) self.assertEqual(run("pull", str(pricing), "7").returncode, 0) self.assertIn("id: 7\n", (pricing / "story-pricing-rules.md").read_text(encoding="utf-8")) self.assertIn("after: [6a]\n", (pricing / "story-pricing-rules.md").read_text(encoding="utf-8")) self.assertEqual(run("pull", str(pricing), "6a").returncode, 0) self.assertIn("id: 6a\n", (pricing / "story-pricing-contract.md").read_text(encoding="utf-8")) self.assertNotIn("drift", json.dumps(self.status_rows(pricing))) self.assertEqual([e["id"] for e in self.next()["blocked"]], ["2a"]) self.add("story-pricing-contract-plan.md", plan("6a", "done"), pricing) self.assertEqual([e["id"] for e in self.next()["ready_to_start"]], ["2a"]) # A bare lettered id names its ticket from any folder, before the title that contains it. for folder, ref in ((self.initiative, "1b.6a"), (pricing, "6a"), (self.initiative, "6a"), (self.epic, "6a")): found = json.loads(run("find", str(folder), ref).stdout) self.assertEqual((found["id"], found["state"]), ("6a", "done")) self.assertEqual(run("mark", str(self.epic), "2a", "in-review").returncode, 0) self.assertIn("ticket: 2a\n", (self.epic / "story-totals-for-6a-plan.md").read_text(encoding="utf-8")) self.assertEqual(self.status_rows(self.epic)[0]["state"], "review") def test_two_files_sharing_an_id_error(self): self.add("story-a.md", ticket("done", 1)) self.add("story-b.md", ticket("draft", 1)) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("share the id 1", json.loads(r.stderr)["error"]) def test_malformed_input_returns_json_error(self): (self.epic / "story-bad.md").write_bytes(ticket("draft", 1).encode().replace(b"# x", b"\xff")) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1, r.stderr) self.assertIn("error", json.loads(r.stderr)) (self.epic / "story-bad.md").unlink() self.seed() (self.root / "_bmad" / "custom" / "ticketing-store-config.toml").write_text("[tickets\nstore = ") r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1, r.stderr) self.assertIn("error", json.loads(r.stderr)) def test_leaf_without_id_sorts_last_and_container_file_is_ignored(self): self.seed() self.add("bug-stray.md", ticket("", kind="bug")) out = json.loads(run("status", str(self.epic)).stdout) self.assertEqual(out["tickets"][-1]["file"], "bug-stray.md") self.assertIsNone(out["tickets"][-1]["id"]) self.assertEqual(out["counts"]["total"], 6) def mark(self, *args): r = run("mark", *args) self.assertEqual(r.returncode, 0, r.stderr) return json.loads(r.stdout) def test_mark_writes_a_plan_beside_the_leaf_and_never_the_leaf(self): self.seed() leaf = (self.epic / "story-scaffold.md").read_text(encoding="utf-8") out = self.mark(str(self.epic), "1", "in-progress", "--assignee", "ann") plan_file = self.epic / "story-scaffold-plan.md" self.assertEqual( out, { "plan": str(plan_file.resolve()), "created": True, "status": "in-progress", "assignee": "ann", "blocked_at": "", "blocked_reason": "", }, ) self.assertEqual((self.epic / "story-scaffold.md").read_text(encoding="utf-8"), leaf) self.assertEqual( plan_file.read_text(encoding="utf-8"), '---\ntitle: "x"\nticket: 1\nstatus: in-progress\nassignee: "ann"\n---\n', ) self.assertEqual(self.files(self.next()["in_progress"]), ["story-scaffold.md"]) def test_mark_clears_the_leafs_blocking_fields_and_takes_a_literal_assignee(self): path = self.epic / "story-scaffold.md" path.write_text(ticket("", 1, blocked_at="2026-09-05", blocked_reason="legal")) out = self.mark(str(self.epic), "1", "draft", "--assignee", "\\1") self.assertEqual(out["assignee"], "\\1") text = (self.epic / "story-scaffold-plan.md").read_text(encoding="utf-8") self.assertNotIn("blocked_at", text) self.assertIn('assignee: "\\\\1"\n', text) self.assertIn("blocked_reason", path.read_text(encoding="utf-8")) self.assertEqual(self.files(self.next()["ready_to_refine"]), ["story-scaffold.md"]) def test_mark_on_a_pulled_leaf_keeps_the_leafs_assignee_in_the_new_plan(self): self.breakdown_epic() run("pull", str(self.epic), "1") self.add( "story-scaffold.md", (self.epic / "story-scaffold.md").read_text().replace("---\n\n", 'assignee: "bob"\n---\n\n', 1), ) self.mark(str(self.epic), "1", "ready-for-dev") text = (self.epic / "story-scaffold-plan.md").read_text(encoding="utf-8") self.assertEqual(text, '---\ntitle: "Scaffold"\nticket: 1\nstatus: ready-for-dev\nassignee: "bob"\n---\n') self.assertEqual( [(e["file"], e["assignee"]) for e in self.next()["ready_to_start"]], [("story-scaffold.md", "bob")] ) def test_find_on_an_entry_with_nothing_yet(self): self.breakdown_epic( self.BREAKDOWN.replace( 'title = "UI shell"', 'title = "UI shell"\ndescription = "The page frame."\nverify = "It renders."\n' 'unknown = "Which grid?"\nreferences = ["SPINE.md#ad-8"]\nnotes = ["Reuse the header."]', ) ) r = run("find", str(self.epic), "2") self.assertEqual(r.returncode, 0, r.stderr) hit = json.loads(r.stdout) epic = self.epic.resolve() self.assertEqual((hit["id"], hit["title"], hit["state"], hit["after"]), (2, "UI shell", "planned", [1])) self.assertEqual( {k: hit[k] for k in ("folder", "description", "verify", "references", "notes", "unknown")}, { "folder": "epic-cart", "description": "The page frame.", "verify": "It renders.", "references": ["SPINE.md#ad-8"], "notes": ["Reuse the header."], "unknown": "Which grid?", }, ) self.assertIsNone(hit["story_file"]) self.assertEqual(hit["plan"], str(epic / "story-ui-shell-plan.md")) self.assertFalse(Path(hit["plan"]).exists()) self.assertEqual(hit["epic_file"], str(epic / "epic-cart.md")) def test_find_names_a_joined_plan_whatever_its_name(self): self.breakdown_epic() self.add("custom-name.md", plan(2, "in-progress")) hit = json.loads(run("find", str(self.epic), "2").stdout) self.assertEqual(hit["plan"], str(self.epic.resolve() / "custom-name.md")) self.assertEqual(hit["status"], "in-progress") def test_find_on_a_refined_entry_names_its_story_file_and_the_plan_after_it(self): self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold, renamed"')) self.add("story-scaffold.md", ticket("", 1)) hit = json.loads(run("find", str(self.epic), "1").stdout) self.assertEqual(hit["story_file"], str(self.epic.resolve() / "story-scaffold.md")) self.assertEqual(hit["plan"], str(self.epic.resolve() / "story-scaffold-plan.md")) def test_find_on_a_pulled_entry_leaves_the_text_to_the_file(self): self.breakdown_epic( self.BREAKDOWN.replace( 'title = "Scaffold"', 'title = "Scaffold"\ndescription = "The service."\nverify = "It runs."\n' 'unknown = "Which host?"\nreferences = ["SPINE.md#ad-8"]\nnotes = ["Reuse the mailer."]', ) ) self.assertEqual(run("pull", str(self.epic), "1").returncode, 0) leaf = self.epic / "story-scaffold.md" leaf.write_text(leaf.read_text(encoding="utf-8").replace("Which host?", "Which region?"), encoding="utf-8") hit = json.loads(run("find", str(self.epic), "1").stdout) self.assertEqual([hit[k] for k in ("description", "verify", "references", "notes")], ["", "", [], []]) self.assertEqual(hit["unknown"], "Which region?") self.assertEqual(hit["story_file"], str(self.epic.resolve() / "story-scaffold.md")) def test_find_names_the_initiative_file_for_a_leaf_directly_under_it(self): solo = self.root / "out" / "initiative-solo" solo.mkdir() (solo / "initiative-solo.md").write_text("---\ntype: initiative\n---\n# Solo\n") (solo / "tickets.toml").write_text('[[entry]]\nid = 1\ntype = "story"\ntitle = "Only"\n') hit = json.loads(run("find", str(solo), "1").stdout) self.assertEqual(hit["epic_file"], str(solo.resolve() / "initiative-solo.md")) self.assertIsNone(hit["story_file"]) def test_find_and_mark_in_a_backlog(self): backlog = self.add_backlog() self.add("bug-x.md", ticket("draft", kind="bug"), backlog) r = run("find", str(backlog), "bug-x") self.assertEqual(r.returncode, 0, r.stderr) hit = json.loads(r.stdout) self.assertIsNone(hit["epic_file"]) self.assertEqual( [hit[k] for k in ("description", "verify", "references", "notes", "unknown")], ["", "", [], [], ""] ) self.assertEqual(hit["plan"], str(backlog.resolve() / "bug-x-plan.md")) self.mark(str(backlog), "bug-x", "in-review") self.assertIn('\nticket: "bug-x"\n', (backlog / "bug-x-plan.md").read_text(encoding="utf-8")) rows = self.status_rows(backlog) self.assertEqual([(t["file"], t["state"]) for t in rows], [("bug-x.md", "review")]) def test_mark_done_on_a_plan_only_entry_clears_blocking_and_keeps_the_plan(self): self.breakdown_epic() path = self.epic / "story-ui-shell-plan.md" self.add(path.name, plan(2, "blocked", assignee="ann", blocked_at="2026-09-05", blocked_reason="legal")) before = path.read_text(encoding="utf-8") out = self.mark(str(self.epic), "2", "done") self.assertEqual((out["created"], out["status"], out["assignee"]), (False, "done", "ann")) text = path.read_text(encoding="utf-8") self.assertIn("\nstatus: done\n", text) self.assertNotIn("blocked_", text) expected = [ line for line in before.splitlines() if not line.startswith(("status:", "blocked_at:", "blocked_reason:")) ] self.assertEqual([line for line in text.splitlines() if not line.startswith("status:")], expected) rows = self.status_rows(self.epic) self.assertEqual(rows[1]["state"], "done") out = self.next() self.assertNotIn(2, [e["id"] for group in ("ready_to_start", "in_progress", "blocked") for e in out[group]]) def test_mark_on_a_build_plan_keeps_its_body_and_other_frontmatter(self): self.breakdown_epic() path = self.epic / "story-scaffold-plan.md" text = plan(1, "in-progress").replace("# x\n", "# x\n\nbody: stays\n") + "## Code Review\n" self.add(path.name, text) self.mark(str(self.epic), "1", "in-review", "--assignee", "ann") after = path.read_text(encoding="utf-8") self.assertEqual( after, text.replace( "status: 'in-progress' # draft | ready-for-dev | in-progress | in-review | built | done", "status: in-review", ).replace("\n---\n\n# x", '\nassignee: "ann"\n---\n\n# x'), ) def test_a_blocked_reason_with_colons_and_quotes_reads_back_exactly(self): self.breakdown_epic() for reason in ('waiting on legal: "terms"', "it's 'odd' \\ and: # not a comment # still not", "Café ✓
next"): out = self.mark(str(self.epic), "2", "blocked", "--blocked", reason) self.assertEqual(out["blocked_reason"], reason) row = self.next()["blocked"][0] self.assertEqual((row["id"], row["status"], row["blocked_reason"]), (2, "blocked", reason)) def test_mark_with_no_plan_creates_one_that_next_reads(self): self.breakdown_epic() self.mark(str(self.epic), "2", "ready-for-dev", "--assignee", "ann") self.assertEqual( (self.epic / "story-ui-shell-plan.md").read_text(encoding="utf-8"), '---\ntitle: "UI shell"\nticket: 2\nstatus: ready-for-dev\nassignee: "ann"\n---\n', ) self.assertEqual( [(e["id"], e["state"], e["status"], e["assignee"]) for e in self.next()["blocked"]][0], (2, "backlog", "ready-for-dev", "ann"), ) def test_mark_blocked_writes_the_design_notes_plan(self): self.breakdown_epic() self.mark(str(self.epic), "UI shell", "blocked", "--blocked", 'waiting on legal: "terms"') today = date.today().isoformat() self.assertEqual( (self.epic / "story-ui-shell-plan.md").read_text(encoding="utf-8"), f'---\ntitle: "UI shell"\nticket: 2\nstatus: blocked\nblocked_at: "{today}"\n' 'blocked_reason: "waiting on legal: \\"terms\\""\n---\n', ) def test_mark_refuses_an_empty_ref_and_never_matches_digits_against_titles(self): self.pricing() self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Phase 2"')) r = run("mark", str(self.initiative), "2", "done") self.assertEqual(r.returncode, 1, r.stdout) self.assertIn("no ticket matches", r.stderr) for ref in ("", " "): r = run("mark", str(self.epic), ref, "done") self.assertEqual(r.returncode, 1, r.stdout) self.assertIn("empty", r.stderr) self.assertEqual(sorted(p.name for p in self.initiative.rglob("*-plan.md")), []) @unittest.skipIf(sys.platform == "win32", "Windows passes arguments as Unicode, so none is undecodable") def test_mark_with_an_undecodable_value_leaves_the_plan_unchanged(self): self.breakdown_epic() path = self.epic / "story-scaffold-plan.md" self.add(path.name, plan(1, "in-progress")) before = path.read_bytes() r = subprocess.run( [sys.executable, str(SCRIPT), "mark", str(self.epic), "1", "blocked", "--blocked", b"\xff"], capture_output=True, check=False, ) self.assertNotEqual(r.returncode, 0, r.stdout) self.assertEqual(path.read_bytes(), before) def test_mark_refuses_a_plan_name_taken_by_another_file(self): self.breakdown_epic() for text in (plan(3), "# notes, not a plan\n"): self.add("story-ui-shell-plan.md", text) r = run("mark", str(self.epic), "2", "done") self.assertEqual(r.returncode, 1, r.stdout) self.assertIn("story-ui-shell-plan.md", json.loads(r.stderr)["error"]) self.assertEqual((self.epic / "story-ui-shell-plan.md").read_text(encoding="utf-8"), text) def test_project_root_flag_finds_the_store_for_tickets_outside_the_project(self): self.write_store("jira") outside = tempfile.TemporaryDirectory() self.addCleanup(outside.cleanup) folder = Path(outside.name) / "epic-cart" folder.mkdir() (folder / "story-scaffold.md").write_text(ticket("", 1)) self.assertEqual(json.loads(run("next", str(folder)).stdout)["store"], "repo") r = run("--project-root", str(self.root), "next", str(folder)) self.assertEqual(r.returncode, 2) r = run("--project-root", str(self.root), "mark", str(folder), "1", "done") self.assertEqual(r.returncode, 2) def test_mark_refuses_on_tracker_store(self): self.write_store("jira") self.seed() r = run("mark", str(self.epic), "1", "done") self.assertEqual(r.returncode, 2) self.assertIn("write verb", r.stderr) def test_next_on_tracker_store_needs_synced_flag(self): self.write_store("linear") self.seed() r = run("next", str(self.epic)) self.assertEqual(r.returncode, 2) self.assertIn("sync", r.stderr) self.assertEqual(self.next("--synced")["store"], "linear") def test_cycle_unknown_prerequisite_and_bad_status_are_errors(self): self.seed() self.add("story-scaffold.md", ticket("draft", 1, after="[3]")) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("cycle", r.stderr) self.add("story-scaffold.md", ticket("draft", 1, after="[9]")) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("names no entry in epic-cart", r.stderr) self.add("story-scaffold.md", ticket("draft", 1, after='["9"]')) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("matches no ticket", r.stderr) self.add("story-scaffold.md", ticket("backlog", 1)) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("status 'backlog'", r.stderr) self.add("story-scaffold.md", ticket("", 1)) (self.epic / "epic-cart.md").write_text("---\ntype: epic\nstatus: backlog\n---\n") r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1) self.assertIn("in-progress, done, dropped", r.stderr) def test_entries_with_no_files_start_without_a_pull(self): self.breakdown_epic() self.add("old-spec.md", "---\ntype: 'feature'\nstatus: 'done'\nsteps:\n - a\n---\n") out = self.next() self.assertNotIn("to_pull", out) self.assertEqual([(e["id"], e["state"], e["file"]) for e in out["ready_to_start"]], [(1, "planned", None)]) self.breakdown_epic(self.BREAKDOWN.replace('title = "Scaffold"', 'title = "Scaffold"\nrefine = true')) out = self.next() self.assertEqual([e["id"] for e in out["ready_to_refine"]], [1]) self.assertEqual(out["ready_to_start"], []) def test_a_plan_sets_the_state_with_or_without_a_story_file(self): self.breakdown_epic() self.add("story-scaffold-plan.md", plan(1, "done")) self.add("story-ui-shell-plan.md", plan(2, "in-review", assignee="ann")) for _ in range(2): out = self.next() self.assertEqual([(e["id"], e["state"], e["assignee"]) for e in out["in_progress"]], [(2, "review", "ann")]) self.assertEqual([e["id"] for e in out["ready_to_start"]], [3]) rows = self.status_rows(self.epic) self.assertEqual([r["state"] for r in rows], ["done", "review", "planned", "planned"]) self.add("story-ui-shell.md", ticket("draft", 2, after="[1]", refined="true", assignee='"bob"')) self.assertEqual(out["in_progress"][0]["file"], "story-ui-shell.md") self.assertEqual(out["in_progress"][0]["status"], "in-review") self.add("story-ui-shell.md", ticket("draft", 2, after="[1]", tracker_status="done")) self.assertEqual(json.loads(run("status", str(self.epic)).stdout)["tickets"][1]["state"], "done") def test_a_built_plan_reads_as_review_and_mark_accepts_built(self): self.breakdown_epic() self.add("story-scaffold-plan.md", plan(1, "built")) out = self.next() self.assertEqual([(e["id"], e["status"], e["state"]) for e in out["in_progress"]], [(1, "built", "review")]) self.assertEqual([e["id"] for e in out["ready_to_start"]], [2, 3]) self.assertEqual(self.mark(str(self.epic), "2", "built")["status"], "built") self.assertEqual(self.status_rows(self.epic)[1]["state"], "review") def test_a_plan_with_blocked_at_blocks_its_entry(self): self.breakdown_epic() self.add("story-scaffold-plan.md", plan(1, "in-progress", blocked_at="2026-09-05", blocked_reason="legal")) out = self.next() self.assertEqual([(e["id"], e["blocked_reason"]) for e in out["blocked"]][0], (1, "legal")) self.assertEqual(out["in_progress"], []) def test_a_leaf_with_blocked_at_and_no_plan_is_blocked_with_its_reason(self): self.add( "story-scaffold.md", ticket("in-progress", 1, blocked_at="2026-09-05", blocked_reason="legal"), ) out = self.next() self.assertEqual([(e["file"], e["blocked_reason"]) for e in out["blocked"]], [("story-scaffold.md", "legal")]) self.assertEqual(out["in_progress"], []) def test_a_plan_leaving_out_a_field_blanks_the_leafs_value(self): self.breakdown_epic() self.add("story-scaffold-plan.md", plan(1, "done")) self.add( "story-ui-shell.md", ticket("in-review", 2, after="[1]", blocked_at="2026-09-05", blocked_reason="legal"), ) self.add("story-ui-shell-plan.md", plan(2)) self.assertNotIn(2, [e["id"] for e in self.next()["blocked"]]) row = self.status_rows(self.epic)[1] self.assertEqual((row["status"], row["state"], row["blocked_reason"]), ("", "backlog", "")) def test_a_quoted_numeric_ticket_joins_its_entry_and_a_doubled_quote_reads_as_one(self): self.breakdown_epic() self.add("story-ui-shell-plan.md", plan(2, "blocked", blocked_reason="can''t")) blocked = self.next()["blocked"] self.assertEqual([(e["id"], e["status"], e["blocked_reason"]) for e in blocked][0], (2, "blocked", "can't")) def test_a_backlog_plan_joins_its_leaf_by_stem_and_a_leaf_without_a_plan_keeps_its_status(self): backlog = self.add_backlog() self.add("bug-x.md", ticket("draft", kind="bug"), backlog) self.add("bug-x-plan.md", plan("bug-x", "in-progress"), backlog) self.add("bug-y.md", ticket("in-review", kind="bug"), backlog) rows = self.status_rows(backlog) self.assertEqual([(r["file"], r["state"]) for r in rows], [("bug-x.md", "in-progress"), ("bug-y.md", "review")]) def test_two_plans_for_one_ticket_error(self): self.breakdown_epic() self.add("story-a-plan.md", plan(1)) self.add("story-b-plan.md", plan(1)) r = run("next", str(self.epic)) self.assertEqual(r.returncode, 1, r.stdout) for name in ("story-a-plan.md", "story-b-plan.md"): self.assertIn(name, json.loads(r.stderr)["error"]) def test_an_orphan_plan_is_a_problem_and_the_rest_of_the_tree_still_reads(self): self.breakdown_epic() for ref in (9, "story-nothing"): self.add("story-a-plan.md", plan(ref)) for command in ("next", "status"): r = run(command, str(self.epic)) self.assertEqual(r.returncode, 0, r.stderr) out = json.loads(r.stdout) self.assertEqual(len(out["problems"]), 1) self.assertIn("story-a-plan.md", out["problems"][0]) self.assertEqual(self.files(self.next()["ready_to_start"]), [None]) def test_a_plan_with_an_unknown_status_blocks_its_ticket_until_mark_fixes_it(self): self.breakdown_epic() self.add("story-scaffold-plan.md", plan(1, "already-satisfied")) out = self.next() self.assertEqual(out["ready_to_start"], []) blocked = {t["id"]: t for t in out["blocked"]} self.assertIn("already-satisfied", blocked[1]["blocked_reason"]) self.assertIn("status 'already-satisfied'", out["problems"][0]) self.assertEqual(run("mark", str(self.epic), "1", "done").returncode, 0) out = self.next() self.assertNotIn("problems", out) self.assertEqual([t["id"] for t in out["ready_to_start"]], [2, 3]) class ActiveInitiativeTests(TreeCase): """With no folder, next, status, and find run on `{tickets.root}/{active_initiative}`.""" def setUp(self): super().setUp() (self.root / "_bmad" / "scripts").mkdir() shutil.copy(CONFIG_UTILS, self.root / "_bmad" / "scripts" / "config_utils.py") self.configure("initiative-checkout") self.seed() def configure(self, initiative, layer="config.toml"): line = f'active_initiative = "{initiative}"\n' if initiative is not None else "" path = self.root / "_bmad" / ("custom" if layer != "config.toml" else "") / layer path.write_text(f'[core]\noutput_folder = "{{project-root}}/out"\n{line}') def elsewhere(self): other = tempfile.TemporaryDirectory() self.addCleanup(other.cleanup) return other.name def ok(self, r): self.assertEqual(r.returncode, 0, r.stderr) return json.loads(r.stdout) def fails(self, r, message): self.assertEqual(r.returncode, 1, r.stdout) self.assertIn(message, json.loads(r.stderr)["error"]) def test_next_with_no_folder_runs_on_the_active_initiative(self): expected = self.ok(run("next", str(self.initiative))) self.assertEqual(self.ok(run("next", cwd=self.epic)), expected) self.assertEqual(expected["folder"], "initiative-checkout") def test_with_no_folder_next_and_status_also_report_the_backlog(self): self.assertNotIn("backlog", self.ok(run("next", cwd=self.root))) backlog = self.add_backlog() self.add("bug-x.md", ticket("ready-for-dev", kind="bug", refined="true"), backlog) out = self.ok(run("next", cwd=self.root)) self.assertEqual(out["folder"], "initiative-checkout") self.assertEqual( (out["backlog"]["folder"], out["backlog"]["ready_to_start"][0]["ref"]), ("backlog", "bug-x.md") ) self.assertEqual(self.ok(run("find", "backlog", "bug-x.md", cwd=self.root))["file"], "bug-x.md") out = self.ok(run("status", cwd=self.root)) self.assertEqual((out["counts"]["total"], out["backlog"]["counts"]), (5, {"total": 1, "backlog": 1})) self.assertNotIn("backlog", self.ok(run("next", str(self.initiative)))) self.add("bug-y.md", ticket("nonsense", kind="bug"), backlog) out = self.ok(run("next", cwd=self.root)) self.assertIn("status", out["backlog"]["error"]) self.assertEqual(len(out["ready_to_refine"]), 2) def test_with_no_folder_a_ticket_the_initiative_lacks_is_found_in_the_backlog(self): backlog = self.add_backlog() self.add("bug-x.md", ticket("draft", 7, kind="bug", tracker_id='"SHOP-7"'), backlog) hit = self.ok(run("find", "bug-x.md", cwd=self.root)) self.assertEqual((hit["folder"], hit["plan"]), ("backlog", str(backlog.resolve() / "bug-x-plan.md"))) self.assertEqual(self.ok(run("find", "SHOP-7", cwd=self.root))["ref"], "bug-x.md") self.fails(run("find", "bug-nope", cwd=self.root), "no ticket matches") self.fails(run("find", str(self.initiative), "bug-x.md"), "no ticket matches") self.ok(run("mark", "bug-x.md", "done", cwd=self.root)) self.assertTrue((backlog / "bug-x-plan.md").is_file()) self.write_store("jira") r = run("mirror", cwd=self.root, stdin=json.dumps([{"tracker_id": "SHOP-7", "tracker_status": "done"}])) self.assertEqual([m["ref"] for m in self.ok(r)["mirrored"]], ["bug-x.md"]) self.assertIn("\ntracker_status: done\n", (backlog / "bug-x.md").read_text(encoding="utf-8")) def test_a_backlog_that_cannot_be_read_is_reported_and_the_initiative_still_answers(self): backlog = self.add_backlog() (backlog / "bug-x.md").write_bytes(b"---\ntype: bug\ntitle: \xff\n---\n") out = self.ok(run("status", cwd=self.root)) self.assertEqual((out["counts"]["total"], out["backlog"]["folder"]), (5, "backlog")) self.assertIn("error", out["backlog"]) self.fails(run("find", "bug-x.md", cwd=self.root), "no ticket matches") self.write_store("jira") r = run("mirror", cwd=self.root, stdin=json.dumps([{"ref": "story-scaffold", "tracker_status": "done"}])) self.assertEqual([m["file"] for m in self.ok(r)["mirrored"]], ["story-scaffold.md"]) def test_an_epic_is_found_by_its_folder_name_under_the_active_initiative(self): out = self.ok(run("--project-root", str(self.root), "status", "epic-cart", cwd=self.elsewhere())) self.assertEqual((out["folder"], out["counts"]["total"]), ("epic-cart", 5)) self.configure(None) self.fails(run("--project-root", str(self.root), "status", "epic-cart", cwd=self.elsewhere()), "not a folder") def test_the_backlog_as_active_initiative_is_reported_once(self): backlog = self.add_backlog() self.add("bug-x.md", ticket("draft", kind="bug"), backlog) self.configure("backlog") out = self.ok(run("status", cwd=self.root)) self.assertEqual(out["folder"], "backlog") self.assertNotIn("backlog", out) def test_project_root_flag_names_the_project_from_anywhere(self): self.write_store(root="out") out = self.ok(run("--project-root", str(self.root), "status", cwd=self.elsewhere())) self.assertEqual(out["folder"], "initiative-checkout") self.assertEqual(out["counts"]["total"], 5) def test_a_relative_folder_is_found_under_the_store_from_anywhere(self): out = self.ok(run("--project-root", str(self.root), "status", "initiative-checkout", cwd=self.elsewhere())) self.assertEqual(out["folder"], "initiative-checkout") out = self.ok( run("--project-root", str(self.root), "next", "out/initiative-checkout/epic-cart", cwd=self.elsewhere()) ) self.assertEqual(out["folder"], "epic-cart") self.fails( run("--project-root", str(self.root), "status", "initiative-gone", cwd=self.elsewhere()), "not a folder" ) def test_root_substitutes_project_root_and_output_folder(self): for root in ("{project-root}/out", "{output_folder}"): self.write_store(root=root) self.assertEqual(self.ok(run("status", cwd=self.root))["folder"], "initiative-checkout", root) def test_find_by_ref_only(self): self.pricing() expected = self.ok(run("find", str(self.initiative), "1.2")) self.assertEqual(self.ok(run("find", "1.2", cwd=self.root)), expected) self.assertEqual(expected["title"], "Pricing rules") def test_a_given_folder_never_reads_the_config(self): (self.root / "_bmad" / "config.toml").write_text("not toml [") self.assertEqual(self.ok(run("next", str(self.epic), cwd=self.root))["folder"], "epic-cart") self.fails(run("next", cwd=self.root), "config.toml") def test_unset_active_initiative_names_the_key(self): for value in (None, ""): self.configure(value) self.fails(run("next", cwd=self.root), "core.active_initiative") def test_no_project_root_found(self): self.fails(run("status", cwd=self.elsewhere()), "no project root") def test_missing_initiative_folder_names_the_resolved_path(self): self.configure("initiative-gone") self.fails(run("next", cwd=self.root), str((self.root / "out" / "initiative-gone").resolve())) def test_store_is_read_from_the_project_found_when_root_lies_outside_it(self): outside = Path(self.elsewhere()) shutil.copytree(self.initiative, outside / "initiative-checkout") self.write_store("linear", outside.as_posix()) r = run("next", cwd=self.root) self.assertEqual(r.returncode, 2, r.stderr) self.assertIn("sync ticket status", r.stderr) self.assertEqual(self.ok(run("status", cwd=self.root))["store"], "linear") def test_the_skill_installs_the_script_and_a_lone_copy_runs(self): bmod = tomllib.loads((SCRIPT.parents[1] / "bmod.toml").read_text(encoding="utf-8")) self.assertIn("scripts/tickets.py", bmod["skill"]["scripts"]) installed = self.root / "_bmad" / "method" / "scripts" installed.mkdir(parents=True) shutil.copy(SCRIPT, installed) r = subprocess.run( [sys.executable, str(installed / "tickets.py"), "next"], text=True, capture_output=True, check=False, cwd=self.root, ) self.assertEqual(self.ok(r)["folder"], "initiative-checkout") def test_mark_with_no_folder_blocks_a_ticket_in_the_active_initiative(self): pricing = self.pricing() r = run("mark", "1.2", "in-progress", "--blocked", "legal", cwd=self.root) self.assertEqual(self.ok(r)["plan"], str(pricing.resolve() / "story-pricing-rules-plan.md")) text = (pricing / "story-pricing-rules-plan.md").read_text(encoding="utf-8") self.assertIn(f'\nblocked_at: "{date.today().isoformat()}"\nblocked_reason: "legal"\n', text) blocked = self.ok(run("next", cwd=self.root))["blocked"] self.assertIn( ("epic-pricing", 2, "in-progress", "legal"), [(e["epic"], e["id"], e["status"], e["blocked_reason"]) for e in blocked], ) def test_user_layer_overrides_the_base_config(self): self.configure("initiative-gone") self.configure("initiative-checkout", layer="config.user.toml") self.assertEqual(self.ok(run("next", cwd=self.root))["folder"], "initiative-checkout") if __name__ == "__main__": unittest.main()