opensource/BMAD-METHOD
3
0
Fork 0
mirror of https://github.com/bmad-code-org/BMAD-METHOD.git synced 2026-08-28 19:20:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

Harden from second review round: file modes, parser edges, JSON contract

Browse Source 
- Atomic writers preserve the target's file mode (mkstemp's 0600 no
  longer flips ticket permissions) and fsync before replace
- _rewrite_depends_on scoped to the frontmatter block
- Zero-indent YAML block sequences accepted by both parsers
- validate flags unknown ticket types; index link targets encoded
- hitl_threshold type-checked from customize.toml; update_ticket gains
  a top-level catch so unexpected errors keep the JSON contract
- Risky bare string values quoted on render; collect_tree skips dot
  folders so archived tickets are not updatable by --id
- Sync test asserts (not skips) on a missing copy; test run() surfaces
  stderr on non-JSON output; regression tests for each fix
This commit is contained in:
Brian Madison
2026-08-02 22:46:25 -05:00
parent bc9878d5d4
commit 12c411aa3e
10 changed files with 196 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/bmm-skills/plan/bmad-ticket/scripts/tests/test_dual_home_sync.py
+8 -5
View File
@@ -29,11 +29,14 @@ class DualHomeSyncTests(unittest.TestCase):
canonical = src / "scripts"
bundled = src / "bmm-skills" / "plan" / "bmad-ticket" / "scripts"
for name in SCRIPTS:
a, b = canonical / name, bundled / name
if not (a.is_file() and b.is_file()):
self.skipTest(f"{name} missing from one home")
self.assertEqual(a.read_bytes(), b.read_bytes(),
f"{name} has drifted between src/scripts and the bundled copy")
with self.subTest(script=name):
a, b = canonical / name, bundled / name
# Once the src root resolves, both homes exist — a missing
# copy is drift, not a layout difference.
self.assertTrue(a.is_file(), f"{name} missing from src/scripts")
self.assertTrue(b.is_file(), f"{name} missing from the bundled copy")
self.assertEqual(a.read_bytes(), b.read_bytes(),
f"{name} has drifted between src/scripts and the bundled copy")
if __name__ == "__main__":
src/bmm-skills/plan/bmad-ticket/scripts/tests/test_ticket_tree.py
+23 -1
View File
@@ -30,7 +30,12 @@ def ticket(tid, ttype, title, status="backlog", deps="[]", covers="[]", extra=""
def run(*args):
proc = subprocess.run([sys.executable, str(SCRIPT), *args],
capture_output=True, text=True)
return proc.returncode, json.loads(proc.stdout)
try:
return proc.returncode, json.loads(proc.stdout)
except json.JSONDecodeError:
raise AssertionError(
f"non-JSON output (exit {proc.returncode})\n"
f"stdout: {proc.stdout!r}\nstderr: {proc.stderr!r}")
class TicketTreeTests(unittest.TestCase):
@@ -231,6 +236,23 @@ class TicketTreeTests(unittest.TestCase):
msgs = " | ".join(e["error"] for e in out["errors"])
self.assertIn("no epic ticket.md", msgs)
def test_zero_indent_block_list_parses(self):
# YAML allows block-sequence items at zero indent — the parser must too.
(self.root / "ALRT-54-zeroindent.md").write_text(
ticket("ALRT-54", "task", "Zero indent").replace(
"depends_on: []", "depends_on:\n- ALRT-13"))
code, out = run("board", "--root", str(self.root))
blocked = {b["id"]: b["waiting_on"] for b in out["blocked"]}
self.assertEqual(blocked["ALRT-54"], ["ALRT-13"])
def test_validate_flags_unknown_type(self):
(self.root / "ALRT-55-badtype.md").write_text(
ticket("ALRT-55", "task", "Bad type").replace("type: task", "type: storyy"))
code, out = run("validate", "--root", str(self.root))
self.assertEqual(code, 1)
self.assertIn("type must be epic or one of",
" | ".join(e["error"] for e in out["errors"]))
def test_scalar_dep_is_not_iterated_charwise(self):
(self.root / "ALRT-53-scalar.md").write_text(
ticket("ALRT-53", "task", "Scalar dep").replace(
src/bmm-skills/plan/bmad-ticket/scripts/tests/test_update_ticket.py
+30 -1
View File
@@ -6,6 +6,8 @@
"""Tests for update_ticket.py — run: uv run python -m unittest discover -s scripts/tests"""
import json
import os
import stat
import subprocess
import sys
import tempfile
@@ -57,7 +59,12 @@ created: 2026-08-01
def run(*args):
proc = subprocess.run([sys.executable, str(SCRIPT), *args],
capture_output=True, text=True)
return proc.returncode, json.loads(proc.stdout)
try:
return proc.returncode, json.loads(proc.stdout)
except json.JSONDecodeError:
raise AssertionError(
f"non-JSON output (exit {proc.returncode})\n"
f"stdout: {proc.stdout!r}\nstderr: {proc.stderr!r}")
class UpdateTicketTests(unittest.TestCase):
@@ -171,6 +178,28 @@ class UpdateTicketTests(unittest.TestCase):
self.assertEqual(code, 1)
self.assertIn("unterminated", out["error"])
def test_file_mode_preserved_across_update(self):
os.chmod(self.story, 0o644)
code, out = run("--path", str(self.story), "--set", "status=in-progress")
self.assertEqual(code, 0)
self.assertEqual(stat.S_IMODE(os.stat(self.story).st_mode), 0o644)
def test_risky_string_value_is_quoted(self):
code, out = run("--path", str(self.bug),
"--set", "discovered_from=see: ALRT-3 [maybe]")
self.assertEqual(code, 0)
self.assertIn('discovered_from: "see: ALRT-3 [maybe]"', self.bug.read_text())
def test_archived_tickets_are_not_updatable_by_id(self):
arch = self.root / ".archive" / "2026-08-01-old"
arch.mkdir(parents=True)
(arch / "ALRT-70-old.md").write_text(
STORY.replace("id: ALRT-12", "id: ALRT-70"), encoding="utf-8")
code, out = run("--root", str(self.root), "--id", "ALRT-70",
"--set", "status=in-progress")
self.assertEqual(code, 1)
self.assertIn("not found", out["error"])
def test_immutable_fields(self):
for spec in ("id=ALRT-99", "type=task", "created=2026-01-01", "schema=2"):
code, out = run("--path", str(self.story), "--set", spec)
src/bmm-skills/plan/bmad-ticket/scripts/ticket_tree.py
+16 -7
View File
@@ -79,8 +79,8 @@ def parse_frontmatter(text):
if rest == "":
items = []
j = i + 1
while j < len(lines) and re.match(r"^\s+-\s+", lines[j]):
items.append(parse_scalar(re.sub(r"^\s+-\s+", "", lines[j])))
while j < len(lines) and re.match(r"^\s*-\s+", lines[j]):
items.append(parse_scalar(re.sub(r"^\s*-\s+", "", lines[j])))
j += 1
if j > i + 1:
fm[key] = items
@@ -210,6 +210,11 @@ def atomic_write(path, content):
try:
with os.fdopen(fd, "w", encoding="utf-8", newline="") as f:
f.write(content)
f.flush()
os.fsync(f.fileno())
# mkstemp creates 0600; keep the target's mode (0644 for new files)
# so a rewrite never tightens permissions on collaborators.
os.chmod(tmp, path.stat().st_mode if path.exists() else 0o644)
os.replace(tmp, path)
except BaseException:
if os.path.exists(tmp):
@@ -242,7 +247,8 @@ def write_index(root, key):
pad = " " * depth
desc = t.get("description")
tail = f" - {desc}" if desc else ""
lines = [f"{pad}* [{t.get('title', t['id'])}]({t['_rel']}){tail}"]
href = t["_rel"].replace(" ", "%20").replace("(", "%28").replace(")", "%29")
lines = [f"{pad}* [{t.get('title', t['id'])}]({href}){tail}"]
if t["type"] == "epic":
for k in sorted(children_of(t, tickets), key=lambda x: id_num(x["id"])):
lines.extend(entry_lines(k, depth + 1))
@@ -474,12 +480,13 @@ def _rewrite_depends_on(path, new_deps):
"""Replace a ticket's depends_on entry (inline or block form) with an
inline list. Only used by archive to drop satisfied edges."""
lines = path.read_text(encoding="utf-8").splitlines(keepends=True)
close = next((i for i in range(1, len(lines)) if lines[i].strip() == "---"), len(lines))
new_line = "depends_on: [" + ", ".join(str(d) for d in new_deps) + "]\n"
for i, line in enumerate(lines):
if re.match(r"^depends_on:", line):
for i in range(1, close): # frontmatter only — a body line never matches
if re.match(r"^depends_on:", lines[i]):
j = i + 1
if line.split(":", 1)[1].strip() == "":
while j < len(lines) and re.match(r"^\s+-\s+", lines[j]):
if lines[i].split(":", 1)[1].strip() == "":
while j < close and re.match(r"^\s*-\s+", lines[j]):
j += 1
lines[i:j] = [new_line]
break
@@ -599,6 +606,8 @@ def cmd_validate(root, args):
if t["_path"].parent != root and t["_path"].parent not in epic_dirs:
err(t, "leaf sits in a folder with no epic ticket.md — invisible to "
"index and board; move it to the bin or an epic folder")
if t["type"] != "epic" and t["type"] not in LEAF_TYPES:
err(t, f"type must be epic or one of {sorted(LEAF_TYPES)}: {t['type']!r}")
if not isinstance(t.get("schema"), int):
err(t, "schema must be an integer")
title = t.get("title")
src/bmm-skills/plan/bmad-ticket/scripts/update_ticket.py
+21 -7
View File
@@ -125,8 +125,8 @@ def parse_frontmatter(lines):
if rest == "":
items = []
j = i + 1
while j < close and re.match(r"^\s+-\s+", lines[j]):
items.append(parse_scalar(re.sub(r"^\s+-\s+", "", lines[j].rstrip("\n"))))
while j < close and re.match(r"^\s*-\s+", lines[j]):
items.append(parse_scalar(re.sub(r"^\s*-\s+", "", lines[j].rstrip("\n"))))
j += 1
entries.append({"key": key, "value": items, "start": start, "end": j})
i = j
@@ -147,9 +147,10 @@ def render(key, value):
if key in QUOTED_FIELDS:
escaped = str(value).replace('"', '\\"')
return f'{key}: "{escaped}"\n'
if key in PLAIN_STR_FIELDS and value == "":
return f'{key}: ""\n'
return f"{key}: {value}\n"
s = str(value)
if (key in PLAIN_STR_FIELDS and s == "") or re.search(r'[:#\[\]{}"\']', s) or s != s.strip():
return f'{key}: "{s.replace(chr(34), chr(92) + chr(34))}"\n'
return f"{key}: {s}\n"
def coerce(field, raw, ticket_type):
@@ -182,6 +183,8 @@ def collect_tree(root):
for p in sorted(root.rglob("*.md")):
if p.name == "index.md":
continue
if any(part.startswith(".") for part in p.relative_to(root).parts):
continue # dot folders (.archive/, .git/) are not the live tree
try:
lines = p.read_text(encoding="utf-8").splitlines(keepends=True)
entries, _ = parse_frontmatter(lines)
@@ -250,7 +253,9 @@ def main():
args.transitions = ",".join(str(t) for t in cfg_transitions)
hitl_threshold = args.hitl_threshold
if hitl_threshold is None:
hitl_threshold = cfg.get("hitl_threshold", DEFAULT_HITL_THRESHOLD)
ht = cfg.get("hitl_threshold", DEFAULT_HITL_THRESHOLD)
hitl_threshold = ht if isinstance(ht, int) and not isinstance(ht, bool) \
else DEFAULT_HITL_THRESHOLD
tree_ids, tree_deps = None, None
if args.path:
@@ -381,6 +386,10 @@ def main():
try:
with os.fdopen(fd, "w", encoding="utf-8", newline="") as f:
f.write("".join(new_lines))
f.flush()
os.fsync(f.fileno())
# mkstemp creates 0600; keep the ticket's own mode.
os.chmod(tmp, path.stat().st_mode)
os.replace(tmp, path)
except BaseException:
if os.path.exists(tmp):
@@ -396,4 +405,9 @@ def main():
if __name__ == "__main__":
main()
try:
main()
except SystemExit:
raise
except Exception as e: # keep the JSON output contract on fs/encoding errors
fail(f"{e.__class__.__name__}: {e}")
src/scripts/tests/test_dual_home_sync.py
+8 -5
View File
@@ -29,11 +29,14 @@ class DualHomeSyncTests(unittest.TestCase):
canonical = src / "scripts"
bundled = src / "bmm-skills" / "plan" / "bmad-ticket" / "scripts"
for name in SCRIPTS:
a, b = canonical / name, bundled / name
if not (a.is_file() and b.is_file()):
self.skipTest(f"{name} missing from one home")
self.assertEqual(a.read_bytes(), b.read_bytes(),
f"{name} has drifted between src/scripts and the bundled copy")
with self.subTest(script=name):
a, b = canonical / name, bundled / name
# Once the src root resolves, both homes exist — a missing
# copy is drift, not a layout difference.
self.assertTrue(a.is_file(), f"{name} missing from src/scripts")
self.assertTrue(b.is_file(), f"{name} missing from the bundled copy")
self.assertEqual(a.read_bytes(), b.read_bytes(),
f"{name} has drifted between src/scripts and the bundled copy")
if __name__ == "__main__":
src/scripts/tests/test_ticket_tree.py
+23 -1
View File
@@ -30,7 +30,12 @@ def ticket(tid, ttype, title, status="backlog", deps="[]", covers="[]", extra=""
def run(*args):
proc = subprocess.run([sys.executable, str(SCRIPT), *args],
capture_output=True, text=True)
return proc.returncode, json.loads(proc.stdout)
try:
return proc.returncode, json.loads(proc.stdout)
except json.JSONDecodeError:
raise AssertionError(
f"non-JSON output (exit {proc.returncode})\n"
f"stdout: {proc.stdout!r}\nstderr: {proc.stderr!r}")
class TicketTreeTests(unittest.TestCase):
@@ -231,6 +236,23 @@ class TicketTreeTests(unittest.TestCase):
msgs = " | ".join(e["error"] for e in out["errors"])
self.assertIn("no epic ticket.md", msgs)
def test_zero_indent_block_list_parses(self):
# YAML allows block-sequence items at zero indent — the parser must too.
(self.root / "ALRT-54-zeroindent.md").write_text(
ticket("ALRT-54", "task", "Zero indent").replace(
"depends_on: []", "depends_on:\n- ALRT-13"))
code, out = run("board", "--root", str(self.root))
blocked = {b["id"]: b["waiting_on"] for b in out["blocked"]}
self.assertEqual(blocked["ALRT-54"], ["ALRT-13"])
def test_validate_flags_unknown_type(self):
(self.root / "ALRT-55-badtype.md").write_text(
ticket("ALRT-55", "task", "Bad type").replace("type: task", "type: storyy"))
code, out = run("validate", "--root", str(self.root))
self.assertEqual(code, 1)
self.assertIn("type must be epic or one of",
" | ".join(e["error"] for e in out["errors"]))
def test_scalar_dep_is_not_iterated_charwise(self):
(self.root / "ALRT-53-scalar.md").write_text(
ticket("ALRT-53", "task", "Scalar dep").replace(
src/scripts/tests/test_update_ticket.py
+30 -1
View File
@@ -6,6 +6,8 @@
"""Tests for update_ticket.py — run: uv run python -m unittest discover -s scripts/tests"""
import json
import os
import stat
import subprocess
import sys
import tempfile
@@ -57,7 +59,12 @@ created: 2026-08-01
def run(*args):
proc = subprocess.run([sys.executable, str(SCRIPT), *args],
capture_output=True, text=True)
return proc.returncode, json.loads(proc.stdout)
try:
return proc.returncode, json.loads(proc.stdout)
except json.JSONDecodeError:
raise AssertionError(
f"non-JSON output (exit {proc.returncode})\n"
f"stdout: {proc.stdout!r}\nstderr: {proc.stderr!r}")
class UpdateTicketTests(unittest.TestCase):
@@ -171,6 +178,28 @@ class UpdateTicketTests(unittest.TestCase):
self.assertEqual(code, 1)
self.assertIn("unterminated", out["error"])
def test_file_mode_preserved_across_update(self):
os.chmod(self.story, 0o644)
code, out = run("--path", str(self.story), "--set", "status=in-progress")
self.assertEqual(code, 0)
self.assertEqual(stat.S_IMODE(os.stat(self.story).st_mode), 0o644)
def test_risky_string_value_is_quoted(self):
code, out = run("--path", str(self.bug),
"--set", "discovered_from=see: ALRT-3 [maybe]")
self.assertEqual(code, 0)
self.assertIn('discovered_from: "see: ALRT-3 [maybe]"', self.bug.read_text())
def test_archived_tickets_are_not_updatable_by_id(self):
arch = self.root / ".archive" / "2026-08-01-old"
arch.mkdir(parents=True)
(arch / "ALRT-70-old.md").write_text(
STORY.replace("id: ALRT-12", "id: ALRT-70"), encoding="utf-8")
code, out = run("--root", str(self.root), "--id", "ALRT-70",
"--set", "status=in-progress")
self.assertEqual(code, 1)
self.assertIn("not found", out["error"])
def test_immutable_fields(self):
for spec in ("id=ALRT-99", "type=task", "created=2026-01-01", "schema=2"):
code, out = run("--path", str(self.story), "--set", spec)
src/scripts/ticket_tree.py
+16 -7
View File
@@ -79,8 +79,8 @@ def parse_frontmatter(text):
if rest == "":
items = []
j = i + 1
while j < len(lines) and re.match(r"^\s+-\s+", lines[j]):
items.append(parse_scalar(re.sub(r"^\s+-\s+", "", lines[j])))
while j < len(lines) and re.match(r"^\s*-\s+", lines[j]):
items.append(parse_scalar(re.sub(r"^\s*-\s+", "", lines[j])))
j += 1
if j > i + 1:
fm[key] = items
@@ -210,6 +210,11 @@ def atomic_write(path, content):
try:
with os.fdopen(fd, "w", encoding="utf-8", newline="") as f:
f.write(content)
f.flush()
os.fsync(f.fileno())
# mkstemp creates 0600; keep the target's mode (0644 for new files)
# so a rewrite never tightens permissions on collaborators.
os.chmod(tmp, path.stat().st_mode if path.exists() else 0o644)
os.replace(tmp, path)
except BaseException:
if os.path.exists(tmp):
@@ -242,7 +247,8 @@ def write_index(root, key):
pad = " " * depth
desc = t.get("description")
tail = f" - {desc}" if desc else ""
lines = [f"{pad}* [{t.get('title', t['id'])}]({t['_rel']}){tail}"]
href = t["_rel"].replace(" ", "%20").replace("(", "%28").replace(")", "%29")
lines = [f"{pad}* [{t.get('title', t['id'])}]({href}){tail}"]
if t["type"] == "epic":
for k in sorted(children_of(t, tickets), key=lambda x: id_num(x["id"])):
lines.extend(entry_lines(k, depth + 1))
@@ -474,12 +480,13 @@ def _rewrite_depends_on(path, new_deps):
"""Replace a ticket's depends_on entry (inline or block form) with an
inline list. Only used by archive to drop satisfied edges."""
lines = path.read_text(encoding="utf-8").splitlines(keepends=True)
close = next((i for i in range(1, len(lines)) if lines[i].strip() == "---"), len(lines))
new_line = "depends_on: [" + ", ".join(str(d) for d in new_deps) + "]\n"
for i, line in enumerate(lines):
if re.match(r"^depends_on:", line):
for i in range(1, close): # frontmatter only — a body line never matches
if re.match(r"^depends_on:", lines[i]):
j = i + 1
if line.split(":", 1)[1].strip() == "":
while j < len(lines) and re.match(r"^\s+-\s+", lines[j]):
if lines[i].split(":", 1)[1].strip() == "":
while j < close and re.match(r"^\s*-\s+", lines[j]):
j += 1
lines[i:j] = [new_line]
break
@@ -599,6 +606,8 @@ def cmd_validate(root, args):
if t["_path"].parent != root and t["_path"].parent not in epic_dirs:
err(t, "leaf sits in a folder with no epic ticket.md — invisible to "
"index and board; move it to the bin or an epic folder")
if t["type"] != "epic" and t["type"] not in LEAF_TYPES:
err(t, f"type must be epic or one of {sorted(LEAF_TYPES)}: {t['type']!r}")
if not isinstance(t.get("schema"), int):
err(t, "schema must be an integer")
title = t.get("title")
src/scripts/update_ticket.py
+21 -7
View File
@@ -125,8 +125,8 @@ def parse_frontmatter(lines):
if rest == "":
items = []
j = i + 1
while j < close and re.match(r"^\s+-\s+", lines[j]):
items.append(parse_scalar(re.sub(r"^\s+-\s+", "", lines[j].rstrip("\n"))))
while j < close and re.match(r"^\s*-\s+", lines[j]):
items.append(parse_scalar(re.sub(r"^\s*-\s+", "", lines[j].rstrip("\n"))))
j += 1
entries.append({"key": key, "value": items, "start": start, "end": j})
i = j
@@ -147,9 +147,10 @@ def render(key, value):
if key in QUOTED_FIELDS:
escaped = str(value).replace('"', '\\"')
return f'{key}: "{escaped}"\n'
if key in PLAIN_STR_FIELDS and value == "":
return f'{key}: ""\n'
return f"{key}: {value}\n"
s = str(value)
if (key in PLAIN_STR_FIELDS and s == "") or re.search(r'[:#\[\]{}"\']', s) or s != s.strip():
return f'{key}: "{s.replace(chr(34), chr(92) + chr(34))}"\n'
return f"{key}: {s}\n"
def coerce(field, raw, ticket_type):
@@ -182,6 +183,8 @@ def collect_tree(root):
for p in sorted(root.rglob("*.md")):
if p.name == "index.md":
continue
if any(part.startswith(".") for part in p.relative_to(root).parts):
continue # dot folders (.archive/, .git/) are not the live tree
try:
lines = p.read_text(encoding="utf-8").splitlines(keepends=True)
entries, _ = parse_frontmatter(lines)
@@ -250,7 +253,9 @@ def main():
args.transitions = ",".join(str(t) for t in cfg_transitions)
hitl_threshold = args.hitl_threshold
if hitl_threshold is None:
hitl_threshold = cfg.get("hitl_threshold", DEFAULT_HITL_THRESHOLD)
ht = cfg.get("hitl_threshold", DEFAULT_HITL_THRESHOLD)
hitl_threshold = ht if isinstance(ht, int) and not isinstance(ht, bool) \
else DEFAULT_HITL_THRESHOLD
tree_ids, tree_deps = None, None
if args.path:
@@ -381,6 +386,10 @@ def main():
try:
with os.fdopen(fd, "w", encoding="utf-8", newline="") as f:
f.write("".join(new_lines))
f.flush()
os.fsync(f.fileno())
# mkstemp creates 0600; keep the ticket's own mode.
os.chmod(tmp, path.stat().st_mode)
os.replace(tmp, path)
except BaseException:
if os.path.exists(tmp):
@@ -396,4 +405,9 @@ def main():
if __name__ == "__main__":
main()
try:
main()
except SystemExit:
raise
except Exception as e: # keep the JSON output contract on fs/encoding errors
fail(f"{e.__class__.__name__}: {e}")