def from_pr()

in sync/upstream.py [0:0]

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

    def from_pr(cls,
                lock: SyncLock,
                git_gecko: Repo,
                git_wpt: Repo,
                pr_id: int,
                body: str | None
                ) -> UpstreamSync | None:
        gecko_commits = []
        bug = None
        integration_branch = None

        if body is None or not cls.has_metadata(body.encode("utf8", "replace")):
            return None

        commits = env.gh_wpt.get_commits(pr_id)

        for gh_commit in commits:
            commit = sync_commit.WptCommit(git_wpt, gh_commit.sha)
            if cls.has_metadata(commit.msg):
                gecko_commits.append(cinnabar(git_gecko).hg2git(commit.metadata["gecko-commit"]))
                commit_bug = env.bz.id_from_url(commit.metadata["bugzilla-url"])
                if bug is not None and commit_bug != bug:
                    logger.error("Got multiple bug numbers in URL from commits")
                    break
                elif bug is None:
                    bug = commit_bug

                if (integration_branch is not None and
                    commit.metadata["integration_branch"] != integration_branch):
                    logger.warning("Got multiple integration branches from commits")
                elif integration_branch is None:
                    integration_branch = commit.metadata["integration_branch"]
            else:
                break

        if not gecko_commits:
            return None

        assert bug
        gecko_base = git_gecko.rev_parse("%s^" % gecko_commits[0])
        gecko_head = git_gecko.rev_parse(gecko_commits[-1])
        wpt_head = commits[-1].sha
        wpt_base = commits[0].sha

        return cls.new(lock, git_gecko, git_wpt, gecko_base, gecko_head,
                       wpt_base, wpt_head, bug, pr_id)