1 # Copyright (c) 2012 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, os, time 6 7 from autotest_lib.client.bin import test, utils 8 from autotest_lib.client.common_lib import error 9 from autotest_lib.client.cros.cellular import mm 10 from autotest_lib.client.cros.networking import shill_proxy 11 12 13 class ResetAuthorizedContext(object): 14 def __init__(self, test): 15 self.test = test 16 17 def __enter__(self): 18 pass 19 20 def __exit__(self, exception, value, traceback): 21 if exception: 22 self.test.SetAuthorized(1) 23 return False 24 25 26 class network_3GRebootStress(test.test): 27 version = 1 28 29 def IsCromo(self, modem_manager): 30 path = modem_manager.path 31 return path.startswith('/org/chromium') 32 33 def CountModems(self): 34 count = len(mm.EnumerateDevices('')) 35 logging.debug('Modem count is %d' % count) 36 return count 37 38 def EnsureModemAbsent(self): 39 utils.poll_for_condition( 40 lambda: self.CountModems() == 0, 41 error.TestFail('Modem failed to disappear'), 42 timeout=shill_proxy.ShillProxy.DEVICE_ENABLE_DISABLE_TIMEOUT) 43 44 def EnsureModemPresent(self): 45 utils.poll_for_condition( 46 lambda: self.CountModems() == 1, 47 error.TestFail('Modem failed to reappear'), 48 timeout=shill_proxy.ShillProxy.DEVICE_ENABLE_DISABLE_TIMEOUT) 49 50 def FindUsbDevicePath(self, modem_manager, modem_path): 51 logging.info('Modem path: %s' % modem_path) 52 53 modem_obj = modem_manager.GetModem(modem_path) 54 props = modem_obj.GetModemProperties() 55 net_device = props['Device'] 56 logging.info('Network device: %s' % net_device) 57 if self.IsCromo(modem_manager): 58 usb_interface_path, _ = os.path.realpath( 59 '/sys/class/net/%s/device' % net_device).rsplit('/', 1) 60 else: 61 usb_interface_path = os.path.realpath(net_device) 62 self.usb_device_path = usb_interface_path 63 logging.info('USB device path: %s' % self.usb_device_path) 64 65 def SetAuthorized(self, authorized): 66 logging.debug('Setting authorized to %d' % authorized) 67 authorized_path = '%s/authorized' % self.usb_device_path 68 authorized_file = open(authorized_path, 'w') 69 authorized_file.write('%d' % authorized) 70 authorized_file.close() 71 72 def ShouldContinue(self): 73 should_continue = True 74 message = 'Starting loop %d' % (self.loops_done + 1) 75 76 if self.max_loops != None: 77 if self.loops_done >= self.max_loops: 78 return False 79 loops_left = self.max_loops - self.loops_done 80 message += '; %d loops left' % loops_left 81 82 if self.max_seconds != None: 83 seconds_done = time.time() - self.start_time 84 if seconds_done >= self.max_seconds: 85 return False 86 seconds_left = self.max_seconds - seconds_done 87 message += '; %d seconds left' % seconds_left 88 89 message += '.' 90 logging.info(message) 91 92 return True 93 94 95 # It takes a Gobi about 1.5 seconds to cycle away and back, on average. 96 # Assume 2, to have a margin of error. (If we run out of time, it's 97 # fine, but we'd like to run the same number of loops every time.) 98 def run_once(self, test_env, max_loops=300, max_seconds=600): 99 with test_env: 100 self.FindUsbDevicePath(test_env.modem_manager, test_env.modem_path) 101 102 self.max_loops = max_loops 103 self.max_seconds = max_seconds 104 105 self.loops_done = 0 106 self.start_time = time.time() 107 108 # Use a context to ensure that we don't leave the modem 109 # unauthorized if we die with an exception. 110 with ResetAuthorizedContext(self): 111 while self.ShouldContinue(): 112 self.SetAuthorized(0) 113 self.EnsureModemAbsent() 114 self.SetAuthorized(1) 115 self.EnsureModemPresent() 116 self.loops_done += 1 117 118 if self.loops_done < self.max_loops: 119 logging.warning('Only got %d loops done in %d seconds. :(', 120 self.loops_done, max_seconds) 121
