def try_push_complete()

in sync/landing.py [0:0]

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

def try_push_complete(git_gecko: Repo,
                      git_wpt: Repo,
                      try_push: TryPush,
                      sync: LandingSync,
                      allow_push: bool = True,
                      accept_failures: bool = False,
                      tasks: Any | None = None,
                      ) -> None:
    """Run after all jobs in a try push are complete.

    This function handles updating the metadata based on the try push, or scheduling
    more jobs. In the case that the metadata has been updated successfully, the try
    push is marked as complete. If there's an error e.g. an infrastructure failure
    the try push is not marked as complete; user action is required to complete the
    handling of the try push (either by passing in accept_failures=True to indicate
    that the failure is not significant or by retyring the try push in which case the
    existing one will be marked as complete)."""

    if try_push.status == "complete":
        logger.warning("Called try_push_complete on a completed try push")
        return None

    if accept_failures:
        try_push.accept_failures = True  # type: ignore

    if tasks is None:
        tasks = try_push.tasks()

    if tasks is None:
        logger.error("Taskgroup id is not yet set")
        return None

    try_result = sync.try_result(tasks=tasks)

    if try_result == TryPushResult.pending:
        logger.info("Try push results are pending")
        return None

    if not try_result == TryPushResult.success:
        if try_result.is_failure():
            if try_result == TryPushResult.infra_fail:
                message = record_build_failures(sync, try_push)
                try_push.infra_fail = True  # type: ignore
                raise AbortError(message)
            elif try_result == TryPushResult.too_many_failures and not try_push.stability:
                message = record_too_many_failures(sync, try_push)
                raise AbortError(message)

        if not try_push.stability:
            update_metadata(sync, try_push)
        else:
            retriggered = tasks.retriggered_wpt_states()
            if not retriggered:
                if try_result == TryPushResult.too_many_failures:
                    record_too_many_failures(sync, try_push)
                    try_push.status = "complete"  # type: ignore
                    return None
                num_new_jobs = tasks.retrigger_failures()
                logger.info(f"{num_new_jobs} new tasks scheduled on try for {sync.bug}")
                if num_new_jobs:
                    assert sync.bug is not None
                    env.bz.comment(sync.bug,
                                   ("Retriggered failing web-platform-test tasks on "
                                    "try before final metadata update."))
                    return None

            update_metadata(sync, try_push, tasks)

    try_push.status = "complete"  # type: ignore

    if try_result == TryPushResult.infra_fail:
        record_infra_fail(sync, try_push)
        return None

    update_landing(git_gecko, git_wpt, allow_push=allow_push)