• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
5import collections
6import dbus
7import dbus.service
8import dbus.mainloop.glib
9import gobject
10import logging
11import os
12import threading
13import 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
19MethodCall = collections.namedtuple("MethodCall", ["method", "argument"])
20
21class 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
92class 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
139if __name__ == '__main__':
140    MockLorgnette().run()
141