1# Lint as: python2, python3 2# Copyright (c) 2012 The Chromium OS Authors. All rights reserved. 3# Use of this source code is governed by a BSD-style license that can be 4# found in the LICENSE file. 5 6from __future__ import absolute_import 7from __future__ import division 8from __future__ import print_function 9 10import dbus 11import logging 12import random 13import time 14 15from six.moves import range 16 17from autotest_lib.client.bin import test, utils 18from autotest_lib.client.common_lib import error 19from autotest_lib.client.cros.cellular import mm1_constants 20from autotest_lib.client.cros.networking import cellular_proxy 21from autotest_lib.client.cros.networking import shill_context 22from autotest_lib.client.cros.networking import shill_proxy 23from autotest_lib.client.cros.networking import mm1_proxy 24 25SERVICE_DISABLE_TIMEOUT = 60 26SERVICE_ENABLE_TIMEOUT = 60 27 28 29class cellular_SafetyDance(test.test): 30 """ 31 Stress tests all connection manager 3G operations. 32 33 This test runs a long series of 3G operations in pseudorandom order. All of 34 these 3G operations must return a convincing result (EINPROGRESS or no 35 error). 36 37 """ 38 version = 1 39 40 def _filterexns(self, fn): 41 v = None 42 try: 43 v = fn() 44 except dbus.exceptions.DBusException as error: 45 if error.get_dbus_name() in self.okerrors: 46 return v, error.get_dbus_message() 47 else: 48 raise error 49 return v, '' 50 51 def _ensure_disabled(self): 52 """ 53 Ensure modem is disabled. 54 55 Raises: 56 error.TestFail if the states are not consistent. 57 """ 58 59 # b/188448918 : QC modems indicate that they are disabled even if they 60 # are enabled. There is no way to know when the disable completed until 61 # b/188448918 is fixed, and MM receives power state indications from the 62 # modem. The sleep can be removed once b/188448918 is fixed. 63 time.sleep(2) 64 65 utils.poll_for_condition( 66 lambda: not self.test_env.modem.IsEnabled(), 67 error.TestFail('Modem failed to enter state Disabled.')) 68 utils.poll_for_condition( 69 lambda: not self.test_env.shill.find_cellular_service_object(), 70 error.TestFail('Service should not be available.'), 71 timeout=SERVICE_DISABLE_TIMEOUT) 72 73 def _enable(self): 74 logging.info('Enable') 75 self._filterexns(lambda: 76 self.test_env.shill.manager.EnableTechnology('cellular')) 77 78 def _disable(self): 79 logging.info('Disable') 80 self._filterexns(lambda: 81 self.test_env.shill.manager.DisableTechnology('cellular')) 82 self._ensure_disabled() 83 84 def _ignoring(self, reason): 85 if ('AlreadyConnected' in reason or 86 'Not connected' in reason or 87 'Bearer already being connected' in reason or 88 'Bearer already being disconnected' in reason or 89 'InProgress' in reason): 90 return True 91 if 'NotSupported' in reason: 92 # We should only ignore this error if we've previously disabled 93 # cellular technology and the service subsequently disappeared 94 # when we tried to connect again. 95 return not self.test_env.shill.find_cellular_service_object() 96 return False 97 98 def _connect(self): 99 logging.info('Connect') 100 try: 101 service = self.test_env.shill.wait_for_cellular_service_object( 102 timeout_seconds=5) 103 except shill_proxy.ShillProxyError: 104 return 105 106 mm_proxy = mm1_proxy.ModemManager1Proxy.get_proxy() 107 if not mm_proxy: 108 raise error.TestFail('Could not get mm_proxy') 109 modem_proxy = mm_proxy.get_modem() 110 modem_proxy.wait_for_states([ 111 mm1_constants.MM_MODEM_STATE_REGISTERED, 112 mm1_constants.MM_MODEM_STATE_CONNECTED 113 ]) 114 115 success, reason = self._filterexns(lambda: 116 self.test_env.shill.connect_service_synchronous( 117 service=service, 118 timeout_seconds= 119 cellular_proxy.CellularProxy.SERVICE_CONNECT_TIMEOUT)) 120 if not success and not self._ignoring(reason): 121 raise error.TestFail('Could not connect: %s' % reason) 122 123 def _disconnect(self): 124 logging.info('Disconnect') 125 try: 126 service = self.test_env.shill.wait_for_cellular_service_object( 127 timeout_seconds=5) 128 except shill_proxy.ShillProxyError: 129 return 130 131 success, reason = self._filterexns(lambda: 132 self.test_env.shill.disconnect_service_synchronous( 133 service=service, 134 timeout_seconds= 135 cellular_proxy.CellularProxy. 136 SERVICE_DISCONNECT_TIMEOUT)) 137 if not success and not self._ignoring(reason): 138 raise error.TestFail('Could not disconnect: %s' % reason) 139 140 def _op(self): 141 n = random.randint(0, len(self.ops) - 1) 142 self.ops[n]() 143 time.sleep(random.randint(5, 20) / 10.0) 144 145 def _run_once_internal(self, ops=30, seed=None): 146 if not seed: 147 seed = int(time.time()) 148 self.okerrors = [ 149 'org.chromium.flimflam.Error.InProgress', 150 'org.chromium.flimflam.Error.AlreadyConnected', 151 'org.chromium.flimflam.Error.AlreadyEnabled', 152 'org.chromium.flimflam.Error.AlreadyDisabled' 153 ] 154 self.ops = [ self._enable, 155 self._disable, 156 self._connect, 157 self._disconnect ] 158 self.device = self.test_env.shill.find_cellular_device_object() 159 if not self.device: 160 raise error.TestFail('Could not find cellular device.') 161 162 # Start in a disabled state. 163 self._disable() 164 logging.info('Seed: %d', seed) 165 random.seed(seed) 166 for _ in range(ops): 167 self._op() 168 169 def run_once(self, test_env, ops=30, seed=None): 170 self.test_env = test_env 171 with test_env, shill_context.ServiceAutoConnectContext( 172 test_env.shill.wait_for_cellular_service_object, False): 173 self._run_once_internal(ops, seed) 174 175 # Enable device to restore autoconnect settings. 176 self._enable() 177 test_env.shill.wait_for_cellular_service_object() 178