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