| # Lint as: python2, python3 |
| # 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. |
| |
| from __future__ import absolute_import |
| from __future__ import division |
| from __future__ import print_function |
| |
| import collections |
| import dbus |
| import dbus.mainloop.glib |
| # AU tests use ToT client code, but ToT -3 client version. |
| try: |
| from gi.repository import GObject |
| except ImportError: |
| import gobject as GObject |
| import time |
| import six |
| |
| from six.moves import map |
| from six.moves import range |
| from six import PY2 |
| |
| 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_READY_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 list(in_dict.items()): |
| 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() |
| 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() |
| 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()) |
| 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 = list( |
| 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() |
| 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() |
| 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() |
| 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() |
| 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() |
| 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() |
| for path in manager_properties[manager_property]: |
| try: |
| test_object = self.get_dbus_object(dbus_type, path) |
| object_properties = test_object.GetProperties() |
| for name, value in list(properties.items()): |
| if (name not in object_properties or |
| self.dbus2primitive(object_properties[name]) != value): |
| break |
| else: |
| return test_object |
| |
| except dbus.exceptions.DBusException as _: |
| # 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 |
| |
| # 'ready' might be an intermittent state; poll for a stable state. |
| for _ in range(self.SERVICE_READY_TIMEOUT): |
| success, state, _ = self.wait_for_property_in( |
| service, |
| self.SERVICE_PROPERTY_STATE, |
| self.SERVICE_CONNECTED_STATES, |
| timeout_seconds=timeout_seconds) |
| if state != 'ready': |
| break |
| time.sleep(1) |
| |
| 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) |
