import sys import tempfile import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from wflib import config, lanes as L from wflib import config as C class ConfigTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name).resolve() def tearDown(self): self.tmp.cleanup() def write(self, text): (self.root / "workflow.toml").write_text(text) def test_find_root_from_nested_dir(self): self.write('format = 1\ntasks = "TASKS.md"\narchive = "a.md"\n') deep = self.root / "a" / "b" deep.mkdir(parents=True) self.assertEqual(C.find_root(deep), self.root) def test_find_root_none(self): with self.assertRaisesRegex(C.ConfigError, "no workflow.toml in .* or above"): C.find_root(self.root) def test_minimal(self): self.write('format = 1\ntasks = "TASKS.md"\narchive = "tasks/archive.md"\n') cfg = C.load(self.root) self.assertEqual((cfg.format, cfg.tasks, cfg.archive), (1, self.root / "TASKS.md", self.root / "tasks/archive.md")) self.assertEqual((cfg.docs, cfg.verify, cfg.done, cfg.ledgers, cfg.anchors_index), ([], [], [], None, None)) def test_full(self): self.write('format = 1\ntasks = "docs/TASKS.md"\narchive = "docs/DONE.md"\ndocs = ["DESIGN.md", "docs/"]\n' 'verify = ["make test"]\ndone = ["update CATALOG"]\nledgers = ".superpowers/sdd"\n' '[anchors]\nindex = "DESIGN.md"\nindex_section = "Subsystems"\nspecs = "docs/specs"\n') cfg = C.load(self.root) self.assertEqual(cfg.docs, [self.root / "DESIGN.md", self.root / "docs"]) self.assertEqual((cfg.verify, cfg.done), (["make test"], ["update CATALOG"])) self.assertEqual(cfg.ledgers, self.root / ".superpowers/sdd") self.assertEqual((cfg.anchors_index, cfg.anchors_section, cfg.anchors_specs), (self.root / "DESIGN.md", "Subsystems", self.root / "docs/specs")) def test_cloud_keys(self): self.write('format = 1\ntasks = "T.md"\narchive = "a.md"\n') cfg = C.load(self.root) self.assertEqual((cfg.cloud, cfg.cloud_include, cfg.cloud_note), (False, [], None)) self.write('format = 1\ntasks = "T.md"\narchive = "a.md"\ncloud = true\ncloud_include = ["out/p"]\n' 'cloud_note = "data in out/p"\n') cfg = C.load(self.root) self.assertEqual((cfg.cloud, cfg.cloud_include, cfg.cloud_note), (True, ["out/p"], "data in out/p")) self.write('format = 1\ntasks = "T.md"\narchive = "a.md"\ncloud = "yes"\n') with self.assertRaisesRegex(C.ConfigError, "'cloud' must be true or false"): C.load(self.root) def test_unknown_key(self): self.write('format = 1\ntasks = "T.md"\narchive = "a.md"\nverfy = []\n') with self.assertRaisesRegex(C.ConfigError, "workflow.toml: unknown key 'verfy'"): C.load(self.root) def test_unknown_anchors_key(self): self.write('format = 1\ntasks = "T.md"\narchive = "a.md"\n[anchors]\nindx = "D.md"\n') with self.assertRaisesRegex(C.ConfigError, r"unknown key 'anchors.indx'"): C.load(self.root) def test_missing_required(self): self.write('format = 1\narchive = "a.md"\n') with self.assertRaisesRegex(C.ConfigError, "workflow.toml: missing 'tasks'"): C.load(self.root) def test_wrong_type(self): self.write('format = 1\ntasks = "T.md"\narchive = "a.md"\nverify = "make"\n') with self.assertRaisesRegex(C.ConfigError, "'verify' must be a list of strings"): C.load(self.root) def test_bad_toml(self): self.write('format = \n') with self.assertRaisesRegex(C.ConfigError, "workflow.toml: "): C.load(self.root) class LaneConfigTest(unittest.TestCase): def load(self, extra: str): d = Path(tempfile.mkdtemp()) self.addCleanup(__import__("shutil").rmtree, d) (d / "workflow.toml").write_text('format = 1\ntasks = "TASKS.md"\narchive = "a.md"\n' + extra) return config.load(d) def test_default_lanes(self): cfg = self.load("") self.assertEqual(cfg.slice_above, "1h") self.assertEqual(cfg.lanes, (L.Lane("fast", ("<1h",), "unblock", None, True), L.Lane("slow", ("1h",), "priority", "fast", False))) self.assertEqual(cfg.area_stale_commits, 20) self.assertEqual(cfg.areas_file, cfg.root / "CLAUDE.md") def test_custom_lanes_replace_default(self): cfg = self.load('slice_above = "5h"\n[lanes.quick]\nefforts = ["<1h", "1h"]\norder = "unblock"\n' '[lanes.long]\nefforts = ["5h"]\nfallback = "quick"\n') self.assertEqual(cfg.lanes, (L.Lane("quick", ("<1h", "1h"), "unblock", None, True), L.Lane("long", ("5h",), "priority", "quick", False))) def test_lane_errors(self): bad = { '[lanes.a]\nefforts = ["<1h", "1h"]\n[lanes.b]\nefforts = ["1h"]\n': "effort '1h' in lanes a and b", '[lanes.a]\nefforts = ["<1h"]\n': "effort '1h' (≤ slice_above) in no lane", '[lanes.a]\nefforts = ["<1h"]\nslices = true\n[lanes.b]\nefforts = ["1h"]\nslices = true\n': "slices = true on more than one lane", '[lanes.a]\nefforts = ["<1h"]\nfallback = "b"\n[lanes.b]\nefforts = ["1h"]\nfallback = "a"\n': "fallback cycle: a ↔ b", '[lanes.a]\nefforts = ["<1h", "1h", "5h"]\n': "lane a: effort '5h' above slice_above '1h'", '[lanes.a]\nefforts = ["<1h", "1h"]\norder = "fifo"\n': "lane a: order 'fifo' (want unblock, priority)", '[lanes.all]\nefforts = ["<1h", "1h"]\n': "lane name 'all' is reserved", '[lanes.A]\nefforts = ["<1h", "1h"]\n': "bad lane name 'A'", '[lanes.a]\nefforts = ["<1h", "1h"]\nspeed = 1\n': "unknown key 'lanes.a.speed'", 'slice_above = "2h"\n': "slice_above '2h' (want <1h, 1h, 5h, 10h, 100h)", 'area_stale_commits = 0\n': "'area_stale_commits' must be a number ≥ 1", } for extra, msg in bad.items(): with self.subTest(msg=msg), self.assertRaises(config.ConfigError) as cm: self.load(extra) self.assertIn(msg, str(cm.exception)) if __name__ == "__main__": unittest.main()