Home | History | Annotate | Download | only in documentscan_AppTestWithFakeLorgnette 
      1 # Copyright 2015 The Chromium OS Authors. All rights reserved.
      2 # Use of this source code is governed by a BSD-style license that can be
      3 # found in the LICENSE file.
      4 
      5 import collections
      6 import dbus
      7 import dbus.service
      8 import dbus.mainloop.glib
      9 import gobject
     10 import logging
     11 import os
     12 import threading
     13 import time
     14 
     15 """ MockLorgnette provides mocked methods from the lorgnette
     16     D-Bus API so that we can perform an image scan operation in
     17     Chrome without access to a physical scanner. """
     18 
     19 MethodCall = collections.namedtuple("MethodCall", ["method", "argument"])
     20 
     21 class LorgnetteManager(dbus.service.Object):
     22     """ The lorgnette DBus Manager object instance.  Methods in this
     23         object are called whenever a DBus RPC method is invoked. """
     24 
     25     SCANNER_NAME = 'scanner1'
     26     SCANNER_MANUFACTURER = 'Chromascanner'
     27     SCANNER_MODEL = 'Fakebits2000'
     28     SCANNER_TYPE = 'Virtual'
     29 
     30     def __init__(self, bus, object_path, scan_image_data):
     31         dbus.service.Object.__init__(self, bus, object_path)
     32         self.method_calls = []
     33         self.scan_image_data = scan_image_data
     34 
     35 
     36     @dbus.service.method('org.chromium.lorgnette.Manager',
     37                          in_signature='', out_signature='a{sa{ss}}')
     38     def ListScanners(self):
     39         """Lists available scanners. """
     40         self.add_method_call('ListScanners', '')
     41         return { self.SCANNER_NAME: {
     42                        'Manufacturer': self.SCANNER_MANUFACTURER,
     43                        'Model': self.SCANNER_MODEL,
     44                        'Type': self.SCANNER_TYPE }}
     45 
     46 
     47     @dbus.service.method('org.chromium.lorgnette.Manager',
     48                          in_signature='sha{sv}', out_signature='')
     49     def ScanImage(self, device, out_fd, scan_properties):
     50         """Writes test image date to |out_fd|.  Do so in chunks since the
     51         entire dataset cannot be successfully written at once.
     52 
     53         @param device string name of the device to scan from.
     54         @param out_fd file handle for the output scan data.
     55         @param scan_properties dict containing parameters for the scan.
     56 
     57         """
     58         self.add_method_call('ScanImage', (device, scan_properties))
     59         scan_output_fd = out_fd.take()
     60         os.write(scan_output_fd, self.scan_image_data)
     61         os.close(scan_output_fd)
     62 
     63         # TODO(pstew): Ensure the timing between return of this method
     64         # and the EOF returned to Chrome at the end of this data stream
     65         # are distinct.  This comes naturally with a real scanner.
     66         time.sleep(1)
     67 
     68 
     69     def add_method_call(self, method, arg):
     70         """Note that a method call was made.
     71 
     72         @param method string the method that was called.
     73         @param arg tuple list of arguments that were called on |method|.
     74 
     75         """
     76         logging.info("Mock Lorgnette method %s called with argument %s",
     77                      method, arg)
     78         self.method_calls.append(MethodCall(method, arg))
     79 
     80 
     81     def get_method_calls(self):
     82         """Provide the method call list, clears this list internally.
     83 
     84         @return list of MethodCall objects
     85 
     86         """
     87         method_calls = self.method_calls
     88         self.method_calls = []
     89         return method_calls
     90 
     91 
     92 class MockLorgnette(threading.Thread):
     93     """This thread object instantiates a mock lorgnette manager and
     94     runs a mainloop that receives DBus API messages. """
     95     LORGNETTE = "org.chromium.lorgnette"
     96     def __init__(self, image_file):
     97         threading.Thread.__init__(self)
     98         gobject.threads_init()
     99         self.image_file = image_file
    100 
    101 
    102     def __enter__(self):
    103         self.start()
    104         return self
    105 
    106 
    107     def __exit__(self, type, value, tb):
    108         self.quit()
    109         self.join()
    110 
    111 
    112     def run(self):
    113         """Runs the main loop."""
    114         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    115         self.bus = dbus.SystemBus()
    116         name = dbus.service.BusName(self.LORGNETTE, self.bus)
    117         with open(self.image_file) as f:
    118             self.image_data = f.read()
    119         self.manager = LorgnetteManager(
    120                 self.bus, '/org/chromium/lorgnette/Manager', self.image_data)
    121         self.mainloop = gobject.MainLoop()
    122         self.mainloop.run()
    123 
    124 
    125     def quit(self):
    126         """Quits the main loop."""
    127         self.mainloop.quit()
    128 
    129 
    130     def get_method_calls(self):
    131         """Returns the method calls that were called on the mock object.
    132 
    133         @return list of MethodCall objects representing the methods called.
    134 
    135          """
    136         return self.manager.get_method_calls()
    137 
    138 
    139 if __name__ == '__main__':
    140     MockLorgnette().run()
    141