def run_push_tasks()

in mozci/data/sources/taskcluster/__init__.py [0:0]

• 55 lines of code
• 8 McCabe index (conditional complexity)

    def run_push_tasks(self, branch, rev):
        try:
            decision_task_id = taskcluster.find_task_id(
                f"gecko.v2.{branch}.revision.{rev}.taskgraph.decision"
            )
        except requests.exceptions.HTTPError as e:
            # If the decision task was not indexed, it means it was broken. So we can
            # assume we didn't run any task for this push.
            if e.response.status_code == 404:
                logger.warning(f"Decision task broken in {rev} on {branch}")
                return []

            raise

        task_data = taskcluster.get_task(decision_task_id)

        results = taskcluster.get_tasks_in_group(task_data["taskGroupId"])

        tasks = []
        for result in results:
            # Skip the decision task.
            if result["status"]["taskId"] == decision_task_id:
                continue

            # Skip "Action" tasks.
            if result["task"]["metadata"]["name"].startswith("Action"):
                continue

            task = {
                "id": result["status"]["taskId"],
                "label": result["task"]["metadata"]["name"],
                "tags": result["task"]["tags"],
                "state": result["status"]["state"],
                "suite": result["task"]["extra"].get("suite", None),
                "platform": result["task"]["extra"].get("treeherder-platform", ""),
                "variant": result["task"]["extra"]
                .get("test-setting", {})
                .get("runtime", {}),
            }

            treeherder = result["task"]["extra"].get("treeherder", {})
            tier = treeherder.get("tier")
            if tier:
                task["tier"] = tier
            job_kind = treeherder.get("jobKind")
            if job_kind:
                task["job_kind"] = job_kind

            # Use the latest run (earlier ones likely had exceptions that
            # caused an automatic retry).
            if task["state"] != "unscheduled":
                run = result["status"]["runs"][-1]

                # Normalize the state to match treeherder's values.
                if task["state"] == "failed":
                    task["state"] = "completed"

                # Derive a result from the reasonResolved.
                reason = run.get("reasonResolved")
                if reason == "completed":
                    task["result"] = "passed"
                elif reason in ("canceled", "superseded", "failed"):
                    task["result"] = reason
                elif reason:
                    task["result"] = "exception"
                else:
                    # Task is not finished, so there is no result yet.
                    assert task["state"] in ("pending", "running", "exception")

                # Compute duration.
                if "started" in run and "resolved" in run:
                    task["duration"] = self.to_ms(run["resolved"]) - self.to_ms(
                        run["started"]
                    )

            tasks.append(task)

        return tasks