are_dependencies_completed.py (1723B)
1 #!/usr/bin/env python3 2 3 # This Source Code Form is subject to the terms of the Mozilla Public 4 # License, v. 2.0. If a copy of the MPL was not distributed with this 5 # file, You can obtain one at http://mozilla.org/MPL/2.0/. 6 7 import argparse 8 import os 9 import sys 10 11 import taskcluster 12 13 queue = taskcluster.Queue({ 14 "rootUrl": os.environ.get("TASKCLUSTER_PROXY_URL", "https://taskcluster.net"), 15 }) 16 17 18 def check_all_dependencies_are_completed(current_task_id): 19 print(f"Fetching task definition of {current_task_id}...") 20 task = queue.task(current_task_id) 21 dependencies_task_ids = task["dependencies"] 22 23 print(f"Fetching status of {len(dependencies_task_ids)} dependencies...") 24 # TODO Make this dict-comprehension async once we go Python 3 25 state_per_task_ids = { 26 task_id: queue.status(task_id)["status"]["state"] 27 for task_id in dependencies_task_ids 28 } 29 print("Statuses fetched.") 30 non_completed_tasks = { 31 task_id: state 32 for task_id, state in state_per_task_ids.items() 33 if state != "completed" 34 } 35 36 if non_completed_tasks: 37 raise ValueError(f"Some tasks are not completed: {non_completed_tasks}") 38 39 40 def main(): 41 parser = argparse.ArgumentParser( 42 description='Errors out if one of the DEPENDENCY_TASK_ID does not have the Taskcluster status "completed"' 43 ) 44 45 parser.add_argument( 46 "current_task_id", 47 metavar="CURRENT_TASK_ID", 48 help="The task ID of the current running task", 49 ) 50 51 result = parser.parse_args() 52 check_all_dependencies_are_completed(result.current_task_id) 53 print("All dependencies are completed. Reporting a green task!") 54 sys.exit(0) 55 56 57 if __name__ == "__main__": 58 main()