| #!/usr/bin/env python3 |
| |
| r""" |
| This module has functions to support various data structures such as the boot_table, valid_boot_list and |
| boot_results_table. |
| """ |
| |
| import glob |
| import json |
| import os |
| import tempfile |
| |
| from robot.libraries.BuiltIn import BuiltIn |
| from tally_sheet import * |
| |
| try: |
| from robot.utils import DotDict |
| except ImportError: |
| import collections |
| |
| import gen_cmd as gc |
| import gen_misc as gm |
| import gen_print as gp |
| import gen_valid as gv |
| import var_funcs as vf |
| |
| # The code base directory will be one level up from the directory containing this module. |
| code_base_dir_path = os.path.dirname(os.path.dirname(__file__)) + os.sep |
| |
| redfish_support_trans_state = int( |
| os.environ.get("REDFISH_SUPPORT_TRANS_STATE", 0) |
| ) or int( |
| BuiltIn().get_variable_value("${REDFISH_SUPPORT_TRANS_STATE}", default=0) |
| ) |
| |
| platform_arch_type = os.environ.get( |
| "PLATFORM_ARCH_TYPE", "" |
| ) or BuiltIn().get_variable_value("${PLATFORM_ARCH_TYPE}", default="power") |
| |
| |
| def create_boot_table(file_path=None, os_host=""): |
| r""" |
| Read the boot table JSON file, convert it to an object and return it. |
| |
| Note that if the user is running without a global OS_HOST robot variable specified, this function will |
| remove all of the "os_" start and end state requirements from the JSON data. |
| |
| Description of argument(s): |
| file_path The path to the boot_table file. If this value is not specified, it will |
| be obtained from the "BOOT_TABLE_PATH" environment variable, if set. |
| Otherwise, it will default to "data/boot_table.json". If this value is a |
| relative path, this function will use the code_base_dir_path as the base |
| directory (see definition above). |
| os_host The host name or IP address of the host associated with the machine being |
| tested. If the user is running without an OS_HOST (i.e. if this argument |
| is blank), we remove os starting and ending state requirements from the |
| boot entries. |
| """ |
| if file_path is None: |
| if redfish_support_trans_state and platform_arch_type != "x86": |
| file_path = os.environ.get( |
| "BOOT_TABLE_PATH", "data/boot_table_redfish.json" |
| ) |
| elif platform_arch_type == "x86": |
| file_path = os.environ.get( |
| "BOOT_TABLE_PATH", "data/boot_table_x86.json" |
| ) |
| else: |
| file_path = os.environ.get( |
| "BOOT_TABLE_PATH", "data/boot_table.json" |
| ) |
| |
| if not file_path.startswith("/"): |
| file_path = code_base_dir_path + file_path |
| |
| # Pre-process the file by removing blank lines and comment lines. |
| temp = tempfile.NamedTemporaryFile() |
| temp_file_path = temp.name |
| |
| cmd_buf = "egrep -v '^[ ]*$|^[ ]*#' " + file_path + " > " + temp_file_path |
| gc.cmd_fnc_u(cmd_buf, quiet=1) |
| |
| boot_file = open(temp_file_path) |
| boot_table = json.load(boot_file, object_hook=DotDict) |
| |
| # If the user is running without an OS_HOST, we remove os starting and ending state requirements from |
| # the boot entries. |
| if os_host == "": |
| for boot in boot_table: |
| state_keys = ["start", "end"] |
| for state_key in state_keys: |
| for sub_state in list(boot_table[boot][state_key]): |
| if sub_state.startswith("os_"): |
| boot_table[boot][state_key].pop(sub_state, None) |
| |
| # For every boot_type we should have a corresponding mfg mode boot type. |
| enhanced_boot_table = DotDict() |
| for key, value in boot_table.items(): |
| enhanced_boot_table[key] = value |
| enhanced_boot_table[key + " (mfg)"] = value |
| |
| return enhanced_boot_table |
| |
| |
| def create_valid_boot_list(boot_table): |
| r""" |
| Return a list of all of the valid boot types (e.g. ['REST Power On', 'REST Power Off', ...]). |
| |
| Description of argument(s): |
| boot_table A boot table such as is returned by the create_boot_table function. |
| """ |
| |
| return list(boot_table.keys()) |
| |
| |
| def read_boot_lists(dir_path="data/boot_lists/"): |
| r""" |
| Read the contents of all the boot lists files found in the given boot lists directory and return |
| dictionary of the lists. |
| |
| Boot lists are simply files containing a boot test name on each line. These files are useful for |
| categorizing and organizing boot tests. For example, there may be a "Power_on" list, a "Power_off" list, |
| etc. |
| |
| The names of the boot list files will be the keys to the top level dictionary. Each dictionary entry is |
| a list of all the boot tests found in the corresponding file. |
| |
| Here is an abbreviated look at the resulting boot_lists dictionary. |
| |
| boot_lists: |
| boot_lists[All]: |
| boot_lists[All][0]: REST Power On |
| boot_lists[All][1]: REST Power Off |
| ... |
| boot_lists[Code_update]: |
| boot_lists[Code_update][0]: BMC oob hpm |
| boot_lists[Code_update][1]: BMC ib hpm |
| ... |
| |
| Description of argument(s): |
| dir_path The path to the directory containing the boot list files. If this value |
| is a relative path, this function will use the code_base_dir_path as the |
| base directory (see definition above). |
| """ |
| |
| if not dir_path.startswith("/"): |
| # Dir path is relative. |
| dir_path = code_base_dir_path + dir_path |
| |
| # Get a list of all file names in the directory. |
| boot_file_names = os.listdir(dir_path) |
| |
| boot_lists = DotDict() |
| for boot_category in boot_file_names: |
| file_path = gm.which(dir_path + boot_category) |
| boot_list = gm.file_to_list(file_path, newlines=0, comments=0, trim=1) |
| boot_lists[boot_category] = boot_list |
| |
| return boot_lists |
| |
| |
| def valid_boot_list(boot_list, valid_boot_types): |
| r""" |
| Verify that each entry in boot_list is a supported boot test. |
| |
| Description of argument(s): |
| boot_list An array (i.e. list) of boot test types (e.g. "REST Power On"). |
| valid_boot_types A list of valid boot types such as that returned by |
| create_valid_boot_list. |
| """ |
| |
| for boot_name in boot_list: |
| boot_name = boot_name.strip(" ") |
| error_message = gv.valid_value( |
| boot_name, valid_values=valid_boot_types, var_name="boot_name" |
| ) |
| if error_message != "": |
| BuiltIn().fail(gp.sprint_error(error_message)) |
| |
| |
| class boot_results: |
| r""" |
| This class defines a boot_results table. |
| """ |
| |
| def __init__( |
| self, boot_table, boot_pass=0, boot_fail=0, obj_name="boot_results" |
| ): |
| r""" |
| Initialize the boot results object. |
| |
| Description of argument(s): |
| boot_table Boot table object (see definition above). The boot table contains all of |
| the valid boot test types. It can be created with the create_boot_table |
| function. |
| boot_pass An initial boot_pass value. This program may be called as part of a |
| larger test suite. As such there may already have been some successful |
| boot tests that we need to keep track of. |
| boot_fail An initial boot_fail value. This program may be called as part of a |
| larger test suite. As such there may already have been some unsuccessful |
| boot tests that we need to keep track of. |
| obj_name The name of this object. |
| """ |
| |
| # Store the method parms as class data. |
| self.__obj_name = obj_name |
| self.__initial_boot_pass = boot_pass |
| self.__initial_boot_fail = boot_fail |
| |
| # Create boot_results_fields for use in creating boot_results table. |
| boot_results_fields = DotDict([("total", 0), ("pass", 0), ("fail", 0)]) |
| # Create boot_results table. |
| self.__boot_results = tally_sheet( |
| "boot type", boot_results_fields, "boot_test_results" |
| ) |
| self.__boot_results.set_sum_fields(["total", "pass", "fail"]) |
| self.__boot_results.set_calc_fields(["total=pass+fail"]) |
| # Create one row in the result table for each kind of boot test in the boot_table (i.e. for all |
| # supported boot tests). |
| for boot_name in list(boot_table.keys()): |
| self.__boot_results.add_row(boot_name) |
| |
| def add_row(self, *args, **kwargs): |
| r""" |
| Add row to tally_sheet class object. |
| |
| Description of argument(s): |
| See add_row method in tally_sheet.py for a description of all arguments. |
| """ |
| self.__boot_results.add_row(*args, **kwargs) |
| |
| def return_total_pass_fail(self): |
| r""" |
| Return the total boot_pass and boot_fail values. This information is comprised of the pass/fail |
| values from the table plus the initial pass/fail values. |
| """ |
| |
| totals_line = self.__boot_results.calc() |
| return ( |
| totals_line["pass"] + self.__initial_boot_pass, |
| totals_line["fail"] + self.__initial_boot_fail, |
| ) |
| |
| def update(self, boot_type, boot_status): |
| r""" |
| Update our boot_results_table. This includes: |
| - Updating the record for the given boot_type by incrementing the pass or fail field. |
| - Calling the calc method to have the totals calculated. |
| |
| Description of argument(s): |
| boot_type The type of boot test just done (e.g. "REST Power On"). |
| boot_status The status of the boot just done. This should be equal to either "pass" |
| or "fail" (case-insensitive). |
| """ |
| |
| self.__boot_results.inc_row_field(boot_type, boot_status.lower()) |
| self.__boot_results.calc() |
| |
| def sprint_report(self, header_footer="\n"): |
| r""" |
| String-print the formatted boot_resuls_table and return them. |
| |
| Description of argument(s): |
| header_footer This indicates whether a header and footer are to be included in the |
| report. |
| """ |
| |
| buffer = "" |
| |
| buffer += gp.sprint(header_footer) |
| buffer += self.__boot_results.sprint_report() |
| buffer += gp.sprint(header_footer) |
| |
| return buffer |
| |
| def print_report(self, header_footer="\n", quiet=None): |
| r""" |
| Print the formatted boot_resuls_table to the console. |
| |
| Description of argument(s): |
| See sprint_report for details. |
| quiet Only print if this value is 0. This function will search upward in the |
| stack to get the default value. |
| """ |
| |
| quiet = int(gm.dft(quiet, gp.get_stack_var("quiet", 0))) |
| |
| gp.qprint(self.sprint_report(header_footer)) |
| |
| def sprint_obj(self): |
| r""" |
| sprint the fields of this object. This would normally be for debug purposes only. |
| """ |
| |
| buffer = "" |
| |
| buffer += "class name: " + self.__class__.__name__ + "\n" |
| buffer += gp.sprint_var(self.__obj_name) |
| buffer += self.__boot_results.sprint_obj() |
| buffer += gp.sprint_var(self.__initial_boot_pass) |
| buffer += gp.sprint_var(self.__initial_boot_fail) |
| |
| return buffer |
| |
| def print_obj(self): |
| r""" |
| Print the fields of this object to stdout. This would normally be for debug purposes. |
| """ |
| |
| gp.gp_print(self.sprint_obj()) |
| |
| |
| def create_boot_results_file_path(pgm_name, openbmc_nickname, master_pid): |
| r""" |
| Create a file path to be used to store a boot_results object. |
| |
| Description of argument(s): |
| pgm_name The name of the program. This will form part of the resulting file name. |
| openbmc_nickname The name of the system. This could be a nickname, a hostname, an IP, |
| etc. This will form part of the resulting file name. |
| master_pid The master process id which will form part of the file name. |
| """ |
| |
| USER = os.environ.get("USER", "") |
| dir_path = "/tmp/" + USER + "/" |
| if not os.path.exists(dir_path): |
| os.makedirs(dir_path) |
| |
| file_name_dict = vf.create_var_dict(pgm_name, openbmc_nickname, master_pid) |
| return vf.create_file_path( |
| file_name_dict, dir_path=dir_path, file_suffix=":boot_results" |
| ) |
| |
| |
| def cleanup_boot_results_file(): |
| r""" |
| Delete all boot results files whose corresponding pids are no longer active. |
| """ |
| |
| # Use create_boot_results_file_path to create a globex to find all of the existing boot results files. |
| globex = create_boot_results_file_path("*", "*", "*") |
| file_list = sorted(glob.glob(globex)) |
| for file_path in file_list: |
| # Use parse_file_path to extract info from the file path. |
| file_dict = vf.parse_file_path(file_path) |
| if gm.pid_active(file_dict["master_pid"]): |
| gp.qprint_timen("Preserving " + file_path + ".") |
| else: |
| gc.cmd_fnc("rm -f " + file_path) |
| |
| |
| def update_boot_history(boot_history, boot_start_message, max_boot_history=10): |
| r""" |
| Update the boot_history list by appending the boot_start_message and by removing all but the last n |
| entries. |
| |
| Description of argument(s): |
| boot_history A list of boot start messages. |
| boot_start_message This is typically a time-stamped line of text announcing the start of a |
| boot test. |
| max_boot_history The max number of entries to be kept in the boot_history list. The |
| oldest entries are deleted to achieve this list size. |
| """ |
| |
| boot_history.append(boot_start_message) |
| |
| # Trim list to max number of entries. |
| del boot_history[: max(0, len(boot_history) - max_boot_history)] |
| |
| |
| def print_boot_history(boot_history, quiet=None): |
| r""" |
| Print the last ten boots done with their time stamps. |
| |
| Description of argument(s): |
| quiet Only print if this value is 0. This function will search upward in the |
| stack to get the default value. |
| """ |
| |
| quiet = int(gm.dft(quiet, gp.get_stack_var("quiet", 0))) |
| |
| # indent 0, 90 chars wide, linefeed, char is "=" |
| gp.qprint_dashes(0, 90) |
| gp.qprintn("Last 10 boots:\n") |
| |
| for boot_entry in boot_history: |
| gp.qprint(boot_entry) |
| gp.qprint_dashes(0, 90) |