bin/plug_ins/Stop/cp_stop_check

#!/usr/bin/env python3

r"""
Check for stop conditions.  Return code of 2 if stop conditions are found.
"""

import argparse
import os
import re

from gen_arg import *  # NOQA
from gen_call_robot import *  # NOQA
from gen_cmd import *  # NOQA
from gen_misc import *  # NOQA
from gen_plug_in_utils import *  # NOQA
from gen_print import *  # NOQA
from gen_valid import *  # NOQA

# Set exit_on_error for gen_valid functions.
set_exit_on_error(True)

# Initialize default plug-in parms..
STOP_REST_FAIL = 0
STOP_COMMAND = ""
stop_test_rc = 2
STOP_VERIFY_HARDWARE_FAIL = 0


# Create parser object to process command line parameters and args.
parser = argparse.ArgumentParser(
    usage="%(prog)s [OPTIONS]",
    description='If the "Stop" plug-in is selected by the user, %(prog)s'
    + " is called by OBMC Boot Test after each boot test.  If %(prog)s returns"
    + " "
    + str(stop_test_rc)
    + ", then OBMC Boot Test will stop.  The user"
    + " may set environment variable STOP_COMMAND to contain any valid bash"
    + " command or program.  %(prog)s will run this stop command.  If the stop"
    + " command returns non-zero, then %(prog)s will return "
    + str(stop_test_rc)
    + ".  %(prog)s recognizes some special values for"
    + ' STOP_COMMAND: 1) "FAIL" means that OBMC Boot Test should stop'
    + ' whenever a boot test fails. 2) "ALL" means that OBMC Boot Test'
    + " should stop after any boot test.  If environment variable"
    + " STOP_REST_FAIL is set, OBMC Boot Test will stop if REST commands are"
    + " no longer working.",
    formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    prefix_chars="-+",
)

# The stock_list will be passed to gen_get_options.  We populate it with the names of stock parm options we
# want.  These stock parms are pre-defined by gen_get_options.
stock_list = [
    ("test_mode", get_plug_default("test_mode", 0)),
    ("quiet", get_plug_default("quiet", 0)),
    ("debug", get_plug_default("debug", 0)),
]


def exit_function(signal_number=0, frame=None):
    r"""
    Execute whenever the program ends normally or with the signals that we catch (i.e. TERM, INT).

    This function will be called by gen_exit_function().
    """

    process_robot_output_files()


def validate_parms():
    r"""
    Validate program parameters, etc.

    This function will be called by gen_setup().
    """

    get_plug_vars()


def stop_check():
    r"""
    Stop this program with the stop check return code.
    """

    if MASTER_PID != PROGRAM_PID:
        save_plug_in_value(stop_check_rc=stop_test_rc)
    exit(stop_test_rc)


def rest_fail():
    r"""
    If STOP_REST_FAIL, then this function will determine whether REST commands to the target are working.  If
    not, this function will stop the program by returning stop_test_rc.
    """

    if not STOP_REST_FAIL:
        return

    REDFISH_SUPPORT_TRANS_STATE = int(
        os.environ.get("REDFISH_SUPPORT_TRANS_STATE", 0)
    ) or int(os.environ.get("AUTOBOOT_REDFISH_SUPPORT_TRANS_STATE", 0))

    if REDFISH_SUPPORT_TRANS_STATE:
        interface = "redfish"
    else:
        interface = "rest"

    print_timen("Checking to see whether %s commands are working." % interface)
    init_robot_out_parms(get_plug_in_package_name() + "." + pgm_name + ".")
    lib_file_path = (
        init_robot_file_path("lib/utils.robot")
        + ":"
        + init_robot_file_path("lib/gen_robot_print.py")
    )
    set_mod_global(lib_file_path)
    timeout = "0 seconds"
    interval = "1 second"
    keyword_string = (
        "${match_state}=  Create Dictionary  %s=1 ;" % interface
        + " ${state}=  Wait State  ${match_state}  "
        + timeout
        + "  "
        + interval
        + "  quiet=${1} ; Rpvar  state"
    )
    set_mod_global(keyword_string)
    cmd_buf = create_robot_cmd_string(
        "extended/run_keyword.robot",
        OPENBMC_HOST,
        SSH_PORT,
        HTTPS_PORT,
        REST_USERNAME,
        REST_PASSWORD,
        OPENBMC_USERNAME,
        OPENBMC_PASSWORD,
        REDFISH_SUPPORT_TRANS_STATE,
        keyword_string,
        lib_file_path,
        quiet,
        test_mode,
        debug,
        outputdir,
        output,
        log,
        report,
        loglevel,
    )
    if not robot_cmd_fnc(cmd_buf):
        print_timen(
            "The caller wishes to stop test execution if %s commands are"
            " failing." % interface
        )
        stop_check()
    print_timen(
        "%s commands are working so no reason as of yet to stop the test."
        % interface
    )


