def get_patch_data()

in bugbot/rules/not_landed.py [0:0]

• 98 lines of code
• 12 McCabe index (conditional complexity)

    def get_patch_data(self, bugs):
        """Get patch information in bugs"""
        patterns = Bugzilla.get_landing_patterns(channels=["nightly", "autoland"])

        def comment_handler(bug, bugid, data):
            # if a comment contains a backout: don't nag
            for comment in bug["comments"]:
                comment = comment["text"].lower()
                if any(pat[0].match(comment) for pat in patterns) and (
                    "backed out" in comment or "backout" in comment
                ):
                    data[bugid]["backout"] = True

        def attachment_id_handler(attachments, bugid, data):
            for a in attachments:
                if (
                    a["content_type"] == "text/x-phabricator-request"
                    and a["is_obsolete"] == 0
                ):
                    data.append(a["id"])

        def attachment_handler(attachments, data):
            for attachment in attachments:
                bugid = str(attachment["bug_id"])
                if bugid in data:
                    data[bugid].append(attachment)
                else:
                    data[bugid] = [attachment]

        def has_blocking_dependencies(attachment):
            rev = PHAB_URL_PAT.search(
                base64.b64decode(attachment["data"]).decode("utf-8")
            ).group(1)
            try:
                revision_data = self.phab.load_revision(rev_id=int(rev))
            except PhabricatorRevisionNotFoundException:
                return None

            stack_graph = revision_data["fields"]["stackGraph"]
            current_revision_phid = revision_data["phid"]
            dependencies = stack_graph[current_revision_phid]

            for dep_phid in dependencies:
                dep_revision_data = self.phab.load_revision(rev_phid=dep_phid)
                dep_status = dep_revision_data["fields"]["status"]["value"]
                if dep_status != "published":
                    return True

            return False

        bugids = list(bugs.keys())
        data = {
            bugid: {
                "backout": False,
                "author": None,
                "count": 0,
                "has_blocking_dependencies": False,
            }
            for bugid in bugids
        }

        # Get the ids of the attachments of interest
        # to avoid to download images, videos, ...
        attachment_ids = []
        Bugzilla(
            bugids=bugids,
            attachmenthandler=attachment_id_handler,
            attachmentdata=attachment_ids,
            attachment_include_fields=["is_obsolete", "content_type", "id"],
        ).get_data().wait()

        # Once we've the ids we can get the data
        attachments_by_bug = {}
        Bugzilla(
            attachmentids=attachment_ids,
            attachmenthandler=attachment_handler,
            attachmentdata=attachments_by_bug,
            attachment_include_fields=[
                "bug_id",
                "data",
                "is_obsolete",
                "content_type",
                "id",
                "creator",
            ],
        ).get_data().wait()

        for bugid, attachments in attachments_by_bug.items():
            res = {"reviewers_phid": set()}
            for attachment in attachments:
                self.handle_attachment(attachment, res)

            if "phab" in res:
                if res["phab"]:
                    data[bugid][
                        "has_blocking_dependencies"
                    ] = has_blocking_dependencies(attachment)
                    data[bugid]["reviewers_phid"] = res["reviewers_phid"]
                    data[bugid]["author"] = res["author"]
                    data[bugid]["count"] = res["count"]

        data = {bugid: v for bugid, v in data.items() if v["author"]}

        if not data:
            return data

        Bugzilla(
            bugids=list(data.keys()),
            commenthandler=comment_handler,
            commentdata=data,
            comment_include_fields=["text"],
        ).get_data().wait()

        data = {
            bugid: v
            for bugid, v in data.items()
            if not v["backout"] and not v["has_blocking_dependencies"]
        }

        return data