common_lib/cros/session_manager.py

# 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.
        """
        self.__flush()
        return self.all_signals_received()


    def __flush(self):
        """Runs the main loop until pending events are done."""
        context = self._main_loop.get_context()
        while context.iteration(False):
            pass


    def reset(self):
        """Prepares the listener to receive a new signal.

        This resets the signal state and flushes any pending signals in order to
        avoid picking up stale signals still lingering in the process' input
        queues.
        """
        self.__flush()
        self.reset_signal_state()


    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()


    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()


    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


class ScreenIsLockedSignalListener(SignalListener):
    """A class to listen for ScreenIsLocked DBus signal.

    The session_manager emits a DBus signal when screen lock operation is
    completed.
    """

    def __init__(self, g_main_loop):
        """Constructor

        @param g_main_loop: glib main loop object.
        """
        super(ScreenIsLockedSignalListener, self).__init__(g_main_loop)
        self._screen_is_locked_received = False
        self.listen_to_signal(self.__handle_signal, 'ScreenIsLocked')


    def reset_signal_state(self):
        """Resets internal signal tracking state."""
        self._screen_is_locked_received = False


    def all_signals_received(self):
        """Returns true when expected signals are all receieved."""
        return self._screen_is_locked_received


    def __handle_signal(self):
        """Callback to be used when ScreenIsLocked signal is received.
        """
        self._screen_is_locked_received = True