def find_build_info()

in mozregression/fetch_build_info.py [0:0]

• 67 lines of code
• 19 McCabe index (conditional complexity)

    def find_build_info(self, push):
        """
        Find build info for an integration build, given a Push, a changeset or a
        date/datetime.

        if `push` is not an instance of Push (e.g. it is a date, datetime, or
        string representing the changeset), a query to json pushes will be
        done.

        Return a :class:`IntegrationBuildInfo` instance.
        """
        if not isinstance(push, Push):
            try:
                push = self.jpushes.push(push)
            except MozRegressionError as exc:
                raise BuildInfoNotFound(str(exc))

        changeset = push.changeset

        tk_routes = self.fetch_config.tk_routes(push)
        try:
            task_id = None
            stored_failure = None
            for tk_route in tk_routes:
                LOG.debug("using taskcluster route %r" % tk_route)
                try:
                    task_id = self.index.findTask(tk_route)["taskId"]
                except TaskclusterFailure as ex:
                    LOG.debug("nothing found via route %r" % tk_route)
                    stored_failure = ex
                    continue
                if task_id:
                    status = self.queue.status(task_id)["status"]
                    break
            if not task_id:
                raise stored_failure
        except TaskclusterFailure:
            raise BuildInfoNotFound(
                "Unable to find build info using the"
                " taskcluster route %r" % self.fetch_config.tk_route(push)
            )

        # find a completed run for that task
        run_id, build_date = None, None
        for run in reversed(status["runs"]):
            if run["state"] == "completed":
                run_id = run["runId"]
                try:
                    build_date = datetime.strptime(run["resolved"], "%Y-%m-%dT%H:%M:%S.%fZ")
                except ValueError:
                    build_date = datetime.strptime(run["resolved"], "%Y-%m-%dT%H:%M:%S.%f+00:00")
                break

        if run_id is None:
            raise BuildInfoNotFound("Unable to find completed runs for task %s" % task_id)
        artifacts = self.queue.listArtifacts(task_id, run_id)["artifacts"]

        # look over the artifacts of that run
        build_url = None
        for a in artifacts:
            name = os.path.basename(a["name"])
            if self.build_regex.search(name):
                meth = self.queue.buildUrl
                if self.fetch_config.tk_needs_auth():
                    meth = self.queue.buildSignedUrl
                build_url = meth("getArtifact", task_id, run_id, a["name"])
                break
        if build_url is None:
            raise BuildInfoNotFound(
                "unable to find a build url for the" " changeset %r" % changeset
            )

        if self.fetch_config.app_name == "gve":
            # Check taskcluster URL to make sure artifact is still around.
            # build_url is an alias that redirects via a 303 status code.
            status_code = requests.head(build_url, allow_redirects=True).status_code
            if status_code != 200:
                error = f"Taskcluster file {build_url} not available (status code: {status_code})."
                raise BuildInfoNotFound(error)
        return IntegrationBuildInfo(
            self.fetch_config,
            build_url=build_url,
            build_date=build_date,
            changeset=changeset,
            repo_url=self.jpushes.repo_url,
            task_id=task_id,
        )