Coverage for src/footman/testing.py: 100%
77 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-06 02:59 +0000
1"""Test your tasks — an in-process CLI runner and a silent recording context.
3Three altitudes, matching how tasks are actually written:
51. **Plain calls.** `@task` returns your function untouched, so a task body
6 is unit-testable by just calling it: `lint(fix=True)`.
72. **Recording.** `recording()` captures the commands a block *would* run —
8 silently, without executing anything — so a test can assert on them:
10 ```python
11 from footman.testing import recording
12 from tasks import lint
14 def test_lint_fix_passes_the_flag():
15 with recording() as steps:
16 lint(fix=True)
17 assert steps[0].command == "ruff check . --fix"
18 ```
203. **CLI-level.** `Runner.invoke` drives argv → exit code → output →
21 structured results, entirely in-process:
23 ```python
24 result = Runner().invoke("--dry-run release 1.2.0 --push")
25 assert result.ok
26 assert result.results[0].task == "release"
27 ```
29Everything here is stdlib-only — the zero-dependency promise holds. The
30pytest fixtures in `footman.pytest_plugin` are thin shims over this module,
31so non-pytest users get the same power.
32"""
34from __future__ import annotations
36import contextlib
37import io
38import os
39import shlex
40import tempfile
41from collections.abc import Generator, Mapping
42from dataclasses import dataclass, field
43from pathlib import Path
44from typing import Any
46from footman import _app, context
47from footman._executor import TaskResult
48from footman.app import App
49from footman.context import Context, Result, use_context
50from footman.registry import Group
52__all__ = [
53 "InvokeResult",
54 "Result",
55 "Runner",
56 "TaskResult",
57 "recording",
58 "use_context",
59]
62@contextlib.contextmanager
63def recording(
64 *, answers: Mapping[Any, Any] | None = None, **overrides: Any
65) -> Generator[list[Result]]:
66 """Capture the commands a block would `run()` — silently, not executing.
68 Yields the live step list; each `run()` call inside the block — hosted
69 toolroom calls included — appends a `Result` instead of executing, and
70 answers with a blank success. Keyword overrides go to the underlying
71 `Context` (e.g. `env={...}`).
73 `answers=` scripts the answers instead. Keys are command prefixes —
74 `"uv tool list"`, or a tuple of tokens — matched against the recorded
75 command (`Result.command`), longest match first. Values: a `str` is
76 stdout with exit 0, an `int` is an exit code, a `Result` sets code and
77 both streams, an exception instance is *raised* by the call (a missing
78 binary), and a list answers consecutive matches in order and refuses by
79 name when it runs dry. Unmatched calls keep the blank success. A
80 non-zero answer takes the real failing lane — returned under
81 `nofail=True`, raised as `RunFailed` otherwise — and a matched call is
82 answered even when it opted out of the record with `recorded=False`.
83 A recorded step also keeps the `env` and `cwd` it would have run with.
85 toolroom's `answers()` is the same table applied one layer down, at the
86 bridge's own seam; nested inside a recording, it wins and the record
87 sees nothing.
88 """
89 # Build the kwargs dict so `overrides` can win over the dry_run/quiet
90 # defaults — passing them as positional defaults made `recording(quiet=False)`
91 # raise "got multiple values for keyword argument" (F51).
92 # One merged dict, then one splat — a caller's override may replace the
93 # defaults (a direct `Context(dry_run=True, **overrides)` would raise on
94 # a duplicate keyword instead).
95 merged: dict[str, Any] = {"dry_run": True, "quiet": True, **overrides}
96 if answers is not None:
97 merged["answers"] = context._Answers(answers)
98 ctx = Context(**merged)
99 with use_context(ctx):
100 yield ctx.steps
103@dataclass
104class InvokeResult:
105 """Everything one `Runner.invoke` produced.
107 Named apart from the run-step `Result` (`footman.Result`, what `run()`
108 returns): this is the outcome of a whole CLI invocation — its exit code,
109 captured streams, and per-task `TaskResult`s."""
111 exit_code: int
112 stdout: str
113 stderr: str
114 results: list[TaskResult] = field(default_factory=list)
116 @property
117 def ok(self) -> bool:
118 return self.exit_code == 0
121@contextlib.contextmanager
122def _isolated(cwd: Path | None) -> Generator[None]:
123 """A throwaway completion cache (and optional cwd) for one invocation.
125 **The brand and its locations are restored too.** A real entry point runs
126 one brand and deliberately never puts the module globals back — the
127 process *is* that CLI, and `run` may not even return (the uv handoff
128 re-execs). A test process is the one place that isn't true: it drives
129 many brands in a row, and the next one must not read the last one's
130 world. `_paths` was already restored here; `_app._brand` was not, so
131 whichever `Runner(App(...))` ran first in a pytest-xdist worker silently
132 decided what every later test saw — three macOS jobs failed on that, and
133 only three, because the workers happened to order things differently.
134 """
135 from footman import _app, _paths
137 with tempfile.TemporaryDirectory(prefix="footman-test-") as tmp:
138 old = os.environ.get("XDG_CACHE_HOME")
139 os.environ["XDG_CACHE_HOME"] = tmp
140 where = _paths.child_args()
141 brand = _app._brand
142 try:
143 if cwd is not None:
144 with contextlib.chdir(cwd):
145 yield
146 else:
147 yield
148 finally:
149 _app._brand = brand
150 _paths.configure_child(*where)
151 if old is None:
152 os.environ.pop("XDG_CACHE_HOME", None)
153 else:
154 os.environ["XDG_CACHE_HOME"] = old
157class Runner:
158 """Drive a footman CLI in-process, capturing output and results.
160 Pass a branded `App` to test a custom CLI (`Runner(App(prog="acme"))`) —
161 error prefixes, `--version`, and hints then use that brand, exactly as
162 they would for real users.
163 """
165 app: App
167 def __init__(self, app: App | None = None) -> None:
168 self.app = app if app is not None else App()
170 def invoke(
171 self,
172 args: str | list[str],
173 *,
174 tasks: Path | Group | None = None,
175 cwd: Path | None = None,
176 stdin: str | bytes | None = None,
177 answers: Mapping[Any, Any] | None = None,
178 ) -> InvokeResult:
179 """Run one command line and return everything it produced.
181 `args` is a string (shlex-split) or an argv list. `tasks` overrides
182 discovery: a `Path` routes through `--tasks-file`, a `Group` skips
183 discovery entirely (an in-memory tree, no files needed). Without it,
184 the normal `tasks.py` cascade from `cwd` applies. `stdin` is what a
185 pipe would have delivered — a `str` is encoded UTF-8 — and its
186 absence means "a terminal": the invocation never reads the test
187 harness's real stream, so `stdin`-bound parameters see exactly what
188 the test says and nothing else. `answers` scripts the world the
189 invocation rehearses in — the same table `recording(answers=…)`
190 takes — and implies `--dry-run`, so the CLI runs end to end against
191 scripted commands. Never raises on task failure — the code is in
192 the `Result`; `KeyboardInterrupt` passes through.
193 """
194 argv = shlex.split(args) if isinstance(args, str) else list(args)
195 if answers is not None and not any(a in ("--dry-run", "-n") for a in argv):
196 argv = ["--dry-run", *argv]
197 out, err = io.StringIO(), io.StringIO()
198 collected: list[TaskResult] = []
199 payload = stdin.encode("utf-8") if isinstance(stdin, str) else stdin
200 previous = context._inject_stdin(payload)
201 table = context._inject_answers(
202 context._Answers(answers) if answers is not None else None
203 )
204 try:
205 return self._invoke(argv, tasks, cwd, out, err, collected)
206 finally:
207 context._inject_answers(table)
208 context._restore_stdin_payload(previous)
210 def _invoke(
211 self,
212 argv: list[str],
213 tasks: Path | Group | None,
214 cwd: Path | None,
215 out: io.StringIO,
216 err: io.StringIO,
217 collected: list[TaskResult],
218 ) -> InvokeResult:
219 with (
220 _isolated(cwd),
221 contextlib.redirect_stdout(out),
222 contextlib.redirect_stderr(err),
223 ):
224 # `App.run` is bypassed below, so point the locations at this
225 # brand's world here — the same call a real entry point makes
226 # before anything reads a path.
227 self.app.brand.install()
228 if isinstance(tasks, Group):
229 # One shared surface with the real CLI (help/version/list/tree/
230 # json all honoured) — no drifting Group-mode re-implementation.
231 code = _app.run_group(
232 tasks, argv, brand=self.app.brand, collect=collected
233 )
234 else:
235 if tasks is not None:
236 argv = [f"--tasks-file={tasks}", *argv]
237 # `_run`, not `run`: bypass the CLI's KeyboardInterrupt->130
238 # wrapper so Ctrl-C reaches pytest (the invoke docstring's
239 # contract). The wrapper stays for real CLI entry.
240 # `handoff=False`: the uv re-exec must never fire from an
241 # embedded invocation — it would execvp the HOST process
242 # (pytest; under pytest-xdist the worker whose stdio is the
243 # test-protocol channel, which the exec kills outright).
244 code = _app._run(
245 argv, brand=self.app.brand, collect=collected, handoff=False
246 )
247 return InvokeResult(code, out.getvalue(), err.getvalue(), collected)