client/site_tests/documentscan_AppTestWithFakeLorgnette/mock_lorgnette.py

# Copyright 2015 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 collections
import dbus
import dbus.service
import dbus.mainloop.glib
import gobject
import logging
import os
import threading
import time

""" MockLorgnette provides mocked methods from the lorgnette
    D-Bus API so that we can perform an image scan operation in
    Chrome without access to a physical scanner. """

MethodCall = collections.namedtuple("MethodCall", ["method", "argument"])

class LorgnetteManager(dbus.service.Object):
    """ The lorgnette DBus Manager object instance.  Methods in this
        object are called whenever a DBus RPC method is invoked. """

    SCANNER_NAME = 'scanner1'
    SCANNER_MANUFACTURER = 'Chromascanner'
    SCANNER_MODEL = 'Fakebits2000'
    SCANNER_TYPE = 'Virtual'

    def __init__(self, bus, object_path, scan_image_data):
        dbus.service.Object.__init__(self, bus, object_path)
        self.method_calls = []
        self.scan_image_data = scan_image_data


    @dbus.service.method('org.chromium.lorgnette.Manager',
                         in_signature='', out_signature='a{sa{ss}}')
    def ListScanners(self):
        """Lists available scanners. """
        self.add_method_call('ListScanners', '')
        return { self.SCANNER_NAME: {
                       'Manufacturer': self.SCANNER_MANUFACTURER,
                       'Model': self.SCANNER_MODEL,
                       'Type': self.SCANNER_TYPE }}


    @dbus.service.method('org.chromium.lorgnette.Manager',
                         in_signature='sha{sv}', out_signature='')
    def ScanImage(self, device, out_fd, scan_properties):
        """Writes test image date to |out_fd|.  Do so in chunks since the
        entire dataset cannot be successfully written at once.

        @param device string name of the device to scan from.
        @param out_fd file handle for the output scan data.
        @param scan_properties dict containing parameters for the scan.

        """
        self.add_method_call('ScanImage', (device, scan_properties))
        scan_output_fd = out_fd.take()
        os.write(scan_output_fd, self.scan_image_data)
        os.close(scan_output_fd)

        # TODO(pstew): Ensure the timing between return of this method
        # and the EOF returned to Chrome at the end of this data stream
        # are distinct.  This comes naturally with a real scanner.
        time.sleep(1)


    def add_method_call(self, method, arg):
        """Note that a method call was made.

        @param method string the method that was called.
        @param arg tuple list of arguments that were called on |method|.

        """
        logging.info("Mock Lorgnette method %s called with argument %s",
                     method, arg)
        self.method_calls.append(MethodCall(method, arg))


    def get_method_calls(self):
        """Provide the method call list, clears this list internally.

        @return list of MethodCall objects

        """
        method_calls = self.method_calls
        self.method_calls = []
        return method_calls


class MockLorgnette(threading.Thread):
    """This thread object instantiates a mock lorgnette manager and
    runs a mainloop that receives DBus API messages. """
    LORGNETTE = "org.chromium.lorgnette"
    def __init__(self, image_file):
        threading.Thread.__init__(self)
        gobject.threads_init()
        self.image_file = image_file


    def __enter__(self):
        self.start()
        return self


    def __exit__(self, type, value, tb):
        self.quit()
        self.join()


    def run(self):
        """Runs the main loop."""
        dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
        self.bus = dbus.SystemBus()
        name = dbus.service.BusName(self.LORGNETTE, self.bus)
        with open(self.image_file) as f:
            self.image_data = f.read()
        self.manager = LorgnetteManager(
                self.bus, '/org/chromium/lorgnette/Manager', self.image_data)
        self.mainloop = gobject.MainLoop()
        self.mainloop.run()


    def quit(self):
        """Quits the main loop."""
        self.mainloop.quit()


    def get_method_calls(self):
        """Returns the method calls that were called on the mock object.

        @return list of MethodCall objects representing the methods called.

         """
        return self.manager.get_method_calls()


if __name__ == '__main__':
    MockLorgnette().run()