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_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()