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