1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
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()
|
