1# -*- coding: utf-8 -*-
2
3'''obexd mock template
4
5This creates the expected methods and properties of the object manager
6org.bluez.obex object (/), the manager object (/org/bluez/obex), but no agents
7or clients.
8
9This supports BlueZ 5 only.
10'''
11
12# This program is free software; you can redistribute it and/or modify it under
13# the terms of the GNU Lesser General Public License as published by the Free
14# Software Foundation; either version 3 of the License, or (at your option) any
15# later version.  See http://www.gnu.org/copyleft/lgpl.html for the full text
16# of the license.
17
18__author__ = 'Philip Withnall'
19__copyright__ = '(c) 2013 Collabora Ltd.'
20
21import tempfile
22import os
23
24import dbus
25
26from dbusmock import OBJECT_MANAGER_IFACE, mockobject
27
28BUS_NAME = 'org.bluez.obex'
29MAIN_OBJ = '/'
30SYSTEM_BUS = False
31IS_OBJECT_MANAGER = True
32
33OBEX_MOCK_IFACE = 'org.bluez.obex.Mock'
34AGENT_MANAGER_IFACE = 'org.bluez.AgentManager1'
35CLIENT_IFACE = 'org.bluez.obex.Client1'
36SESSION_IFACE = 'org.bluez.obex.Session1'
37PHONEBOOK_ACCESS_IFACE = 'org.bluez.obex.PhonebookAccess1'
38TRANSFER_IFACE = 'org.bluez.obex.Transfer1'
39TRANSFER_MOCK_IFACE = 'org.bluez.obex.transfer1.Mock'
40
41
42def load(mock, _parameters):
43    mock.AddObject('/org/bluez/obex', AGENT_MANAGER_IFACE, {}, [
44        ('RegisterAgent', 'os', '', ''),
45        ('UnregisterAgent', 'o', '', ''),
46    ])
47
48    obex = mockobject.objects['/org/bluez/obex']
49    obex.AddMethods(CLIENT_IFACE, [
50        ('CreateSession', 'sa{sv}', 'o', CreateSession),
51        ('RemoveSession', 'o', '', RemoveSession),
52    ])
53
54
55@dbus.service.method(CLIENT_IFACE,
56                     in_signature='sa{sv}', out_signature='o')
57def CreateSession(self, destination, args):
58    '''OBEX method to create a new transfer session.
59
60    The destination must be the address of the destination Bluetooth device.
61    The given arguments must be a map from well-known keys to values,
62    containing at least the ‘Target’ key, whose value must be ‘PBAP’ (other
63    keys and values are accepted by the real daemon, but not by this mock
64    daemon at the moment). If the target is missing or incorrect, an
65    Unsupported error is returned on the bus.
66
67    Returns the path of a new Session object.
68    '''
69
70    if 'Target' not in args or args['Target'].upper() != 'PBAP':
71        raise dbus.exceptions.DBusException(
72            'Non-PBAP targets are not currently supported by this python-dbusmock template.',
73            name=OBEX_MOCK_IFACE + '.Unsupported')
74
75    # Find the first unused session ID.
76    client_path = '/org/bluez/obex/client'
77    session_id = 0
78    while client_path + '/session' + str(session_id) in mockobject.objects:
79        session_id += 1
80
81    path = client_path + '/session' + str(session_id)
82    properties = {
83        'Source': dbus.String('FIXME', variant_level=1),
84        'Destination': dbus.String(destination, variant_level=1),
85        'Channel': dbus.Byte(0, variant_level=1),
86        'Target': dbus.String('FIXME', variant_level=1),
87        'Root': dbus.String('FIXME', variant_level=1),
88    }
89
90    self.AddObject(path,
91                   SESSION_IFACE,
92                   # Properties
93                   properties,
94                   # Methods
95                   [
96                       ('GetCapabilities', '', 's', ''),  # Currently a no-op
97                   ])
98
99    session = mockobject.objects[path]
100    session.AddMethods(PHONEBOOK_ACCESS_IFACE, [
101        ('Select', 'ss', '', ''),  # Currently a no-op
102        # Currently a no-op
103        ('List', 'a{sv}', 'a(ss)', 'ret = dbus.Array(signature="(ss)")'),
104        # Currently a no-op
105        ('ListFilterFields', '', 'as', 'ret = dbus.Array(signature="(s)")'),
106        ('PullAll', 'sa{sv}', 'sa{sv}', PullAll),
107        ('GetSize', '', 'q', 'ret = 0'),
108    ])
109
110    manager = mockobject.objects['/']
111    manager.EmitSignal(OBJECT_MANAGER_IFACE, 'InterfacesAdded',
112                       'oa{sa{sv}}', [
113                           dbus.ObjectPath(path),
114                           {SESSION_IFACE: properties},
115                       ])
116
117    return path
118
119
120@dbus.service.method(CLIENT_IFACE,
121                     in_signature='o', out_signature='')
122def RemoveSession(self, session_path):
123    '''OBEX method to remove an existing transfer session.
124
125    This takes the path of the transfer Session object and removes it.
126    '''
127
128    manager = mockobject.objects['/']
129
130    # Remove all the session's transfers.
131    transfer_id = 0
132    while session_path + '/transfer' + str(transfer_id) in mockobject.objects:
133        transfer_path = session_path + '/transfer' + str(transfer_id)
134        transfer_id += 1
135
136        self.RemoveObject(transfer_path)
137
138        manager.EmitSignal(OBJECT_MANAGER_IFACE, 'InterfacesRemoved',
139                           'oas', [
140                               dbus.ObjectPath(transfer_path),
141                               [TRANSFER_IFACE],
142                           ])
143
144    # Remove the session itself.
145    self.RemoveObject(session_path)
146
147    manager.EmitSignal(OBJECT_MANAGER_IFACE, 'InterfacesRemoved',
148                       'oas', [
149                           dbus.ObjectPath(session_path),
150                           [SESSION_IFACE, PHONEBOOK_ACCESS_IFACE],
151                       ])
152
153
154@dbus.service.method(PHONEBOOK_ACCESS_IFACE,
155                     in_signature='sa{sv}', out_signature='sa{sv}')
156def PullAll(self, target_file, filters):
157    '''OBEX method to start a pull transfer of a phone book.
158
159    This doesn't complete the transfer; code to mock up activating and
160    completing the transfer must be provided by the test driver, as it’s
161    too complex and test-specific to put here.
162
163    The target_file is the absolute path for a file which will have zero or
164    more vCards, separated by new-line characters, written to it if the method
165    is successful (and the transfer is completed). This target_file is actually
166    emitted in a TransferCreated signal, which is a special part of the mock
167    interface designed to be handled by the test driver, which should then
168    populate that file and call UpdateStatus on the Transfer object. The test
169    driver is responsible for deleting the file once the test is complete.
170
171    The filters parameter is a map of filters to be applied to the results
172    device-side before transmitting them back to the adapter.
173
174    Returns a tuple containing the path for a new Transfer D-Bus object
175    representing the transfer, and a map of the initial properties of that
176    Transfer object.
177    '''
178
179    # Find the first unused session ID.
180    session_path = self.path
181    transfer_id = 0
182    while session_path + '/transfer' + str(transfer_id) in mockobject.objects:
183        transfer_id += 1
184
185    transfer_path = session_path + '/transfer' + str(transfer_id)
186
187    # Create a new temporary file to transfer to.
188    with tempfile.NamedTemporaryFile(suffix='.vcf',
189                                     prefix='tmp-bluez5-obex-PullAll_',
190                                     delete=False) as temp_file:
191        filename = os.path.abspath(temp_file.name)
192
193    props = {
194        'Status': dbus.String('queued', variant_level=1),
195        'Session': dbus.ObjectPath(session_path,
196                                   variant_level=1),
197        'Name': dbus.String(target_file, variant_level=1),
198        'Filename': dbus.String(filename, variant_level=1),
199        'Transferred': dbus.UInt64(0, variant_level=1),
200    }
201
202    self.AddObject(transfer_path,
203                   TRANSFER_IFACE,
204                   # Properties
205                   props,
206                   # Methods
207                   [
208                       ('Cancel', '', '', ''),  # Currently a no-op
209                   ])
210
211    transfer = mockobject.objects[transfer_path]
212    transfer.AddMethods(TRANSFER_MOCK_IFACE, [
213        ('UpdateStatus', 'b', '', UpdateStatus),
214    ])
215
216    manager = mockobject.objects['/']
217    manager.EmitSignal(OBJECT_MANAGER_IFACE, 'InterfacesAdded',
218                       'oa{sa{sv}}', [
219                           dbus.ObjectPath(transfer_path),
220                           {TRANSFER_IFACE: props},
221                       ])
222
223    # Emit a behind-the-scenes signal that a new transfer has been created.
224    manager.EmitSignal(OBEX_MOCK_IFACE, 'TransferCreated', 'sa{sv}s',
225                       [transfer_path, filters, filename])
226
227    return (transfer_path, props)
228
229
230@dbus.service.signal(OBEX_MOCK_IFACE, signature='sa{sv}s')
231def TransferCreated(_self, _path, _filters, _transfer_filename):
232    '''Mock signal emitted when a new Transfer object is created.
233
234    This is not part of the BlueZ OBEX interface; it is purely for use by test
235    driver code. It is emitted by the PullAll method, and is intended to be
236    used as a signal to call UpdateStatus on the newly created Transfer
237    (potentially after a timeout).
238
239    The path is of the new Transfer object, and the filters are as provided to
240    PullAll. The transfer filename is the full path and filename of a newly
241    created empty temporary file which the test driver should write to.
242
243    The test driver should then call UpdateStatus on the Transfer object each
244    time a write is made to the transfer file. The final call to UpdateStatus
245    should mark the transfer as complete.
246
247    The test driver is responsible for deleting the transfer file once the test
248    is complete.
249
250    FIXME: Ideally the UpdateStatus method would only be used for completion,
251    and all intermediate updates would be found by watching the size of the
252    transfer file. However, that means adding a dependency on an inotify
253    package, which seems a little much.
254    '''
255
256
257@dbus.service.method(TRANSFER_MOCK_IFACE,
258                     in_signature='b', out_signature='')
259def UpdateStatus(self, is_complete):
260    '''Mock method to update the transfer status.
261
262    If is_complete is False, this marks the transfer is active; otherwise it
263    marks the transfer as complete. It is an error to call this method after
264    calling it with is_complete as True.
265
266    In both cases, it updates the number of bytes transferred to be the current
267    size of the transfer file (whose filename was emitted in the
268    TransferCreated signal).
269    '''
270    status = 'complete' if is_complete else 'active'
271    transferred = os.path.getsize(self.props[TRANSFER_IFACE]['Filename'])
272
273    self.props[TRANSFER_IFACE]['Status'] = status
274    self.props[TRANSFER_IFACE]['Transferred'] = dbus.UInt64(transferred, variant_level=1)
275
276    self.EmitSignal(dbus.PROPERTIES_IFACE, 'PropertiesChanged', 'sa{sv}as', [
277        TRANSFER_IFACE,
278        {
279            'Status': dbus.String(status, variant_level=1),
280            'Transferred': dbus.UInt64(transferred, variant_level=1),
281        },
282        [],
283    ])
284