Home | History | Annotate | Download | only in enterprise_CFM_USBPeripheralRebootStress 
      1 # Copyright 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 logging, time, random
      6 
      7 from autotest_lib.client.common_lib import error
      8 from autotest_lib.server.cros.cfm import cfm_base_test
      9 
     10 CMD = "usb-devices | grep ^P:"
     11 FAILED_TEST_LIST = list()
     12 IDLE_TIME = 30
     13 LONG_TIMEOUT = 20
     14 MEETS_BETWEEN_REBOOT = 5
     15 
     16 
     17 class enterprise_CFM_USBPeripheralRebootStress(cfm_base_test.CfmBaseTest):
     18     """Stress test of USB devices in CfM mode by warm rebooting CfM
     19     multiple times with joining/leaving meetings ."""
     20     version = 1
     21 
     22 
     23     def _peripherals_sanity_test(self):
     24         """Checks for connected peripherals."""
     25         if not self.cfm_facade.get_mic_devices():
     26              FAILED_TEST_LIST.append('No mic detected')
     27         if not self.cfm_facade.get_speaker_devices():
     28              FAILED_TEST_LIST.append('No speaker detected')
     29         if not self.cfm_facade.get_camera_devices():
     30              FAILED_TEST_LIST.append('No camera detected')
     31         if not self.cfm_facade.get_preferred_mic():
     32              FAILED_TEST_LIST.append('No preferred mic')
     33         if not self.cfm_facade.get_preferred_speaker():
     34              FAILED_TEST_LIST.append('No preferred speaker')
     35         if not self.cfm_facade.get_preferred_camera():
     36              FAILED_TEST_LIST.append('No preferred camera')
     37 
     38 
     39     def _run_hangout_session(self, hangout, original_list):
     40         """Start a hangout session and end the session after random time.
     41 
     42         @raises error.TestFail if any of the checks fail.
     43         """
     44         hangout_name = hangout
     45         logging.info('Session name: %s', hangout_name)
     46         logging.info('Now joining session.........')
     47         self.cfm_facade.start_new_hangout_session(hangout_name)
     48         # Minimum sleep time of 3 secs is needed to allow Hangout app to
     49         # stabilize after starting a session before getting peripheral info.
     50         time.sleep(random.randrange(3, LONG_TIMEOUT))
     51         if not self._compare_cmd_output(original_list):
     52             raise error.TestFail(
     53                 'After joining meeting list of USB devices is not the same.')
     54         self._peripherals_sanity_test()
     55         self.cfm_facade.end_hangout_session()
     56         logging.info('Stopping session................')
     57 
     58 
     59     def _compare_cmd_output(self, original_list):
     60         """Compare output of linux cmd."""
     61         only_in_original = []
     62         only_in_new = []
     63         new_output = self._host.run(CMD).stdout.rstrip()
     64         new_list= new_output.splitlines()
     65         if not set(new_list) == set(original_list):
     66             only_in_original = list(set(original_list) - set(new_list))
     67             only_in_new = list(set(new_list) - set(original_list))
     68             if only_in_original:
     69                 logging.info('These are devices not in the new list')
     70                 for _device in only_in_original:
     71                     logging.info('    %s', _device)
     72             if only_in_new:
     73                 logging.info('These are devices not in the original list')
     74                 for _device in only_in_new:
     75                     logging.info('    %s', _device)
     76         return set(new_list) == set(original_list)
     77 
     78 
     79     def run_once(self, hangout, reboot_cycles):
     80         """
     81         Main function to run autotest.
     82 
     83         @hangout: Name of meeting that DUT will join/leave.
     84         @param reboot_cycles: Number of times CfM reboots during the test.
     85         """
     86 
     87         usb_original_output = self._host.run(CMD).stdout.rstrip()
     88         logging.info('The initial usb devices:\n %s', usb_original_output)
     89         usb_original_list = usb_original_output.splitlines()
     90 
     91         logging.info("Performing in total %d reboot cycles...", reboot_cycles)
     92         for cycle in range(reboot_cycles):
     93             logging.info("Performing reboot cycle #%d.", cycle)
     94             self._host.reboot()
     95             time.sleep(random.randrange(1, IDLE_TIME))
     96             self.cfm_facade.restart_chrome_for_cfm()
     97             self.cfm_facade.wait_for_hangouts_telemetry_commands()
     98             if not self._compare_cmd_output(usb_original_list):
     99                 raise error.TestFail(
    100                     "After reboot list of USB devices is not the same.")
    101 
    102             for attempt in range(random.randrange(1, MEETS_BETWEEN_REBOOT)):
    103                 logging.info('Join meeting for-loop: #%d', attempt)
    104                 self._run_hangout_session(hangout, usb_original_list)
    105                 if FAILED_TEST_LIST:
    106                      raise error.TestFail(
    107                          'Test failed because of following reasons: %s'
    108                          % ', '.join(map(str, FAILED_TEST_LIST)))
    109 
    110