def do_landable()

in sync/command.py [0:0]

• 70 lines of code
• 18 McCabe index (conditional complexity)

def do_landable(git_gecko,
                git_wpt,
                prev_wpt_head: str | None = None,
                include_incomplete: bool = False,
                include_all: bool = True,
                retrigger: bool = False,
                blocked: bool = True,
                **kwargs: Any
                ) -> None:
    from . import update
    from .sync import LandableStatus
    from .downstream import DownstreamAction, DownstreamSync
    from .landing import current, load_sync_point, landable_commits, unlanded_with_type

    current_landing = current(git_gecko, git_wpt)

    if current_landing:
        print("Current landing will update head to %s" % current_landing.wpt_commits.head.sha1)
        prev_wpt_head = current_landing.wpt_commits.head.sha1
    else:
        sync_point = load_sync_point(git_gecko, git_wpt)
        print("Last sync was to commit %s" % sync_point["upstream"])
        prev_wpt_head = sync_point["upstream"]

    landable = landable_commits(git_gecko, git_wpt, prev_wpt_head,
                                include_incomplete=include_incomplete)

    if landable is None:
        print("Next landing will not add any new commits")
        wpt_head = None
    else:
        wpt_head, commits = landable
        print("Next landing will update wpt head to %s, adding %i new PRs" %
              (wpt_head, len(commits)))

    if include_all or retrigger:
        unlandable = unlanded_with_type(git_gecko, git_wpt, wpt_head, prev_wpt_head)
        count = 0
        for pr, _, status in unlandable:
            count += 1
            if blocked and status in (LandableStatus.ready,
                                      LandableStatus.upstream,
                                      LandableStatus.skip):
                continue
            msg = status.reason_str()
            if status == LandableStatus.missing_try_results:
                sync = DownstreamSync.for_pr(git_gecko, git_wpt, pr)
                assert isinstance(sync, DownstreamSync)
                next_action = sync.next_action
                reason = next_action.reason_str()
                if next_action == DownstreamAction.wait_try:
                    latest_try_push = sync.latest_try_push
                    assert latest_try_push is not None
                    reason = "{} {}".format(reason,
                                            latest_try_push.treeherder_url)
                elif next_action == DownstreamAction.manual_fix:
                    latest_try_push = sync.latest_try_push
                    assert latest_try_push is not None
                    reason = "Manual fixup required {}".format(
                        latest_try_push.treeherder_url)
                msg = f"{msg} ({reason})"
            elif status == LandableStatus.error:
                sync = DownstreamSync.for_pr(git_gecko, git_wpt, pr)
                assert sync is not None
                if sync.error:
                    err_msg = sync.error["message"] or ""
                    err_msg = err_msg.splitlines()[0] if err_msg else err_msg
                    msg = f"{msg} ({err_msg})"
            print(f"{pr}: {msg}")

        print("%i PRs are unlandable:" % count)

        if retrigger:
            errors = update.retrigger(git_gecko, git_wpt, unlandable)
            if errors:
                print("The following PRs have errors:\n%s" % "\n".join(
                    str(item) for item in errors))