• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
5import dbus
6import logging
7
8from autotest_lib.client.bin import utils
9
10
11DBUS_INTERFACE_OBJECT_MANAGER = 'org.freedesktop.DBus.ObjectManager'
12DBUS_ERROR_SERVICEUNKNOWN = 'org.freedesktop.DBus.Error.ServiceUnknown'
13
14
15def 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
50def 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
90def 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