blob: a4e7461b2cfcc8b4d2f365b5982a34773a43fadc [file] [log] [blame]
# 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 dbus
from autotest_lib.client.bin import utils
from autotest_lib.client.cros import constants
def connect():
"""Create and return a DBus connection to session_manager.
Connects to the session manager over the DBus system bus. Returns
appropriately configured DBus interface object.
@return a dbus.Interface object connection to the session_manager.
bus = dbus.SystemBus()
proxy = bus.get_object('org.chromium.SessionManager',
return dbus.Interface(proxy, 'org.chromium.SessionManagerInterface')
class SignalListener(object):
"""A class to listen for ownership-related DBus signals.
The session_manager emits a couple of DBus signals when certain events
related to device ownership occur. This class provides a way to
listen for them and check on their status.
_got_new_key = False
_got_new_policy = False
_main_loop = None
def __init__(self, g_main_loop):
@param g_mail_loop: glib main loop object.
self._main_loop = g_main_loop
def listen_for_new_key_and_policy(self):
"""Set to listen for signals indicating new owner key and device policy.
self.__listen_to_signal(self.__handle_new_key, 'SetOwnerKeyComplete')
def wait_for_signals(self, desc,
"""Block for |timeout| seconds waiting for the signals to come in.
@param desc: string describing the high-level reason you're waiting
for the signals.
@param timeout: maximum seconds to wait for the signals.
@raises TimeoutError if the timeout is hit.
condition=lambda: self.__received_signals(),
desc='Initial policy push complete.',
def __received_signals(self):
"""Run main loop until all pending events are done, checks for signals.
Runs self._main_loop until it says it has no more events pending,
then returns the state of the internal variables tracking whether
desired signals have been received.
@return True if both signals have been handled, False otherwise.
context = self._main_loop.get_context()
while context.iteration(False):
return self._got_new_key and self._got_new_policy
def __reset_signal_state(self):
"""Resets internal signal tracking state."""
self._got_new_policy = self._got_new_key = False
def __listen_to_signal(self, callback, signal):
"""Connect a callback to a given session_manager dbus signal.
Sets up a signal receiver for signal, and calls the provided callback
when it comes in.
@param callback: a callable to call when signal is received.
@param signal: the signal to listen for.
bus = dbus.SystemBus()
def __handle_new_key(self, success):
"""Callback to be used when a new key signal is received."""
self._got_new_key = (success == 'success')
def __handle_new_policy(self, success):
"""Callback to be used when a new policy signal is received."""
self._got_new_policy = (success == 'success')