1# Copyright (c) 2017 The Chromium OS Authors. All rights reserved. 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4 5import glob 6import logging 7import os 8import time 9 10from autotest_lib.client.common_lib import error 11from autotest_lib.client.common_lib.cros import perf_stat_lib 12from autotest_lib.server.cros.cfm import cfm_base_test 13from autotest_lib.server.cros import cfm_jmidata_log_collector 14 15_BASE_DIR = '/home/chronos/user/Storage/ext/' 16_EXT_ID = 'ikfcpmgefdpheiiomgmhlmmkihchmdlj' 17_JMI_DIR = '/0*/File\ System/000/t/00/*' 18_JMI_SOURCE_DIR = _BASE_DIR + _EXT_ID + _JMI_DIR 19_PA_LOGS_PATTERN = _BASE_DIR + _EXT_ID + '/def/File\ System/primary/p/00/0*' 20 21_USB_DIR = '/sys/bus/usb/devices' 22_AUTOZOOM_IS_RUNNING_STRING = 'AutoZoom running successfully.' 23 24_LONG_TIMEOUT = 15 25 26class enterprise_CFM_AutoZoomSanity(cfm_base_test.CfmBaseTest): 27 """Auto Zoom Sanity test.""" 28 version = 1 29 30 def get_data_from_jmifile(self, data_type, jmidata): 31 """ 32 Gets data from jmidata log for given data type. 33 34 @param data_type: Type of data to be retrieved from jmi data log. 35 @param jmidata: Raw jmi data log to parse. 36 @returns Data for given data type from jmidata log. 37 """ 38 return cfm_jmidata_log_collector.GetDataFromLogs( 39 self, data_type, jmidata) 40 41 42 def get_file_to_parse(self): 43 """ 44 Copy jmi logs from client to test's results directory. 45 46 @returns The newest jmi log file. 47 """ 48 self._host.get_file(_JMI_SOURCE_DIR, self.resultsdir) 49 source_jmi_files = self.resultsdir + '/0*' 50 if not source_jmi_files: 51 raise error.TestNAError('JMI data file not found.') 52 newest_file = max(glob.iglob(source_jmi_files), key=os.path.getctime) 53 return newest_file 54 55 56 def verify_cfm_sent_resolution(self): 57 """Check / verify CFM sent video resolution data from JMI logs.""" 58 jmi_file = self.get_file_to_parse() 59 jmifile_to_parse = open(jmi_file, 'r') 60 jmidata = jmifile_to_parse.read() 61 62 cfm_sent_res_list = self.get_data_from_jmifile( 63 'video_sent_frame_height', jmidata) 64 percentile_95 = perf_stat_lib.get_kth_percentile( 65 cfm_sent_res_list, 0.95) 66 67 self.output_perf_value(description='video_sent_frame_height', 68 value=cfm_sent_res_list, 69 units='resolution', 70 higher_is_better=True) 71 self.output_perf_value(description='95th percentile res sent', 72 value=percentile_95, 73 units='resolution', 74 higher_is_better=True) 75 76 # TODO(dkaeding): Add logic to examine the cfm sent resolution and 77 # take appropriate action. 78 logging.info('95th percentile of outgoing video resolution: %s', 79 percentile_95) 80 81 82 def check_verify_callgrok_logs(self): 83 """Verify needed information in callgrok logs.""" 84 # TODO(dkaeding): Implement this method. 85 return NotImplemented 86 87 88 def verify_autozoom_running_in_packaged_app_logs(self): 89 """Checks logs from the device to verify that AutoZoom is running.""" 90 self.save_all_packaged_app_logs() 91 pa_log_files = glob.glob(os.path.join(self.debugdir, 92 'packaged_app_log*.txt')) 93 for log_file in pa_log_files: 94 with open(log_file, 'r') as fhandle: 95 if _AUTOZOOM_IS_RUNNING_STRING in fhandle.read(): 96 return 97 raise error.TestFail('AutoZoom not running on device.') 98 99 def get_usb_device_dirs(self): 100 """Gets usb device dirs from _USB_DIR path. 101 102 @returns list with number of device dirs else None 103 """ 104 usb_dir_list = list() 105 cmd = 'ls %s' % _USB_DIR 106 cmd_output = self._host.run(cmd).stdout.strip().split('\n') 107 for d in cmd_output: 108 usb_dir_list.append(os.path.join(_USB_DIR, d)) 109 return usb_dir_list 110 111 112 def file_exists_on_host(self, path): 113 """ 114 Checks if file exists on host. 115 116 @param path: File path 117 @returns True or False 118 """ 119 return self._host.run('ls %s' % path, 120 ignore_status=True).exit_status == 0 121 122 123 def check_peripherals(self, peripheral_dict): 124 """ 125 Check and verify correct peripherals are attached. 126 127 @param peripheral_dict: dict of peripherals that should be connected 128 """ 129 usb_dir_list = self.get_usb_device_dirs() 130 peripherals_found = list() 131 for d_path in usb_dir_list: 132 file_name = os.path.join(d_path, 'product') 133 if self.file_exists_on_host(file_name): 134 peripherals_found.append(self._host.run( 135 'cat %s' % file_name).stdout.strip()) 136 137 logging.info('Attached peripherals: %s', peripherals_found) 138 139 for peripheral in peripheral_dict: 140 if peripheral not in peripherals_found: 141 raise error.TestFail('%s not found.' % peripheral) 142 143 144 def run_once(self, session_length, peripheral_dict): 145 """Runs the sanity test.""" 146 self.cfm_facade.wait_for_telemetry_commands() 147 self.check_peripherals(peripheral_dict) 148 self.cfm_facade.start_meeting_session() 149 time.sleep(_LONG_TIMEOUT) 150 self.cfm_facade.end_meeting_session() 151 self.verify_autozoom_running_in_packaged_app_logs() 152 153