Coverage for src/footman/discover.py: 94%

1"""Load the task cascade and merge it into one command tree. 

2 

3In a monorepo you rarely want a single tasks file. footman collects every 

4`tasks.py` from the repo root down to your current directory and merges them 

5top-down: a new name **appends**, a name already present **overrides** (the 

6folder nearest your cwd wins), and a command group present at both levels 

7**merges** (its tasks overlaid the same way). Each task remembers the folder of 

8the file that defined it, so it runs from there regardless of where you stand. 

9 

10The registry raises on a duplicate name, so the merge can't be done by importing 

11every file into one registry — each file is imported into a fresh registry and 

12the resulting trees are overlaid here. 

13""" 

14 

15from __future__ import annotations 

16 

17import importlib.util 

18import sys 

19from pathlib import Path 

20 

21from footman import registry 

22from footman.registry import Group, Task 

23 

24# Attribute stamped on every task fn: the directory of the file that defined 

25# it. The scheduler uses it as the task's working directory. 

26DEFINING_DIR = "_footman_dir" 

27SHADOWED = "_footman_shadowed" 

28 

29 

30class TasksImportError(Exception): 

31 """A tasks file failed to import; names the file and keeps the cause.""" 

32 

33 def __init__(self, path: Path, original: BaseException) -> None: 

34 self.path = path 

35 self.original = original 

36 super().__init__(f"{path}: {type(original).__name__}: {original}") 

37 

38 

39class FinalizeError(Exception): 

40 """A `@finalize` hook raised while editing the discovered tree.""" 

41 

42 def __init__(self, name: str, original: BaseException) -> None: 

43 self.name = name 

44 self.original = original 

45 super().__init__(f"@finalize {name!r}: {type(original).__name__}: {original}") 

46 

47 

48def _import_file(path: Path, index: int) -> Group: 

49 """Import *path* into a fresh registry and return the populated tree.""" 

50 registry.reset() 

51 spec = importlib.util.spec_from_file_location(f"footman_tasks_{index}", path) 

52 if spec is None or spec.loader is None: 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true

53 raise TasksImportError(path, ImportError("cannot load tasks file")) 

54 module = importlib.util.module_from_spec(spec) 

55 sys.modules[spec.name] = module 

56 parent = str(path.parent) 

57 # Search this file's own dir first for sibling helpers (move-to-front, not 

58 # insert-if-absent — a shared dir on sys.path must not shadow it), snapshot 

59 # sys.path/sys.modules, and evict the direct siblings it imports afterwards. 

60 # Otherwise two cascade files each doing `import helpers` share whoever 

61 # imported first (F14/D8). Restoring sys.path also stops it accumulating 

62 # across the many load_tree calls an in-process runner makes. 

63 saved_path = sys.path[:] 

64 before = set(sys.modules) 

65 if parent in sys.path: 

66 sys.path.remove(parent) 

67 sys.path.insert(0, parent) 

68 try: 

69 spec.loader.exec_module(module) 

70 except Exception as exc: 

71 raise TasksImportError(path, exc) from exc 

72 finally: 

73 sys.path[:] = saved_path 

74 _evict_siblings(before, Path(parent)) 

75 return registry.root 

76 

77 

78def _evict_siblings(before: set[str], parent: Path) -> None: 

79 """Drop modules a cascade file imported that live directly in its dir. 

80 

81 A sibling `helpers.py` (`parent/helpers.py`) or a package one level down 

82 (`parent/pkg/__init__.py`) — so the next file gets its own copy rather than 

83 whoever-imported-first-wins. Deeper imports and editable-installed packages 

84 live elsewhere on disk and are deliberately left (D8). 

85 """ 

86 for name in set(sys.modules) - before: 

87 file = getattr(sys.modules.get(name), "__file__", None) 

88 if file is None: 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true

89 continue 

90 f = Path(file) 

91 sibling = f.parent == parent 

92 package = f.name == "__init__.py" and f.parent.parent == parent 

