#!/usr/bin/env python r""" This module has functions to support various data structures such as the boot_table, valid_boot_list and boot_results_table. """ import os import tempfile import json import glob from tally_sheet import * from robot.libraries.BuiltIn import BuiltIn try: from robot.utils import DotDict except ImportError: import collections import gen_print as gp import gen_valid as gv import gen_misc as gm import gen_cmd as gc 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 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: 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.svalid_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)