Home | History | Annotate | Download | only in enterprise_CFM_AutoZoomSanity 
      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 
      5 import glob
      6 import logging
      7 import os
      8 import time
      9 
     10 from autotest_lib.client.common_lib import error
     11 from autotest_lib.client.common_lib.cros import perf_stat_lib
     12 from autotest_lib.server.cros.cfm import cfm_base_test
     13 from 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 _USB_DIR = '/sys/bus/usb/devices'
     20 
     21 
     22 class enterprise_CFM_AutoZoomSanity(cfm_base_test.CfmBaseTest):
     23     """Auto Zoom Sanity test."""
     24     version = 1
     25 
     26     def get_data_from_jmifile(self, data_type, jmidata):
     27         """
     28         Gets data from jmidata log for given data type.
     29 
     30         @param data_type: Type of data to be retrieved from jmi data log.
     31         @param jmidata: Raw jmi data log to parse.
     32         @returns Data for given data type from jmidata log.
     33         """
     34         return cfm_jmidata_log_collector.GetDataFromLogs(
     35                 self, data_type, jmidata)
     36 
     37 
     38     def get_file_to_parse(self):
     39         """
     40         Copy jmi logs from client to test's results directory.
     41 
     42         @returns The newest jmi log file.
     43         """
     44         self._host.get_file(_JMI_SOURCE_DIR, self.resultsdir)
     45         source_jmi_files = self.resultsdir + '/0*'
     46         if not source_jmi_files:
     47             raise error.TestNAError('JMI data file not found.')
     48         newest_file = max(glob.iglob(source_jmi_files), key=os.path.getctime)
     49         return newest_file
     50 
     51 
     52     def verify_cfm_sent_resolution(self):
     53         """Check / verify CFM sent video resolution data from JMI logs."""
     54         jmi_file = self.get_file_to_parse()
     55         jmifile_to_parse = open(jmi_file, 'r')
     56         jmidata = jmifile_to_parse.read()
     57 
     58         cfm_sent_res_list = self.get_data_from_jmifile(
     59                 'video_sent_frame_height', jmidata)
     60         percentile_95 = perf_stat_lib.get_kth_percentile(
     61                 cfm_sent_res_list, 0.95)
     62 
     63         self.output_perf_value(description='video_sent_frame_height',
     64                                value=cfm_sent_res_list,
     65                                units='resolution',
     66                                higher_is_better=True)
     67         self.output_perf_value(description='95th percentile res sent',
     68                                value=percentile_95,
     69                                units='resolution',
     70                                higher_is_better=True)
     71 
     72         # TODO(dkaeding): Add logic to examine the cfm sent resolution and
     73         # take appropriate action.
     74         logging.info('95th percentile of outgoing video resolution: %s',
     75                      percentile_95)
     76 
     77 
     78     def check_verify_callgrok_logs(self):
     79         """Verify needed information in callgrok logs."""
     80         # TODO(dkaeding): Implement this method.
     81         return NotImplemented
     82 
     83 
     84     def get_usb_device_dirs(self):
     85         """Gets usb device dirs from _USB_DIR path.
     86 
     87         @returns list with number of device dirs else None
     88         """
     89         usb_dir_list = list()
     90         cmd = 'ls %s' % _USB_DIR
     91         cmd_output = self._host.run(cmd).stdout.strip().split('\n')
     92         for d in cmd_output:
     93             usb_dir_list.append(os.path.join(_USB_DIR, d))
     94         return usb_dir_list
     95 
     96 
     97     def file_exists_on_host(self, path):
     98         """
     99         Checks if file exists on host.
    100 
    101         @param path: File path
    102         @returns True or False
    103         """
    104         return self._host.run('ls %s' % path,
    105                               ignore_status=True).exit_status == 0
    106 
    107 
    108     def check_peripherals(self, peripheral_dict):
    109         """
    110         Check and verify correct peripherals are attached.
    111 
    112         @param peripheral_dict: dict of peripherals that should be connected
    113         """
    114         usb_dir_list = self.get_usb_device_dirs()
    115         peripherals_found = list()
    116         for d_path in usb_dir_list:
    117             file_name = os.path.join(d_path, 'product')
    118             if self.file_exists_on_host(file_name):
    119                 peripherals_found.append(self._host.run(
    120                         'cat %s' % file_name).stdout.strip())
    121 
    122         logging.info('Attached peripherals: %s', peripherals_found)
    123 
    124         for peripheral in peripheral_dict:
    125             if peripheral not in peripherals_found:
    126                 raise error.TestFail('%s not found.' % peripheral)
    127 
    128 
    129     def run_once(self, session_length, peripheral_dict):
    130         """Runs the sanity test."""
    131         self.cfm_facade.wait_for_meetings_telemetry_commands()
    132         self.check_peripherals(peripheral_dict)
    133         self.cfm_facade.start_meeting_session()
    134         time.sleep(session_length)
    135         self.cfm_facade.end_meeting_session()
    136         self.verify_cfm_sent_resolution()
    137         self.check_verify_callgrok_logs()
    138