Coverage for src/footman/executor.py: 91%
272 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 13:38 +0000
1"""Bind resolved segments to their task functions and run them.
3The splitter validated the command line against the manifest and produced
4string-valued `Segment` objects. Here — on the execution
5path, with the user's module imported — we resolve each segment to its real
6function, coerce the strings to the annotated types, and call it.
8Coercion covers what the manifest grammar promises: `int`/`float`, `Path`,
9`Enum`/`Literal` choices, `list[...]` (repeatable), and `*args` variadic
10(which also receives anything after `--`). A task "fails" if it raises or
11returns a non-zero `int` exit code; failures stop the chain unless
12`--keep-going` is set.
13"""
15from __future__ import annotations
17import enum
18import inspect
19import io
20import os
21import time
22from dataclasses import dataclass, field
23from pathlib import Path, PurePath
24from types import MappingProxyType
25from typing import Any
27from footman import coerce, context, registry
28from footman.context import (
29 Context,
30 Failed,
31 Result,
32 RunFailed,
33 _current,
34 context_param_name,
35)
36from footman.discover import defining_dir
37from footman.manifest import resolved_signature
38from footman.registry import Group, Task
39from footman.split import ChainError, Segment
42@dataclass
43class TaskResult:
44 """Outcome of running one segment."""
46 task: str
47 ok: bool
48 code: int = 0
49 returned: Any = None
50 error: BaseException | None = None
51 duration: float = 0.0
52 output: str = ""
53 steps: list[Result] = field(default_factory=list)
54 cancelled: bool = False # failed only because fail-fast killed it mid-run
57def resolve(root: Group, path: list[str]) -> Task:
58 """Walk *path* (`["docs", "build"]`) to its task function.
60 A path that lands on a runnable group (`["lint"]`) resolves to that group's
61 default action — the function `@group.default` registered.
62 """
63 node = root
64 for name in path[:-1]:
65 node = node.groups[name]
66 last = path[-1]
67 if last in node.tasks:
68 return node.tasks[last]
69 group = node.groups[last]
70 if group.default_task is None: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true
71 raise KeyError(last) # a non-runnable group is never a segment target
72 return group.default_task
75_MISSING = object()
78def _wants_context(fn: Any) -> bool:
79 """True when a validator accepts a second positional argument — the sibling
80 parameters coerced so far. Decided by *inspecting* the signature, never by
81 catching a `TypeError` from the call, so a real arity error raised inside the
82 validator is not mistaken for the one-argument form."""
83 try:
84 params = inspect.signature(fn).parameters.values()
85 except (TypeError, ValueError):
86 return False # a builtin/C callable with no signature — treat as one-arg
87 positional = 0
88 for p in params:
89 if p.kind is p.VAR_POSITIONAL:
90 return True
91 if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD): 91 ↛ 88line 91 didn't jump to line 88 because the condition on line 91 was always true
92 positional += 1
93 return positional >= 2
96def _run_checks(
97 value: Any, peeled: coerce.Peeled, label: str, params: dict[str, Any] | None = None
98) -> Any:
99 """Apply `check(fn)` validators to one coerced value (element-level).
101 A validator declaring a second argument receives the sibling parameters
102 already coerced (those to its left in the signature), read-only — so it can
103 validate against another input, e.g. a version against the current release of
104 the package named in an earlier parameter."""
105 view: MappingProxyType[str, Any] | None = None
106 for fn in peeled.checks:
107 try:
108 if _wants_context(fn):
109 if view is None: 109 ↛ 111line 109 didn't jump to line 111 because the condition on line 109 was always true
110 view = MappingProxyType(dict(params) if params else {})
111 fn(value, view)
112 else:
113 fn(value)
114 except ValueError as exc:
115 raise ValueError(f"{label}: {exc}") from exc
116 return value
119def _validate_value(value: Any, peeled: coerce.Peeled, label: str) -> Any:
120 """Validate a value the splitter never saw (env fallback, variadic /
121 passthrough token) against the constraints it would have enforced eagerly
122 for a CLI token (choices, bounds, path)."""
123 choices = coerce.all_choices(peeled.element)
124 if choices is not None: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 shown = str(value.value) if isinstance(value, enum.Enum) else str(value)
126 tags = coerce.element_tags(peeled.element)
127 # A mixed union (`Literal['a','b'] | int`) accepts a choice member OR a
128 # value that coerces to one of its tags — reject only when neither fits.
129 type_ok = bool(tags) and coerce.coerce_scalar(str(value), tags)[0]
130 if shown not in choices and not type_ok:
131 extra = f", or {coerce.type_phrase(tags)}" if tags else ""
132 raise ValueError(
133 f"{label} must be one of {'|'.join(choices)}{extra} (got {value!r})"
134 )
135 if (
136 peeled.bounds is not None
137 and isinstance(value, (int, float))
138 and not isinstance(value, bool)
139 ):
140 lo, hi = peeled.bounds
141 # Negated form rejects NaN (compares False to everything), matching the
142 # splitter's eager bounds check; identical to </> for real numbers.
143 if (lo is not None and not (value >= lo)) or (
144 hi is not None and not (value <= hi)
145 ):
146 raise ValueError(f"{label} must be between {lo} and {hi} (got {value!r})")
147 if peeled.path_req is not None and isinstance(value, PurePath): 147 ↛ 148line 147 didn't jump to line 148 because the condition on line 147 was never true
148 tests = {"exists": Path.exists, "file": Path.is_file, "dir": Path.is_dir}
149 if not tests[peeled.path_req](Path(value)):
150 raise ValueError(f"{label}: {value} does not satisfy {peeled.path_req}")
151 return value
154def _coerce_extra(
155 token: str, peeled: coerce.Peeled, label: str, params: dict[str, Any] | None = None
156) -> Any:
157 """Coerce + validate one token the splitter never validated (an env
158 fallback or a `--` passthrough value): strict coercion, then the same
159 choices / bounds / path / check(fn) checks a CLI token gets."""
160 try:
161 value = coerce.coerce_token(token, peeled.element)
162 except ValueError as exc:
163 raise ValueError(f"{label} {exc}") from exc
164 return _run_checks(_validate_value(value, peeled, label), peeled, label, params)
167def _env_value(
168 param: inspect.Parameter,
169 peeled: coerce.Peeled,
170 params: dict[str, Any] | None = None,
171) -> Any:
172 """The env-fallback path for an absent option: CLI beats env beats default.
174 The env string flows through the same coercion, bounds, choices, and
175 `check(fn)` validators a CLI token would — it just runs at binding time
176 (the splitter never sees the environment).
177 """
178 raw = os.environ.get(peeled.env) if peeled.env is not None else None
179 if raw is None:
180 return _MISSING
181 label = f"--{param.name.replace('_', '-')} (from ${peeled.env})"
183 def one(token: str) -> Any:
184 return _coerce_extra(token, peeled, label, params)
186 if peeled.multiple:
187 parts = [raw] if peeled.nosplit else [p for p in raw.split(",") if p] or [raw]
188 return [one(p) for p in parts]
189 return one(raw)
192def _prompt_param(
193 cli: str,
194 peeled: coerce.Peeled,
195 ctx: Context | None,
196 params: dict[str, Any] | None = None,
197) -> Any:
198 """Resolve a defaultless `ask()` parameter by prompting, coercing the answer
199 through the same pipeline as a CLI token and re-asking on a bad value. Off a
200 terminal or under `--no-input`/`--json` it raises instead — the value must
201 then be supplied on the command line."""
202 marker = peeled.ask
203 assert marker is not None # bind only calls this when ask() is present
204 if (ctx is not None and ctx.no_input) or not context._stdin_is_tty():
205 raise ValueError(
206 f"--{cli} is required and nothing supplied it — pass --{cli} "
207 f"(a terminal is needed to prompt; --no-input and --json never ask)."
208 )
209 choices = coerce.all_choices(peeled.element)
210 hint = f" ({'/'.join(choices)})" if choices else ""
211 text = marker.prompt or f"{cli}{hint}: "
212 while True:
213 raw = context._prompt_core(text, secret=marker.secret)
214 if choices is not None and raw not in choices:
215 out = context.real_stderr()
216 out.write(f" choose one of {', '.join(choices)}\n")
217 out.flush()
218 continue
219 try:
220 value = coerce.coerce_token(raw, peeled.element)
221 return _run_checks(value, peeled, f"--{cli}", params)
222 except ValueError as exc:
223 out = context.real_stderr()
224 out.write(f" {exc}\n")
225 out.flush()
228def _left_siblings(
229 sig: inspect.Signature, current: inspect.Parameter, kwargs: dict[str, Any]
230) -> dict[str, Any]:
231 """The effective values of the parameters to *current*'s left — a provided
232 value where one was resolved, else the parameter's own default — so a
233 contextual `check` reads what the body will actually receive, never a copy of
234 the default that can drift out of sync."""
235 view: dict[str, Any] = {}
236 for p in sig.parameters.values(): 236 ↛ 243line 236 didn't jump to line 243 because the loop on line 236 didn't complete
237 if p.name == current.name:
238 break
239 if p.name in kwargs:
240 view[p.name] = kwargs[p.name]
241 elif p.default is not inspect.Parameter.empty:
242 view[p.name] = p.default
243 return view
246def bind(
247 seg: Segment,
248 fn: Task,
249 ctx: Context | None = None,
250 forwarded: dict[str, Any] | None = None,
251) -> tuple[list[Any], dict[str, Any]]:
252 """Turn a segment's string values into `(*args, **kwargs)` for *fn*.
254 Coercion (union member selection, list handling, one-or-many collapse) goes
255 through `footman.coerce`, the same module the manifest and splitter use.
256 `check(fn)` validators run here on the coerced values, and absent options
257 fall back to their `env()` variable before their default.
259 *forwarded* carries values a dispatching task passed down via the `forward`
260 marker. Precedence is CLI value > forwarded > env > default: a forwarded
261 value overrides only a parameter that *has* a default (it never rescues a
262 required one — a prerequisite must still be independently runnable).
263 """
264 sig = resolved_signature(fn)
265 empty = inspect.Parameter.empty
266 var_args: list[Any] = []
267 kwargs: dict[str, Any] = {}
269 for param in sig.parameters.values():
270 # The parameters bound to this one's left, at their effective values,
271 # for a contextual check(fn, params).
272 siblings = _left_siblings(sig, param, kwargs)
273 if param.kind is inspect.Parameter.VAR_POSITIONAL:
274 extra = [*seg.variadic, *(seg.passthrough or [])]
275 if param.annotation is empty: 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true
276 var_args = list(extra)
277 else:
278 peeled = coerce.peel(param.annotation)
279 label = f"<{param.name}>"
280 var_args = [_coerce_extra(v, peeled, label, siblings) for v in extra]
281 continue
283 cli = registry.cli_name(param.name)
284 if cli not in seg.values:
285 # A forwarded value overrides a defaulted parameter (never a
286 # required one — the guard on `param.default`), ahead of env/default.
287 if (
288 forwarded is not None
289 and param.name in forwarded
290 and param.default is not empty
291 ):
292 kwargs[param.name] = forwarded[param.name]
293 continue
294 if param.annotation is not empty:
295 peeled = coerce.peel(param.annotation)
296 if peeled.env is not None:
297 value = _env_value(param, peeled, siblings)
298 if value is not _MISSING:
299 kwargs[param.name] = value
300 continue
301 # ask(): prompt for a required (defaultless) param the CLI and
302 # env didn't fill — CLI > env > default > prompt, so a default
303 # short-circuits it and the prompt is the last resort.
304 if peeled.ask is not None and param.default is empty:
305 kwargs[param.name] = _prompt_param(cli, peeled, ctx, siblings)
306 continue
307 raw = seg.values[cli]
308 if isinstance(raw, bool): # a flag, already resolved by the splitter
309 kwargs[param.name] = raw
310 continue
311 if param.annotation is empty: 311 ↛ 312line 311 didn't jump to line 312 because the condition on line 311 was never true
312 kwargs[param.name] = raw
313 continue
315 peeled = coerce.peel(param.annotation)
316 label = f"--{cli}"
317 if peeled.mapping:
318 result: dict[Any, Any] = {}
319 for key, value in raw:
320 k = coerce.coerce_one(key, peeled.key)
321 v = _run_checks(
322 coerce.coerce_one(value, peeled.element), peeled, label, siblings
323 )
324 if peeled.value_multiple:
325 result.setdefault(k, []).append(v)
326 else:
327 result[k] = v
328 kwargs[param.name] = result
329 elif peeled.multiple:
330 items = raw if isinstance(raw, list) else [raw]
331 kwargs[param.name] = [
332 _run_checks(
333 coerce.coerce_one(v, peeled.element), peeled, label, siblings
334 )
335 for v in items
336 ]
337 else:
338 kwargs[param.name] = _run_checks(
339 coerce.coerce_one(raw, peeled.element), peeled, label, siblings
340 )
342 # Positional-only params (`def build(target, /)`) cannot be passed by
343 # keyword, so move the leading run of them out of kwargs into positional
344 # args, in signature order. A defaultless one is splitter-enforced present,
345 # so a `hole` (a skipped optional) is only ever filled by an existing
346 # default and never leaves a gap before a supplied later param.
347 pos: list[Any] = []
348 hole: list[Any] = []
349 ctx_name = context_param_name(sig)
350 for param in sig.parameters.values():
351 if param.kind is not inspect.Parameter.POSITIONAL_ONLY:
352 break # positional-only params always lead the signature
353 if param.name == ctx_name: 353 ↛ 354line 353 didn't jump to line 354 because the condition on line 353 was never true
354 continue # run_task injects ctx as the first positional itself
355 if param.name in kwargs:
356 pos += hole
357 hole = []
358 pos.append(kwargs.pop(param.name))
359 elif param.default is not empty: 359 ↛ 350line 359 didn't jump to line 350 because the condition on line 359 was always true
360 hole.append(param.default)
362 # `--` passthrough always has a home now: a task's *args, and/or the run
363 # context (`passthrough()` / `ctx.passthrough`). So it is never an error.
364 return [*pos, *var_args], kwargs
367def forward_map(
368 fn: Task, seg: Segment, received: dict[str, Any] | None = None
369) -> dict[str, Any]:
370 """The `forward`-marked parameter values *fn* passes to what it dispatches.
372 Read from the segment's CLI value or the parameter's default — never by
373 prompting, so building the map is side-effect free. Only defaulted
374 parameters contribute; a required one is never forwarded (matching `bind`).
376 A value *fn* itself *received* via forwarding wins over its segment/default,
377 so a forwarded value chains through a callee that re-declares the marker.
378 """
379 sig = resolved_signature(fn)
380 empty = inspect.Parameter.empty
381 out: dict[str, Any] = {}
382 for param in sig.parameters.values():
383 if param.annotation is empty or param.default is empty:
384 continue
385 peeled = coerce.peel(param.annotation)
386 if not peeled.forward:
387 continue
388 if received is not None and param.name in received:
389 out[param.name] = received[param.name]
390 continue
391 cli = registry.cli_name(param.name)
392 if cli not in seg.values:
393 out[param.name] = param.default
394 continue
395 raw = seg.values[cli]
396 if isinstance(raw, bool): 396 ↛ 398line 396 didn't jump to line 398 because the condition on line 396 was always true
397 out[param.name] = raw
398 elif peeled.multiple:
399 items = raw if isinstance(raw, list) else [raw]
400 out[param.name] = [coerce.coerce_one(v, peeled.element) for v in items]
401 else:
402 out[param.name] = coerce.coerce_one(raw, peeled.element)
403 return out
406def _call(
407 fn: Task, args: list[Any], kwargs: dict[str, Any]
408) -> tuple[int, Any, BaseException | None]:
409 try:
410 returned = fn(*args, **kwargs)
411 except SystemExit as exc:
412 # A non-int, non-None code is Python's `sys.exit("message")` idiom: the
413 # object is the reason the interpreter would print to stderr. Carry it as
414 # the failure error so it renders (stderr + --json) like any other failure,
415 # instead of vanishing into a bare "exited with code 1". An int/None code
416 # (the "fail with code N" idiom) has no message to surface.
417 has_reason = not isinstance(exc.code, int) and exc.code is not None
418 code = 1 if has_reason else (exc.code if isinstance(exc.code, int) else 0)
419 return code, None, (exc if has_reason else None)
420 except Failed as exc:
421 # `footman.fail("reason", code=…)`: a deliberate stop. Honour its code and
422 # carry the reason as the error, rendered verbatim (see _app / context).
423 return exc.code, None, exc
424 except RunFailed as exc:
425 # A `run()` command failed: propagate its own exit code, not a flat 1,
426 # so `fm` mirrors the command's code (docs/ci.md's "exited N" contract).
427 return (exc.result.code or 1), None, exc
428 except Exception as exc: # a failed task must not crash the runner
429 return 1, None, exc
430 if isinstance(returned, int) and not isinstance(returned, bool):
431 return returned, returned, None
432 return 0, returned, None
435class Unavailable(Exception):
436 """A `@requires`-gated task was asked to run; the message is the reason."""
439def run_task(
440 fn: Task, seg: Segment, ctx: Context, forwarded: dict[str, Any] | None = None
441) -> TaskResult:
442 """Bind *seg* to *fn* and run it within *ctx* (contextvar set for run()).
444 `ctx` is injected as the first argument if the task declares a `ctx`
445 parameter. Output routing (per-task buffering for parallel/`--json`) is the
446 caller's job via `ctx.sink`; here we just capture its final value.
447 *forwarded* carries `forward`-marked values from a dispatching task.
448 """
449 # `@requires` availability is re-checked live at the moment of execution —
450 # the manifest's cached answer is only ever a listing annotation.
451 if (reason := registry.availability(fn)) is not None:
452 return TaskResult(task=seg.task, ok=False, code=2, error=Unavailable(reason))
453 try:
454 args, kwargs = bind(seg, fn, ctx, forwarded)
455 except ChainError:
456 raise # e.g. passthrough with no *args — reported by the app layer
457 except Exception as exc: # a coercion failure (e.g. a custom-type constructor)
458 return _result(seg, 2, None, exc, 0.0)
460 if context_param_name(resolved_signature(fn)):
461 args = [ctx, *args] # ctx is the first positional parameter
463 ctx.fn = fn # what inherited() reads to find the shadowed task
464 ctx.interactive = registry.is_interactive(fn) # arms the prompt guard
465 ctx.atomic = registry.is_atomic(fn) # its subprocesses opt out of the kill
466 if ctx.cwd is None and (home := defining_dir(fn)) is not None:
467 ctx.cwd = Path(home) # run from the folder that defined the task
469 token = _current.set(ctx)
470 ctx.in_task = True # a mid-body prompt()/confirm()/select() is now guarded
471 start = time.perf_counter()
472 try:
473 code, returned, error = _call(fn, args, kwargs)
474 finally:
475 _current.reset(token)
476 duration = time.perf_counter() - start
477 output = ctx.sink.getvalue() if isinstance(ctx.sink, io.StringIO) else ""
478 result = _result(seg, code, returned, error, duration, output, ctx.steps)
479 # A task that failed while fail-fast was already aborting the run wasn't a
480 # genuine failure — it was cut off. Report that honestly, not as "failed".
481 if not result.ok and context._aborting.is_set():
482 result.cancelled = True
483 return result
486def _result(
487 seg: Segment,
488 code: int,
489 returned: Any,
490 error: BaseException | None,
491 duration: float,
492 output: str = "",
493 steps: list[Result] | None = None,
494) -> TaskResult:
495 return TaskResult(
496 task=seg.task,
497 ok=error is None and code == 0,
498 # Honor an explicit non-zero code (run_task passes 2 for bind/coercion
499 # refusals); only synthesize 1 when an error carries no code of its own.
500 code=code if code != 0 else (1 if error is not None else 0),
501 returned=returned,
502 error=error,
503 duration=duration,
504 output=output,
505 steps=steps or [],
506 )
509def run_chain(
510 root: Group,
511 segments: list[Segment],
512 *,
513 keep_going: bool = False,
514 capture: bool = False,
515 ctx_config: dict[str, Any] | None = None,
516) -> list[TaskResult]:
517 """Run a chain sequentially (a thin shim over the DAG scheduler)."""
518 from footman import schedule
520 return schedule.run_plan(
521 root,
522 segments,
523 sequential=True,
524 keep_going=keep_going,
525 capture=capture,
526 ctx_config=ctx_config,
527 )