44
44
if TYPE_CHECKING :
45
45
from textual .widgets .tree import TreeNode
46
46
from typing import NoReturn
47
+ import networkx as nx
48
+
49
+
50
+ class Task (NamedTuple ):
51
+ """A task."""
52
+
53
+ task : PTask
54
+ reasons_rerun : list [str ]
47
55
48
56
49
57
class Module (NamedTuple ):
@@ -185,7 +193,7 @@ def _select_tasks_by_expressions_and_marker(session: Session) -> list[PTask]:
185
193
return [task for task in session .tasks if task .signature in remaining ]
186
194
187
195
188
- def _organize_tasks (tasks : list [PTaskWithPath ]) -> list [Module ]:
196
+ def _organize_tasks (tasks : list [PTaskWithPath ], dag : nx . DiGraph ) -> list [Module ]:
189
197
"""Organize tasks in a dictionary.
190
198
191
199
The dictionary has file names as keys and then a dictionary with task names and
@@ -197,11 +205,22 @@ def _organize_tasks(tasks: list[PTaskWithPath]) -> list[Module]:
197
205
module = name_to_module .get (
198
206
task .path .as_posix (), Module (task .path .as_posix (), [])
199
207
)
208
+ reasons = _find_reasons_to_rerun (task , dag )
209
+ task_wrap = Task (task = task , reasons_rerun = [])
200
210
module .tasks .append (task )
201
211
name_to_module [module .name ] = module
202
212
return [name_to_module [name ] for name in sorted (name_to_module )]
203
213
204
214
215
+ def _find_reasons_to_rerun (task : PTask , dag : nx .DiGraph ) -> list [str ]:
216
+ """Find the reasons to rerun a task."""
217
+ reasons = []
218
+ for task in dag .nodes :
219
+ if node .task == task :
220
+ reasons .append (node .name )
221
+ return reasons
222
+
223
+
205
224
def _print_collected_tasks (
206
225
dictionary : dict [Path , list [PTaskWithPath ]],
207
226
show_nodes : bool ,
0 commit comments