You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
746 lines
30 KiB
746 lines
30 KiB
# 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 collections
|
|
import dbus
|
|
import dbus.mainloop.glib
|
|
import gobject
|
|
import time
|
|
|
|
from autotest_lib.client.cros import dbus_util
|
|
|
|
|
|
class ShillProxyError(Exception):
|
|
"""Exceptions raised by ShillProxy and its children."""
|
|
pass
|
|
|
|
|
|
class ShillProxyTimeoutError(ShillProxyError):
|
|
"""Timeout exception raised by ShillProxy and its children."""
|
|
def __init__(self, desc):
|
|
super(ShillProxyTimeoutError, self).__init__(
|
|
'Timed out waiting for condition %s.' % desc)
|
|
|
|
|
|
class ShillProxy(object):
|
|
"""A wrapper around a DBus proxy for shill."""
|
|
|
|
# Core DBus error names
|
|
DBUS_ERROR_UNKNOWN_OBJECT = 'org.freedesktop.DBus.Error.UnknownObject'
|
|
# Shill error names
|
|
ERROR_ALREADY_CONNECTED = 'org.chromium.flimflam.Error.AlreadyConnected'
|
|
ERROR_FAILURE = 'org.chromium.flimflam.Error.Failure'
|
|
ERROR_INCORRECT_PIN = 'org.chromium.flimflam.Error.IncorrectPin'
|
|
ERROR_IN_PROGRESS = 'org.chromium.flimflam.Error.InProgress'
|
|
ERROR_NOT_CONNECTED = 'org.chromium.flimflam.Error.NotConnected'
|
|
ERROR_NOT_SUPPORTED = 'org.chromium.flimflam.Error.NotSupported'
|
|
ERROR_PIN_BLOCKED = 'org.chromium.flimflam.Error.PinBlocked'
|
|
|
|
|
|
DBUS_INTERFACE = 'org.chromium.flimflam'
|
|
DBUS_SERVICE_UNKNOWN = 'org.freedesktop.DBus.Error.ServiceUnknown'
|
|
DBUS_TYPE_DEVICE = 'org.chromium.flimflam.Device'
|
|
DBUS_TYPE_IPCONFIG = 'org.chromium.flimflam.IPConfig'
|
|
DBUS_TYPE_MANAGER = 'org.chromium.flimflam.Manager'
|
|
DBUS_TYPE_PROFILE = 'org.chromium.flimflam.Profile'
|
|
DBUS_TYPE_SERVICE = 'org.chromium.flimflam.Service'
|
|
|
|
ENTRY_FIELD_NAME = 'Name'
|
|
ENTRY_FIELD_TYPE = 'Type'
|
|
|
|
MANAGER_PROPERTY_ACTIVE_PROFILE = 'ActiveProfile'
|
|
MANAGER_PROPERTY_DEVICES = 'Devices'
|
|
MANAGER_PROPERTY_NO_AUTOCONNECT_TECHNOLOGIES = 'NoAutoConnectTechnologies'
|
|
MANAGER_PROPERTY_ENABLED_TECHNOLOGIES = 'EnabledTechnologies'
|
|
MANAGER_PROPERTY_PROHIBITED_TECHNOLOGIES = 'ProhibitedTechnologies'
|
|
MANAGER_PROPERTY_UNINITIALIZED_TECHNOLOGIES = 'UninitializedTechnologies'
|
|
MANAGER_PROPERTY_PROFILES = 'Profiles'
|
|
MANAGER_PROPERTY_SERVICES = 'Services'
|
|
MANAGER_PROPERTY_DEFAULT_SERVICE = 'DefaultService'
|
|
MANAGER_PROPERTY_ALL_SERVICES = 'ServiceCompleteList'
|
|
MANAGER_PROPERTY_DHCPPROPERTY_HOSTNAME = 'DHCPProperty.Hostname'
|
|
MANAGER_PROPERTY_DHCPPROPERTY_VENDORCLASS = 'DHCPProperty.VendorClass'
|
|
MANAGER_PROPERTY_WIFI_GLOBAL_FT_ENABLED = 'WiFi.GlobalFTEnabled'
|
|
|
|
MANAGER_OPTIONAL_PROPERTY_MAP = {
|
|
MANAGER_PROPERTY_DHCPPROPERTY_HOSTNAME: dbus.String,
|
|
MANAGER_PROPERTY_DHCPPROPERTY_VENDORCLASS: dbus.String,
|
|
MANAGER_PROPERTY_WIFI_GLOBAL_FT_ENABLED: dbus.Boolean
|
|
}
|
|
|
|
PROFILE_PROPERTY_ENTRIES = 'Entries'
|
|
PROFILE_PROPERTY_NAME = 'Name'
|
|
|
|
OBJECT_TYPE_PROPERTY_MAP = {
|
|
'Device': ( DBUS_TYPE_DEVICE, MANAGER_PROPERTY_DEVICES ),
|
|
'Profile': ( DBUS_TYPE_PROFILE, MANAGER_PROPERTY_PROFILES ),
|
|
'Service': ( DBUS_TYPE_SERVICE, MANAGER_PROPERTY_SERVICES ),
|
|
'AnyService': ( DBUS_TYPE_SERVICE, MANAGER_PROPERTY_ALL_SERVICES )
|
|
}
|
|
|
|
DEVICE_ENUMERATION_TIMEOUT = 30
|
|
DEVICE_ENABLE_DISABLE_TIMEOUT = 60
|
|
SERVICE_DISCONNECT_TIMEOUT = 5
|
|
|
|
SERVICE_PROPERTY_AUTOCONNECT = 'AutoConnect'
|
|
SERVICE_PROPERTY_DEVICE = 'Device'
|
|
SERVICE_PROPERTY_GUID = 'GUID'
|
|
SERVICE_PROPERTY_HEX_SSID = 'WiFi.HexSSID'
|
|
SERVICE_PROPERTY_HIDDEN = 'WiFi.HiddenSSID'
|
|
SERVICE_PROPERTY_MODE = 'Mode'
|
|
SERVICE_PROPERTY_NAME = 'Name'
|
|
SERVICE_PROPERTY_PASSPHRASE = 'Passphrase'
|
|
SERVICE_PROPERTY_PROFILE = 'Profile'
|
|
SERVICE_PROPERTY_SAVE_CREDENTIALS = 'SaveCredentials'
|
|
# Unless you really care whether a network is WPA (TSN) vs. WPA-2
|
|
# (RSN), you should use SERVICE_PROPERTY_SECURITY_CLASS.
|
|
SERVICE_PROPERTY_SECURITY_RAW = 'Security'
|
|
SERVICE_PROPERTY_SECURITY_CLASS = 'SecurityClass'
|
|
SERVICE_PROPERTY_SSID = 'SSID'
|
|
SERVICE_PROPERTY_STRENGTH = 'Strength'
|
|
SERVICE_PROPERTY_STATE = 'State'
|
|
SERVICE_PROPERTY_STATIC_IP_CONFIG = 'StaticIPConfig'
|
|
SERVICE_PROPERTY_TYPE = 'Type'
|
|
|
|
# EAP related properties.
|
|
SERVICE_PROPERTY_EAP_EAP = 'EAP.EAP'
|
|
SERVICE_PROPERTY_EAP_INNER_EAP = 'EAP.InnerEAP'
|
|
SERVICE_PROPERTY_EAP_IDENTITY = 'EAP.Identity'
|
|
SERVICE_PROPERTY_EAP_PASSWORD = 'EAP.Password'
|
|
SERVICE_PROPERTY_EAP_CA_CERT_PEM = 'EAP.CACertPEM'
|
|
SERVICE_PROPERTY_CLIENT_CERT_ID = 'EAP.CertID'
|
|
SERVICE_PROPERTY_EAP_KEY_MGMT = 'EAP.KeyMgmt'
|
|
SERVICE_PROPERTY_EAP_PIN = 'EAP.PIN'
|
|
SERVICE_PROPERTY_PRIVATE_KEY_ID = 'EAP.KeyID'
|
|
SERVICE_PROPERTY_USE_SYSTEM_CAS = 'EAP.UseSystemCAs'
|
|
|
|
# OpenVPN related properties.
|
|
SERVICE_PROPERTY_OPENVPN_CA_CERT_PEM = 'OpenVPN.CACertPEM'
|
|
SERVICE_PROPERTY_OPENVPN_PASSWORD = 'OpenVPN.Password'
|
|
SERVICE_PROPERTY_OPENVPN_PKCS11_ID = 'OpenVPN.Pkcs11.ID'
|
|
SERVICE_PROPERTY_OPENVPN_PKCS11_PIN = 'OpenVPN.Pkcs11.PIN'
|
|
SERVICE_PROPERTY_OPENVPN_PROVIDER_HOST = 'Provider.Host'
|
|
SERVICE_PROPERTY_OPENVPN_PROVIDER_TYPE = 'Provider.Type'
|
|
SERVICE_PROPERTY_OPENVPN_REMOTE_CERT_EKU = 'OpenVPN.RemoteCertEKU'
|
|
SERVICE_PROPERTY_OPENVPN_USER = 'OpenVPN.User'
|
|
SERVICE_PROPERTY_OPENVPN_VERB = 'OpenVPN.Verb'
|
|
SERVICE_PROPERTY_OPENVPN_VERIFY_HASH = 'OpenVPN.VerifyHash'
|
|
SERVICE_PROPERTY_OPENVPN_VERIFY_X509_NAME = 'OpenVPN.VerifyX509Name'
|
|
SERVICE_PROPERTY_OPENVPN_VERIFY_X509_TYPE = 'OpenVPN.VerifyX509Type'
|
|
|
|
# L2TP VPN related properties.
|
|
SERVICE_PROPERTY_L2TP_CA_CERT_PEM = 'L2TPIPsec.CACertPEM'
|
|
SERVICE_PROPERTY_L2TP_CLIENT_CERT_ID = 'L2TPIPsec.ClientCertID'
|
|
SERVICE_PROPERTY_L2TP_CLIENT_CERT_SLOT = 'L2TPIPsec.ClientCertSlot'
|
|
SERVICE_PROPERTY_L2TP_PASSWORD = 'L2TPIPsec.Password'
|
|
SERVICE_PROPERTY_L2TP_PIN = 'L2TPIPsec.PIN'
|
|
SERVICE_PROPERTY_L2TP_PSK = 'L2TPIPsec.PSK'
|
|
SERVICE_PROPERTY_L2TP_USER = 'L2TPIPsec.User'
|
|
SERVICE_PROPERTY_L2TP_XAUTH_PASSWORD = 'L2TPIPsec.XauthPassword'
|
|
SERVICE_PROPERTY_L2TP_XAUTH_USER = 'L2TPIPsec.XauthUser'
|
|
|
|
# Mapping of service property to (dbus-type, additional kwargs).
|
|
SERVICE_PROPERTY_MAP = {
|
|
SERVICE_PROPERTY_AUTOCONNECT: (dbus.Boolean, {}),
|
|
SERVICE_PROPERTY_DEVICE: (dbus.ObjectPath, {}),
|
|
SERVICE_PROPERTY_GUID: (dbus.String, {}),
|
|
SERVICE_PROPERTY_HEX_SSID: (dbus.String, {}),
|
|
SERVICE_PROPERTY_HIDDEN: (dbus.Boolean, {}),
|
|
SERVICE_PROPERTY_MODE: (dbus.String, {}),
|
|
SERVICE_PROPERTY_NAME: (dbus.String, {}),
|
|
SERVICE_PROPERTY_PASSPHRASE: (dbus.String, {}),
|
|
SERVICE_PROPERTY_PROFILE: (dbus.ObjectPath, {}),
|
|
SERVICE_PROPERTY_SAVE_CREDENTIALS: (dbus.Boolean, {}),
|
|
SERVICE_PROPERTY_SECURITY_RAW: (dbus.String, {}),
|
|
SERVICE_PROPERTY_SECURITY_CLASS: (dbus.String, {}),
|
|
SERVICE_PROPERTY_SSID: (dbus.String, {}),
|
|
SERVICE_PROPERTY_STRENGTH: (dbus.Byte, {}),
|
|
SERVICE_PROPERTY_STATE: (dbus.String, {}),
|
|
SERVICE_PROPERTY_TYPE: (dbus.String, {}),
|
|
SERVICE_PROPERTY_STATIC_IP_CONFIG: (dbus.Dictionary,
|
|
{'signature' : 'sv'}),
|
|
|
|
SERVICE_PROPERTY_EAP_EAP: (dbus.String, {}),
|
|
SERVICE_PROPERTY_EAP_INNER_EAP: (dbus.String, {}),
|
|
SERVICE_PROPERTY_EAP_IDENTITY: (dbus.String, {}),
|
|
SERVICE_PROPERTY_EAP_PASSWORD: (dbus.String, {}),
|
|
SERVICE_PROPERTY_EAP_CA_CERT_PEM: (dbus.Array, {}),
|
|
SERVICE_PROPERTY_CLIENT_CERT_ID: (dbus.String, {}),
|
|
SERVICE_PROPERTY_EAP_KEY_MGMT: (dbus.String, {}),
|
|
SERVICE_PROPERTY_EAP_PIN: (dbus.String, {}),
|
|
SERVICE_PROPERTY_PRIVATE_KEY_ID: (dbus.String, {}),
|
|
SERVICE_PROPERTY_USE_SYSTEM_CAS: (dbus.Boolean, {}),
|
|
|
|
SERVICE_PROPERTY_OPENVPN_CA_CERT_PEM: (dbus.Array, {}),
|
|
SERVICE_PROPERTY_OPENVPN_PASSWORD: (dbus.String, {}),
|
|
SERVICE_PROPERTY_OPENVPN_PKCS11_ID: (dbus.String, {}),
|
|
SERVICE_PROPERTY_OPENVPN_PKCS11_PIN: (dbus.String, {}),
|
|
SERVICE_PROPERTY_OPENVPN_PROVIDER_HOST: (dbus.String, {}),
|
|
SERVICE_PROPERTY_OPENVPN_PROVIDER_TYPE: (dbus.String, {}),
|
|
SERVICE_PROPERTY_OPENVPN_REMOTE_CERT_EKU: (dbus.String, {}),
|
|
SERVICE_PROPERTY_OPENVPN_USER: (dbus.String, {}),
|
|
SERVICE_PROPERTY_OPENVPN_VERB: (dbus.String, {}),
|
|
SERVICE_PROPERTY_OPENVPN_VERIFY_HASH: (dbus.String, {}),
|
|
SERVICE_PROPERTY_OPENVPN_VERIFY_X509_NAME: (dbus.String, {}),
|
|
SERVICE_PROPERTY_OPENVPN_VERIFY_X509_TYPE: (dbus.String, {}),
|
|
|
|
SERVICE_PROPERTY_L2TP_CA_CERT_PEM: (dbus.Array, {}),
|
|
SERVICE_PROPERTY_L2TP_CLIENT_CERT_ID: (dbus.String, {}),
|
|
SERVICE_PROPERTY_L2TP_CLIENT_CERT_SLOT: (dbus.String, {}),
|
|
SERVICE_PROPERTY_L2TP_PASSWORD: (dbus.String, {}),
|
|
SERVICE_PROPERTY_L2TP_PIN: (dbus.String, {}),
|
|
SERVICE_PROPERTY_L2TP_PSK: (dbus.String, {}),
|
|
SERVICE_PROPERTY_L2TP_USER: (dbus.String, {}),
|
|
SERVICE_PROPERTY_L2TP_XAUTH_PASSWORD: (dbus.String, {}),
|
|
SERVICE_PROPERTY_L2TP_XAUTH_USER: (dbus.String, {})
|
|
}
|
|
|
|
SERVICE_CONNECTED_STATES = ['portal', 'no-connectivity', 'redirect-found',
|
|
'portal-suspected', 'online', 'ready']
|
|
SUPPORTED_WIFI_STATION_TYPES = {'managed': 'managed',
|
|
'ibss': 'adhoc',
|
|
None: 'managed'}
|
|
|
|
DEVICE_PROPERTY_ADDRESS = 'Address'
|
|
DEVICE_PROPERTY_EAP_AUTHENTICATION_COMPLETED = 'EapAuthenticationCompleted'
|
|
DEVICE_PROPERTY_EAP_AUTHENTICATOR_DETECTED = 'EapAuthenticatorDetected'
|
|
DEVICE_PROPERTY_IP_CONFIG = 'IpConfig'
|
|
DEVICE_PROPERTY_INTERFACE = 'Interface'
|
|
DEVICE_PROPERTY_NAME = 'Name'
|
|
DEVICE_PROPERTY_POWERED = 'Powered'
|
|
DEVICE_PROPERTY_RECEIVE_BYTE_COUNT = 'ReceiveByteCount'
|
|
DEVICE_PROPERTY_SCANNING = 'Scanning'
|
|
DEVICE_PROPERTY_TRANSMIT_BYTE_COUNT = 'TransmitByteCount'
|
|
DEVICE_PROPERTY_TYPE = 'Type'
|
|
|
|
TECHNOLOGY_CELLULAR = 'cellular'
|
|
TECHNOLOGY_ETHERNET = 'ethernet'
|
|
TECHNOLOGY_VPN = 'vpn'
|
|
TECHNOLOGY_WIFI = 'wifi'
|
|
|
|
VALUE_POWERED_ON = True
|
|
VALUE_POWERED_OFF = False
|
|
|
|
POLLING_INTERVAL_SECONDS = 0.2
|
|
|
|
# Default log level used in connectivity tests.
|
|
LOG_LEVEL_FOR_TEST = -4
|
|
|
|
# Default log scopes used in connectivity tests.
|
|
LOG_SCOPES_FOR_TEST_COMMON = [
|
|
'connection',
|
|
'dbus',
|
|
'device',
|
|
'link',
|
|
'manager',
|
|
'portal',
|
|
'service'
|
|
]
|
|
|
|
# Default log scopes used in connectivity tests for specific technologies.
|
|
LOG_SCOPES_FOR_TEST = {
|
|
TECHNOLOGY_CELLULAR: LOG_SCOPES_FOR_TEST_COMMON + ['cellular'],
|
|
TECHNOLOGY_ETHERNET: LOG_SCOPES_FOR_TEST_COMMON + ['ethernet'],
|
|
TECHNOLOGY_VPN: LOG_SCOPES_FOR_TEST_COMMON + ['vpn'],
|
|
TECHNOLOGY_WIFI: LOG_SCOPES_FOR_TEST_COMMON + ['wifi'],
|
|
}
|
|
|
|
UNKNOWN_METHOD = 'org.freedesktop.DBus.Error.UnknownMethod'
|
|
|
|
|
|
@staticmethod
|
|
def str2dbus(dbus_class, value):
|
|
"""Typecast string property values to dbus types.
|
|
|
|
This mostly makes it easy to special case Boolean constructors
|
|
to interpret strings like 'false' and '0' as False.
|
|
|
|
@param dbus_class: DBus class object.
|
|
@param value: value to pass to constructor.
|
|
|
|
"""
|
|
if isinstance(dbus_class, dbus.Boolean):
|
|
return dbus_class(value.lower() in ('true','1'))
|
|
else:
|
|
return dbus_class(value)
|
|
|
|
|
|
@staticmethod
|
|
def service_properties_to_dbus_types(in_dict):
|
|
"""Convert service properties to dbus types.
|
|
|
|
@param in_dict: Dictionary containing service properties.
|
|
@return DBus variant dictionary containing service properties.
|
|
|
|
"""
|
|
dbus_dict = {}
|
|
for key, value in in_dict.iteritems():
|
|
if key not in ShillProxy.SERVICE_PROPERTY_MAP:
|
|
raise ShillProxyError('Unsupported property %s' % (key))
|
|
(dbus_type, kwargs) = ShillProxy.SERVICE_PROPERTY_MAP[key]
|
|
dbus_dict[key] = dbus_type(value, variant_level=1, **kwargs)
|
|
return dbus_dict
|
|
|
|
|
|
@classmethod
|
|
def dbus2primitive(cls, value):
|
|
"""Typecast values from dbus types to python types.
|
|
|
|
@param value: dbus object to convert to a primitive.
|
|
|
|
"""
|
|
return dbus_util.dbus2primitive(value)
|
|
|
|
|
|
@staticmethod
|
|
def get_dbus_property(interface, property_key):
|
|
"""get property on a dbus Interface
|
|
|
|
@param interface dbus Interface to receive new setting
|
|
@param property_key string name of property on interface
|
|
@return python typed object representing property value or None
|
|
|
|
"""
|
|
properties = interface.GetProperties(utf8_strings=True)
|
|
if property_key in properties:
|
|
return ShillProxy.dbus2primitive(properties[property_key])
|
|
else:
|
|
return None
|
|
|
|
|
|
@staticmethod
|
|
def set_dbus_property(interface, property_key, value):
|
|
"""set property on a dbus Interface
|
|
|
|
@param interface dbus Interface to receive new setting
|
|
@param property_key string name of property on interface
|
|
@param value string value to set for property on interface from string
|
|
|
|
"""
|
|
properties = interface.GetProperties(utf8_strings=True)
|
|
if property_key not in properties:
|
|
raise ShillProxyError('No property %s found in %s' %
|
|
(property_key, interface.object_path))
|
|
else:
|
|
dbus_class = properties[property_key].__class__
|
|
interface.SetProperty(property_key,
|
|
ShillProxy.str2dbus(dbus_class, value))
|
|
|
|
|
|
@staticmethod
|
|
def set_optional_dbus_property(interface, property_key, value):
|
|
"""set an optional property on a dbus Interface.
|
|
|
|
This method can be used for properties that are optionally listed
|
|
in the profile. It skips the initial check of the property
|
|
being in the interface.GetProperties list.
|
|
|
|
@param interface dbus Interface to receive new setting
|
|
@param property_key string name of property on interface
|
|
@param value string value to set for property on interface from string
|
|
|
|
"""
|
|
if property_key not in ShillProxy.MANAGER_OPTIONAL_PROPERTY_MAP:
|
|
raise ShillProxyError('Unsupported property %s' %
|
|
(property_key))
|
|
else:
|
|
dbus_class = ShillProxy.MANAGER_OPTIONAL_PROPERTY_MAP[property_key]
|
|
interface.SetProperty(property_key,
|
|
ShillProxy.str2dbus(dbus_class, value))
|
|
|
|
|
|
@classmethod
|
|
def get_proxy(cls, bus=None, timeout_seconds=10):
|
|
"""Create a Proxy, retrying if necessary.
|
|
|
|
This method creates a proxy object of the required subclass of
|
|
ShillProxy. A call to SomeSubclassOfShillProxy.get_proxy() will return
|
|
an object of type SomeSubclassOfShillProxy.
|
|
|
|
Connects to shill over D-Bus. If shill is not yet running,
|
|
retry until it is, or until |timeout_seconds| expires.
|
|
|
|
After connecting to shill, this method will verify that shill
|
|
is answering RPCs. No timeout is applied to the test RPC, so
|
|
this method _may_ block indefinitely.
|
|
|
|
@param bus D-Bus bus to use, or specify None and this object will
|
|
create a mainloop and bus.
|
|
@param timeout_seconds float number of seconds to try connecting
|
|
A value <= 0 will cause the method to return immediately,
|
|
without trying to connect.
|
|
@return a ShillProxy instance if we connected, or None otherwise
|
|
|
|
"""
|
|
end_time = time.time() + timeout_seconds
|
|
connection = None
|
|
while connection is None and time.time() < end_time:
|
|
try:
|
|
# We create instance of class on which this classmethod was
|
|
# called. This way, calling SubclassOfShillProxy.get_proxy()
|
|
# will get a proxy of the right type.
|
|
connection = cls(bus=bus)
|
|
except dbus.exceptions.DBusException as e:
|
|
if e.get_dbus_name() != ShillProxy.DBUS_SERVICE_UNKNOWN:
|
|
raise ShillProxyError('Error connecting to shill')
|
|
else:
|
|
# Wait a moment before retrying
|
|
time.sleep(ShillProxy.POLLING_INTERVAL_SECONDS)
|
|
|
|
if connection is None:
|
|
return None
|
|
|
|
# Although shill is connected to D-Bus at this point, it may
|
|
# not have completed initialization just yet. Call into shill,
|
|
# and wait for the response, to make sure that it is truly up
|
|
# and running. (Shill will not service D-Bus requests until
|
|
# initialization is complete.)
|
|
connection.get_profiles()
|
|
return connection
|
|
|
|
|
|
def __init__(self, bus=None):
|
|
if bus is None:
|
|
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
|
|
bus = dbus.SystemBus()
|
|
self._bus = bus
|
|
self._manager = self.get_dbus_object(self.DBUS_TYPE_MANAGER, '/')
|
|
|
|
|
|
def configure_service_by_guid(self, guid, properties={}):
|
|
"""Configure a service identified by its GUID.
|
|
|
|
@param guid string unique identifier of service.
|
|
@param properties dictionary of service property:value pairs.
|
|
|
|
"""
|
|
config = properties.copy()
|
|
config[self.SERVICE_PROPERTY_GUID] = guid
|
|
self.configure_service(config)
|
|
|
|
|
|
def configure_service(self, config):
|
|
"""Configure a service with given properties.
|
|
|
|
@param config dictionary of service property:value pairs.
|
|
@return DBus object interface representing configured Service.
|
|
|
|
"""
|
|
# Convert configuration values to dbus variant typed values.
|
|
dbus_config = ShillProxy.service_properties_to_dbus_types(config)
|
|
path = self.manager.ConfigureService(dbus_config)
|
|
return self.get_dbus_object(self.DBUS_TYPE_SERVICE, path)
|
|
|
|
|
|
def configure_service_for_profile(self, path, config):
|
|
"""Configure a service in the given profile with given properties.
|
|
|
|
@param path string path of profile for which service should be
|
|
configured.
|
|
@param config dictionary of service property:value pairs.
|
|
|
|
"""
|
|
# Convert configuration values to dbus variant typed values.
|
|
dbus_config = ShillProxy.service_properties_to_dbus_types(config)
|
|
self.manager.ConfigureServiceForProfile(dbus.ObjectPath(path),
|
|
dbus_config)
|
|
|
|
|
|
def set_logging(self, level, scopes):
|
|
"""Set the logging in shill to the specified |level| and |scopes|.
|
|
|
|
@param level int log level to set to in shill.
|
|
@param scopes list of strings of log scopes to set to in shill.
|
|
|
|
"""
|
|
self.manager.SetDebugLevel(level)
|
|
self.manager.SetDebugTags('+'.join(scopes))
|
|
|
|
|
|
def set_logging_for_test(self, technology):
|
|
"""Set the logging in shill for a test of the specified |technology|.
|
|
|
|
Set the log level to |LOG_LEVEL_FOR_TEST| and the log scopes to the
|
|
ones defined in |LOG_SCOPES_FOR_TEST| for |technology|. If |technology|
|
|
is not found in |LOG_SCOPES_FOR_TEST|, the log scopes are set to
|
|
|LOG_SCOPES_FOR_TEST_COMMON|.
|
|
|
|
@param technology string representing the technology type of a test
|
|
that the logging in shill is to be customized for.
|
|
|
|
"""
|
|
scopes = self.LOG_SCOPES_FOR_TEST.get(technology,
|
|
self.LOG_SCOPES_FOR_TEST_COMMON)
|
|
self.set_logging(self.LOG_LEVEL_FOR_TEST, scopes)
|
|
|
|
|
|
def wait_for_property_in(self, dbus_object, property_name,
|
|
expected_values, timeout_seconds):
|
|
"""Wait till a property is in a list of expected values.
|
|
|
|
Block until the property |property_name| in |dbus_object| is in
|
|
|expected_values|, or |timeout_seconds|.
|
|
|
|
@param dbus_object DBus proxy object as returned by
|
|
self.get_dbus_object.
|
|
@param property_name string property key in dbus_object.
|
|
@param expected_values iterable set of values to return successfully
|
|
upon seeing.
|
|
@param timeout_seconds float number of seconds to return if we haven't
|
|
seen the appropriate property value in time.
|
|
@return tuple(successful, final_value, duration)
|
|
where successful is True iff we saw one of |expected_values| for
|
|
|property_name|, final_value is the member of |expected_values| we
|
|
saw, and duration is how long we waited to see that value.
|
|
|
|
"""
|
|
start_time = time.time()
|
|
duration = lambda: time.time() - start_time
|
|
|
|
update_queue = collections.deque()
|
|
signal_receiver = lambda key, value: update_queue.append((key, value))
|
|
receiver_ref = self._bus.add_signal_receiver(
|
|
signal_receiver,
|
|
signal_name='PropertyChanged',
|
|
dbus_interface=dbus_object.dbus_interface,
|
|
path=dbus_object.object_path)
|
|
try:
|
|
# Check to make sure we're not already in a target state.
|
|
try:
|
|
properties = self.dbus2primitive(
|
|
dbus_object.GetProperties(utf8_strings=True))
|
|
last_value = properties.get(property_name, '(no value found)')
|
|
if last_value in expected_values:
|
|
return True, last_value, duration()
|
|
|
|
except dbus.exceptions.DBusException:
|
|
return False, '(object reference became invalid)', duration()
|
|
|
|
context = gobject.MainLoop().get_context()
|
|
while duration() < timeout_seconds:
|
|
# Dispatch all pending events.
|
|
while context.iteration(False):
|
|
pass
|
|
|
|
while update_queue:
|
|
updated_property, value = map(self.dbus2primitive,
|
|
update_queue.popleft())
|
|
if property_name != updated_property:
|
|
continue
|
|
|
|
last_value = value
|
|
if not last_value in expected_values:
|
|
continue
|
|
|
|
return True, last_value, duration()
|
|
|
|
time.sleep(0.2) # Give that CPU a break. CPUs love breaks.
|
|
finally:
|
|
receiver_ref.remove()
|
|
|
|
return False, last_value, duration()
|
|
|
|
|
|
@property
|
|
def manager(self):
|
|
""" @return DBus proxy object representing the shill Manager. """
|
|
return self._manager
|
|
|
|
|
|
def get_active_profile(self):
|
|
"""Get the active profile in shill.
|
|
|
|
@return dbus object representing the active profile.
|
|
|
|
"""
|
|
properties = self.manager.GetProperties(utf8_strings=True)
|
|
return self.get_dbus_object(
|
|
self.DBUS_TYPE_PROFILE,
|
|
properties[self.MANAGER_PROPERTY_ACTIVE_PROFILE])
|
|
|
|
|
|
def get_dbus_object(self, type_str, path):
|
|
"""Return the DBus object of type |type_str| at |path| in shill.
|
|
|
|
@param type_str string (e.g. self.DBUS_TYPE_SERVICE).
|
|
@param path path to object in shill (e.g. '/service/12').
|
|
@return DBus proxy object.
|
|
|
|
"""
|
|
return dbus.Interface(
|
|
self._bus.get_object(self.DBUS_INTERFACE, path,
|
|
introspect=False),
|
|
type_str)
|
|
|
|
|
|
def get_devices(self):
|
|
"""Return the list of devices as dbus Interface objects"""
|
|
properties = self.manager.GetProperties(utf8_strings=True)
|
|
return [self.get_dbus_object(self.DBUS_TYPE_DEVICE, path)
|
|
for path in properties[self.MANAGER_PROPERTY_DEVICES]]
|
|
|
|
|
|
def get_profiles(self):
|
|
"""Return the list of profiles as dbus Interface objects"""
|
|
properties = self.manager.GetProperties(utf8_strings=True)
|
|
return [self.get_dbus_object(self.DBUS_TYPE_PROFILE, path)
|
|
for path in properties[self.MANAGER_PROPERTY_PROFILES]]
|
|
|
|
|
|
def get_service(self, params):
|
|
"""
|
|
Get the shill service that matches |params|.
|
|
|
|
@param params dict of strings understood by shill to describe
|
|
a service.
|
|
@return DBus object interface representing a service.
|
|
|
|
"""
|
|
dbus_params = self.service_properties_to_dbus_types(params)
|
|
path = self.manager.GetService(dbus_params)
|
|
return self.get_dbus_object(self.DBUS_TYPE_SERVICE, path)
|
|
|
|
|
|
def get_service_for_device(self, device):
|
|
"""Attempt to find a service that manages |device|.
|
|
|
|
@param device a dbus object interface representing a device.
|
|
@return Dbus object interface representing a service if found. None
|
|
otherwise.
|
|
|
|
"""
|
|
properties = self.manager.GetProperties(utf8_strings=True)
|
|
all_services = properties.get(self.MANAGER_PROPERTY_ALL_SERVICES,
|
|
None)
|
|
if not all_services:
|
|
return None
|
|
|
|
for service_path in all_services:
|
|
service = self.get_dbus_object(self.DBUS_TYPE_SERVICE,
|
|
service_path)
|
|
properties = service.GetProperties(utf8_strings=True)
|
|
device_path = properties.get(self.SERVICE_PROPERTY_DEVICE, None)
|
|
if device_path == device.object_path:
|
|
return service
|
|
|
|
return None
|
|
|
|
|
|
def find_object(self, object_type, properties):
|
|
"""Find a shill object with the specified type and properties.
|
|
|
|
Return the first shill object of |object_type| whose properties match
|
|
all that of |properties|.
|
|
|
|
@param object_type string representing the type of object to be
|
|
returned. Valid values are those object types defined in
|
|
|OBJECT_TYPE_PROPERTY_MAP|.
|
|
@param properties dict of strings understood by shill to describe
|
|
a service.
|
|
@return DBus object interface representing the object found or None
|
|
if no matching object is found.
|
|
|
|
"""
|
|
if object_type not in self.OBJECT_TYPE_PROPERTY_MAP:
|
|
return None
|
|
|
|
dbus_type, manager_property = self.OBJECT_TYPE_PROPERTY_MAP[object_type]
|
|
manager_properties = self.manager.GetProperties(utf8_strings=True)
|
|
for path in manager_properties[manager_property]:
|
|
try:
|
|
test_object = self.get_dbus_object(dbus_type, path)
|
|
object_properties = test_object.GetProperties(utf8_strings=True)
|
|
for name, value in properties.iteritems():
|
|
if (name not in object_properties or
|
|
self.dbus2primitive(object_properties[name]) != value):
|
|
break
|
|
else:
|
|
return test_object
|
|
|
|
except dbus.exceptions.DBusException, e:
|
|
# This could happen if for instance, you're enumerating services
|
|
# and test_object was removed in shill between the call to get
|
|
# the manager properties and the call to get the service
|
|
# properties. This causes failed method invocations.
|
|
continue
|
|
return None
|
|
|
|
|
|
def find_matching_service(self, properties, only_visible=True):
|
|
"""Find a service object that matches the given properties.
|
|
|
|
This re-implements the manager DBus method FindMatchingService.
|
|
The advantage of doing this here is that FindMatchingServices does
|
|
not exist on older images, which will cause tests to fail.
|
|
|
|
@param properties dict of strings understood by shill to describe
|
|
a service.
|
|
@param only_visible if set to True, restrict the search to services
|
|
that are currently visible.
|
|
|
|
"""
|
|
return self.find_object('Service' if only_visible else 'AnyService',
|
|
properties)
|
|
|
|
|
|
def connect_service_synchronous(self, service, timeout_seconds):
|
|
"""Connect a service and wait for its state to become connected.
|
|
|
|
@param service DBus service object to connect.
|
|
@param timeout_seconds number of seconds to wait for service to go
|
|
enter a connected state.
|
|
@return True if the service connected successfully.
|
|
|
|
"""
|
|
try:
|
|
service.Connect()
|
|
except dbus.exceptions.DBusException as e:
|
|
if e.get_dbus_name() != self.ERROR_ALREADY_CONNECTED:
|
|
raise e
|
|
success, _, _ = self.wait_for_property_in(
|
|
service, self.SERVICE_PROPERTY_STATE,
|
|
self.SERVICE_CONNECTED_STATES,
|
|
timeout_seconds=timeout_seconds)
|
|
return success
|
|
|
|
|
|
def disconnect_service_synchronous(self, service, timeout_seconds):
|
|
"""Disconnect a service and wait for its state to go idle.
|
|
|
|
@param service DBus service object to disconnect.
|
|
@param timeout_seconds number of seconds to wait for service to go idle.
|
|
@return True if the service disconnected successfully.
|
|
|
|
"""
|
|
try:
|
|
service.Disconnect()
|
|
except dbus.exceptions.DBusException as e:
|
|
if e.get_dbus_name() not in [self.ERROR_IN_PROGRESS,
|
|
self.ERROR_NOT_CONNECTED]:
|
|
raise e
|
|
success, _, _ = self.wait_for_property_in(
|
|
service, self.SERVICE_PROPERTY_STATE, ['idle'],
|
|
timeout_seconds=timeout_seconds)
|
|
return success
|
|
|
|
|
|
def get_default_interface_name(self):
|
|
"""Retrieve the name of the default interface.
|
|
|
|
Default interface is determined via the Manager's default service.
|
|
|
|
@return Device name string, or None.
|
|
"""
|
|
service_path = self.get_dbus_property(self.manager,
|
|
self.MANAGER_PROPERTY_DEFAULT_SERVICE)
|
|
if not service_path:
|
|
return None
|
|
service = self.get_dbus_object(self.DBUS_TYPE_SERVICE, service_path)
|
|
device_path = self.get_dbus_property(service,
|
|
self.SERVICE_PROPERTY_DEVICE)
|
|
if not device_path:
|
|
return None
|
|
device = self.get_dbus_object(self.DBUS_TYPE_DEVICE, device_path)
|
|
return self.get_dbus_property(device, self.DEVICE_PROPERTY_INTERFACE)
|