1# Copyright (c) 2014 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 contextlib 6import dbus 7import logging 8import sys 9import traceback 10 11import common 12from autotest_lib.client.bin import utils 13from autotest_lib.client.common_lib import error 14from autotest_lib.client.cros import backchannel 15from autotest_lib.client.cros import upstart 16from autotest_lib.client.cros.cellular import mm 17from autotest_lib.client.cros.cellular.pseudomodem import pseudomodem_context 18from autotest_lib.client.cros.networking import cellular_proxy 19from autotest_lib.client.cros.networking import shill_context 20from autotest_lib.client.cros.networking import shill_proxy 21 22 23class CellularTestEnvironment(object): 24 """Setup and verify cellular test environment. 25 26 This context manager configures the following: 27 - Sets up backchannel. 28 - Shuts down other devices except cellular. 29 - Shill and MM logging is enabled appropriately for cellular. 30 - Initializes members that tests should use to access test environment 31 (eg. |shill|, |modem_manager|, |modem|). 32 - modemfwd is stopped to prevent the modem from rebooting underneath 33 us. 34 35 Then it verifies the following is valid: 36 - The backchannel is using an Ethernet device. 37 - The SIM is inserted and valid. 38 - There is one and only one modem in the device. 39 - The modem is registered to the network. 40 - There is a cellular service in shill and it's not connected. 41 42 Don't use this base class directly, use the appropriate subclass. 43 44 Setup for over-the-air tests: 45 with CellularOTATestEnvironment() as test_env: 46 # Test body 47 48 Setup for pseudomodem tests: 49 with CellularPseudoMMTestEnvironment( 50 pseudomm_args=({'family': '3GPP'})) as test_env: 51 # Test body 52 53 """ 54 55 def __init__(self, use_backchannel=True, shutdown_other_devices=True, 56 modem_pattern='', skip_modem_reset=False): 57 """ 58 @param use_backchannel: Set up the backchannel that can be used to 59 communicate with the DUT. 60 @param shutdown_other_devices: If True, shutdown all devices except 61 cellular. 62 @param modem_pattern: Search string used when looking for the modem. 63 64 """ 65 # Tests should use this main loop instead of creating their own. 66 self.mainloop = dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) 67 self.bus = dbus.SystemBus(mainloop=self.mainloop) 68 69 self.shill = None 70 self.modem_manager = None 71 self.modem = None 72 self.modem_path = None 73 self._backchannel = None 74 75 self._modem_pattern = modem_pattern 76 self._skip_modem_reset = skip_modem_reset 77 78 self._nested = None 79 self._context_managers = [] 80 if use_backchannel: 81 self._backchannel = backchannel.Backchannel() 82 self._context_managers.append(self._backchannel) 83 if shutdown_other_devices: 84 self._context_managers.append( 85 shill_context.AllowedTechnologiesContext( 86 [shill_proxy.ShillProxy.TECHNOLOGY_CELLULAR])) 87 88 89 @contextlib.contextmanager 90 def _disable_shill_autoconnect(self): 91 self._enable_shill_cellular_autoconnect(False) 92 yield 93 self._enable_shill_cellular_autoconnect(True) 94 95 96 def __enter__(self): 97 try: 98 if upstart.has_service('modemfwd') and upstart.is_running('modemfwd'): 99 upstart.stop_job('modemfwd') 100 # Temporarily disable shill autoconnect to cellular service while 101 # the test environment is setup to prevent a race condition 102 # between disconnecting the modem in _verify_cellular_service() 103 # and shill autoconnect. 104 with self._disable_shill_autoconnect(): 105 self._nested = contextlib.nested(*self._context_managers) 106 self._nested.__enter__() 107 108 self._initialize_shill() 109 110 # Perform SIM verification now to ensure that we can enable the 111 # modem in _initialize_modem_components(). ModemManager does not 112 # allow enabling a modem without a SIM. 113 self._verify_sim() 114 self._initialize_modem_components() 115 116 self._setup_logging() 117 118 self._verify_backchannel() 119 self._wait_for_modem_registration() 120 self._verify_cellular_service() 121 122 return self 123 except (error.TestError, dbus.DBusException, 124 shill_proxy.ShillProxyError) as e: 125 except_type, except_value, except_traceback = sys.exc_info() 126 lines = traceback.format_exception(except_type, except_value, 127 except_traceback) 128 logging.error('Error during test initialization:\n' + 129 ''.join(lines)) 130 self.__exit__(*sys.exc_info()) 131 raise error.TestError('INIT_ERROR: %s' % str(e)) 132 except: 133 self.__exit__(*sys.exc_info()) 134 raise 135 136 137 def __exit__(self, exception, value, traceback): 138 if upstart.has_service('modemfwd'): 139 upstart.restart_job('modemfwd') 140 if self._nested: 141 return self._nested.__exit__(exception, value, traceback) 142 self.shill = None 143 self.modem_manager = None 144 self.modem = None 145 self.modem_path = None 146 147 148 def _get_shill_cellular_device_object(self): 149 return utils.poll_for_condition( 150 lambda: self.shill.find_cellular_device_object(), 151 exception=error.TestError('Cannot find cellular device in shill. ' 152 'Is the modem plugged in?'), 153 timeout=shill_proxy.ShillProxy.DEVICE_ENUMERATION_TIMEOUT) 154 155 156 def _enable_modem(self): 157 modem_device = self._get_shill_cellular_device_object() 158 try: 159 modem_device.Enable() 160 except dbus.DBusException as e: 161 if (e.get_dbus_name() != 162 shill_proxy.ShillProxy.ERROR_IN_PROGRESS): 163 raise 164 165 utils.poll_for_condition( 166 lambda: modem_device.GetProperties()['Powered'], 167 exception=error.TestError( 168 'Failed to enable modem.'), 169 timeout=shill_proxy.ShillProxy.DEVICE_ENABLE_DISABLE_TIMEOUT) 170 171 172 def _enable_shill_cellular_autoconnect(self, enable): 173 shill = cellular_proxy.CellularProxy.get_proxy(self.bus) 174 shill.manager.SetProperty( 175 shill_proxy.ShillProxy. 176 MANAGER_PROPERTY_NO_AUTOCONNECT_TECHNOLOGIES, 177 '' if enable else 'cellular') 178 179 180 def _is_unsupported_error(self, e): 181 return (e.get_dbus_name() == 182 shill_proxy.ShillProxy.ERROR_NOT_SUPPORTED or 183 (e.get_dbus_name() == 184 shill_proxy.ShillProxy.ERROR_FAILURE and 185 'operation not supported' in e.get_dbus_message())) 186 187 188 def _reset_modem(self): 189 modem_device = self._get_shill_cellular_device_object() 190 try: 191 # MBIM modems do not support being reset. 192 self.shill.reset_modem(modem_device, expect_service=False) 193 except dbus.DBusException as e: 194 if not self._is_unsupported_error(e): 195 raise 196 197 198 def _initialize_shill(self): 199 """Get access to shill.""" 200 # CellularProxy.get_proxy() checks to see if shill is running and 201 # responding to DBus requests. It returns None if that's not the case. 202 self.shill = cellular_proxy.CellularProxy.get_proxy(self.bus) 203 if self.shill is None: 204 raise error.TestError('Cannot connect to shill, is shill running?') 205 206 207 def _initialize_modem_components(self): 208 """Reset the modem and get access to modem components.""" 209 # Enable modem first so shill initializes the modemmanager proxies so 210 # we can call reset on it. 211 self._enable_modem() 212 if not self._skip_modem_reset: 213 self._reset_modem() 214 215 # PickOneModem() makes sure there's a modem manager and that there is 216 # one and only one modem. 217 self.modem_manager, self.modem_path = \ 218 mm.PickOneModem(self._modem_pattern) 219 self.modem = self.modem_manager.GetModem(self.modem_path) 220 if self.modem is None: 221 raise error.TestError('Cannot get modem object at %s.' % 222 self.modem_path) 223 224 225 def _setup_logging(self): 226 self.shill.set_logging_for_cellular_test() 227 self.modem_manager.SetDebugLogging() 228 229 230 def _verify_sim(self): 231 """Verify SIM is valid. 232 233 Make sure a SIM in inserted and that it is not locked. 234 235 @raise error.TestError if SIM does not exist or is locked. 236 237 """ 238 modem_device = self._get_shill_cellular_device_object() 239 props = modem_device.GetProperties() 240 241 # No SIM in CDMA modems. 242 family = props[ 243 cellular_proxy.CellularProxy.DEVICE_PROPERTY_TECHNOLOGY_FAMILY] 244 if (family == 245 cellular_proxy.CellularProxy. 246 DEVICE_PROPERTY_TECHNOLOGY_FAMILY_CDMA): 247 return 248 249 # Make sure there is a SIM. 250 if not props[cellular_proxy.CellularProxy.DEVICE_PROPERTY_SIM_PRESENT]: 251 raise error.TestError('There is no SIM in the modem.') 252 253 # Make sure SIM is not locked. 254 lock_status = props.get( 255 cellular_proxy.CellularProxy.DEVICE_PROPERTY_SIM_LOCK_STATUS, 256 None) 257 if lock_status is None: 258 raise error.TestError('Failed to read SIM lock status.') 259 locked = lock_status.get( 260 cellular_proxy.CellularProxy.PROPERTY_KEY_SIM_LOCK_ENABLED, 261 None) 262 if locked is None: 263 raise error.TestError('Failed to read SIM LockEnabled status.') 264 elif locked: 265 raise error.TestError( 266 'SIM is locked, test requires an unlocked SIM.') 267 268 269 def _verify_backchannel(self): 270 """Verify backchannel is on an ethernet device. 271 272 @raise error.TestError if backchannel is not on an ethernet device. 273 274 """ 275 if self._backchannel is None: 276 return 277 278 if not self._backchannel.is_using_ethernet(): 279 raise error.TestError('An ethernet connection is required between ' 280 'the test server and the device under test.') 281 282 283 def _wait_for_modem_registration(self): 284 """Wait for the modem to register with the network. 285 286 @raise error.TestError if modem is not registered. 287 288 """ 289 utils.poll_for_condition( 290 self.modem.ModemIsRegistered, 291 exception=error.TestError( 292 'Modem failed to register with the network.'), 293 timeout=cellular_proxy.CellularProxy.SERVICE_REGISTRATION_TIMEOUT) 294 295 296 def _verify_cellular_service(self): 297 """Make sure a cellular service exists. 298 299 The cellular service should not be connected to the network. 300 301 @raise error.TestError if cellular service does not exist or if 302 there are multiple cellular services. 303 304 """ 305 service = self.shill.wait_for_cellular_service_object() 306 307 try: 308 service.Disconnect() 309 except dbus.DBusException as e: 310 if (e.get_dbus_name() != 311 cellular_proxy.CellularProxy.ERROR_NOT_CONNECTED): 312 raise 313 success, state, _ = self.shill.wait_for_property_in( 314 service, 315 cellular_proxy.CellularProxy.SERVICE_PROPERTY_STATE, 316 ('idle',), 317 cellular_proxy.CellularProxy.SERVICE_DISCONNECT_TIMEOUT) 318 if not success: 319 raise error.TestError( 320 'Cellular service needs to start in the "idle" state. ' 321 'Current state is "%s". ' 322 'Modem disconnect may have failed.' % 323 state) 324 325 326class CellularOTATestEnvironment(CellularTestEnvironment): 327 """Setup and verify cellular over-the-air (OTA) test environment. """ 328 def __init__(self, **kwargs): 329 super(CellularOTATestEnvironment, self).__init__(**kwargs) 330 331 332class CellularPseudoMMTestEnvironment(CellularTestEnvironment): 333 """Setup and verify cellular pseudomodem test environment. """ 334 def __init__(self, pseudomm_args=None, **kwargs): 335 """ 336 @param pseudomm_args: Tuple of arguments passed to the pseudomodem, see 337 pseudomodem_context.py for description of each argument in the 338 tuple: (flags_map, block_output, bus) 339 340 """ 341 kwargs["skip_modem_reset"] = True 342 super(CellularPseudoMMTestEnvironment, self).__init__(**kwargs) 343 self._context_managers.append( 344 pseudomodem_context.PseudoModemManagerContext( 345 True, bus=self.bus, *pseudomm_args)) 346