def add_pr()

in sync/landing.py [0:0]

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

    def add_pr(self,
               pr_id: int,
               sync: DownstreamSync | UpstreamSync,
               wpt_commits: list[WptCommit],
               copy: bool = True,
               prev_wpt_head: str | None = None,
               ) -> Commit | None:
        if len(wpt_commits) > 1:
            assert all(item.pr() == pr_id for item in wpt_commits)

        # Assume we can always use the author of the first commit
        author = first_non_merge(wpt_commits).author

        git_work_wpt = self.wpt_worktree.get()
        git_work_gecko = self.gecko_worktree.get()

        pr = env.gh_wpt.get_pull(int(pr_id))

        metadata = {
            "wpt-pr": pr_id,
            "wpt-commits": ", ".join(item.sha1 for item in wpt_commits)
        }

        message = b"""Bug %d [wpt PR %d] - %s, a=testonly

Automatic update from web-platform-tests\n%s
"""
        message = message % ((sync and sync.bug) or self.bug,
                             pr.number,
                             pr.title.encode("utf8"),
                             b"\n--\n".join(item.msg for item in wpt_commits) + b"\n--")

        message = sync_commit.try_filter(message)

        upstream_changed = set()
        diffs = wpt_commits[-1].commit.diff(wpt_commits[0].commit.parents[0])
        for diff in diffs:
            new_path = diff.b_path
            if new_path:
                upstream_changed.add(new_path)

        logger.info("Upstream files changed:\n%s" % "\n".join(sorted(upstream_changed)))

        # If this is originally an UpstreamSync and no new changes were introduced to the GH PR
        # then we can safely skip and not need to re-apply these changes. Compare the hash of
        # the upstreamed gecko commits against the final hash in the PR.
        if isinstance(sync, upstream.UpstreamSync):
            commit_is_local = False
            pr_head = sync.pr_head
            if sync.wpt_commits.head.sha1 == pr_head:
                commit_is_local = True
            else:
                # Check if we rebased locally without pushing the rebase;
                # this is a thing we used to do to check the PR would merge
                try:
                    ref_log = sync.git_wpt.references[sync.branch_name].log()
                except Exception:
                    # If we can't read the reflog just skip this
                    pass
                else:
                    commit_is_local = any(entry.newhexsha == pr_head for entry in ref_log)
            if commit_is_local:
                logger.info("Upstream sync doesn't introduce any gecko changes")
                return None

        if copy:
            commit = self.copy_pr(git_work_gecko, git_work_wpt, pr, wpt_commits,
                                  message, author, metadata)
        else:
            commit = self.move_pr(git_work_gecko, git_work_wpt, pr, wpt_commits,
                                  message, author, prev_wpt_head, metadata)

        if commit is not None:
            self.gecko_commits.head = commit  # type: ignore

        return commit