Coverage for src/footman/split.py: 97%
215 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 13:38 +0000
1"""Separator-free chain splitting, driven purely by the manifest.
3`fm build lint --fix test` is split into three independent segments with no
4separator at all — duty's muscle memory, but with real flags and positionals.
5The manifest gives the splitter exact knowledge of every task's shape, which
6makes the split deterministic under six rules (see `NOTES`):
81. params with defaults are options, never positionals (the load-bearing rule);
92. required positionals are consumed by exact arity, eagerly validated;
103. options bind to their own segment;
114. list options repeat the flag (`--tag a --tag b`);
125. variadic / `--` passthrough segments are terminal; `+` is the always
13 available explicit boundary;
146. globals precede the first task name.
16Every error names the task, states the expectation, and proposes the fix —
17error messages are product surface here, not diagnostics.
18"""
20from __future__ import annotations
22import difflib
23from collections.abc import Iterable
24from dataclasses import dataclass, field
25from pathlib import Path
26from typing import Any
28from footman import coerce
31def _did_you_mean(word: str, known: Iterable[str]) -> str:
32 """A ` — did you mean 'x'?` suffix when *word* closely matches a known name.
34 Empty when nothing is close, so a genuine typo never gets false-confident
35 advice. The one idiom behind every not-found message (task, option, choice).
36 """
37 close = difflib.get_close_matches(word, list(known), n=1)
38 return f" — did you mean {close[0]!r}?" if close else ""
41def _misplaced_global(token: str) -> str | None:
42 """The teaching message when *token* is really one of the GLOBALS.
44 A global option found after a task name is a position mistake, not an
45 unknown name — so name the real problem and its fix instead of guessing
46 at close matches. Only fires for *unknown* task options: a task param
47 that shares a global's name still wins by position, as it should.
48 """
49 name = token.split("=", 1)[0]
50 if name not in _GLOBAL_KIND:
51 return None
52 canon = _CANON.get(name, name)
53 label = name if name == canon else f"{name} ({canon})"
54 return f"{label} is a global option — it goes before the first task name"
57class ChainError(Exception):
58 """A malformed command line, carrying a teaching message for the user."""
61# Global options bind to `fm` itself and must precede the first task name
62# (`--help`/`-h` is the one exception: anywhere before `--`, it wins).
63# (canonical, short alias, kind, value-hint, help)
64GLOBALS: list[tuple[str, str | None, str, str | None, str]] = [
65 ("--help", "-h", "flag", None, "help for {prog}, or the named group/task"),
66 ("--version", "-V", "flag", None, "print the version and exit"),
67 ("--list", "-l", "flag", None, "list tasks (flat)"),
68 ("--tree", None, "flag", None, "list tasks grouped by command group"),
69 ("--where", None, "option", "TASK", "print the task's source file:line"),
70 ("--dry-run", "-n", "flag", None, "print the parsed plan without running"),
71 ("--keep-going", "-k", "flag", None, "run every branch even if one fails"),
72 ("--fail-fast", None, "flag", None, "stop at the first failure"),
73 ("--sequential", "-s", "flag", None, "run one at a time (default: parallel)"),
74 ("--jobs", "-j", "option", "N", "max parallel tasks (default: cores - 1)"),
75 ("--yes", "-y", "flag", None, "assume yes to every confirm() gate"),
76 ("--no-input", None, "flag", None, "never prompt; error if input is required"),
77 ("--quiet", "-q", "flag", None, "suppress the per-task summary"),
78 ("--verbose", "-v", "flag", None, "replay captured output even on success"),
79 ("--color", None, "option", "WHEN", "when to colour: always|never|auto (default)"),
80 ("--no-color", None, "flag", None, "disable ANSI colour (same as --color=never)"),
81 ("--no-progress", None, "flag", None, "no progress bar, eta, or timing capture"),
82 ("--json", None, "flag", None, "stdout is one JSON document (captures output)"),
83 ("--timings", None, "flag", None, "show per-task durations"),
84 ("--directory", "-C", "option", "PATH", "run as if launched from PATH"),
85 ("--tasks-file", "-f", "option", "PATH", "only this tasks file, no tasks cascade"),
86 ("--config", None, "option", "PATH", "only this config file, no config cascade"),
87 # "option?": the value is optional — bare `--install-completion` /
88 # `--setup-completion` detect the invoking shell.
89 ("--install-completion", None, "option?", "[SHELL]", "install shell completion"),
90 ("--setup-completion", None, "option?", "[SHELL]", "print completion for eval"),
91 (
92 "--uninstall-completion",
93 None,
94 "option?",
95 "[SHELL]",
96 "remove the completion hook",
97 ),
98]
99_GLOBAL_KIND = {name: kind for name, _, kind, _, _ in GLOBALS}
100_GLOBAL_KIND.update({alias: kind for _, alias, kind, _, _ in GLOBALS if alias})
101_CANON = {alias: name for name, alias, _, _, _ in GLOBALS if alias}
104@dataclass
105class Segment:
106 """One resolved task invocation within a chain."""
108 task: str # dotted path, e.g. "docs.build"
109 path: list[str] # ["docs", "build"]
110 values: dict[str, Any] = field(default_factory=dict) # cli-name -> value
111 variadic: list[str] = field(default_factory=list)
112 passthrough: list[str] | None = None
115def _required_label(p: dict) -> str:
116 """Label a required option for the missing-option error — a flag teaches
117 both its `--x` and `--no-x` forms."""
118 name = f"--{p['name']}"
119 return f"{name} (or --no-{p['name']})" if p["kind"] == "flag" else name
122def _suggest_only(choices: list | None, dynamic: dict | None) -> bool:
123 """Whether a completer only *suggests* (never rejects): a soft completer
124 (`strict=False`), or a strict one whose candidate list is empty — the
125 completer genuinely returned nothing, and a *failing* strict completer
126 aborts the manifest build instead, so rejecting every value would brick
127 the task."""
128 return bool(dynamic) and (not dynamic.get("strict") or not choices)
131def _check(
132 where: str,
133 label: str,
134 value: str,
135 *,
136 choices: list | None = None,
137 types: list | None = None,
138 dynamic: dict | None = None,
139 path: str | None = None,
140 bounds: tuple | None = None,
141) -> None:
142 """Validate one string against choices or type tags; raise a taught error."""
143 if choices is not None:
144 if _suggest_only(choices, dynamic):
145 return
146 if value in choices:
147 return # an exact choice needs no further type/bounds checks
148 # A union like `Literal['fast','slow'] | int` carries both choices and
149 # types: accept a value that matches either, and only teach both when
150 # neither fits.
151 if not (types and coerce.coerce_scalar(value, types)[0]):
152 listing = "|".join(choices) if choices else "(none available)"
153 extra = f", or {coerce.type_phrase(types)}" if types else ""
154 hint = _did_you_mean(value, choices)
155 raise ChainError(
156 f"{where}: {label} must be one of {listing}{extra} "
157 f"(got {value!r}){hint}"
158 )
159 # matched via the type path -> fall through to bounds/path below
160 elif types and not coerce.coerce_scalar(value, types)[0]:
161 expected = coerce.type_phrase(types)
162 raise ChainError(f"{where}: {label} expects {expected} (got {value!r})")
163 if path is not None:
164 _check_path(where, label, value, path)
165 if bounds is not None:
166 _check_bounds(where, label, value, types, bounds)
169_PATH_PHRASE = {
170 "exists": ("an existing path", Path.exists),
171 "file": ("an existing file", Path.is_file),
172 "dir": ("an existing directory", Path.is_dir),
173}
176def _check_path(where: str, label: str, value: str, req: str) -> None:
177 phrase, test = _PATH_PHRASE[req]
178 if not test(Path(value)):
179 raise ChainError(f"{where}: {label} must be {phrase} (got {value!r})")
182def _check_bounds(
183 where: str, label: str, value: str, types: list | None, bounds: tuple
184) -> None:
185 ok, number = coerce.coerce_scalar(value, types or ["int", "float"])
186 if not ok or isinstance(number, bool) or not isinstance(number, (int, float)): 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true
187 return # the types check above already taught the type error
188 lo, hi = bounds
189 # Negated comparisons so NaN (which compares False to everything, so `< lo`
190 # and `> hi` are both False) is rejected, not silently accepted; identical
191 # to the plain comparisons for every real number.
192 if (lo is not None and not (number >= lo)) or (
193 hi is not None and not (number <= hi)
194 ):
195 expect = (
196 f"at least {lo}"
197 if hi is None
198 else f"at most {hi}"
199 if lo is None
200 else f"between {lo} and {hi}"
201 )
202 raise ChainError(f"{where}: {label} must be {expect} (got {value!r})")
205def _validate(where: str, p: dict, value: str) -> None:
206 """Eagerly validate a choice/typed value; raise a taught error if wrong."""
207 label = (
208 f"<{p['name']}>" if p["kind"] in ("argument", "variadic") else f"--{p['name']}"
209 )
210 bounds = (p.get("min"), p.get("max")) if "min" in p or "max" in p else None
211 _check(
212 where,
213 label,
214 value,
215 choices=p.get("choices"),
216 types=p.get("types"),
217 dynamic=p.get("dynamic"),
218 path=p.get("path"),
219 bounds=bounds,
220 )
223def _parse_globals(argv: list[str], i: int) -> tuple[list[str], int]:
224 globals_: list[str] = []
225 while i < len(argv) and argv[i].startswith("-") and argv[i] != "--":
226 name = argv[i].split("=", 1)[0]
227 if name not in _GLOBAL_KIND:
228 raise ChainError(
229 f"unknown global option {name} "
230 f"(global options go before the first task)"
231 )
232 kind = _GLOBAL_KIND[name]
233 if kind == "flag" and "=" in argv[i]:
234 raise ChainError(f"{_CANON.get(name, name)} is a flag and takes no value")
235 globals_.append(_CANON.get(name, name) + argv[i][len(name) :])
236 i += 1
237 if kind == "option" and "=" not in globals_[-1]:
238 if i >= len(argv): 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 raise ChainError(f"{name} expects a value")
240 globals_.append(argv[i])
241 i += 1
242 elif kind == "option?" and "=" not in globals_[-1]:
243 # Optional value: consume the next word only when one is present
244 # and not option-shaped; normalise to --name=value so downstream
245 # can tell "given with value" from "given bare".
246 if i < len(argv) and not argv[i].startswith("-"):
247 globals_[-1] += f"={argv[i]}"
248 i += 1
249 return globals_, i
252def split_chain(tree: dict, argv: list[str]) -> tuple[list[str], list[Segment]]:
253 """Split *argv* into leading globals and a list of resolved segments."""
254 globals_, i = _parse_globals(argv, 0)
255 segments: list[Segment] = []
257 while i < len(argv):
258 node, path = tree, []
259 while i < len(argv) and argv[i] in node["groups"]:
260 path.append(argv[i])
261 node = node["groups"][argv[i]]
262 i += 1
263 if i < len(argv) and argv[i] in node["tasks"]:
264 task = node["tasks"][argv[i]]
265 path.append(argv[i])
266 i += 1
267 elif "default" in node:
268 # A runnable group (one with `@group.default`) resolves to its
269 # default action: `fm lint` / `fm lint --fix` run it. `path` stays
270 # the group's; `i` is not advanced, so the flag loop below parses
271 # the default's options. A default takes no positionals, so a
272 # trailing non-flag token (`fm lint test`) ends the segment and
273 # opens a fresh target on the next pass.
274 task = node["default"]
275 else:
276 got = argv[i] if i < len(argv) else "(end of line)"
277 if i < len(argv) and (misplaced := _misplaced_global(got)) is not None:
278 raise ChainError(misplaced)
279 scope = " ".join(path)
280 where = f"{scope}: " if scope else ""
281 names = list(node["groups"]) + list(node["tasks"])
282 hint = _did_you_mean(got, names) if i < len(argv) else ""
283 known = ", ".join(names)
284 raise ChainError(
285 f"{where}expected a task name, got {got!r}{hint} (know: {known})"
286 )
288 opts = {
289 "--" + p["name"]: p
290 for p in task["params"]
291 if p["kind"] in ("flag", "option")
292 }
293 # Exact-arity positionals, then a single trailing consumer for the rest:
294 # a typed multiple/one-or-many positional, or a `*args` variadic.
295 fixed = [
296 p
297 for p in task["params"]
298 if p["kind"] == "argument" and not p.get("multiple")
299 ]
300 rest = next(
301 (
302 p
303 for p in task["params"]
304 if (p["kind"] == "argument" and p.get("multiple"))
305 or p["kind"] == "variadic"
306 ),
307 None,
308 )
309 seg = Segment(task=".".join(path), path=list(path))
310 filled = 0
311 rest_count = 0
313 while i < len(argv):
314 tok = argv[i]
315 if tok == "+": # explicit segment boundary
316 i += 1
317 break
318 if tok == "--": # passthrough is terminal for the whole line
319 seg.passthrough = argv[i + 1 :]
320 i = len(argv)
321 break
322 if tok.startswith("--"):
323 i = _consume_option(seg, opts, argv, i)
324 elif filled < len(fixed):
325 _consume_positional(seg, tree, fixed[filled], tok)
326 filled += 1
327 i += 1
328 elif rest is not None:
329 if rest["kind"] == "variadic":
330 _validate(seg.task, rest, tok) # eager, like every positional
331 seg.variadic.append(tok)
332 else:
333 _consume_positional(seg, tree, rest, tok)
334 rest_count += 1
335 i += 1
336 else:
337 break # arity satisfied: the next word starts a new segment
339 missing = [f"<{p['name']}>" for p in fixed[filled:]]
340 if rest is not None and rest["kind"] == "argument" and rest_count == 0:
341 missing.append(f"<{rest['name']}>")
342 if missing:
343 raise ChainError(
344 f"{seg.task}: missing required argument(s): {', '.join(missing)}"
345 )
347 # Required options — a mapping or bool with no default. (Dicts are only
348 # ever options; a bool is a flag, so teach both --x and --no-x forms.)
349 missing_opts = [
350 _required_label(p)
351 for p in task["params"]
352 if p.get("required") and p["name"] not in seg.values
353 ]
354 if missing_opts:
355 raise ChainError(
356 f"{seg.task}: missing required option(s): {', '.join(missing_opts)}"
357 )
358 segments.append(seg)
360 return globals_, segments
363def _consume_option(seg: Segment, opts: dict, argv: list[str], i: int) -> int:
364 tok = argv[i]
365 name = tok.split("=", 1)[0]
366 negated = False
367 p = opts.get(name)
368 if p is None and name.startswith("--no-"):
369 candidate = "--" + name[len("--no-") :]
370 if candidate in opts and opts[candidate]["kind"] == "flag": 370 ↛ 372line 370 didn't jump to line 372 because the condition on line 370 was always true
371 p, negated = opts[candidate], True
372 if p is None:
373 if (misplaced := _misplaced_global(name)) is not None:
374 raise ChainError(f"{seg.task}: {misplaced}")
375 forms = list(opts) + [
376 f"--no-{opts[k]['name']}" for k in opts if opts[k]["kind"] == "flag"
377 ]
378 hint = _did_you_mean(name, forms) or (
379 " (task options come right after their task; "
380 "globals go before the first task)"
381 )
382 raise ChainError(f"{seg.task}: unknown option {name}{hint}")
384 cli = p["name"]
385 if p["kind"] == "flag":
386 if "=" in tok: 386 ↛ 387line 386 didn't jump to line 387 because the condition on line 386 was never true
387 raise ChainError(f"{seg.task}: --{cli} is a flag and takes no value")
388 seg.values[cli] = not negated
389 return i + 1
391 # value-bearing option
392 if "=" in tok:
393 value = tok.split("=", 1)[1]
394 i += 1
395 else:
396 i += 1
397 if i >= len(argv): 397 ↛ 398line 397 didn't jump to line 398 because the condition on line 397 was never true
398 raise ChainError(f"{seg.task}: {name} expects a value")
399 if argv[i] == "--":
400 raise ChainError(
401 f"{seg.task}: {name} expects a value, but found '--' — give "
402 f"{name} a value, or use {name}=-- if the literal is intended"
403 )
404 value = argv[i]
405 i += 1
406 if p.get("mapping"):
407 for pair in _values(p, value):
408 _consume_pair(seg, p, cli, pair)
409 elif p.get("multiple"):
410 for part in _values(p, value):
411 _validate(seg.task, p, part)
412 seg.values.setdefault(cli, []).append(part)
413 else:
414 _validate(seg.task, p, value)
415 seg.values[cli] = value
416 return i
419def _values(p: dict, value: str) -> list[str]:
420 """Comma-split parts of a list/dict value, unless the param opts out.
422 Called only for collection params, so splitting is the default; a `nosplit`
423 param (values may contain commas) takes the whole token verbatim.
424 """
425 if p.get("nosplit"):
426 return [value]
427 return [part for part in value.split(",") if part] or [value]
430def _consume_pair(seg: Segment, p: dict, cli: str, pair: str) -> None:
431 """Parse and validate one `KEY=VALUE` token for a dict parameter."""
432 if "=" not in pair:
433 raise ChainError(f"{seg.task}: --{cli} expects KEY=VALUE (got {pair!r})")
434 key, value = pair.split("=", 1)
435 bounds = (p.get("min"), p.get("max")) if "min" in p or "max" in p else None
436 _check(seg.task, f"--{cli} key", key, types=p.get("key_types"))
437 _check(
438 seg.task,
439 f"--{cli} value",
440 value,
441 choices=p.get("value_choices"),
442 types=p.get("value_types"),
443 path=p.get("path"),
444 bounds=bounds,
445 )
446 seg.values.setdefault(cli, []).append((key, value))
449def _consume_positional(seg: Segment, tree: dict, p: dict, tok: str) -> None:
450 if (
451 "choices" in p
452 and tok not in p["choices"]
453 and not _suggest_only(p["choices"], p.get("dynamic"))
454 and not (p.get("types") and coerce.coerce_scalar(tok, p["types"])[0])
455 and (tok in tree["tasks"] or tok in tree["groups"])
456 ):
457 raise ChainError(
458 f"{seg.task}: <{p['name']}> must be one of "
459 f"{'|'.join(p['choices'])} — {tok!r} looks like the next task; "
460 f"did you forget <{p['name']}>?"
461 )
462 if p.get("multiple"):
463 for part in _values(p, tok):
464 _validate(seg.task, p, part)
465 seg.values.setdefault(p["name"], []).append(part)
466 else:
467 _validate(seg.task, p, tok)
468 seg.values[p["name"]] = tok