1# Lint as: python2, python3 2# Copyright 2014 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 6from __future__ import absolute_import 7from __future__ import division 8from __future__ import print_function 9 10import dbus 11import logging 12import six 13 14import common 15 16from autotest_lib.client.bin import utils 17from autotest_lib.client.common_lib import seven 18 19 20DBUS_INTERFACE_OBJECT_MANAGER = 'org.freedesktop.DBus.ObjectManager' 21DBUS_ERROR_SERVICEUNKNOWN = 'org.freedesktop.DBus.Error.ServiceUnknown' 22 23 24def dbus2primitive(value): 25 """Convert values from dbus types to python types. 26 27 @param value: dbus object to convert to a primitive. 28 29 """ 30 if isinstance(value, dbus.Boolean): 31 return bool(value) 32 elif isinstance(value, int): 33 return int(value) 34 elif isinstance(value, dbus.UInt16): 35 return seven.ensure_long(value) 36 elif isinstance(value, dbus.UInt32): 37 return seven.ensure_long(value) 38 elif isinstance(value, dbus.UInt64): 39 return seven.ensure_long(value) 40 elif isinstance(value, float): 41 return float(value) 42 elif isinstance(value, str): 43 return str(value) 44 elif isinstance(value, six.text_type): 45 return str(value.encode('utf-8')) 46 elif isinstance(value, list): 47 return [dbus2primitive(x) for x in value] 48 elif isinstance(value, tuple): 49 return tuple([dbus2primitive(x) for x in value]) 50 elif isinstance(value, dict): 51 return dict([(dbus2primitive(k), dbus2primitive(v)) 52 for k, v in value.items()]) 53 else: 54 logging.error('Failed to convert dbus object of class: %r', 55 value.__class__.__name__) 56 return value 57 58 59def get_objects_with_interface(service_name, object_manager_path, 60 dbus_interface, path_prefix=None, 61 bus=None): 62 """Get objects that have a particular interface via a property manager. 63 64 @param service_name: string remote service exposing the object manager 65 to query (e.g. 'org.chromium.peerd'). 66 @param object_manager_path: string DBus path of object manager on remote 67 service (e.g. '/org/chromium/peerd') 68 @param dbus_interface: string interface of object we're interested in. 69 @param path_prefix: string prefix of DBus path to filter for. If not 70 None, we'll return only objects in the remote service whose 71 paths start with this prefix. 72 @param bus: dbus.Bus object, defaults to dbus.SystemBus(). Note that 73 normally, dbus.SystemBus() multiplexes a single DBus connection 74 among its instances. 75 @return dict that maps object paths to dicts of interface name to properties 76 exposed by that interface. This is similar to the structure 77 returned by org.freedesktop.DBus.ObjectManaber.GetManagedObjects(). 78 79 """ 80 if bus is None: 81 bus = dbus.SystemBus() 82 object_manager = dbus.Interface( 83 bus.get_object(service_name, object_manager_path), 84 dbus_interface=DBUS_INTERFACE_OBJECT_MANAGER) 85 objects = dbus2primitive(object_manager.GetManagedObjects()) 86 logging.debug('Saw objects %r', objects) 87 # Filter by interface. 88 objects = [(path, interfaces) 89 for path, interfaces in six.iteritems(objects) 90 if dbus_interface in interfaces] 91 if path_prefix is not None: 92 objects = [(path, interfaces) 93 for path, interfaces in objects 94 if path.startswith(path_prefix)] 95 objects = dict(objects) 96 logging.debug('Filtered objects: %r', objects) 97 return objects 98 99def get_dbus_object(bus, service_name, object_manager_path, timeout=None): 100 """Keeps trying to get the a DBus object until a timeout expires. 101 Useful if a test should wait for a system daemon to start up. 102 103 @param bus: dbus.Bus object. 104 @param service_name: string service to look up (e.g. 'org.chromium.peerd'). 105 @param object_manager_path: string DBus path of object manager on remote 106 service (e.g. '/org/chromium/peerd') 107 @param timeout: maximum time in seconds to wait for the bus object. 108 @return The DBus object or None if the timeout expired. 109 110 """ 111 112 def try_get_object(): 113 try: 114 return bus.get_object(service_name, object_manager_path) 115 except dbus.exceptions.DBusException as e: 116 # Only handle DBUS_ERROR_SERVICEUNKNOWN, which is thrown when the 117 # service is not running yet. Otherwise, rethrow. 118 if e.get_dbus_name() == DBUS_ERROR_SERVICEUNKNOWN: 119 return None 120 raise 121 122 return utils.poll_for_condition( 123 condition=try_get_object, 124 desc='Get bus object "%s" / "%s"' % (service_name, 125 object_manager_path), 126 timeout=timeout or 0) 127