client/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():
    """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',
                           '/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):
        """Constructor

        @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')
        self.__listen_to_signal(self.__handle_new_policy,
                                'PropertyChangeComplete')


    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='Initial policy push complete.',
            timeout=constants.DEFAULT_OWNERSHIP_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._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()
        bus.add_signal_receiver(
            handler_function=callback,
            signal_name=signal,
            dbus_interface='org.chromium.Chromium',
            bus_name=None,
            path='/org/chromium/SessionManager')


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