# Copyright 2017 The Chromium OS Authors. All rights reserved. # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. import random import logging import time from autotest_lib.client.common_lib import error from autotest_lib.server.cros.cfm import cfm_base_test from autotest_lib.client.common_lib.cros.manual import audio_helper from autotest_lib.client.common_lib.cros.manual import cfm_helper from autotest_lib.client.common_lib.cros.manual import meet_helper from autotest_lib.client.common_lib.cros.manual import video_helper from autotest_lib.client.common_lib.cros.manual import get_usb_devices ATRUS = "Hangouts Meet speakermic" CORE_DIR_LINES = 5 FAILURE_REASON = { 'telemetry': ['No hangouts or meet telemetry API available', 'telemetry api', 'RPC: cfm_main_screen.', 'Failed RPC', 'cfm_main_screen', 'Webview with screen param', 'autotest_lib.client.common_lib.error.TestFail'], 'chrome':[], 'kernel':[], 'atrus':[] } NUM_AUDIO_STREAM_IN_MEETING = 3 LOG_CHECK_LIST = ['check_kernel_errorlog', 'check_video_errorlog', 'check_audio_errorlog', 'check_chrome_errorlog', 'check_atrus_errorlog', 'check_usb_stability'] TEST_DELAY = 0 class enterprise_CFM_Test(cfm_base_test.CfmBaseTest): """Executes multiple tests on CFM device based on control file, after each test, perform mulitple verifications based on control file. Test flow can be controlled by control file, such as abort on failure, or continue on failure. """ version = 1 def gpio_test(self): """ Powercycle USB port on Guado. """ if self.run_meeting_test: if self.random: min_meetings = random.randrange(-1, self.gpio_min_meets) else: min_meetings = self.gpio_min_meets if self.meets_last_gpio <= min_meetings: logging.debug('Skip gpio test.') return True, None if cfm_helper.check_is_platform(self.client, 'guado'): status, errmsg = cfm_helper.gpio_usb_test(self.client, self.gpio_list, self.puts, self.gpio_pause, 'guado') self.gpio_no += 1 self.meets_last_gpio = 0 else: logging.info('Skip gpio_test for non-guado CfM.') return True, None ## workaround for bug b/69261543 if self.is_in_meet: self.is_camera_muted = self.cfm_facade.is_camera_muted() return status, errmsg def meeting_test(self): """ Join/leave meeting. """ if self.is_in_meet: status, errmsg = meet_helper.leave_meeting(self.cfm_facade, self.is_meeting) if status: self.is_camera_muted = True self.is_in_meet = False return True, None else: if self.recover_on_failure or self.meets_last_reboot == 0: self.is_in_meet = False else: self.is_in_meet = True return False, errmsg else: status, errmsg = meet_helper.join_meeting(self.cfm_facade, self.is_meeting, self.meeting_code) if status: self.is_camera_muted = False self.is_in_meet = True self.meet_no += 1 self.meets_last_reboot += 1 self.meets_last_gpio += 1 return True, None else: if self.recover_on_failure or self.meets_last_reboot == 0: self.is_in_meet = True else: self.is_in_meet = False return False, errmsg def reboot_test(self): """ Reboot CfM. """ if self.run_meeting_test: if self.random: min_meetings = random.randrange(-1, self.reboot_min_meets) else: min_meetings = self.reboot_min_meets if self.meets_last_reboot < min_meetings: logging.info('Skip reboot CfM test') return True, None try: self.client.reboot() time.sleep(self.reboot_timeout) except Exception as e: errmsg = 'Reboot test fails for %s' % self.ip_addr logging.exception(errmsg) self.reboot_log = cfm_helper.check_last_reboot(self.client) self.chrome_file = cfm_helper.check_chrome_logfile(self.client) return False, errmsg self.reboot_no += 1 self.meets_last_reboot = 0 if 'meeting_test' in self.action_config: self.reboot_log = cfm_helper.check_last_reboot(self.client) if list(set(self.verification_config) & set(LOG_CHECK_LIST)): self.log_checking_point = cfm_helper.find_last_log(self.client, self.speaker) return self.restart_chrome_and_meeting(True) self.reboot_log = cfm_helper.check_last_reboot(self.client) self.chrome_file = cfm_helper.check_chrome_logfile(self.client) if list(set(self.verification_config) & set(LOG_CHECK_LIST)): self.log_checking_point = cfm_helper.find_last_log(self.client, self.speaker) return True, None def restart_chrome_and_meeting(self, recovery): """ Restart Chrome and Join/Start meeting if previous state is in meeting. """ result, errmsg = meet_helper.restart_chrome(self.cfm_facade, self.is_meeting, recovery) logging.info('restart chrome result:%s, msg = %s', result,errmsg) if not result: logging.info('restart chrome result: False, msg = %s', errmsg) self.chrome_file = cfm_helper.check_chrome_logfile(self.client) return False, errmsg if self.is_in_meet: logging.info('start meeting if needed') self.is_in_meet = False self.chrome_file = cfm_helper.check_chrome_logfile(self.client) test_result, ret_msg = self.meeting_test() if test_result: try: self.is_camera_muted = self.cfm_facade.is_camera_muted() self.is_mic_muted = self.cfm_facade.is_mic_muted() except Exception as e: errmsg = 'Fail to run telemetry api to check camera..' logging.exception(errmsg) return False, errmsg self.chrome_file = cfm_helper.check_chrome_logfile(self.client) return True, None # TODO(mzhuo): Adding resetusb test. def reset_usb_test(self): """ Reset USB port """ return True, None def mute_unmute_camera_test(self): """ Mute or unmute camera. """ if not self.camera: logging.info('Skip mute/unmute camera testing.') return True, None if self.is_in_meet: if self.is_camera_muted: status, errmsg = meet_helper.mute_unmute_camera( self.cfm_facade, True) if status: self.is_camera_muted = False else: return False, errmsg else: status, errmsg = meet_helper.mute_unmute_camera( self.cfm_facade, False) if status: self.is_camera_muted = True else: return False, errmsg return True, None def mute_unmute_mic_test(self): """ Mute or unmute microphone. """ if not self.speaker and not cfm_helper.check_is_platform(self.client, 'buddy'): logging.info('Skip mute/unmute microphone testing.') return True, None if self.is_in_meet: if self.is_mic_muted: status, errmsg = meet_helper.mute_unmute_mic(self.cfm_facade, True) if status: self.is_mic_muted = False else: return False, errmsg else: status, errmsg = meet_helper.mute_unmute_mic(self.cfm_facade, False) if status: self.is_mic_muted = True else: return False, errmsg return True, None def speaker_volume_test(self): """ Update speaker volume. """ if not self.speaker and not cfm_helper.check_is_platform(self.client, 'buddy'): logging.info('Skip update volume of speaker testing.') return True, None if self.is_in_meet: test_result, ret_msg = meet_helper.speaker_volume_test( self.cfm_facade, self.vol_change_step, self.vol_change_mode, self.random) if test_result: self.speaker_volume = int(ret_msg) return True, None else: return False, ret_msg else: return True, None # TODO(mzhuo): Adding test to turn on/off monite. def flap_monitor_test(self): """ Connect or disconnect monitor. """ return True, None def check_usb_enumeration(self): """ Verify all usb devices which were enumerated originally are enumerated. """ return cfm_helper.check_usb_enumeration(self.client, self.puts) def check_usb_inf_init(self): """ Verify all usb devices which were enumerated originally have valid interfaces: video interface, audio interface or touch interface. """ return cfm_helper.check_usb_interface_initializion(self.client, self.puts) def check_v4l2_interface(self): """ Verify camera has v4l2 file handler created. """ if not self.camera: return True, None return video_helper.check_v4l2_interface(self.client, self.camera, self.name_camera) def check_audio_card(self): """ Verify speaker/microphone has audip file handler created. """ if not self.speaker: return True, None return audio_helper.check_soundcard_by_name(self.client, self.name_speaker) def check_cras_speaker(self): """ Verify cras server detects speaker. """ if not self.speaker: return True, None return audio_helper.check_speaker_exist_cras(self.client, self.name_speaker) def check_cras_mic(self): """ Verify cras server detects microphone. """ if not self.speaker: return True, None return audio_helper.check_microphone_exist_cras(self.client, self.name_speaker) def check_cras_mic_mute(self): """ Verify cras shows mic muted or unmuted as expected. """ if not self.speaker or not self.is_in_meet: return True, None return audio_helper.check_cras_mic_mute(self.client, self.cfm_facade) def check_cras_pspeaker(self): """ Verify cras shows correct preferred speaker. """ if not self.speaker: return True, None return audio_helper.check_is_preferred_speaker(self.client, self.name_speaker) def check_cras_speaker_vol(self): """ Verify cras shows correct volume for speaker. """ if not self.speaker or not self.is_in_meet: return True, None return audio_helper.check_default_speaker_volume(self.client, self.cfm_facade) def check_cras_pmic(self): """ Verify cras shows correct preferred microphone. """ if not self.speaker: return True, None return audio_helper.check_is_preferred_mic(self.client, self.name_speaker) # TODO(mzhuo): add verification for preferred camera def check_prefer_camera(self): """ Verify preferred camera is correct. """ return True, None #TODO(mzhuo): add verification to verify camera is muted or unmuted #in video stack in kernel space. def check_camera_mute(self): """ Verify camera is muted or unmuted as expected. """ return True, None def check_video_stream(self): """ Verify camera is streaming or not streaming as expected. """ if not self.camera: return True, None return video_helper.check_video_stream(self.client, self.is_camera_muted, self.camera, self.name_camera) def check_audio_stream(self): """ Verify speaker is streaming or not streaming as expected. """ if not self.speaker: return True, None return audio_helper.check_audio_stream(self.client, self.is_in_meet) # TODO(mzhuo): Adding verification for speaker in Hotrod App def check_hotrod_speaker(self): """ Verify hotrod shows all speakers. """ return True, None # TODO(mzhuo): Adding verification for mic in Hotrod App def check_hotrod_mic(self): """ Verify hotrod shows all microphone. """ return True, None # TODO(mzhuo): Adding verification for camera in Hotrod App def check_hotrod_camera(self): """ Verify hotrod shows all cameras. """ return True, None # TODO(mzhuo): Adding verification for speaker in Hotrod App def check_hotrod_pspeaker(self): """ Verify hotrod selects correct preferred speaker. """ return True, None # TODO(mzhuo): Adding verification for mic in Hotrod App def check_hotrod_pmic(self): """ Verify hotrod selects correct preferred microphone. """ return True, None # TODO(mzhuo): Adding verification for camera in Hotrod App def check_hotrod_pcamera(self): """ Verify hotrod selects correct preferred camera. """ return True, None #TODO(mzhuo): Adding verififaction in hotrod layer for speaker volume def check_hotrod_speaker_vol(self): """ Verify hotrod can set volume for speaker. """ return True, None #TODO(mzhuo): Adding verififaction in hotrod layer for mic mute status def check_hotrod_mic_state(self): """ Verify hotrod can mute or unmute microphone. """ return True, None #TODO(mzhuo): Adding verififaction in hotrod layer for camera status def check_hotrod_camera_state(self): """ Verify hotrod can mute or unmute camera. """ return True, None def check_kernel_errorlog(self): """ Check /var/log/message does not contain any element in error_key_words['kernel']. """ return cfm_helper.check_log(self.client, self.log_checking_point, self.error_key_words, 'kernel', 'messages') def check_chrome_errorlog(self): """ Check /var/log/chrome/chrome does not contain any element in error_key_words['chrome']. """ return cfm_helper.check_log(self.client, self.log_checking_point, self.error_key_words, 'chrome', 'chrome') def check_atrus_errorlog(self): """ Check /var/log/atrus.log does not contain any element in error_key_words['atrus']. """ if self.current_test in ['gpio_test', 'resetusb_test']: return True, None if not self.name_speaker: return True, None if self.name_speaker not in ATRUS: logging.info('Speaker %s', self.name_speaker) return True, None if cfm_helper.check_is_platform(self.client, 'guado'): return cfm_helper.check_log(self.client, self.log_checking_point, self.error_key_words, 'atrus', 'atrus') else: return True, None def check_video_errorlog(self): """ Check /var/log/message does not contain any element in error_key_words['video']. """ return cfm_helper.check_log(self.client, self.log_checking_point, self.error_key_words, 'video', 'messages') def check_audio_errorlog(self): """ Check /var/log/message does not contain any element in error_key_words['audio']. """ return cfm_helper.check_log(self.client, self.log_checking_point, self.error_key_words, 'audio', 'messages') def check_usb_errorlog(self): """ Check /var/log/message does not contain any element in error_key_words['usb']. """ return cfm_helper.check_log(self.client, self.log_checking_point, self.error_key_words, 'usb', 'messages') def check_usb_stability(self): """ Check if no disruptive test done, USB device should not go offline. """ if self.current_test in ['gpio_test', 'reboot_test', 'resetusb_test']: return True, None return cfm_helper.check_log(self.client, self.log_checking_point, self.error_key_words, 'usb_stability', 'messages') def check_process_crash(self): """ check no process crashing. """ test_result, self.cdlines = cfm_helper.check_process_crash(self.client, self.cdlines) return test_result, str(self.cdlines) #TODO(mzhuo): Adding verififaction to check whether there is kernel panic def check_kernel_panic(self): """ Check no kernel panic reported. """ return True, None def check_chrome_restarted(self): """ Check whether chrome is killed and restarted. """ if self.chrome_file == cfm_helper.check_chrome_logfile(self.client): return True, None else: self.chrome_file = cfm_helper.check_chrome_logfile(self.client) return False, 'Chrome is restarted unexpectly.' def check_cfm_rebooted(self): """ Check whether CfM is rebooted. """ logging.info('Last reboot: %s', self.reboot_log) if self.reboot_log == cfm_helper.check_last_reboot(self.client): return True, None else: self.reboot_log = cfm_helper.check_last_reboot(self.client) return False, 'CfM is rebooted unexpectly.' def initialize_action_check_config(self, action_config, verification_config, fixedmode): """ Initialize action list based on control file. @param action_config: dict defines the number of test should be done for each test @param fixedmode: if True all tests are executed in fixed order; if False all tests are executed in random order. """ self.action_config = [] if action_config['meeting_test'] == 1: self.action_config = ['meeting_test'] if not self.camera: action_config['mute_unmute_camera_test'] = 0 verification_config['check_v4l2_interface'] = False verification_config['check_video_stream'] = False verification_config['check_video_errorlog'] = False if not self.speaker: if not cfm_helper.check_is_platform(self.client, 'buddy'): action_config['mute_unmute_mic_test'] = 0 action_config['speaker_volume_test'] = 0 verification_config['check_audio_card'] = False verification_config['check_cras_speaker'] = False verification_config['check_cras_mic'] = False verification_config['check_cras_pspeaker'] = False verification_config['check_cras_pmic'] = False verification_config['check_audio_stream'] = False verification_config['check_audio_errorlog'] = False verification_config['check_cras_speaker_vol'] = False verification_config['check_cras_mic_mute'] = False not_in_meeting_action = ['meeting_test', 'gpio_test', 'reboot_test'] if fixedmode: for action, nof_times in action_config.iteritems(): if not action in not_in_meeting_action: self.action_config.extend(nof_times * [action]) else: for action, nof_times in action_config.iteritems(): if not action in not_in_meeting_action and nof_times > 0: dup_test = max(1, random.randrange(0, nof_times)) for _ in range(dup_test): self.action_config.insert(max(1, random.randrange(-1, len(self.action_config))), action) if action_config['meeting_test'] == 1: self.action_config.append('meeting_test') for action, nof_times in action_config.iteritems(): if action == 'meeting_test': continue if action in not_in_meeting_action and nof_times > 0: dup_test = max(1, random.randrange(0, nof_times)) for _ in range(dup_test): self.action_config.insert(max(0, random.randrange(-1, len(self.action_config))), action) logging.info('Test list = %s', self.action_config) self.verification_config = [v for v in verification_config.keys() if verification_config[v]] logging.info('Verification list = %s', self.verification_config) def initialize_test(self, test_config, action_config, verification_config, error_key_words, test_flow_control): """ Initialize the list of tests and verifications; and polulate data needed for test: USB devices: which can be retrieved from control file or automatically detected by script; Test loop, meeting mode, meeting code, test flow contro variables. """ self.gpio_pause = test_config['gpiopause'] self.reboot_timeout = test_config['reboot_timeout'] self.vol_change_step = test_config['vol_change_step'] self.vol_change_mode = test_config['vol_change_mode'] self.gpio_list = test_config['gpio_list'] self.is_meeting = test_config['is_meeting'] self.meeting_code = test_config ['meeting_code'] self.reboot_min_meets = test_config['reboot_after_min_meets'] self.gpio_min_meets = test_config['gpio_after_min_meets'] self.run_meeting_test = action_config['meeting_test'] self.random = test_flow_control['random_mode'] self.recover_on_failure = test_flow_control['recovery_on_fatal_failure'] self.error_key_words = error_key_words if test_config['puts']: self.puts = test_config['puts'].split(',') else: self.puts = None if verification_config['check_process_crash']: cfm_helper.clear_core_file(self.client) self.action_fun = { 'gpio_test': self.gpio_test, 'meeting_test': self.meeting_test, 'reboot_test': self.reboot_test, 'reset_usb_test': self.reset_usb_test, 'mute_unmute_camera_test': self.mute_unmute_camera_test, 'mute_unmute_mic_test': self.mute_unmute_mic_test, 'speaker_volume_test': self.speaker_volume_test, 'flap_monitor_test': self.flap_monitor_test } self.veri_fun = { 'check_usb_enumeration': self.check_usb_enumeration, 'check_usb_inf_init': self.check_usb_inf_init, 'check_v4l2_interface': self.check_v4l2_interface, 'check_audio_card': self.check_audio_card, 'check_cras_speaker': self.check_cras_speaker, 'check_cras_mic': self.check_cras_mic, 'check_cras_pspeaker': self.check_cras_pspeaker, 'check_cras_pmic': self.check_cras_pmic, 'check_cras_speaker_vol': self.check_cras_speaker_vol, 'check_cras_mic_mute': self.check_cras_mic_mute, 'check_prefer_camera': self.check_prefer_camera, 'check_camera_mute': self.check_camera_mute, 'check_audio_stream': self.check_audio_stream, 'check_video_stream': self.check_video_stream, 'check_hotrod_speaker': self.check_hotrod_speaker, 'check_hotrod_mic': self.check_hotrod_mic, 'check_hotrod_camera': self.check_hotrod_camera, 'check_hotrod_pspeaker': self.check_hotrod_pspeaker, 'check_hotrod_pmic': self.check_hotrod_pmic, 'check_hotrod_pcamera': self.check_hotrod_pcamera, 'check_hotrod_speaker_vol': self.check_hotrod_speaker_vol, 'check_hotrod_mic_state': self.check_hotrod_mic_state, 'check_hotrod_camera_state': self.check_hotrod_camera_state, 'check_usb_errorlog': self.check_usb_errorlog, 'check_kernel_errorlog': self.check_kernel_errorlog, 'check_video_errorlog': self.check_video_errorlog, 'check_audio_errorlog': self.check_audio_errorlog, 'check_chrome_errorlog': self.check_chrome_errorlog, 'check_atrus_errorlog': self.check_atrus_errorlog, 'check_usb_stability': self.check_usb_stability, 'check_process_crash': self.check_process_crash, 'check_kernel_panic': self.check_kernel_panic, 'check_cfm_rebooted':self.check_cfm_rebooted, 'check_chrome_restarted':self.check_chrome_restarted } self.usb_data = [] self.speaker = None self.speaker_volume = None self.camera = None self.name_speaker = None self.mimo_sis = None self.mimo_display = None self.is_in_meet = False self.is_camera_muted = True self.is_mic_muted = False self.meets_last_reboot = 0 self.meets_last_gpio = 0 self.meet_no = 0 self.reboot_no = 0 self.gpio_no = 0 self.cdlines = CORE_DIR_LINES usb_data = cfm_helper.retrieve_usb_devices(self.client) if not usb_data: logging.info('\n\nEnterprise_CFM_Test_Failed.') raise error.TestFail('Fails to find any usb devices on CfM.') peripherals = cfm_helper.extract_peripherals_for_cfm(usb_data) if not peripherals: logging.info('\n\nEnterprise_CFM_Test_Failed.') raise error.TestFail('Fails to find any periphereal on CfM.') if not self.puts: self.puts = peripherals.keys() else: if [put for put in self.puts if not put in peripherals.keys()]: logging.info('Fails to find target device %s', put) logging.info('\nEnterprise_CFM_Test_Failed.') raise error.TestFail('Fails to find device') for _put in self.puts: if _put in get_usb_devices.CAMERA_MAP.keys(): self.camera = _put if _put in get_usb_devices.SPEAKER_MAP.keys(): self.speaker = _put if self.camera: self.name_camera = get_usb_devices.get_device_prod(self.camera) logging.info('Camera under test: %s %s', self.camera, self.name_camera) if self.speaker: self.name_speaker = get_usb_devices.get_device_prod(self.speaker) logging.info('Speaker under test: %s %s', self.speaker, self.name_speaker) if not test_flow_control['skip_cfm_check']: if cfm_helper.check_is_platform(self.client, 'guado'): if not cfm_helper.check_peripherals_for_cfm(peripherals): logging.info('Sanity Check on CfM fails.') logging.info('\n\nEnterprise_CFM_Test_Failed.') raise error.TestFail('Sanity Check on CfM fails.') self.ip_addr = cfm_helper.get_mgmt_ipv4(self.client) logging.info('CfM %s passes sanity check, will start test.', self.ip_addr) self.initialize_action_check_config(action_config, verification_config, True) if list(set(self.verification_config) & set(LOG_CHECK_LIST)): self.log_checking_point = cfm_helper.find_last_log(self.client, self.speaker) self.chrome_file = cfm_helper.check_chrome_logfile(self.client) self.reboot_log = cfm_helper.check_last_reboot(self.client) def recovery_routine(self, failure_msg): """ Telemetry api often returns API timeout, or javascript timeout due to various reasons. To workout the problem before the fix is ready script checks if errmsg from test failure or verification failure has keywords defined in FAILURE_REASON['telemetry']. If yes, script checks whether CfM is rebooted or Chrome browser is rebooted unexpectly. If no, script restarts chrome. If yes no chrome restart is done to preserve valid failure state. @param failure_msg: failure message returned from test failure or verification failure @return True if recovery is successfully done, False, if recovery is not done, or fails. """ loop_result = False to_be_recovered = False if not self.check_chrome_restarted(): self.chrome_file = cfm_helper.check_chrome_logfile(self.client) return False if not self.check_cfm_rebooted(): self.reboot_log = cfm_helper.check_last_reboot(self.client) return False for _err in FAILURE_REASON['telemetry']: if _err in failure_msg: to_be_recovered = True break if to_be_recovered: logging.info('Restart Chrome to recover......') result, emsg = self.restart_chrome_and_meeting(True) if result: loop_result = True if list(set(self.verification_config) & set(LOG_CHECK_LIST)): self.log_checking_point = cfm_helper.find_last_log( self.client, self.speaker) return loop_result def process_test_result(self, loop_result, loop_no, test_no, failed_tests, failed_verifications, failed_tests_loop, failed_verifications_loop, test_flow_control, test_config, finished_tests_verifications, test_done): """ Proceess test result data, and print out test report. @params loop_result: True when all tests and verifications pass, False if any test or verification fails. @param loop_no: sequence number of the loop. @param test_no: sequence number of the test. @param failed_tests: failed tests. @param failed_verifications: failed verifications. @param failed_tests_loop: failed tests in the loop. @param failed_verifications_loop: failed verifications in the loop. @param test_flow_control: variable of flow control defined in control file @param test_config: variable of test config defined in control file @param finished_tests_verifications: data to keep track number of tests and verifications performed. @param test_done: True if all loops are done; False otherwose. """ if 'reboot_test' in finished_tests_verifications.keys(): finished_tests_verifications['reboot_test'] = self.reboot_no if not loop_result and not test_done: logging.info('\n\nVerification_or_Test_Fail on %s for loop NO:' ' %d, Test: %d', self.ip_addr, loop_no, test_no) if failed_tests_loop: logging.info('----- Failed_Tests: %s', failed_tests_loop) if failed_verifications_loop: logging.info('----- Failed_Verifications: %s', failed_verifications_loop) if test_flow_control['report']: logging.info('\n\n\n----------------Summary---------------') logging.info('---Loop %d, Test: %d, Meet: %d, Reboot: %d, Gpio: %s', loop_no, test_no, self.meet_no, self.reboot_no, self.gpio_no) for testname, counter in failed_tests.iteritems(): logging.info('----Test: %s, Failed times: %d, Total Run: %d', testname, counter, finished_tests_verifications[testname]) for veriname, counter in failed_verifications.iteritems(): logging.info('----Verification: %s, Failed times: %d,' 'Total Run: %d', veriname, counter, finished_tests_verifications[veriname]) if self.random: time.sleep(random.randrange(0, test_config['loop_timeout'])) else: time.sleep(test_config['loop_timeout']) if not test_done: if list(set(self.verification_config) & set(LOG_CHECK_LIST)): self.log_checking_point = cfm_helper.find_last_log(self.client, self.speaker) def run_once(self, host, run_test_only, test_config, action_config, verification_config, error_key_words, test_flow_control): """Runs the test.""" logging.info('Start_Test_Script:Enterprise_CFM_Test') self.client = host if test_flow_control['reboot_before_start']: try: self.client.reboot() except Exception as e: errmsg = ('Reboot test fails for %s.' '% self.ip_addr') logging.exception(errmsg) raise error.TestFail(errmsg) if action_config['meeting_test'] > 0: result, errmsg = meet_helper.restart_chrome(self.cfm_facade, test_config['is_meeting'], True) if not result: logging.info('Restart chrome fails, msg = %s', errmsg) raise error.TestFail(errmsg) self.initialize_test(test_config, action_config, verification_config, error_key_words, test_flow_control) test_no = 0 failed_tests = {} failed_verifications = {} finished_tests_verifications = {} test_failure_reason = [] verification_failure_reason = [] for loop_no in xrange(1, test_config['repeat'] + 1): logging.info('=============%s:Test_Loop_No:%d=============', self.ip_addr, loop_no) logging.info('Action list: %s', self.action_config) failed_tests_loop = [] failed_verifications_loop = [] for test in self.action_config: loop_result = True if not test in finished_tests_verifications.keys(): finished_tests_verifications[test] = 1 else: finished_tests_verifications[test] += 1 self.current_test = test logging.info('\nStart_test:%s', test) test_result, test_msg = self.action_fun[test]() test_no += 1 if test_result: logging.info('Test_Result:%s:SUCCESS', test) else: logging.info('Test_Result:%s:FAILURE:%s', test, test_msg) test_failure_reason.append(test_msg) failed_tests_loop.append(test) loop_result = False if not test in failed_tests.keys(): failed_tests[test] = 1 else: failed_tests[test] += 1 logging.info('\n%s:Test_failure:%s:%s', self.ip_addr, test, test_msg) if self.recover_on_failure or self.meets_last_reboot < 1: loop_result = self.recovery_routine(test_msg) if self.random: time.sleep(random.randrange(test_config['min_timeout'], test_config['action_timeout'])) else: time.sleep(test_config['min_timeout']) for verification in self.verification_config: if not verification in finished_tests_verifications.keys(): finished_tests_verifications[verification] = 1 else: finished_tests_verifications[verification] += 1 logging.info('\nStart_verification:%s', verification) veri_result, veri_msg = self.veri_fun[verification]() if veri_result: logging.info('Verification_Result:%s:SUCCESS', verification) else: logging.info('Verification_Result:%s:FAILURE:%s', verification, veri_msg) verification_failure_reason.append(veri_msg) failed_verifications_loop.append(verification) if not verification in failed_verifications.keys(): failed_verifications[verification] = 1 else: failed_verifications[verification] += 1 logging.info('%s:Verification_fail:%s:%s', self.ip_addr, verification, veri_msg) loop_result = False if self.recover_on_failure: loop_result = self.recovery_routine(veri_msg) self.process_test_result(loop_result, loop_no, test_no, failed_tests, failed_verifications, failed_tests_loop, failed_verifications_loop, test_flow_control, test_config, finished_tests_verifications, False) if not loop_result: if test_flow_control['abort_on_failure']: logging.info('Enterprise_CFM_Test_Failed.') time.sleep(test_config['debug_timeout']) logging.info('Enterprise_CFM_Test_Finished.') raise error.TestFail( 'Test_or_Verification_fails after {}.'.format(test)) else: logging.info('Enterprise_CFM_Test_Failure_Detected.') if self.random: self.initialize_action_check_config(action_config, verification_config, False) logging.info('Enterprise_CFM_Test_Finished.') self.process_test_result(loop_result, loop_no, test_no, failed_tests, failed_verifications, failed_tests_loop, failed_verifications_loop, test_flow_control, test_config, finished_tests_verifications, True) if test_failure_reason: logging.debug('Test failure reason %s', test_failure_reason) if verification_failure_reason: logging.debug('Verification failure reason %s', verification_failure_reason)