• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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_meetings_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