def gecko_push()

in sync/upstream.py [0:0]

• 40 lines of code
• 8 McCabe index (conditional complexity)

def gecko_push(git_gecko: Repo,
               git_wpt: Repo,
               repository_name: str,
               hg_rev: str,
               raise_on_error: bool = False,
               base_rev: Any | None = None,
               ) -> tuple[set[UpstreamSync], set[UpstreamSync], set] | None:
    rev = git_gecko.rev_parse(cinnabar(git_gecko).hg2git(hg_rev))
    last_sync_point, prev_commit = UpstreamSync.prev_gecko_commit(git_gecko,
                                                                  repository_name)

    assert last_sync_point.commit is not None
    if base_rev is None and git_gecko.is_ancestor(rev, last_sync_point.commit.commit):
        logger.info("Last sync point moved past commit")
        return None

    with SyncLock("upstream", None) as lock:
        assert isinstance(lock, SyncLock)
        updated = updated_syncs_for_push(git_gecko,
                                         git_wpt,
                                         prev_commit,
                                         sync_commit.GeckoCommit(git_gecko, rev))

        if updated is None:
            return set(), set(), set()

        create_endpoints, update_syncs = updated

        pushed_syncs, failed_syncs = update_sync_prs(lock,
                                                     git_gecko,
                                                     git_wpt,
                                                     create_endpoints,
                                                     update_syncs,
                                                     raise_on_error=raise_on_error)

        landable_syncs = {item for item in UpstreamSync.load_by_status(git_gecko, git_wpt, "open")
                          if item.error is None}
        if TYPE_CHECKING:
            landable = cast(Set[UpstreamSync], landable_syncs)
        else:
            landable = landable_syncs
        landed_syncs = try_land_syncs(lock, landable)

        # TODO
        if not git_gecko.is_ancestor(rev, last_sync_point.commit.commit):
            with last_sync_point.as_mut(lock):
                last_sync_point.commit = rev.hexsha  # type: ignore

    return pushed_syncs, landed_syncs, failed_syncs