def run_rtas_for_os()

in cortado/rtas/_cli.py [0:0]

• 24 lines of code
• 10 McCabe index (conditional complexity)

def run_rtas_for_os():
    configure_logging()

    if len(sys.argv) == 2:
        pool_size = int(sys.argv[1])
    else:
        pool_size = 1

    current_os = get_current_os()

    registry = get_registry()  # load all modules
    rtas_for_os = [rta for rta in registry.values() if current_os in rta.platforms]

    log.info(f"RTAs for `{current_os}`: {len(rtas_for_os)}")

    rta_names = [r.name for r in rtas_for_os]

    log.info(f"Parallel processes to run RTAs: {pool_size}")

    with Pool(pool_size) as p:
        errors = p.map(_run_rta_in_process, rta_names)

    names_and_errors = list(zip(rta_names, errors))

    error_counter = Counter()  # type: ignore
    error_counter.update([("Failed" if e else "Succeeded") for _, e in names_and_errors])  # type: ignore

    for name, error in sorted(names_and_errors):
        if errors:
            log.info(f"Failure: `{name}`, Error: `{str(error)}`")
        else:
            log.info(f"Success: `{name}`")

    results = ", ".join([f"{k}={v}" for k, v in error_counter.items()])  # type: ignore
    log.info(f"RTA execution results: {results}")