302 lines
13 KiB
Python
302 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
from discussion_framework.cli import main
|
|
|
|
|
|
class CliTest(unittest.TestCase):
|
|
def test_init_validate_and_status(self) -> None:
|
|
with TempDir() as root:
|
|
self.assertEqual(main(["--root", str(root), "init"]), 0)
|
|
self.assertTrue((root / "discussion" / "index.ndjson").exists())
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
self.assertEqual(main(["--root", str(root), "validate"]), 0)
|
|
self.assertIn("validation passed", stdout.getvalue())
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
self.assertEqual(main(["--root", str(root), "status"]), 0)
|
|
self.assertIn("discussions: 0", stdout.getvalue())
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
self.assertEqual(main(["--root", str(root), "status", "--json"]), 0)
|
|
self.assertEqual(json.loads(stdout.getvalue())["total"], 0)
|
|
|
|
def test_start_creates_discussion_and_agenda(self) -> None:
|
|
with TempDir() as root:
|
|
main(["--root", str(root), "init"])
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
self.assertEqual(
|
|
main(
|
|
[
|
|
"--root",
|
|
str(root),
|
|
"start",
|
|
"pbs-autocomplete-parameter-names",
|
|
"PBS autocomplete parameter names",
|
|
"--tag",
|
|
"compiler,pbs",
|
|
"--tag",
|
|
"lsp",
|
|
"--domain-owner",
|
|
"compiler/pbs",
|
|
"--pain",
|
|
"Completions show arg0 and arg1.",
|
|
"--context",
|
|
"Stdlib methods lack authored parameter names.",
|
|
"--question",
|
|
"Where should canonical parameter names live?",
|
|
]
|
|
),
|
|
0,
|
|
)
|
|
|
|
discussion = root / "discussion"
|
|
records = [json.loads(line) for line in (discussion / "index.ndjson").read_text(encoding="utf-8").splitlines()]
|
|
meta = records[0]
|
|
item = records[1]
|
|
agenda_file = discussion / "workflow" / "agendas" / "AGD-0001-pbs-autocomplete-parameter-names.md"
|
|
|
|
self.assertIn("started DSC-0001 with AGD-0001", stdout.getvalue())
|
|
self.assertEqual(meta["next_id"]["DSC"], 2)
|
|
self.assertEqual(meta["next_id"]["AGD"], 2)
|
|
self.assertEqual(item["ticket"], "pbs-autocomplete-parameter-names")
|
|
self.assertEqual(item["tags"], ["compiler", "pbs", "lsp"])
|
|
self.assertEqual(item["agendas"][0]["file"], agenda_file.name)
|
|
self.assertTrue(agenda_file.exists())
|
|
agenda_text = agenda_file.read_text(encoding="utf-8")
|
|
self.assertIn("Domain owner: `compiler/pbs`", agenda_text)
|
|
self.assertIn("Completions show arg0 and arg1.", agenda_text)
|
|
|
|
def test_root_auto_detects_existing_discussion_directory(self) -> None:
|
|
with TempDir() as root:
|
|
main(["--root", str(root), "init"])
|
|
cwd = Path.cwd()
|
|
try:
|
|
import os
|
|
|
|
os.chdir(root / "discussion")
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
self.assertEqual(main(["init"]), 0)
|
|
self.assertIn("already initialized", stdout.getvalue())
|
|
finally:
|
|
os.chdir(cwd)
|
|
|
|
self.assertFalse((root / "discussion" / "discussion").exists())
|
|
|
|
def test_semantic_decision_and_plan_commands(self) -> None:
|
|
with TempDir() as root:
|
|
main(["--root", str(root), "init"])
|
|
self.assertEqual(
|
|
main(
|
|
[
|
|
"--root",
|
|
str(root),
|
|
"start",
|
|
"demo-flow",
|
|
"Demo Flow",
|
|
"--tag",
|
|
"compiler",
|
|
]
|
|
),
|
|
0,
|
|
)
|
|
self.assertEqual(main(["--root", str(root), "accept-agenda", "DSC-0001"]), 0)
|
|
self.assertEqual(
|
|
main(["--root", str(root), "add-decision", "DSC-0001", "--title", "Demo Decision"]),
|
|
0,
|
|
)
|
|
self.assertEqual(main(["--root", str(root), "accept-decision", "DSC-0001"]), 0)
|
|
self.assertEqual(
|
|
main(["--root", str(root), "add-plan", "DSC-0001", "--title", "Demo Plan"]),
|
|
0,
|
|
)
|
|
self.assertEqual(main(["--root", str(root), "mark-plan", "PLN-0001", "in_progress"]), 0)
|
|
self.assertEqual(main(["--root", str(root), "mark-plan", "PLN-0001", "done"]), 0)
|
|
|
|
discussion = root / "discussion"
|
|
records = [json.loads(line) for line in (discussion / "index.ndjson").read_text(encoding="utf-8").splitlines()]
|
|
meta = records[0]
|
|
item = records[1]
|
|
decision = discussion / "workflow" / "decisions" / "DEC-0001-demo-decision.md"
|
|
plan = discussion / "workflow" / "plans" / "PLN-0001-demo-plan.md"
|
|
|
|
self.assertEqual(meta["next_id"]["DEC"], 2)
|
|
self.assertEqual(meta["next_id"]["PLN"], 2)
|
|
self.assertEqual(item["status"], "in_progress")
|
|
self.assertEqual(item["agendas"][0]["status"], "accepted")
|
|
self.assertEqual(item["decisions"][0]["status"], "accepted")
|
|
self.assertEqual(item["decisions"][0]["ref_agenda"], "AGD-0001")
|
|
self.assertEqual(item["plans"][0]["status"], "done")
|
|
self.assertEqual(item["plans"][0]["ref_decisions"], ["DEC-0001"])
|
|
self.assertTrue(decision.exists())
|
|
self.assertTrue(plan.exists())
|
|
self.assertIn("status: accepted", decision.read_text(encoding="utf-8"))
|
|
self.assertIn("status: done", plan.read_text(encoding="utf-8"))
|
|
|
|
def test_housekeep_with_lesson_removes_workflow_artifacts(self) -> None:
|
|
with TempDir() as root:
|
|
main(["--root", str(root), "init"])
|
|
discussion = root / "discussion"
|
|
(discussion / "workflow" / "agendas" / "AGD-0001-demo.md").write_text("agenda", encoding="utf-8")
|
|
(discussion / "workflow" / "decisions" / "DEC-0001-demo.md").write_text("decision", encoding="utf-8")
|
|
(discussion / "workflow" / "plans" / "PLN-0001-demo.md").write_text("plan", encoding="utf-8")
|
|
lesson = discussion / "lessons" / "DSC-0001-demo"
|
|
lesson.mkdir()
|
|
lesson_file = lesson / "LSN-0001-demo.md"
|
|
lesson_file.write_text("---\nid: LSN-0001\n---\n# Demo\n", encoding="utf-8")
|
|
|
|
records = [
|
|
{"type": "meta", "next_id": {"DSC": 2, "AGD": 2, "DEC": 2, "PLN": 2, "LSN": 1, "CLSN": 1}},
|
|
{
|
|
"type": "discussion",
|
|
"id": "DSC-0001",
|
|
"status": "in_progress",
|
|
"ticket": "demo",
|
|
"title": "Demo",
|
|
"created_at": "2026-05-08",
|
|
"updated_at": "2026-05-08",
|
|
"tags": [],
|
|
"agendas": [{"id": "AGD-0001", "file": "AGD-0001-demo.md", "status": "accepted"}],
|
|
"decisions": [{"id": "DEC-0001", "file": "DEC-0001-demo.md", "status": "accepted"}],
|
|
"plans": [{"id": "PLN-0001", "file": "PLN-0001-demo.md", "status": "done"}],
|
|
"lessons": [],
|
|
},
|
|
]
|
|
(discussion / "index.ndjson").write_text(
|
|
"\n".join(json.dumps(record, separators=(",", ":")) for record in records) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
self.assertEqual(
|
|
main(["--root", str(root), "housekeep", "DSC-0001", "--lesson", str(lesson_file)]),
|
|
0,
|
|
)
|
|
|
|
updated = (discussion / "index.ndjson").read_text(encoding="utf-8").splitlines()
|
|
meta = json.loads(updated[0])
|
|
item = json.loads(updated[1])
|
|
self.assertEqual(meta["next_id"]["LSN"], 2)
|
|
self.assertEqual(item["status"], "done")
|
|
self.assertEqual(item["agendas"], [])
|
|
self.assertEqual(item["decisions"], [])
|
|
self.assertEqual(item["plans"], [])
|
|
self.assertEqual(item["lessons"][0]["id"], "LSN-0001")
|
|
self.assertFalse((discussion / "workflow" / "agendas" / "AGD-0001-demo.md").exists())
|
|
|
|
def test_validate_allows_done_discussion_with_historical_workflow_entries(self) -> None:
|
|
with TempDir() as root:
|
|
main(["--root", str(root), "init"])
|
|
discussion = root / "discussion"
|
|
lesson = discussion / "lessons" / "DSC-0001-demo"
|
|
lesson.mkdir()
|
|
lesson_file = lesson / "LSN-0001-demo.md"
|
|
lesson_file.write_text("---\nid: LSN-0001\n---\n# Demo\n", encoding="utf-8")
|
|
|
|
records = [
|
|
{"type": "meta", "next_id": {"DSC": 2, "AGD": 2, "DEC": 2, "PLN": 2, "LSN": 2, "CLSN": 1}},
|
|
{
|
|
"type": "discussion",
|
|
"id": "DSC-0001",
|
|
"status": "done",
|
|
"ticket": "demo",
|
|
"title": "Demo",
|
|
"created_at": "2026-05-08",
|
|
"updated_at": "2026-05-08",
|
|
"tags": [],
|
|
"agendas": [{"id": "AGD-0001", "file": "AGD-0001-demo.md", "status": "done"}],
|
|
"decisions": [{"id": "DEC-0001", "file": "DEC-0001-demo.md", "status": "accepted"}],
|
|
"plans": [{"id": "PLN-0001", "file": "PLN-0001-demo.md", "status": "done"}],
|
|
"lessons": [
|
|
{
|
|
"id": "LSN-0001",
|
|
"file": "discussion/lessons/DSC-0001-demo/LSN-0001-demo.md",
|
|
"status": "done",
|
|
}
|
|
],
|
|
},
|
|
]
|
|
(discussion / "index.ndjson").write_text(
|
|
"\n".join(json.dumps(record, separators=(",", ":")) for record in records) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
self.assertEqual(main(["--root", str(root), "validate"]), 0)
|
|
self.assertIn("validation passed", stdout.getvalue())
|
|
|
|
def test_inspect_context_next_actions_and_backup(self) -> None:
|
|
with TempDir() as root:
|
|
main(["--root", str(root), "init"])
|
|
discussion = root / "discussion"
|
|
agenda = discussion / "workflow" / "agendas" / "AGD-0001-demo.md"
|
|
agenda.write_text("agenda", encoding="utf-8")
|
|
records = [
|
|
{"type": "meta", "next_id": {"DSC": 2, "AGD": 2, "DEC": 1, "PLN": 1, "LSN": 1, "CLSN": 1}},
|
|
{
|
|
"type": "discussion",
|
|
"id": "DSC-0001",
|
|
"status": "open",
|
|
"ticket": "demo",
|
|
"title": "Demo",
|
|
"created_at": "2026-05-08",
|
|
"updated_at": "2026-05-08",
|
|
"tags": ["studio"],
|
|
"agendas": [{"id": "AGD-0001", "file": "AGD-0001-demo.md", "status": "open"}],
|
|
"decisions": [],
|
|
"plans": [],
|
|
"lessons": [],
|
|
},
|
|
]
|
|
(discussion / "index.ndjson").write_text(
|
|
"\n".join(json.dumps(record, separators=(",", ":")) for record in records) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
self.assertEqual(main(["--root", str(root), "inspect", "DSC-0001", "--json"]), 0)
|
|
self.assertEqual(json.loads(stdout.getvalue())["artifacts"]["agendas"][0]["id"], "AGD-0001")
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
self.assertEqual(main(["--root", str(root), "context", "DSC-0001"]), 0)
|
|
self.assertIn("Suggested next actions", stdout.getvalue())
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
self.assertEqual(main(["--root", str(root), "next-actions"]), 0)
|
|
self.assertIn("accept-agenda", stdout.getvalue())
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
self.assertEqual(main(["--root", str(root), "backup-index"]), 0)
|
|
self.assertTrue(any((discussion / ".backups").iterdir()))
|
|
|
|
|
|
class TempDir:
|
|
def __enter__(self) -> Path:
|
|
import tempfile
|
|
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
return Path(self._tmp.name)
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
self._tmp.cleanup()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|