blob: 521ee93f7a95d94542cec99eb676da692916f3c7 [file] [log] [blame]
# 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_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
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 }}
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)
# 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.
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|.
""""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):
self.image_file = image_file
def __enter__(self):
return self
def __exit__(self, type, value, tb):
def run(self):
"""Runs the main loop."""
self.bus = dbus.SystemBus()
name = dbus.service.BusName(self.LORGNETTE, self.bus)
with open(self.image_file) as f:
self.image_data =
self.manager = LorgnetteManager(
self.bus, '/org/chromium/lorgnette/Manager', self.image_data)
self.mainloop = gobject.MainLoop()
def quit(self):
"""Quits the main loop."""
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__':