def esel_stop_check():
    r"""
    Run the esel_stop_check program to determine whether any eSEL entries found warrant stopping the test
    run.  See esel_stop_check help text for details.
    """

    if STOP_ESEL_STOP_FILE_PATH == "":
        return

    cmd_buf = (
        "esel_stop_check --esel_stop_file_path=" + STOP_ESEL_STOP_FILE_PATH
    )
    shell_rc, out_buf = shell_cmd(cmd_buf, show_err=0)
    if shell_rc == stop_test_rc:
        print_timen(
            "The caller wishes to stop test execution based on the presence of"
            " certain esel entries."
        )
        stop_check()


def pel_stop_check():
    r"""
    Determine whether any PEL entries found warrant stopping the test
    run.
    """

    if STOP_PEL_STOP_FILE_PATH == "":
        return

    pel_txt_file_path = (
        os.environ.get("AUTOBOOT_FFDC_DIR_PATH", "")
        + os.environ.get("AUTOBOOT_FFDC_PREFIX", "")
        + "PEL_logs_list.json"
    )

    if not os.path.isfile(pel_txt_file_path):
        qprint_timen(
            "The following file was not present so no further"
            + " action will be taken."
        )
        qprint_var(pel_txt_file_path)
        return

    default_stop_dir_path = ""

    # If pel_stop_file_path is unqualified and cannot be found, pre-pend
    # default_stop_dir_path for the user.
    pel_stop_file_path = os.environ.get("STOP_PEL_STOP_FILE_PATH", "")
    if not os.path.isfile(pel_stop_file_path) and os.path.isfile(
        default_stop_dir_path + pel_stop_file_path
    ):
        pel_stop_file_path = default_stop_dir_path + pel_stop_file_path
        qprint_timen("Using default stop file path.")
        qprint_var(pel_stop_file_path)

    # First, read the file in and convert it to a list.
    pel_stop_list = file_to_list(pel_stop_file_path, newlines=0, comments=0)

    if len(pel_stop_list) == 0:
        print_timen(
            "There are no records to process in " + pel_stop_file_path + "."
        )
        return

    pel_all_list = file_to_list(pel_txt_file_path, newlines=0, comments=0)

    if len(pel_all_list) == 0:
        print_timen(
            "There are no records to process in " + pel_txt_file_path + "."
        )
        return

    for stop_pel in pel_stop_list:
        for pel_all in pel_all_list:
            pel_match = re.search(".*SRC.*" + stop_pel + ".*", pel_all)
            if pel_match:
                print_timen(
                    "The caller wishes to stop test execution based on "
                    + "the presence of certain PEL entries."
                )
                stop_check()


def main():
    gen_setup()

    print_plug_in_header()

    if STOP_COMMAND.upper() == "FAIL":
        if AUTOBOOT_BOOT_SUCCESS == "0":
            print_timen("The caller wishes to stop after each boot failure.")
            stop_check()
    elif STOP_COMMAND.upper() == "ALL":
        print_timen("The caller wishes to stop after each boot test.")
        stop_check()
    elif len(STOP_COMMAND) > 0:
        shell_rc, out_buf = shell_cmd(STOP_COMMAND, quiet=quiet, show_err=0)
        if shell_rc != 0:
            print_timen("The caller wishes to stop test execution.")
            stop_check()

    rest_fail()

    esel_stop_check()

    pel_stop_check()

    if STOP_VERIFY_HARDWARE_FAIL:
        hardware_error_found = restore_plug_in_value(0, "Verify_hardware")
        if hardware_error_found:
            print_timen(
                "The caller wishes to stop test execution when the"
                " Verify_hardware plug-in detects a hardware error."
            )
            stop_check()

    qprint_timen("The caller does not wish to stop the test run.")


main()