Bug 1636271 - [taskgraph] Create utility function for listing all tasks in a task group, r=tomprince

Creates a new utility function in 'taskgraph.util.taskcluster' to return task
definitions from a task group. Also refactors the existing
'list_task_group_incomplete_tasks' function to use it.

Differential Revision: https://phabricator.services.mozilla.com/D74407

taskcluster/taskgraph/actions/cancel_all.py

@@ -12,7 +12,7 @@ import os
 import requests
 
 from taskgraph.util.taskcluster import (
-    list_task_group_incomplete_tasks,
+    list_task_group_incomplete_task_ids,
     cancel_task,
     CONCURRENCY,
 )
@@ -48,7 +48,8 @@ def cancel_all_action(parameters, graph_config, input, task_group_id, task_id):
             raise
 
     own_task_id = os.environ.get('TASK_ID', '')
-    to_cancel = [t for t in list_task_group_incomplete_tasks(task_group_id) if t != own_task_id]
+    to_cancel = [t for t in list_task_group_incomplete_task_ids(task_group_id) if t != own_task_id]
+
     logger.info("Cancelling {} tasks".format(len(to_cancel)))
     with futures.ThreadPoolExecutor(CONCURRENCY) as e:
         cancel_futs = [e.submit(do_cancel_task, t) for t in to_cancel]

taskcluster/taskgraph/util/taskcluster.py

@@ -306,17 +306,23 @@ def send_email(address, subject, content, link, use_proxy=False):
     })
 
 
-def list_task_group_incomplete_tasks(task_group_id):
-    """Generate the incomplete tasks in a task group"""
+def list_task_group_tasks(task_group_id):
+    """Generate the tasks in a task group"""
     params = {}
     while True:
         url = liburls.api(get_root_url(False), 'queue', 'v1',
                           'task-group/{}/list'.format(task_group_id))
         resp = _do_request(url, force_get=True, params=params).json()
-        for task in [t['status'] for t in resp['tasks']]:
-            if task['state'] in ['running', 'pending', 'unscheduled']:
-                yield task['taskId']
+        for task in resp['tasks']:
+            yield task
         if resp.get('continuationToken'):
             params = {'continuationToken': resp.get('continuationToken')}
         else:
             break
+
+
+def list_task_group_incomplete_task_ids(task_group_id):
+    states = ('running', 'pending', 'unscheduled')
+    for task in [t['status'] for t in list_task_group_tasks(task_group_id)]:
+        if task['state'] in states:
+            yield task['taskId']