def updates_for_backout()

in sync/upstream.py [0:0]

• 48 lines of code
• 16 McCabe index (conditional complexity)

def updates_for_backout(git_gecko: Repo,
                        git_wpt: Repo,
                        commit: GeckoCommit,
                        ) -> tuple[CreateSyncs, UpdateSyncs]:
    backed_out_commits, bugs = commit.wpt_commits_backed_out()
    backed_out_commit_shas = {item.sha1 for item in backed_out_commits}

    create_syncs: CreateSyncs = {None: []}
    update_syncs: UpdateSyncs = {}

    for backed_out_commit in backed_out_commits:
        syncs: list[SyncProcess] = []
        backed_out_bug = backed_out_commit.bug
        if backed_out_bug:
            syncs = UpstreamSync.for_bug(git_gecko,
                                         git_wpt,
                                         backed_out_bug,
                                         statuses={"open", "incomplete"},
                                         flat=True)
            if len(syncs) not in (0, 1):
                raise ValueError("Lookup of upstream syncs for bug %s returned syncs: %r" %
                                 (len(syncs), syncs))
        if syncs:
            sync = syncs.pop()
            assert isinstance(sync, UpstreamSync)
            if commit in sync.gecko_commits:
                # This commit was already processed
                backed_out_commit_shas = set()
                return {}, {}
            if backed_out_commit in sync.upstreamed_gecko_commits:
                backed_out_commit_shas.remove(backed_out_commit.sha1)
                assert sync.bug is not None
                update_syncs[sync.bug] = (sync, commit)

    if backed_out_commit_shas:
        # This backout covers something other than known open syncs, so we need to
        # create a new sync especially for it
        # TODO: we should check for this already existing before we process the backout
        # Need to create a bug for this backout
        backout_bug = None
        for bug in bugs:
            open_bug_syncs = UpstreamSync.for_bug(git_gecko,
                                                  git_wpt,
                                                  bug,
                                                  statuses={"open", "incomplete"},
                                                  flat=False)
            if bug not in update_syncs and not open_bug_syncs:
                backout_bug = bug
                break
        if backout_bug is None:
            # TODO: Turn create_syncs into a class
            new_no_bug = create_syncs[None]
            assert isinstance(new_no_bug, list)
            new_no_bug.append(Endpoints(commit))
        else:
            create_syncs[backout_bug] = Endpoints(commit)
    return create_syncs, update_syncs