Coverage for src/footman/_suggest.py: 85%
96 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-06 02:59 +0000
1"""Fresh values for a dynamic completer — spawned by the completion hot path.
3A dynamic completer (`suggest(fn)`) queries live state: git branches, release
4candidates, deploy targets. Serving the manifest's *baked* snapshot for a
5build-critical answer is as wrong as answering from an empty cache, so when TAB
6lands on a dynamic parameter `_complete` spawns this process to run that one
7completer fresh.
9It lives out of the hot path precisely because it imports the framework and the
10user's code — the thing a TAB press must never do. Isolation is the point: a
11slow or crashing completer dies here, bounded by the caller's timeout, and the
12hot path degrades to no candidates rather than a hung keystroke.
13"""
15from __future__ import annotations
17import contextlib
18import inspect
19import io
20import sys
21from typing import TYPE_CHECKING, Any
23if TYPE_CHECKING: # runtime imports stay deferred: the TAB path spawns cheap
24 from pathlib import Path
27def _maybe_reexec(files: list[Path]) -> None:
28 """Continue in the environment that owns this completer — the project's
29 when the lock rule claims the directory, a script file's when it
30 carries its own dependencies. Both rules live in `_script`, shared
31 with the refresh child, and are asked in the run path's order:
32 project first, script only where no project has spoken. No healing
33 here — a keystroke is waiting, and the detached refresh owns that.
34 The interpreter flags come along too — `-P` on the spawn and not on
35 the re-exec would be a hole, since this replaces the process with the
36 argv it is handed."""
37 from pathlib import Path
39 from footman import _script
41 argv = ["-P", "-m", "footman._suggest", *sys.argv[1:]]
42 root = _script.project_home(Path.cwd())
43 if root is not None:
44 _script.project_reexec(root, argv, heal=False)
45 return
46 _script.maybe_reexec(files, argv)
49def _fresh(completer: Any) -> list[str]:
50 """Run *completer* with its own stdout/stderr muted, so its chatter can't
51 leak into the value channel the hot path reads."""
52 with (
53 contextlib.redirect_stdout(io.StringIO()),
54 contextlib.redirect_stderr(io.StringIO()),
55 ):
56 return [str(v) for v in completer.fn()]
59def _values(param: str, path: list[str], g: dict[str, object]) -> list[str]:
60 """The fresh output of *param*'s completer on the task at *path*.
62 Rediscovers the same tasks the manifest was built from (honouring
63 `-f`/`--config`), walks to the task, peels the parameter, and runs its
64 completer. Any miss — no tasks file, a renamed task, a plain parameter —
65 is an empty list, never an error.
66 """
67 from footman import _app, _coerce, _discover, _manifest, registry
69 found = _app.resolve_task_files(g, on_warning=lambda *a: None, on_note=None)
70 files = found.files
71 if not files or not path: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true
72 return []
73 _maybe_reexec(files) # before any user code is imported
74 # Plugin mounts are authored in the tasks files, so discovery alone
75 # rebuilds the whole tree — a completer on a mounted task included.
76 base = registry.Group("root")
77 root = _discover.load_tree(files, base=base)
79 node: registry.Group | None = root
80 for name in path[:-1]: # descend the groups 80 ↛ 81line 80 didn't jump to line 81 because the loop on line 80 never started
81 node = node.groups.get(name) if node else None
82 task = node.tasks.get(path[-1]) if node else None
83 if task is None:
84 return []
85 for p in _manifest.resolved_signature(task).parameters.values():
86 if (
87 registry.cli_name(p.name) != param
88 or p.annotation is inspect.Parameter.empty
89 ):
90 continue
91 completer = _coerce.peel(p.annotation).completer
92 if completer is None: 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true
93 return []
94 return _fresh(completer)
95 return []
98def _global_values(name: str, g: dict[str, object]) -> list[str]:
99 """The fresh output of the completer on the plugin global *name*.
101 The globals ride the tree's contributions, so the same discovery that
102 finds a task's completer finds an option's: rediscover, match the cli
103 name, peel the annotation, run what it carries. Any miss — no tasks
104 file, an unmounted owner, a plain option — is an empty list."""
105 from footman import _app, _coerce, _discover, registry
107 found = _app.resolve_task_files(g, on_warning=lambda *a: None, on_note=None)
108 files = found.files
109 if not files: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true
110 return []
111 _maybe_reexec(files) # before any user code is imported
112 base = registry.Group("root")
113 root = _discover.load_tree(files, base=base)
114 for opt in root.contributions.get("globals", ()):
115 if opt.name != name:
116 continue
117 completer = _coerce.peel(opt.annotation).completer
118 if completer is None:
119 return []
120 return _fresh(completer)
121 return []
124def main(argv: list[str]) -> int:
125 param: str | None = None
126 globopt: str | None = None
127 path: list[str] = []
128 g: dict[str, object] = {}
129 i = 0
130 while i < len(argv):
131 arg = argv[i]
132 if arg == "--param" and i + 1 < len(argv):
133 param, i = argv[i + 1], i + 2
134 elif arg == "--global" and i + 1 < len(argv): 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true
135 globopt, i = argv[i + 1], i + 2
136 elif arg == "--path" and i + 1 < len(argv):
137 path.append(argv[i + 1])
138 i += 2
139 elif arg == "--tasks-file" and i + 1 < len(argv): 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true
140 g["tasks_file"], i = argv[i + 1], i + 2
141 elif arg == "--config" and i + 1 < len(argv): 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true
142 g["config"], i = argv[i + 1], i + 2
143 elif arg == "--where" and i + 1 < len(argv): 143 ↛ 155line 143 didn't jump to line 155 because the condition on line 143 was always true
144 # This CLI's resolved locations (and brand context), handed over
145 # by the hot path — a child inherits the environment but not the
146 # brand. Length-prefixed: the count leads the words, so growing
147 # `child_args` can never eat the flags that follow — the arity
148 # drifted once per release until it was written down.
149 from footman import _paths
151 count = int(argv[i + 1])
152 _paths.configure_child(*argv[i + 2 : i + 2 + count])
153 i += 2 + count
154 else:
155 i += 1
156 real = sys.stdout
157 try:
158 # This process's stdout IS the candidate channel, and the calls below
159 # import the user's code before any candidate is written — so a tasks
160 # file that prints at import time (or a plugin imported alongside it)
161 # would have its chatter served to the shell as completions. Muted
162 # for the whole computation — the policy `_fresh` already applies to
163 # the completer body, applied to the import that precedes it — and
164 # only the finished candidates touch the real stream. The re-exec a
165 # script file may trigger in there is unaffected: the redirect
166 # rebinds `sys.stdout`, never fd 1, and exec replaces the process.
167 with (
168 contextlib.redirect_stdout(io.StringIO()),
169 contextlib.redirect_stderr(io.StringIO()),
170 ):
171 if globopt is not None: 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true
172 values = _global_values(globopt, g)
173 elif param is not None: 173 ↛ 176line 173 didn't jump to line 176 because the condition on line 173 was always true
174 values = _values(param, path, g)
175 else:
176 return 0
177 except Exception:
178 return 0 # any failure → no candidates; the hot path falls back to empty
179 if values: 179 ↛ 185line 179 didn't jump to line 185 because the condition on line 179 was always true
180 # The candidate protocol is newline-delimited, so a value carrying a
181 # newline would split into two bogus candidates — the second being
182 # whatever came after the break. A completion token has no legal
183 # newline; the value's first line is the whole candidate.
184 real.write("\n".join(v.splitlines()[0] for v in values if v.strip()) + "\n")
185 return 0
188if __name__ == "__main__":
189 raise SystemExit(main(sys.argv[1:]))