1# Lint as: python2, python3 2# Copyright (c) 2013 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 6import dbus 7import logging 8 9from autotest_lib.client.bin import utils 10from autotest_lib.client.cros.networking import shill_proxy 11 12 13class CellularProxy(shill_proxy.ShillProxy): 14 """Wrapper around shill dbus interface used by cellular tests.""" 15 16 # Properties exposed by shill. 17 DEVICE_PROPERTY_DBUS_OBJECT = 'DBus.Object' 18 DEVICE_PROPERTY_MODEL_ID = 'Cellular.ModelID' 19 DEVICE_PROPERTY_MANUFACTURER = 'Cellular.Manufacturer' 20 DEVICE_PROPERTY_OUT_OF_CREDITS = 'Cellular.OutOfCredits' 21 DEVICE_PROPERTY_SIM_LOCK_STATUS = 'Cellular.SIMLockStatus' 22 DEVICE_PROPERTY_SIM_PRESENT = 'Cellular.SIMPresent' 23 DEVICE_PROPERTY_TECHNOLOGY_FAMILY = 'Cellular.Family' 24 DEVICE_PROPERTY_TECHNOLOGY_FAMILY_CDMA = 'CDMA' 25 DEVICE_PROPERTY_TECHNOLOGY_FAMILY_GSM = 'GSM' 26 SERVICE_PROPERTY_LAST_GOOD_APN = 'Cellular.LastGoodAPN' 27 28 # APN info property names. 29 APN_INFO_PROPERTY_APN = 'apn' 30 31 # Keys into the dictionaries exposed as properties. 32 PROPERTY_KEY_SIM_LOCK_TYPE = 'LockType' 33 PROPERTY_KEY_SIM_LOCK_ENABLED = 'LockEnabled' 34 PROPERTY_KEY_SIM_LOCK_RETRIES_LEFT = 'RetriesLeft' 35 36 # Valid values taken by properties exposed by shill. 37 VALUE_SIM_LOCK_TYPE_PIN = 'sim-pin' 38 VALUE_SIM_LOCK_TYPE_PUK = 'sim-puk' 39 40 # Various timeouts in seconds. 41 SERVICE_CONNECT_TIMEOUT = 60 42 SERVICE_DISCONNECT_TIMEOUT = 60 43 SERVICE_REGISTRATION_TIMEOUT = 60 44 SLEEP_INTERVAL = 0.1 45 46 def set_logging_for_cellular_test(self): 47 """Set the logging in shill for a test of cellular technology. 48 49 Set the log level to |ShillProxy.LOG_LEVEL_FOR_TEST| and the log scopes 50 to the ones defined in |ShillProxy.LOG_SCOPES_FOR_TEST| for 51 |ShillProxy.TECHNOLOGY_CELLULAR|. 52 53 """ 54 self.set_logging_for_test(self.TECHNOLOGY_CELLULAR) 55 56 57 def find_cellular_service_object(self): 58 """Returns the first dbus object found that is a cellular service. 59 60 @return DBus object for the first cellular service found. None if no 61 service found. 62 63 """ 64 return self.find_object('Service', { 65 'Type': self.TECHNOLOGY_CELLULAR, 66 'Connectable': True 67 }) 68 69 70 def wait_for_cellular_service_object( 71 self, timeout_seconds=SERVICE_REGISTRATION_TIMEOUT): 72 """Waits for the cellular service object to show up. 73 74 @param timeout_seconds: Amount of time to wait for cellular service. 75 @return DBus object for the first cellular service found. 76 @raises ShillProxyError if no cellular service is found within the 77 registration timeout period. 78 79 """ 80 return utils.poll_for_condition( 81 lambda: self.find_cellular_service_object(), 82 exception=shill_proxy.ShillProxyTimeoutError( 83 'Failed to find cellular service object'), 84 timeout=timeout_seconds) 85 86 87 def find_cellular_device_object(self): 88 """Returns the first dbus object found that is a cellular device. 89 90 @return DBus object for the first cellular device found. None if no 91 device found. 92 93 """ 94 return self.find_object('Device', {'Type': self.TECHNOLOGY_CELLULAR}) 95 96 97 def reset_modem(self, modem, expect_device=True, expect_powered=True, 98 expect_service=True): 99 """Reset |modem|. 100 101 Do, in sequence, 102 (1) Ensure that the current device object disappears. 103 (2) If |expect_device|, ensure that the device reappears. 104 (3) If |expect_powered|, ensure that the device is powered. 105 (4) If |expect_service|, ensure that the service reappears. 106 107 This function does not check the service state for the device after 108 reset. 109 110 @param modem: DBus object for the modem to reset. 111 @param expect_device: If True, ensure that a DBus object reappears for 112 the same modem after the reset. 113 @param expect_powered: If True, ensure that the modem is powered on 114 after the reset. 115 @param expect_service: If True, ensure that a service managing the 116 reappeared modem also reappears. 117 118 @return (device, service) 119 device: DBus object for the reappeared Device after the reset. 120 service: DBus object for the reappeared Service after the reset. 121 Either of these may be None, if the object is not expected to 122 reappear. 123 124 @raises ShillProxyError if any of the conditions (1)-(4) fail. 125 126 """ 127 logging.info('Resetting modem') 128 # Obtain identifying information about the modem. 129 properties = modem.GetProperties() 130 # NOTE: Using the Model ID means that this will break if we have two 131 # identical cellular modems in a DUT. Fortunately, we only support one 132 # modem at a time. 133 model_id = properties.get(self.DEVICE_PROPERTY_MODEL_ID) 134 if not model_id: 135 raise shill_proxy.ShillProxyError( 136 'Failed to get identifying information for the modem.') 137 old_modem_path = modem.object_path 138 old_modem_mm_object = properties.get(self.DEVICE_PROPERTY_DBUS_OBJECT) 139 if not old_modem_mm_object: 140 raise shill_proxy.ShillProxyError( 141 'Failed to get the mm object path for the modem.') 142 143 manufacturer = properties.get(self.DEVICE_PROPERTY_MANUFACTURER) 144 if not manufacturer: 145 raise shill_proxy.ShillProxyError( 146 'Failed to get the manufacturer for the modem.') 147 148 # On Qualcomm modems, rebooting the modem causes ModemManager to also 149 # restart, because when the Qrtr services are removed, qc-netmgr 150 # restarts ModemManager. 151 mm_rebooted = "QUALCOMM" in manufacturer 152 modem.Reset() 153 154 # (1) Wait for the old modem to disappear 155 utils.poll_for_condition(lambda: self._is_old_modem_gone( 156 old_modem_path, old_modem_mm_object), 157 exception=shill_proxy.ShillProxyTimeoutError( 158 'Old modem disappeared'), 159 timeout=60) 160 161 # (2) Wait for the device to reappear 162 if not expect_device: 163 return None, None 164 # The timeout here should be sufficient for our slowest modem to 165 # reappear. 166 new_modem = utils.poll_for_condition( 167 lambda: self._get_reappeared_modem( 168 model_id, old_modem_mm_object, mm_rebooted), 169 exception=shill_proxy.ShillProxyTimeoutError( 170 'The modem reappeared after reset.'), 171 timeout=60) 172 173 # (3) Check powered state of the device 174 if not expect_powered: 175 return new_modem, None 176 success, _, _ = self.wait_for_property_in(new_modem, 177 self.DEVICE_PROPERTY_POWERED, 178 [self.VALUE_POWERED_ON], 179 timeout_seconds=15) 180 if not success: 181 raise shill_proxy.ShillProxyError( 182 'After modem reset, new modem failed to enter powered ' 183 'state.') 184 185 # (4) Check that service reappears 186 if not expect_service: 187 return new_modem, None 188 new_service = self.get_service_for_device(new_modem) 189 if not new_service: 190 raise shill_proxy.ShillProxyError( 191 'Failed to find a shill service managing the reappeared ' 192 'device.') 193 return new_modem, new_service 194 195 196 def disable_modem_for_test_setup(self, timeout_seconds=10): 197 """ 198 Disables all cellular modems. 199 200 Use this method only for setting up tests. Do not use this method to 201 test disable functionality because this method repeatedly attempts to 202 disable the cellular technology until it succeeds (ignoring all DBus 203 errors) since the DisableTechnology() call may fail for various reasons 204 (eg. an enable is in progress). 205 206 @param timeout_seconds: Amount of time to wait until the modem is 207 disabled. 208 @raises ShillProxyError if the modems fail to disable within 209 |timeout_seconds|. 210 211 """ 212 def _disable_cellular_technology(self): 213 try: 214 self._manager.DisableTechnology(self.TECHNOLOGY_CELLULAR) 215 return True 216 except dbus.DBusException as e: 217 return False 218 219 utils.poll_for_condition( 220 lambda: _disable_cellular_technology(self), 221 exception=shill_proxy.ShillProxyTimeoutError( 222 'Failed to disable cellular technology.'), 223 timeout=timeout_seconds) 224 modem = self.find_cellular_device_object() 225 self.wait_for_property_in(modem, self.DEVICE_PROPERTY_POWERED, 226 [self.VALUE_POWERED_OFF], 227 timeout_seconds=timeout_seconds) 228 229 230 def _is_old_modem_gone(self, modem_path, modem_mm_object): 231 """Tests if the DBus object for modem disappears after Reset. 232 233 @param modem_path: The DBus path for the modem object that must vanish. 234 @param modem_mm_object: The modemmanager object path reported by the 235 old modem. This is unique everytime a new modem is (re)exposed. 236 237 @return True if the object disappeared, false otherwise. 238 239 """ 240 device = self.get_dbus_object(self.DBUS_TYPE_DEVICE, modem_path) 241 try: 242 properties = device.GetProperties() 243 # DBus object exists, perhaps a reappeared device? 244 return (properties.get(self.DEVICE_PROPERTY_DBUS_OBJECT) != 245 modem_mm_object) 246 except dbus.DBusException as e: 247 if e.get_dbus_name() == self.DBUS_ERROR_UNKNOWN_OBJECT: 248 return True 249 return False 250 251 252 def _get_reappeared_modem(self, model_id, old_modem_mm_object, mm_reboot): 253 """Check that a vanished modem reappeers. 254 255 @param model_id: The model ID reported by the vanished modem. 256 @param old_modem_mm_object: The previously reported modemmanager object 257 path for this modem. 258 @param mm_reboot: indicates when modemmanager was rebooted. 259 When modemmanager reboots, the previous modem object name has 260 no importance. 261 262 @return The reappeared DBus object, if any. None otherwise. 263 264 """ 265 # TODO(pprabhu) This will break if we have multiple cellular devices 266 # in the system at the same time. 267 device = self.find_cellular_device_object() 268 if not device: 269 return None 270 properties = device.GetProperties() 271 if (model_id == properties.get(self.DEVICE_PROPERTY_MODEL_ID) and 272 (mm_reboot or 273 (old_modem_mm_object != properties.get( 274 self.DEVICE_PROPERTY_DBUS_OBJECT) 275 and '/' in properties.get(self.DEVICE_PROPERTY_DBUS_OBJECT)))): 276 return device 277 return None 278