Coverage for src/footman/_suggest.py: 75%

56 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-07-24 13:38 +0000

1"""Fresh values for a dynamic completer — spawned by the completion hot path. 

2 

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. 

8 

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""" 

14 

15from __future__ import annotations 

16 

17import contextlib 

18import inspect 

19import io 

20import sys 

21 

22 

23def _values(param: str, path: list[str], g: dict[str, object]) -> list[str]: 

24 """The fresh output of *param*'s completer on the task at *path*. 

25 

26 Rediscovers the same tasks the manifest was built from (honouring 

27 `-f`/`--config`), walks to the task, peels the parameter, and runs its 

28 completer. Any miss — no tasks file, a renamed task, a plain parameter — 

29 is an empty list, never an error. 

30 """ 

31 from footman import _app, coerce, compose, discover, manifest, registry 

32 

33 files, cfg = _app.resolve_task_files(g, on_warning=lambda *a: None, on_note=None) 

34 if not files or not path: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 return [] 

36 base = registry.Group("root") 

37 plugins = cfg.get("plugins") 

38 if isinstance(plugins, list): # a completer may live on a plugin task 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true

39 compose.mount_plugins(base, plugins) 

40 root = discover.load_tree(files, base=base) 

41 

42 node: registry.Group | None = root 

43 for name in path[:-1]: # descend the groups 43 ↛ 44line 43 didn't jump to line 44 because the loop on line 43 never started

44 node = node.groups.get(name) if node else None 

45 task = node.tasks.get(path[-1]) if node else None 

46 if task is None: 

47 return [] 

48 for p in manifest.resolved_signature(task).parameters.values(): 

49 if ( 

50 registry.cli_name(p.name) != param 

51 or p.annotation is inspect.Parameter.empty 

52 ): 

53 continue 

54 completer = coerce.peel(p.annotation).completer 

55 if completer is None: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true

56 return [] 

57 # Mute the completer's own stdout/stderr so its chatter can't leak into 

58 # the value channel the hot path reads. 

59 with ( 

60 contextlib.redirect_stdout(io.StringIO()), 

61 contextlib.redirect_stderr(io.StringIO()), 

62 ): 

63 return [str(v) for v in completer.fn()] 

64 return [] 

65 

66 

67def main(argv: list[str]) -> int: 

68 param: str | None = None 

69 path: list[str] = [] 

70 g: dict[str, object] = {} 

71 i = 0 

72 while i < len(argv): 

73 arg = argv[i] 

74 if arg == "--param" and i + 1 < len(argv): 

75 param, i = argv[i + 1], i + 2 

76 elif arg == "--path" and i + 1 < len(argv): 76 ↛ 79line 76 didn't jump to line 79 because the condition on line 76 was always true

77 path.append(argv[i + 1]) 

78 i += 2 

79 elif arg == "--tasks-file" and i + 1 < len(argv): 

80 g["tasks_file"], i = argv[i + 1], i + 2 

81 elif arg == "--config" and i + 1 < len(argv): 

82 g["config"], i = argv[i + 1], i + 2 

83 else: 

84 i += 1 

85 if param is None: 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true

86 return 0 

87 try: 

88 values = _values(param, path, g) 

89 except Exception: 

90 return 0 # any failure → no candidates; the hot path falls back to empty 

91 if values: 91 ↛ 93line 91 didn't jump to line 93 because the condition on line 91 was always true

92 sys.stdout.write("\n".join(values) + "\n") 

93 return 0 

94 

95 

96if __name__ == "__main__": 

97 raise SystemExit(main(sys.argv[1:]))