def landable_commits()

in sync/landing.py [0:0]

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

def landable_commits(git_gecko: Repo,
                     git_wpt: Repo,
                     prev_wpt_head: str,
                     wpt_head: str | None = None,
                     include_incomplete: bool = False
                     ) -> tuple[str, LandableCommits] | None:
    """Get the list of commits that are able to land.

    :param prev_wpt_head: The sha1 of the previous wpt commit landed to gecko.
    :param wpt_head: The sha1 of the latest possible commit to land to gecko,
                     or None to use the head of the master branch
    :param include_incomplete: By default we don't attempt to land anything that
                               hasn't completed a metadata update. This flag disables
                               that and just lands everything up to the specified commit."""
    if wpt_head is None:
        wpt_head = "origin/master"
    pr_commits = unlanded_wpt_commits_by_pr(git_gecko, git_wpt, prev_wpt_head, wpt_head)
    landable_commits = []
    for pr, commits in pr_commits:
        last = False
        if not pr:
            # Assume this was some trivial fixup:
            continue

        first_commit = first_non_merge(commits)
        if not first_commit:
            # If we only have a merge commit just use that; it doesn't come from gecko anyway
            first_commit = commits[-1]

        def upstream_sync(bug_number):
            syncs = upstream.UpstreamSync.for_bug(git_gecko,
                                                  git_wpt,
                                                  bug_number,
                                                  flat=True)
            for sync in syncs:
                if sync.merge_sha == commits[-1].sha1 and not sync.wpt_commits:
                    # TODO: this shouldn't be mutating here
                    with SyncLock("upstream", None) as lock:
                        assert isinstance(lock, SyncLock)
                        with sync.as_mut(lock):
                            # If we merged with a merge commit, the set of commits
                            # here will be empty
                            sync.set_wpt_base(sync_commit.WptCommit(git_wpt,
                                                                    commits[0].sha1 + "~").sha1)

                # Only check the first commit since later ones could be added in the PR
                sync_revs = {item.canonical_rev for item in sync.upstreamed_gecko_commits}
                if any(commit.metadata.get("gecko-commit") in sync_revs for commit in commits):
                    break
            else:
                sync = None
            return sync

        sync = None
        sync = load.get_pr_sync(git_gecko, git_wpt, pr)
        if isinstance(sync, downstream.DownstreamSync):
            if sync and "affected-tests" in sync.data and sync.data["affected-tests"] is None:
                del sync.data["affected-tests"]
        if not include_incomplete:
            if not sync:
                # TODO: schedule a downstream sync for this pr
                logger.info("PR %s has no corresponding sync" % pr)
                last = True
            elif (isinstance(sync, downstream.DownstreamSync) and
                  sync.landable_status not in (LandableStatus.ready, LandableStatus.skip)):
                logger.info(f"PR {pr}: {sync.landable_status.reason_str()}")
                last = True
            if last:
                break
        assert isinstance(sync, (upstream.UpstreamSync, downstream.DownstreamSync))
        landable_commits.append((pr, sync, commits))

    if not landable_commits:
        logger.info("No new commits are landable")
        return None

    wpt_head = landable_commits[-1][2][-1].sha1
    logger.info("Landing up to commit %s" % wpt_head)

    return wpt_head, landable_commits