#!/usr/bin/python2 from picasso import Picasso from . import data_manager, tester_logger, TaskStopException from devices import cpu, mem, eth, hba, nvdimm, bmc, tpm, fan,\ psu, nvme, fpga, i2c,\ arions1, arions2,\ radagasts1, radagasts2 from collections import OrderedDict from .common_fn import get_current_time import time import re class SystemTester(object): def __init__(self, dialog_config): self.ext_loopback = False self.dialog_config = dialog_config self.test_stop_on_err = False self.test_loop = 1 self.thread_list = [] self.diag_ui = Picasso() self.question_exec_ext_loopback = 'Do you want to execute the external loopback test?' self.question_confirm_cabling = 'Are the required cables connected for external loopback test?' def __str__(self): return 'System test' def start(self): dialog_list = ['test_menu', 'test_loop', 'test_options'] # Select test items test_info_resp, test_checked_list = self.set_system_test_items( dialog_list[0]) if test_info_resp is False: return False # question for Chelsio to test external loopback ext_re = re.compile('cx5|t6|rocky|boris|radagast') for card in test_checked_list: if ext_re.search(str(card)): if 'ok' == self.diag_ui.show_yesno_dialog(self.question_exec_ext_loopback): self.ext_loopback = True if self.diag_ui.show_yesno_dialog( self.question_confirm_cabling, yes = 'Yes & Run', no = 'Cancel') == 'ok' else False break # Set test duration time test_loop_resp, loop_cnt = self.set_system_test_loop( dialog_list[1]) if test_loop_resp is False: return False if (loop_cnt > 99): loop_cnt = 99 if (loop_cnt == 0): loop_cnt = 60 * 60 * 24 * 365 # Set test option (stop on error) test_option_resp, test_option_info = self.set_system_test_options( dialog_list[2]) if test_option_resp is False: return False test_stop_on_err = test_option_info[0] # make a dictionary to store all selected device by device type test_dev_dict = OrderedDict() for device_id_list in test_checked_list: for dev in device_id_list: dev_info = dev.split('@') if len(dev_info) != 2: tester_logger.error( "Err: Parse device id({}) failed".format()) return False if dev_info[0] not in test_dev_dict: test_dev_dict[dev_info[0]] = [] test_dev_dict[dev_info[0]].append(dev_info[1]) # init test_result in selected module for dev_type in test_dev_dict.keys(): module = data_manager._module_dict.get(dev_type, None) module.clean_test_result() # update log name log_timestamp = "{}".format(time.strftime("%Y%m%d_%H%M%S")) tester_logger.update_log_name('TestSystem', log_timestamp) tester_logger.info("\nSystem Test Start: {}\n".format(get_current_time())) for loop_idx in xrange(loop_cnt): tester_logger.info( "======================== LOOP {} ===================" "========".format(loop_idx + 1)) for dev_type in test_dev_dict.keys(): module = data_manager._module_dict.get(dev_type, None) if module is None: tester_logger.error('Err: Module({}) doesn\'t exist.'.format(dev_type)) module.is_tested = False module.clean_test_result() try: self.system_test_worker(test_dev_dict, test_stop_on_err, self.ext_loopback) except TaskStopException as e: tester_logger.info(e) break # test summary try: tester_logger.info('') tester_logger.info('System Test Summary:') for dev_type in test_dev_dict.keys(): module = data_manager._module_dict.get(dev_type, None) if module is None: tester_logger.error('Err: No module exist') return False module.test_summary() tester_logger.info('') tester_logger.info("\nSystem Test Stop: {}\n".format(get_current_time())) tester_logger.info(tester_logger.get_log_file()) if test_stop_on_err: for dev_type in test_dev_dict.keys(): module = data_manager._module_dict.get(dev_type, None) if module is None: tester_logger.error('Err: No module exist') return False if not all(module.test_result): return False except AttributeError: tester_logger.error("Summary function hasn't implement yet") return True def stop(self): pass def system_test_worker(self, test_dev_dict, test_stop_on_err, external_loopback): data_manager.test_stop_on_err = test_stop_on_err data_manager.external_loopback = external_loopback for dev_type in test_dev_dict.keys(): try: # get module, ex: cpu, mem, hba... module = data_manager._module_dict.get(dev_type, None) if module is None: module = globals()[dev_type]() data_manager._module_dict[dev_type] = module # dev_id, ex: processor0(cpu), dimm1(mem)... for dev_id in test_dev_dict[dev_type]: ret = module.test(dev_id) if ret is True: continue elif test_stop_on_err is True: tester_logger.info( 'Stop on error: ({}) test failed'.format(dev_type)) return False except (TypeError, KeyError): tester_logger.error( "Err: Device ({}) hasn't implement yet".format(dev_type)) if test_stop_on_err is True: return False continue return True def set_system_test_items(self, dialog_item): test_info_resp, test_checked_list = self.diag_ui.show_dialog_from_file( self.dialog_config, dialog_item) if ((test_info_resp == 'cancel') or (test_info_resp == 'esc') or (test_info_resp == 'ok' and not len(test_checked_list))): return False, None return True, test_checked_list def set_system_test_loop(self, dialog_item): test_loop_resp, loop_info = self.diag_ui.show_dialog_from_file( self.dialog_config, dialog_item) if ((test_loop_resp == 'cancel') or (test_loop_resp == 'esc') or not loop_info.isdigit() or int(loop_info) < 0): return False, None return True, int(loop_info) def set_system_test_options(self, dialog_item): test_option_resp, test_option_info = self.diag_ui.show_dialog_from_file( self.dialog_config, dialog_item) if (test_option_resp == 'cancel' or test_option_resp == 'esc'): return False, None return True, test_option_info if __name__ == '__main__': system_tester = SystemTester() system_tester.start()