93 if sibling or package: 

94 del sys.modules[name] 

95 

96 

97def _tag(group: Group, directory: str) -> None: 

98 """Stamp every task in *group* (recursively) with its defining directory.""" 

99 for fn in group.tasks.values(): 

100 setattr(fn, DEFINING_DIR, directory) 

101 for sub in group.groups.values(): 101 ↛ 102line 101 didn't jump to line 102 because the loop on line 101 never started

102 _tag(sub, directory) 

103 

104 

105def _overlay(base: Group, overlay: Group, directory: str) -> None: 

106 """Merge *overlay* onto *base* in place: local (overlay) wins by name.""" 

107 for name, fn in overlay.tasks.items(): 

108 setattr(fn, DEFINING_DIR, directory) 

109 # Keep the task this one shadows reachable: `inherited()` calls it, 

110 # `--where` lists it, `--help` shows its options. Without this the 

111 # parent's function is simply dropped and a leaf can only 

112 # *replace* the root's task, never extend it. 

113 if (previous := base.tasks.get(name)) is not None and previous is not fn: 

114 setattr(fn, SHADOWED, previous) 

115 base.groups.pop(name, None) # a local task shadows an inherited group 

116 base.tasks[name] = fn 

117 for name, sub in overlay.groups.items(): 

118 if name in base.groups: 

119 base.groups[name].help = sub.help or base.groups[name].help 

120 _overlay(base.groups[name], sub, directory) 

121 else: 

122 base.tasks.pop(name, None) # a local group shadows an inherited task 

123 _tag(sub, directory) 

124 base.groups[name] = sub 

125 

126 

127def load_tree(files: list[Path], base: Group | None = None) -> Group: 

128 """Import each file (root first) and overlay them into one merged tree. 

129 

130 *base* seeds the tree (config-mounted plugin groups go there), so 

131 anything a tasks file defines overlays it — user names win over plugins 

132 exactly as nearer cascade files win over farther ones. 

133 """ 

134 merged = base if base is not None else Group("root") 

135 finalizers: list[registry.Finalizer] = [] 

136 try: 

137 for index, path in enumerate(files): 

138 tree = _import_file(path, index) 

139 _overlay(merged, tree, str(path.parent)) 

140 # Collect each file's @finalize hooks in cascade order, before the 

141 # next _import_file resets the registry. 

142 finalizers.extend(tree.finalizers) 

143 finally: 

144 # Leave no global state behind — even when a file registered some tasks 

145 # and then raised, which would otherwise strand ghost tasks in 

146 # registry.root for the rest of the process (F62). 

147 registry.reset() 

148 # Run the hooks on the fully-merged tree, cascade order (root first, the 

149 # folder nearest cwd last), each seeing the previous edits — so a subfolder 

150 # refines what root did. Discovery-time, so the edits reach the manifest. 

151 view = registry.Tasks(merged) 

152 for run in finalizers: 

153 try: 

154 run(view) 

155 except Exception as exc: # a bad hook names itself, never a bare traceback 

156 raise FinalizeError(getattr(run, "__name__", repr(run)), exc) from exc 

157 return merged 

158 

159 

160def load_single(path: Path) -> Group: 

161 """Load exactly one tasks file (the `-f/--tasks-file` escape hatch).""" 

162 return load_tree([path]) 

163 

164 

165def defining_dir(fn: Task) -> str | None: 

166 """The folder the task was defined in, if the cascade tagged it.""" 

167 return getattr(fn, DEFINING_DIR, None) 

168 

169 

170def shadowed(fn: Task) -> Task | None: 

171 """The task *fn* overrides — the same name, one cascade level up.""" 

172 return getattr(fn, SHADOWED, None) 

173 

174 

175def shadow_chain(fn: Task) -> list[Task]: 

176 """*fn* and every task it shadows, nearest first.""" 

177 chain = [fn] 

178 while (previous := shadowed(chain[-1])) is not None: 

179 chain.append(previous) 

180 return chain