def audit()

in exec/src/klio_exec/commands/audit.py [0:0]

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

def audit(job_dir, config_obj):
    tw = io.TerminalWriter()
    tw.hasmarkup = True
    tw.write("Auditing your Klio job...\n")

    try:
        audit_steps = _get_audit_steps(job_dir, config_obj, tw)
        for step in audit_steps:
            step.before_tests()

        pytest_failed = _run_pytest(tw)

        tw.sep("=", "audit session starts", bold=True)

        start = time.time()

        # TODO: Currently, we're running all of the audit steps in
        # the same Python process and there's no isolation. Do we want
        # to run the tests once per step instead?
        for step in audit_steps:
            step.after_tests()

            if not any([step.errored, step.warned]):
                msg = "[{}]: audit step passed!\n".format(step.AUDIT_STEP_NAME)
                tw.write(msg, green=True)

        end = time.time()

        failed_steps = [step for step in audit_steps if step.errored]
        warned_steps = [step for step in audit_steps if step.warned]

        error_count = len(failed_steps)
        warning_count = len(warned_steps)

        finished_msg = "{0} errors, {1} warnings in {2:.3f} seconds".format(
            error_count, warning_count, end - start
        )
        color = "green"
        if error_count:
            color = "red"
        elif warning_count:
            color = "yellow"
        kwargs = {
            color: True,
            "bold": True,
        }
        tw.sep("=", finished_msg, **kwargs)

    except Exception:
        # use logging instead of tw to easily get traceback info
        logging.error(
            "Unable to run the audit command due to:\n", exc_info=True
        )

        raise SystemExit(1)

    if not pytest_failed and not error_count:
        if not warning_count:
            tw.write("Good job! Your job does not exhibit any known issues.\n")
        else:
            tw.write(
                "Cool! Your job has warnings, but no errors. "
                "Please check the warnings.\n",
                yellow=True,
            )
    else:
        tw.sep(
            "You have errors in your job. Please fix and try again.\n",
            red=True,
        )
        raise SystemExit(1)