# Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import dbus
import logging
import time
from autotest_lib.client.cros.networking import shill_proxy
class CellularProxy(shill_proxy.ShillProxy):
"""Wrapper around shill dbus interface used by cellular tests."""
# Properties exposed by shill.
DEVICE_PROPERTY_DBUS_OBJECT = 'DBus.Object'
DEVICE_PROPERTY_MODEL_ID = 'Cellular.ModelID'
DEVICE_PROPERTY_OUT_OF_CREDITS = 'Cellular.OutOfCredits'
DEVICE_PROPERTY_SIM_LOCK_STATUS = 'Cellular.SIMLockStatus'
DEVICE_PROPERTY_SIM_PRESENT = 'Cellular.SIMPresent'
DEVICE_PROPERTY_TECHNOLOGY_FAMILY = 'Cellular.Family'
DEVICE_PROPERTY_TECHNOLOGY_FAMILY_CDMA = 'CDMA'
DEVICE_PROPERTY_TECHNOLOGY_FAMILY_GSM = 'GSM'
SERVICE_PROPERTY_LAST_GOOD_APN = 'Cellular.LastGoodAPN'
# APN info property names.
APN_INFO_PROPERTY_APN = 'apn'
# Keys into the dictionaries exposed as properties.
PROPERTY_KEY_SIM_LOCK_TYPE = 'LockType'
PROPERTY_KEY_SIM_LOCK_ENABLED = 'LockEnabled'
PROPERTY_KEY_SIM_LOCK_RETRIES_LEFT = 'RetriesLeft'
# Valid values taken by properties exposed by shill.
VALUE_SIM_LOCK_TYPE_PIN = 'sim-pin'
VALUE_SIM_LOCK_TYPE_PUK = 'sim-puk'
# Various timeouts in seconds.
SERVICE_CONNECT_TIMEOUT = 60
SERVICE_DISCONNECT_TIMEOUT = 60
SERVICE_REGISTRATION_TIMEOUT = 60
SLEEP_INTERVAL = 0.1
def set_logging_for_cellular_test(self):
"""Set the logging in shill for a test of cellular technology.
Set the log level to |ShillProxy.LOG_LEVEL_FOR_TEST| and the log scopes
to the ones defined in |ShillProxy.LOG_SCOPES_FOR_TEST| for
|ShillProxy.TECHNOLOGY_CELLULAR|.
"""
self.set_logging_for_test(self.TECHNOLOGY_CELLULAR)
def find_cellular_service_object(self):
"""Returns the first dbus object found that is a cellular service.
@return DBus object for the first cellular service found. None if no
service found.
"""
return self.find_object('Service', {'Type': self.TECHNOLOGY_CELLULAR})
def wait_for_cellular_service_object(
self, timeout_seconds=SERVICE_REGISTRATION_TIMEOUT):
"""Waits for the cellular service object to show up.
@param timeout_seconds: Amount of time to wait for cellular service.
@return DBus object for the first cellular service found.
@raises ShillProxyError if no cellular service is found within the
registration timeout period.
"""
CellularProxy._poll_for_condition(
lambda: self.find_cellular_service_object() is not None,
'Failed to find cellular service object',
timeout=timeout_seconds)
return self.find_cellular_service_object()
def find_cellular_device_object(self):
"""Returns the first dbus object found that is a cellular device.
@return DBus object for the first cellular device found. None if no
device found.
"""
return self.find_object('Device', {'Type': self.TECHNOLOGY_CELLULAR})
def reset_modem(self, modem, expect_device=True, expect_powered=True,
expect_service=True):
"""Reset |modem|.
Do, in sequence,
(1) Ensure that the current device object disappears.
(2) If |expect_device|, ensure that the device reappears.
(3) If |expect_powered|, ensure that the device is powered.
(4) If |expect_service|, ensure that the service reappears.
This function does not check the service state for the device after
reset.
@param modem: DBus object for the modem to reset.
@param expect_device: If True, ensure that a DBus object reappears for
the same modem after the reset.
@param expect_powered: If True, ensure that the modem is powered on
after the reset.
@param expect_service: If True, ensure that a service managing the
reappeared modem also reappears.
@return (device, service)
device: DBus object for the reappeared Device after the reset.
service: DBus object for the reappeared Service after the reset.
Either of these may be None, if the object is not expected to
reappear.
@raises ShillProxyError if any of the conditions (1)-(4) fail.
"""
logging.info('Resetting modem')
# Obtain identifying information about the modem.
properties = modem.GetProperties(utf8_strings=True)
# NOTE: Using the Model ID means that this will break if we have two
# identical cellular modems in a DUT. Fortunately, we only support one
# modem at a time.
model_id = properties.get(self.DEVICE_PROPERTY_MODEL_ID)
if not model_id:
raise shill_proxy.ShillProxyError(
'Failed to get identifying information for the modem.')
old_modem_path = modem.object_path
old_modem_mm_object = properties.get(self.DEVICE_PROPERTY_DBUS_OBJECT)
if not old_modem_mm_object:
raise shill_proxy.ShillProxyError(
'Failed to get the mm object path for the modem.')
modem.Reset()
# (1) Wait for the old modem to disappear
CellularProxy._poll_for_condition(
lambda: self._is_old_modem_gone(old_modem_path,
old_modem_mm_object),
'Old modem disappeared',
timeout=60)
# (2) Wait for the device to reappear
if not expect_device:
return None, None
# The timeout here should be sufficient for our slowest modem to
# reappear.
CellularProxy._poll_for_condition(
lambda: self._get_reappeared_modem(model_id,
old_modem_mm_object),
desc='The modem reappeared after reset.',
timeout=60)
new_modem = self._get_reappeared_modem(model_id, old_modem_mm_object)
# (3) Check powered state of the device
if not expect_powered:
return new_modem, None
success, _, _ = self.wait_for_property_in(new_modem,
self.DEVICE_PROPERTY_POWERED,
[self.VALUE_POWERED_ON],
timeout_seconds=10)
if not success:
raise shill_proxy.ShillProxyError(
'After modem reset, new modem failed to enter powered '
'state.')
# (4) Check that service reappears
if not expect_service:
return new_modem, None
new_service = self.get_service_for_device(new_modem)
if not new_service:
raise shill_proxy.ShillProxyError(
'Failed to find a shill service managing the reappeared '
'device.')
return new_modem, new_service
def disable_modem_for_test_setup(self, timeout_seconds=10):
"""
Disables all cellular modems.
Use this method only for setting up tests. Do not use this method to
test disable functionality because this method repeatedly attempts to
disable the cellular technology until it succeeds (ignoring all DBus
errors) since the DisableTechnology() call may fail for various reasons
(eg. an enable is in progress).
@param timeout_seconds: Amount of time to wait until the modem is
disabled.
@raises ShillProxyError if the modems fail to disable within
|timeout_seconds|.
"""
def _disable_cellular_technology(self):
try:
self._manager.DisableTechnology(self.TECHNOLOGY_CELLULAR)
return True
except dbus.DBusException as e:
return False
CellularProxy._poll_for_condition(
lambda: _disable_cellular_technology(self),
'Failed to disable cellular technology.',
timeout=timeout_seconds)
modem = self.find_cellular_device_object()
self.wait_for_property_in(modem, self.DEVICE_PROPERTY_POWERED,
[self.VALUE_POWERED_OFF],
timeout_seconds=timeout_seconds)
# TODO(pprabhu) Use utils.poll_for_condition instead
@staticmethod
def _poll_for_condition(condition, desc, timeout=10):
"""Poll till |condition| is satisfied.
@param condition: A function taking no arguments. The condition is
met when the return value can be cast to the bool True.
@param desc: The description given when we timeout waiting for
|condition|.
"""
start_time = time.time()
while True:
value = condition()
if value:
return value
if(time.time() + CellularProxy.SLEEP_INTERVAL - start_time >
timeout):
raise shill_proxy.ShillProxyError(
'Timed out waiting for condition %s.' % desc)
time.sleep(CellularProxy.SLEEP_INTERVAL)
def _is_old_modem_gone(self, modem_path, modem_mm_object):
"""Tests if the DBus object for modem disappears after Reset.
@param modem_path: The DBus path for the modem object that must vanish.
@param modem_mm_object: The modemmanager object path reported by the
old modem. This is unique everytime a new modem is (re)exposed.
@return True if the object disappeared, false otherwise.
"""
device = self.get_dbus_object(self.DBUS_TYPE_DEVICE, modem_path)
try:
properties = device.GetProperties()
# DBus object exists, perhaps a reappeared device?
return (properties.get(self.DEVICE_PROPERTY_DBUS_OBJECT) !=
modem_mm_object)
except dbus.DBusException as e:
if e.get_dbus_name() == self.DBUS_ERROR_UNKNOWN_OBJECT:
return True
return False
def _get_reappeared_modem(self, model_id, old_modem_mm_object):
"""Check that a vanished modem reappers.
@param model_id: The model ID reported by the vanished modem.
@param old_modem_mm_object: The previously reported modemmanager object
path for this modem.
@return The reappeared DBus object, if any. None otherwise.
"""
# TODO(pprabhu) This will break if we have multiple cellular devices
# in the system at the same time.
device = self.find_cellular_device_object()
if not device:
return None
properties = device.GetProperties(utf8_strings=True)
if (model_id == properties.get(self.DEVICE_PROPERTY_MODEL_ID) and
(old_modem_mm_object !=
properties.get(self.DEVICE_PROPERTY_DBUS_OBJECT))):
return device
return None