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