def updated_syncs_for_push()

in sync/upstream.py [0:0]

• 67 lines of code
• 21 McCabe index (conditional complexity)

def updated_syncs_for_push(git_gecko: Repo,
                           git_wpt: Repo,
                           first_commit: GeckoCommit,
                           head_commit: GeckoCommit,
                           ) -> tuple[CreateSyncs, UpdateSyncs] | None:
    # TODO: Check syncs with pushes that no longer exist on autoland
    all_commits = wpt_commits(git_gecko, first_commit, head_commit)
    if not all_commits:
        logger.info("No new commits affecting wpt found")
        return None
    else:
        logger.info("Got %i commits since the last sync point" % len(all_commits))

    commits = remove_complete_backouts(all_commits)

    if not commits:
        logger.info("No commits remain after removing backout pairs")
        return None

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

    for commit in commits:
        assert isinstance(commit, GeckoCommit)
        if commit.upstream_sync(git_gecko, git_wpt) is not None:
            # This commit was already processed e.g. by a manual invocation, so skip
            continue
        if commit.is_backout:
            create, update = updates_for_backout(git_gecko, git_wpt, commit)
            create_syncs.update(create)
            update_syncs.update(update)
        if commit.is_downstream or commit.is_landing:
            continue
        else:
            bug = commit.bug
            if bug is None:
                continue
            sync: SyncProcess | None = None
            if bug in update_syncs:
                sync, _ = update_syncs[bug]
            else:
                statuses = ["open", "incomplete"]
                syncs = UpstreamSync.for_bug(git_gecko, git_wpt, bug, statuses=statuses,
                                             flat=True)
                if len(syncs) not in (0, 1):
                    logger.warning("Lookup of upstream syncs for bug %s returned syncs: %r" %
                                   (len(syncs), syncs))
                    # Try to pick the most recent sync
                    for status in statuses:
                        status_syncs = [s for s in syncs if s.status == status]
                        if status_syncs:
                            status_syncs.sort(key=lambda x: int(x.process_name.obj_id))
                            sync = status_syncs.pop()
                            break
                if syncs:
                    sync = syncs[0]

            if sync:
                assert isinstance(sync, UpstreamSync)
                if commit not in sync.gecko_commits:
                    update_syncs[bug] = (sync, commit)
                elif sync.pr is None:
                    head = sync.gecko_commits.head
                    assert isinstance(head, GeckoCommit)
                    update_syncs[bug] = (sync, head)
            else:
                if bug is None:
                    create_syncs[None].append(Endpoints(commit))
                elif bug in create_syncs:
                    bug_endpoints = create_syncs[bug]
                    assert isinstance(bug_endpoints, Endpoints)
                    bug_endpoints.head = commit
                else:
                    create_syncs[bug] = Endpoints(commit)

    return create_syncs, update_syncs