workflow: initial public history

1 files changed, 304 insertions, 0 deletions

diff --git a/tests/test_check.py b/tests/test_check.py
new file mode 100644
index 0000000..e4f3d3f
--- /dev/null
+++ b/tests/test_check.py
@@ -0,0 +1,304 @@
+import os
+import sys
+import time
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from wflib import check as K
+from wflib import config as C
+
+TOML = 'format = 1\ntasks = "TASKS.md"\narchive = "tasks/archive.md"\ndocs = ["DESIGN.md", "docs/"]\n'
+ANCHORS = '[anchors]\nindex = "DESIGN.md"\nindex_section = "Subsystems"\nspecs = "docs/specs"\n'
+ARCHIVE = "# Archive (newest first)\n\n- 2026-09-01 **t-done** Done thing — ok\n"
+DESIGN = """\
+# Design
+
+## Subsystems
+
+### terrain
+
+Heightmap. [spec](docs/specs/terrain.md#terrain)
+
+### input
+
+Keys.
+
+## Notes
+
+Free text.
+"""
+SPEC = '# Terrain spec\n\n<a id="terrain"></a>\n## Terrain\n\nText.\n'
+CLEAN = """\
+# Tasks
+
+## Awaiting your decision
+
+- **a-key**: Key needed. Which one?
+
+## Pending
+
+- **t-one** [P1] (1h): One.
+  - After: [[t-done]]
+  Ref: DESIGN.md#terrain, docs/specs/terrain.md
+
+- **t-two** [P2] (5h) (blocked: [[a-key]]): Two.
+  - After: [[t-one]]
+
+## Needs human
+
+## Deferred
+"""
+
+
+class Base(unittest.TestCase):
+    toml = TOML
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self.tmp.name).resolve()
+        (self.root / "tasks").mkdir()
+        (self.root / "docs" / "specs").mkdir(parents=True)
+        (self.root / "workflow.toml").write_text(self.toml)
+        (self.root / "tasks" / "archive.md").write_text(ARCHIVE)
+        (self.root / "DESIGN.md").write_text(DESIGN)
+        (self.root / "docs" / "specs" / "terrain.md").write_text(SPEC)
+        self.tasks(CLEAN)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def tasks(self, text):
+        (self.root / "TASKS.md").write_text(text)
+
+    def pending(self, items):
+        self.tasks("## Awaiting your decision\n\n- **a-key**: Key. Q?\n\n## Pending\n\n" + items +
+                   "\n## Needs human\n\n## Deferred\n")
+
+    def run_check(self):
+        errors, warnings = K.check(C.load(self.root))
+        return [str(e) for e in errors], [str(w) for w in warnings]
+
+    def errors(self):
+        return self.run_check()[0]
+
+
+class TasksCheckTest(Base):
+    def test_clean(self):
+        self.assertEqual(self.run_check(), ([], []))
+
+    def test_after_prose_and_deferred(self):
+        self.tasks("## Awaiting your decision\n\n## Pending\n\n"
+                   "- **t-a** [P1] (1h): A.\n  After: when ready [[t-d]]\n\n"
+                   "## Needs human\n\n## Deferred\n\n- **t-d** [P3] (1h): D.\n")
+        w = self.run_check()[1]
+        self.assertEqual(len(w), 2, w)
+        self.assertTrue(any("has prose" in x and ":6:" in x for x in w), w)
+        self.assertTrue(any("'t-d' is deferred" in x for x in w), w)
+
+    def test_after_clean_links(self):
+        self.pending("- **t-a** [P1] (1h): A.\n  After: [[t-done]], [[t-b]]\n\n- **t-b** [P1] (1h): B.\n")
+        self.assertEqual(self.run_check()[1], [])
+
+    def test_duplicate_id(self):
+        self.pending("- **t-a** [P1] (1h): A.\n\n- **t-a** [P1] (1h): Again.\n")
+        self.assertEqual(self.errors(), ["TASKS.md:9: t-a: duplicate id (also line 7)"])
+
+    def test_id_reused_from_archive(self):
+        self.pending("- **t-done** [P1] (1h): A.\n")
+        self.assertEqual(self.errors(), ["TASKS.md:7: t-done: id already used in the archive (ids are never reused)"])
+
+    def test_bad_grammar(self):
+        self.pending("- **T-Foo** [P1] (1h): A.\n")
+        self.assertEqual(self.errors(), ["TASKS.md:7: T-Foo: bad id (want t-… or a-…, lowercase a-z 0-9 -)"])
+
+    def test_wrong_kind_for_section(self):
+        self.pending("- **a-x**: A. Q?\n")
+        self.assertEqual(self.errors(), ["TASKS.md:7: a-x: a- item in Pending (a- items belong in Awaiting, t- elsewhere)"])
+
+    def test_task_in_awaiting(self):
+        self.tasks("## Awaiting your decision\n\n- **t-x** [P1] (1h): A.\n\n## Pending\n\n## Needs human\n\n## Deferred\n")
+        self.assertEqual(self.errors(),
+                         ["TASKS.md:3: t-x: t- item in Awaiting your decision (a- items belong in Awaiting, t- elsewhere)"])
+
+    def test_missing_priority_and_effort(self):
+        self.pending("- **t-a**: A.\n")
+        self.assertEqual(self.errors(), ["TASKS.md:7: t-a: no priority [P0]-[P3]",
+                                         "TASKS.md:7: t-a: no effort (want <1h, 1h, 5h, 10h, 100h)"])
+
+    def test_bad_priority_and_effort(self):
+        self.pending("- **t-a** [P7] (2h): A.\n")
+        self.assertEqual(self.errors(), ["TASKS.md:7: t-a: priority P7 (want P0-P3)",
+                                         "TASKS.md:7: t-a: effort '2h' (want <1h, 1h, 5h, 10h, 100h)"])
+
+    def test_dangling_link(self):
+        self.pending("- **t-a** [P1] (1h): A.\n  - see [[t-none]]\n")
+        self.assertEqual(self.errors(), ["TASKS.md:8: [[t-none]]: no such id in TASKS or archive"])
+
+    def test_dangling_link_in_prose_section(self):
+        self.tasks(CLEAN + "\n## Notes\n\nSee [[a-gone]] and `[[t-example]]`.\n")
+        self.assertEqual(self.errors(), ["TASKS.md:22: [[a-gone]]: no such id in TASKS or archive"])
+
+    def test_blocked_on_missing_awaiting_item(self):
+        self.pending("- **t-a** [P1] (1h) (blocked: [[a-none]]): A.\n")
+        self.assertEqual(self.errors(), ["TASKS.md:7: [[a-none]]: no such id in TASKS or archive",
+                                         "TASKS.md:7: t-a: blocked on 'a-none', which is not an open Awaiting item"])
+
+    def test_after_cycle(self):
+        self.pending("- **t-a** [P1] (1h): A.\n  - After: [[t-b]]\n\n- **t-b** [P1] (1h): B.\n  - After: [[t-a]]\n")
+        self.assertIn("TASKS.md:7: t-a: After: cycle t-a → t-b → t-a", self.errors())
+
+    def test_item_before_its_open_dependency(self):
+        self.pending("- **t-a** [P1] (1h): A.\n  - After: [[t-b]]\n\n- **t-b** [P1] (1h): B.\n")
+        self.assertEqual(self.errors(), ["TASKS.md:7: t-a: placed before 't-b', which it is After:"])
+
+    def test_after_bare_id_is_flagged(self):
+        self.pending("- **t-a** [P1] (1h): A.\n  After: t-b\n\n- **t-b** [P1] (1h): B.\n")
+        self.assertEqual(self.errors(), ["TASKS.md:8: t-a: After: 't-b' is not a link (want [[t-b]])"])
+
+    def test_slices_bare_id_is_flagged(self):
+        self.pending("- **t-a** [P1] (1h): A.\n  - Slices: [[t-b]], t-c\n\n- **t-b** [P1] (1h): B.\n")
+        self.assertEqual(self.errors(), ["TASKS.md:8: t-a: Slices: 't-c' is not a link (want [[t-c]])"])
+
+    def test_ref_path_missing(self):
+        self.pending("- **t-a** [P1] (1h): A.\n  Ref: docs/none.md\n")
+        self.assertEqual(self.errors(), ["TASKS.md:8: t-a: Ref 'docs/none.md' does not exist"])
+
+    def test_ref_anchor_missing(self):
+        self.pending("- **t-a** [P1] (1h): A.\n  Ref: DESIGN.md#nope\n")
+        self.assertEqual(self.errors(), ["TASKS.md:8: t-a: Ref 'DESIGN.md#nope': no such anchor"])
+
+    def test_old_numbered_items(self):
+        self.pending("1. **[P1] Old** (Effort: 1h) — goal.\n   - Steps\n")
+        self.assertEqual(self.errors(), ["TASKS.md:7: old numbered item (wf migrate)"])
+
+    def test_malformed_header_has_line(self):
+        self.pending("- **t-ok** [P1] (1h): Fine.\n\n- **t-x** [P1] (1h) no colon\n")
+        self.assertEqual(self.errors(),
+                         ["TASKS.md:9: t-x: bad header: want '- **id** [Pn] (effort) [(status)]: Title. Goal.'"])
+
+    def test_item_after_prose_is_flagged(self):
+        self.pending("- **t-a** [P1] (1h): A.\nstray\n- **t-b** [P1] (1h): B.\n")
+        self.assertEqual(self.errors(), ["TASKS.md:9: t-b: item after a flush-left prose line (line 8): indent or move the prose"])
+
+    def test_missing_section(self):
+        self.tasks("## Pending\n\n## Needs human\n\n## Deferred\n")
+        self.assertEqual(self.errors(), ["TASKS.md: no '## Awaiting your decision' section"])
+
+    def test_number_refs_warn(self):
+        self.pending("- **t-a** [P1] (1h): A, see #12.\n")
+        self.assertEqual(self.run_check(), ([], ["TASKS.md:7: '#12': number ref (tasks have ids: [[t-…]])"]))
+
+    def test_empty_progress_note_warns(self):
+        self.pending("- **t-a** [P1] (1h) (in progress: ): A.\n")
+        self.assertEqual(self.run_check(), ([], ["TASKS.md:7: t-a: in progress without a branch or note"]))
+
+    def test_doc_link_to_unknown_id(self):
+        (self.root / "docs" / "note.md").write_text("# N\n\nSee [[t-one]], [[t-done]], [[t-lost]], [[wiki-page]].\n")
+        self.assertEqual(self.errors(), ["docs/note.md:3: [[t-lost]]: no such id in TASKS or archive"])
+
+
+class StaleAwaitingTest(Base):
+    def commit(self, date):
+        import os
+        import subprocess
+        env = {**os.environ, "GIT_AUTHOR_DATE": date, "GIT_COMMITTER_DATE": date,
+               "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t"}
+        for cmd in (["init", "-q"], ["add", "-A"], ["commit", "-q", "-m", "x"]):
+            subprocess.run(["git", "-C", str(self.root), *cmd], check=True, env=env, capture_output=True)
+
+    def test_old_unreferenced_awaiting_item_warns(self):
+        self.tasks(CLEAN.replace("- **a-key**: Key needed. Which one?",
+                                 "- **a-key**: Key needed. Which one?\n\n- **a-old**: Old question. Still open?"))
+        self.commit("2020-01-01T00:00:00")
+        errors, warnings = self.run_check()
+        self.assertEqual(errors, [])
+        self.assertEqual(len(warnings), 1)
+        self.assertRegex(warnings[0], r"^TASKS.md:7: a-old: waiting \d+ days, no task references it$")
+
+    def test_fresh_item_does_not_warn(self):
+        self.tasks(CLEAN.replace("- **a-key**: Key needed. Which one?",
+                                 "- **a-key**: Key needed. Which one?\n\n- **a-new**: New question. Open?"))
+        import datetime
+        self.commit(datetime.datetime.now().isoformat(timespec="seconds"))
+        self.assertEqual(self.run_check(), ([], []))
+
+
+class ConfigCheckTest(Base):
+    def test_missing_tasks_file(self):
+        (self.root / "TASKS.md").unlink()
+        self.assertEqual(self.errors(), ["workflow.toml: tasks 'TASKS.md' does not exist"])
+
+    def test_missing_archive_and_docs(self):
+        (self.root / "tasks" / "archive.md").unlink()
+        (self.root / "DESIGN.md").unlink()
+        self.pending("- **t-a** [P1] (1h): A.\n")
+        self.assertEqual(self.errors(), ["workflow.toml: archive 'tasks/archive.md' does not exist",
+                                         "workflow.toml: docs 'DESIGN.md' does not exist"])
+
+
+class AnchorsCheckTest(Base):
+    toml = TOML + ANCHORS
+
+    def test_clean(self):
+        self.assertEqual(self.run_check(), ([], []))
+
+    def test_task_anchor_not_in_index_section(self):
+        self.pending("- **t-a** [P1] (1h): A.\n  Ref: DESIGN.md#notes\n")
+        self.assertEqual(self.errors(), ["TASKS.md:8: t-a: Ref 'DESIGN.md#notes' is not a heading under 'Subsystems'"])
+
+    def test_duplicate_index_slug(self):
+        (self.root / "DESIGN.md").write_text(DESIGN.replace("### input", "### Terrain"))
+        self.assertEqual(self.errors(), ["DESIGN.md:9: two 'Subsystems' headings give anchor 'terrain' (also line 5)"])
+
+    def test_index_link_to_missing_file_and_anchor(self):
+        (self.root / "DESIGN.md").write_text(DESIGN.replace("Keys.", "Keys. [a](docs/none.md) [b](docs/specs/terrain.md#nope) [c](https://x.y/z)"))
+        self.assertEqual(self.errors(), ["DESIGN.md:11: link 'docs/none.md': file does not exist",
+                                         "DESIGN.md:11: link 'docs/specs/terrain.md#nope': no such anchor"])
+
+    def test_explicit_spec_anchor_without_index_entry(self):
+        (self.root / "docs" / "specs" / "combat.md").write_text('# Combat\n\n<a id="combat"></a>\n## Combat\n')
+        self.assertEqual(self.errors(), ["docs/specs/combat.md:3: explicit anchor 'combat' has no heading under 'Subsystems' in DESIGN.md"])
+
+
+class ProblemTest(unittest.TestCase):
+    def test_key_ignores_line(self):
+        a = K.Problem("TASKS.md", 7, "t-a", "x")
+        b = K.Problem("TASKS.md", 9, "t-a", "x")
+        self.assertEqual(a.key, b.key)
+        self.assertEqual(str(a), "TASKS.md:7: t-a: x")
+        self.assertEqual(str(K.Problem("workflow.toml", None, "", "y")), "workflow.toml: y")
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class MergedToolWorktreeTest(unittest.TestCase):
+    def git(self, root, *a):
+        import subprocess
+        subprocess.run(["git", "-C", str(root), "-c", "user.name=t", "-c", "user.email=t@t", *a],
+                       check=True, capture_output=True)
+
+    def test_flags_only_merged_worktrees(self):
+        with tempfile.TemporaryDirectory() as d:
+            root = Path(d) / "tool"
+            root.mkdir()
+            self.git(root, "init", "-b", "master")
+            self.git(root, "commit", "--allow-empty", "-m", "a")
+            self.git(root, "worktree", "add", str(Path(d) / "fresh"), "-b", "fresh")
+            self.git(root, "worktree", "add", str(Path(d) / "done"), "-b", "done")
+            self.git(root, "worktree", "add", str(Path(d) / "open"), "-b", "open")
+            self.git(Path(d) / "open", "commit", "--allow-empty", "-m", "b")
+            self.git(root, "commit", "--allow-empty", "-m", "c")
+            old = time.time() - 3 * 3600
+            for n in ("done", "open"):
+                os.utime(Path(d) / n / ".git", (old, old))
+            got = K.merged_tool_worktrees(root)
+            names = sorted(p.message.split("branch ")[1].split(")")[0] for p in got)
+            self.assertEqual(names, ["done"])  # fresh (<2h) skipped
+
+    def test_not_git(self):
+        with tempfile.TemporaryDirectory() as d:
+            self.assertEqual(K.merged_tool_worktrees(Path(d)), [])