from __future__ import annotations import contextlib import io import json import os from pathlib import Path import subprocess import sys 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_parallel_add_plan_allocates_unique_ids(self) -> None: with TempDir() as root: main(["--root", str(root), "init"]) self.assertEqual(main(["--root", str(root), "start", "parallel-demo", "Parallel Demo"]), 0) self.assertEqual(main(["--root", str(root), "accept-agenda", "DSC-0001"]), 0) self.assertEqual(main(["--root", str(root), "add-decision", "DSC-0001", "--title", "Parallel Decision"]), 0) self.assertEqual(main(["--root", str(root), "accept-decision", "DSC-0001"]), 0) script = "from discussion_framework.cli import main; import sys; raise SystemExit(main(sys.argv[1:]))" env = os.environ.copy() src = str(Path(__file__).resolve().parents[1] / "src") env["PYTHONPATH"] = src + os.pathsep + env["PYTHONPATH"] if env.get("PYTHONPATH") else src processes = [ subprocess.Popen( [ sys.executable, "-c", script, "--root", str(root), "add-plan", "DSC-0001", "--title", f"Parallel Plan {number}", ], env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) for number in range(1, 4) ] results = [process.communicate(timeout=10) + (process.returncode,) for process in processes] self.assertEqual([returncode for *_streams, returncode in results], [0, 0, 0], results) discussion = root / "discussion" records = [json.loads(line) for line in (discussion / "index.ndjson").read_text(encoding="utf-8").splitlines()] meta = records[0] plans = records[1]["plans"] self.assertEqual(meta["next_id"]["PLN"], 4) self.assertEqual(sorted(plan["id"] for plan in plans), ["PLN-0001", "PLN-0002", "PLN-0003"]) for plan in plans: self.assertTrue((discussion / "workflow" / "plans" / plan["file"]).exists()) 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()