| # 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(bus_loop): |
| """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. |
| |
| @param bus_loop: An externally-owned DBusGMainLoop. |
| |
| @return a dbus.Interface object connection to the session_manager. |
| """ |
| bus = dbus.SystemBus(mainloop=bus_loop) |
| proxy = bus.get_object('org.chromium.SessionManager', |
| '/org/chromium/SessionManager') |
| return dbus.Interface(proxy, 'org.chromium.SessionManagerInterface') |
| |
| |
| class SignalListener(object): |
| """A class to listen for DBus signals from the session manager. |
| |
| The session_manager emits several DBus signals when different events |
| of interest occur. This class provides a framework for derived classes |
| to use to listen for certain signals. |
| """ |
| |
| def __init__(self, g_main_loop): |
| """Constructor |
| |
| @param g_mail_loop: glib main loop object. |
| """ |
| self._main_loop = g_main_loop |
| |
| |
| def wait_for_signals(self, desc, |
| timeout=constants.DEFAULT_OWNERSHIP_TIMEOUT): |
| """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. |
| """ |
| utils.poll_for_condition( |
| condition=lambda: self.__received_signals(), |
| desc=desc, |
| timeout=timeout) |
| self.reset_signal_state() |
| |
| |
| 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): |
| pass |
| return self.all_signals_received() |
| |
| |
| def reset_signal_state(self): |
| """Resets internal signal tracking state.""" |
| raise NotImplementedError() |
| |
| |
| def all_signals_received(self): |
| """Resets internal signal tracking state.""" |
| raise NotImplementedError() |
| |
| |
| 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() |
| bus.add_signal_receiver( |
| handler_function=callback, |
| signal_name=signal, |
| dbus_interface='org.chromium.SessionManagerInterface', |
| bus_name=None, |
| path='/org/chromium/SessionManager') |
| |
| |
| |
| class OwnershipSignalListener(SignalListener): |
| """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. |
| """ |
| |
| def __init__(self, g_main_loop): |
| """Constructor |
| |
| @param g_mail_loop: glib main loop object. |
| """ |
| super(OwnershipSignalListener, self).__init__(g_main_loop) |
| self._listen_for_new_key = False |
| self._got_new_key = False |
| self._listen_for_new_policy = False |
| self._got_new_policy = False |
| |
| |
| def listen_for_new_key_and_policy(self): |
| """Set to listen for signals indicating new owner key and device policy. |
| """ |
| self._listen_for_new_key = self._listen_for_new_policy = True |
| self.listen_to_signal(self.__handle_new_key, 'SetOwnerKeyComplete') |
| self.listen_to_signal(self.__handle_new_policy, |
| 'PropertyChangeComplete') |
| self.reset_signal_state() |
| |
| |
| def listen_for_new_policy(self): |
| """Set to listen for signal indicating new device policy. |
| """ |
| self._listen_for_new_key = False |
| self._listen_for_new_policy = True |
| self.listen_to_signal(self.__handle_new_policy, |
| 'PropertyChangeComplete') |
| self.reset_signal_state() |
| |
| |
| def reset_signal_state(self): |
| """Resets internal signal tracking state.""" |
| self._got_new_key = not self._listen_for_new_key |
| self._got_new_policy = not self._listen_for_new_policy |
| |
| |
| def all_signals_received(self): |
| """Returns true when expected signals are all receieved.""" |
| return self._got_new_key and self._got_new_policy |
| |
| |
| def __handle_new_key(self, success): |
| """Callback to be used when a new key signal is received. |
| |
| @param success: the string 'success' if the key was generated. |
| """ |
| self._got_new_key = (success == 'success') |
| |
| |
| def __handle_new_policy(self, success): |
| """Callback to be used when a new policy signal is received. |
| |
| @param success: the string 'success' if the policy was stored. |
| """ |
| self._got_new_policy = (success == 'success') |
| |
| |
| |
| class SessionSignalListener(SignalListener): |
| """A class to listen for SessionStateChanged DBus signals. |
| |
| The session_manager emits a DBus signal whenever a user signs in, when |
| the user session begins termination, and when the session is terminated. |
| This class allows this signal to be polled for |
| """ |
| |
| def __init__(self, g_main_loop): |
| """Constructor |
| |
| @param g_mail_loop: glib main loop object. |
| """ |
| super(SessionSignalListener, self).__init__(g_main_loop) |
| self._new_state = None |
| self._expected_state = None |
| |
| |
| def listen_for_session_state_change(self, expected): |
| """Set to listen for state changed signal with payload == |expected|. |
| |
| @param expected: string representing the state transition we expect. |
| One of 'started', 'stopping', or 'stopped'. |
| """ |
| if expected not in {'started', 'stopping', 'stopped'}: |
| raise ValueError("expected must be one of 'started', 'stopping'," + |
| " or 'stopped'.") |
| self.listen_to_signal(self.__handle_signal, 'SessionStateChanged') |
| self._expected_state = expected |
| |
| |
| def reset_signal_state(self): |
| """Resets internal signal tracking state.""" |
| self._new_state = None |
| |
| |
| def all_signals_received(self): |
| """Returns true when expected signals are all receieved.""" |
| return self._new_state == self._expected_state |
| |
| |
| def __handle_signal(self, state): |
| """Callback to be used when a new state-change signal is received. |
| |
| @param state: the state transition being signaled. |
| """ |
| self._new_state = state |