def download_logs()

in sync/trypush.py [0:0]

• 33 lines of code
• 11 McCabe index (conditional complexity)

    def download_logs(self, wpt_taskgroup: TaskGroupView | TryPushTasks,
                      first_only: bool = False) -> TaskGroupView:
        """Download all the wptreport logs for the current try push

        :return: List of paths to logs
        """
        # Allow passing either TryPushTasks or the actual TaskGroupView
        if hasattr(wpt_taskgroup, "wpt_tasks"):
            assert isinstance(wpt_taskgroup, TryPushTasks)
            wpt_tasks = wpt_taskgroup.wpt_tasks
        else:
            assert isinstance(wpt_taskgroup, TaskGroupView)
            wpt_tasks = wpt_taskgroup

        exclude = set()

        def included(t: dict[str, dict[str, Any]]) -> bool:
            # if a name is on the excluded list, only download SUCCESS job logs
            name = t.get("task", {}).get("metadata", {}).get("name")
            state = t.get("status", {}).get("state")
            output = name not in exclude or state == tc.SUCCESS

            # Add this name to exclusion list, so that we don't download the repeated run
            # logs in a stability try run
            if first_only and name is not None and name not in exclude:
                exclude.add(name)

            return output

        if self.try_rev is None:
            if wpt_tasks:
                logger.info("Got try push with no rev; setting it from a task")
                try_rev = (next(iter(wpt_tasks))
                           .get("task", {})
                           .get("payload", {})
                           .get("env", {})
                           .get("GECKO_HEAD_REV"))
                if try_rev:
                    self.try_rev = try_rev  # type: ignore
                else:
                    raise ValueError("Unknown try rev for %s" % self.process_name)

        include_tasks = wpt_tasks.filter(included)
        logger.info("Downloading logs for try revision %s" % self.try_rev)
        file_names = ["wptreport.json"]
        include_tasks.download_logs(self.log_path(), file_names)
        return include_tasks