From f37facb3bd7c53d3b7629cb53b22a05a952bfa27 Mon Sep 17 00:00:00 2001 From: Marcel Holtmann Date: Mon, 10 Dec 2012 23:01:18 +0100 Subject: [PATCH] test: Add test scripts from obexd repository --- test/exchange-business-cards | 19 ++++ test/ftp-client | 183 +++++++++++++++++++++++++++++++ test/get-obex-capabilities | 19 ++++ test/list-folders | 38 +++++++ test/map-client | 207 +++++++++++++++++++++++++++++++++++ test/opp-client | 121 ++++++++++++++++++++ test/pbap-client | 157 ++++++++++++++++++++++++++ test/simple-obex-agent | 132 ++++++++++++++++++++++ 8 files changed, 876 insertions(+) create mode 100755 test/exchange-business-cards create mode 100755 test/ftp-client create mode 100755 test/get-obex-capabilities create mode 100755 test/list-folders create mode 100755 test/map-client create mode 100755 test/opp-client create mode 100755 test/pbap-client create mode 100755 test/simple-obex-agent diff --git a/test/exchange-business-cards b/test/exchange-business-cards new file mode 100755 index 000000000..aace07b65 --- /dev/null +++ b/test/exchange-business-cards @@ -0,0 +1,19 @@ +#!/usr/bin/python + +import sys +import dbus + +bus = dbus.SessionBus() +client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"), + "org.bluez.obex.Client") + +if (len(sys.argv) < 4): + print "Usage: %s " % (sys.argv[0]) + sys.exit(1) + +print "Creating Session" +path = client.CreateSession(sys.argv[1], { "Target": "OPP" }) +opp = dbus.Interface(bus.get_object("org.bluez.obex.client", path), + "org.bluez.obex.ObjectPush") + +opp.ExchangeBusinessCards(sys.argv[2], sys.argv[3]) diff --git a/test/ftp-client b/test/ftp-client new file mode 100755 index 000000000..064a81ae7 --- /dev/null +++ b/test/ftp-client @@ -0,0 +1,183 @@ +#!/usr/bin/python + +import gobject + +import sys +import dbus +import dbus.service +import dbus.mainloop.glib +import os.path +from optparse import OptionParser + +def parse_options(): + parser.add_option("-d", "--device", dest="device", + help="Device to connect", metavar="DEVICE") + parser.add_option("-c", "--chdir", dest="new_dir", + help="Change current directory to DIR", metavar="DIR") + parser.add_option("-l", "--list", action="store_true", dest="list_dir", + help="List the current directory") + parser.add_option("-g", "--get", dest="get_file", + help="Get FILE", metavar="FILE") + parser.add_option("-p", "--put", dest="put_file", + help="Put FILE", metavar="FILE") + parser.add_option("-y", "--copy", dest="copy_file", + help="Copy FILE", metavar="FILE") + parser.add_option("-m", "--move", dest="move_file", + help="Move FILE", metavar="FILE") + parser.add_option("-n", "--destname", dest="dest_file", + help="Destination FILE", metavar="FILE") + parser.add_option("-r", "--remove", dest="remove_file", + help="Remove FILE", metavar="FILE") + parser.add_option("-v", "--verbose", action="store_true", + dest="verbose") + + return parser.parse_args() + +class FtpClient: + def __init__(self, session_path, verbose=False): + self.progress = 0 + self.transfer_path = None + self.transfer_size = 0 + self.verbose = verbose + bus = dbus.SessionBus() + obj = bus.get_object("org.bluez.obex.client", session_path) + self.session = dbus.Interface(obj, "org.bluez.obex.Session") + self.ftp = dbus.Interface(obj, "org.bluez.obex.FileTransfer") + bus.add_signal_receiver(self.transfer_complete, + dbus_interface="org.bluez.obex.Transfer", + signal_name="Complete", + path_keyword="path") + bus.add_signal_receiver(self.transfer_error, + dbus_interface="org.bluez.obex.Transfer", + signal_name="Error", + path_keyword="path") + if self.verbose: + bus.add_signal_receiver(self.transfer_progress, + dbus_interface="org.bluez.obex.Transfer", + signal_name="PropertyChanged", + path_keyword="path") + + def create_transfer_reply(self, reply): + (path, properties) = reply + self.transfer_path = path + self.transfer_size = properties["Size"] + if self.verbose: + print "Transfer created: %s" % path + + def generic_reply(self): + if self.verbose: + print "Operation succeeded" + + def error(self, err): + print err + mainloop.quit() + + def transfer_complete(self, path): + if path != self.transfer_path: + return + if self.verbose: + print "Transfer finished" + mainloop.quit() + + def transfer_error(self, code, message, path): + if path != self.transfer_path: + return + print "Transfer finished with error %s: %s" % (code, message) + mainloop.quit() + + def transfer_progress(self, prop, value, path): + if path != self.transfer_path: + return + + if prop != "Progress": + return + + speed = (value - self.progress) / 1000 + print "Transfer progress %d/%d at %d kBps" % (value, + self.transfer_size, + speed) + self.progress = value + + def change_folder(self, new_dir): + for node in new_dir.split("/"): + self.ftp.ChangeFolder(node) + + def list_folder(self): + for i in self.ftp.ListFolder(): + if i["Type"] == "folder": + print "%s/" % (i["Name"]) + else: + print "%s" % (i["Name"]) + + def put_file(self, filename): + self.ftp.PutFile(os.path.abspath(filename), + os.path.basename(filename), + reply_handler=self.create_transfer_reply, + error_handler=self.error) + + def get_file(self, filename): + self.ftp.GetFile(os.path.abspath(filename), + os.path.basename(filename), + reply_handler=self.create_transfer_reply, + error_handler=self.error) + + def remove_file(self, filename): + self.ftp.Delete(filename, + reply_handler=self.generic_reply, + error_handler=self.error) + + def move_file(self, filename, destname): + self.ftp.MoveFile(filename, destname, + reply_handler=self.generic_reply, + error_handler=self.error) + + def copy_file(self, filename, destname): + self.ftp.CopyFile(filename, destname, + reply_handler=self.generic_reply, + error_handler=self.error) + +if __name__ == '__main__': + + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + + parser = OptionParser() + + (options, args) = parse_options() + + if not options.device: + parser.print_help() + sys.exit(0) + + bus = dbus.SessionBus() + mainloop = gobject.MainLoop() + + client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"), + "org.bluez.obex.Client") + + print "Creating Session" + path = client.CreateSession(options.device, { "Target": "ftp" }) + + ftp_client = FtpClient(path, options.verbose) + + if options.new_dir: + ftp_client.change_folder(options.new_dir) + + if options.list_dir: + ftp_client.list_folder() + + if options.get_file: + ftp_client.get_file(options.get_file) + + if options.put_file: + ftp_client.put_file(options.put_file) + + if options.move_file: + ftp_client.move_file(options.move_file, options.dest_file) + + if options.copy_file: + ftp_client.copy_file(options.copy_file, options.dest_file) + + if options.remove_file: + ftp_client.remove_file(options.remove_file) + + mainloop.run() diff --git a/test/get-obex-capabilities b/test/get-obex-capabilities new file mode 100755 index 000000000..a06f2eb91 --- /dev/null +++ b/test/get-obex-capabilities @@ -0,0 +1,19 @@ +#!/usr/bin/python + +import sys +import dbus + +bus = dbus.SessionBus() +client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"), + "org.bluez.obex.Client") + +if (len(sys.argv) < 3): + print "Usage: %s " % (sys.argv[0]) + sys.exit(1) + +print "Creating Session" +session_path = client.CreateSession(sys.argv[1], { "Target": sys.argv[2] }) +session = dbus.Interface(bus.get_object("org.bluez.obex.client", session_path), + "org.bluez.obex.Session") + +print session.GetCapabilities() diff --git a/test/list-folders b/test/list-folders new file mode 100755 index 000000000..ecd456cb9 --- /dev/null +++ b/test/list-folders @@ -0,0 +1,38 @@ +#!/usr/bin/python + +import sys +import dbus + + +def list_folder(folder): + bus = dbus.SessionBus() + client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"), + "org.bluez.obex.Client") + + path = client.CreateSession(sys.argv[1], { "Target": "ftp" }) + + ftp = dbus.Interface(bus.get_object("org.bluez.obex.client", path), + "org.bluez.obex.FileTransfer") + + if folder: + for node in folder.split("/"): + ftp.ChangeFolder(node) + + for i in ftp.ListFolder(): + if i["Type"] == "folder": + print "%s/" % (i["Name"]) + else: + print "%s" % (i["Name"]) + + +if __name__ == '__main__': + + if len(sys.argv) < 2: + print "Usage: %s [folder]" % (sys.argv[0]) + sys.exit(1) + + folder = None + if len(sys.argv) == 3: + folder = sys.argv[2] + + list_folder(folder) diff --git a/test/map-client b/test/map-client new file mode 100755 index 000000000..756ebb8c9 --- /dev/null +++ b/test/map-client @@ -0,0 +1,207 @@ +#!/usr/bin/python + +import gobject + +import sys +import os +import dbus +import dbus.mainloop.glib +from optparse import OptionParser + +from pprint import pformat + +def unwrap(x): + """Hack to unwrap D-Bus values, so that they're easier to read when + printed. Taken from d-feet """ + + if isinstance(x, list): + return map(unwrap, x) + + if isinstance(x, tuple): + return tuple(map(unwrap, x)) + + if isinstance(x, dict): + return dict([(unwrap(k), unwrap(v)) for k, v in x.iteritems()]) + + for t in [unicode, str, long, int, float, bool]: + if isinstance(x, t): + return t(x) + + return x + +def parse_options(): + parser.add_option("-d", "--device", dest="device", + help="Device to connect", metavar="DEVICE") + parser.add_option("-c", "--chdir", dest="new_dir", + help="Change current directory to DIR", metavar="DIR") + parser.add_option("-l", "--lsdir", action="store_true", dest="ls_dir", + help="List folders in current directory") + parser.add_option("-v", "--verbose", action="store_true", dest="verbose") + parser.add_option("-L", "--lsmsg", action="store", dest="ls_msg", + help="List messages in supplied CWD subdir") + parser.add_option("-g", "--get", action="store", dest="get_msg", + help="Get message contents") + parser.add_option("--get-properties", action="store", dest="get_msg_properties", + help="Get message properties") + parser.add_option("--mark-read", action="store", dest="mark_msg_read", + help="Marks the messages as read") + parser.add_option("--mark-unread", action="store", dest="mark_msg_unread", + help="Marks the messages as unread") + parser.add_option("--mark-deleted", action="store", dest="mark_msg_deleted", + help="Deletes the message from the folder") + parser.add_option("--mark-undeleted", action="store", dest="mark_msg_undeleted", + help="Undeletes the message") + parser.add_option("-u", "--update-inbox", action="store_true", dest="update_inbox", + help="Checks for new mails") + + return parser.parse_args() + +def set_folder(session, new_dir): + session.SetFolder(new_dir) + +class MapClient: + def __init__(self, session_path, verbose=False): + self.progress = 0 + self.transfer_path = None + self.props = dict() + self.verbose = verbose + self.path = session_path + bus = dbus.SessionBus() + obj = bus.get_object("org.bluez.obex.client", session_path) + self.session = dbus.Interface(obj, "org.bluez.obex.Session") + self.map = dbus.Interface(obj, "org.bluez.obex.MessageAccess") + bus.add_signal_receiver(self.transfer_complete, + dbus_interface="org.bluez.obex.Transfer", + signal_name="Complete", + path_keyword="path") + bus.add_signal_receiver(self.transfer_error, + dbus_interface="org.bluez.obex.Transfer", + signal_name="Error", + path_keyword="path") + + def create_transfer_reply(self, reply): + (path, properties) = reply + self.transfer_path = path + self.props[path] = properties + if self.verbose: + print "Transfer created: %s (file %s)" % (path, + properties["Filename"]) + + def generic_reply(self): + if self.verbose: + print "Operation succeeded" + + def error(self, err): + print err + mainloop.quit() + + def transfer_complete(self, path): + if path != self.transfer_path: + return + if self.verbose: + print "Transfer finished" + properties = self.props.get(path) + if properties == None: + return + f = open(properties["Filename"], "r") + os.remove(properties["Filename"]) + print f.readlines() + + def transfer_error(self, code, message, path): + if path != self.transfer_path: + return + print "Transfer finished with error %s: %s" % (code, message) + mainloop.quit() + + def set_folder(self, new_dir): + self.map.SetFolder(new_dir) + + def list_folders(self): + for i in self.map.ListFolders(dict()): + print "%s/" % (i["Name"]) + + def list_messages(self, folder): + ret = self.map.ListMessages(folder, dict()) + print pformat(unwrap(ret)) + + def get_message(self, handle): + self.map.ListMessages("", dict()) + path = self.path + "/message" + handle + obj = bus.get_object("org.bluez.obex.client", path) + msg = dbus.Interface(obj, "org.bluez.obex.Message") + msg.Get("", True, reply_handler=self.create_transfer_reply, + error_handler=self.error) + + def get_message_properties(self, handle): + self.map.ListMessages("", dict()) + path = self.path + "/message" + handle + obj = bus.get_object("org.bluez.obex.client", path) + msg = dbus.Interface(obj, "org.bluez.obex.Message") + ret = msg.GetProperties() + print pformat(unwrap(ret)) + + def set_message_property(self, handle, prop, flag): + self.map.ListMessages("", dict()) + path = self.path + "/message" + handle + obj = bus.get_object("org.bluez.obex.client", path) + msg = dbus.Interface(obj, "org.bluez.obex.Message") + msg.SetProperty (prop, flag); + + def update_inbox(self): + self.map.UpdateInbox() + + +if __name__ == '__main__': + + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + + parser = OptionParser() + + (options, args) = parse_options() + + if not options.device: + parser.print_help() + exit(0) + + bus = dbus.SessionBus() + mainloop = gobject.MainLoop() + + client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"), + "org.bluez.obex.Client") + + print "Creating Session" + path = client.CreateSession(options.device, { "Target": "map" }) + + map_client = MapClient(path, options.verbose) + + if options.new_dir: + map_client.set_folder(options.new_dir) + + if options.ls_dir: + map_client.list_folders() + + if options.ls_msg is not None: + map_client.list_messages(options.ls_msg) + + if options.get_msg is not None: + map_client.get_message(options.get_msg) + + if options.get_msg_properties is not None: + map_client.get_message_properties(options.get_msg_properties) + + if options.mark_msg_read is not None: + map_client.set_message_property(options.mark_msg_read, "Read", True) + + if options.mark_msg_unread is not None: + map_client.set_message_property(options.mark_msg_unread, "Read", False) + + if options.mark_msg_deleted is not None: + map_client.set_message_property(options.mark_msg_deleted, "Deleted", True) + + if options.mark_msg_undeleted is not None: + map_client.set_message_property(options.mark_msg_undeleted, "Deleted", False) + + if options.update_inbox: + map_client.update_inbox() + + mainloop.run() diff --git a/test/opp-client b/test/opp-client new file mode 100755 index 000000000..3d23dfbc4 --- /dev/null +++ b/test/opp-client @@ -0,0 +1,121 @@ +#!/usr/bin/python + +import sys +import dbus +import gobject +import dbus.mainloop.glib +import os.path +from optparse import OptionParser + +def parse_options(): + parser.add_option("-d", "--device", dest="device", + help="Device to connect", metavar="DEVICE") + parser.add_option("-p", "--pull", dest="pull_to_file", + help="Pull vcard and store in FILE", metavar="FILE") + parser.add_option("-s", "--send", dest="send_file", + help="Send FILE", metavar="FILE") + parser.add_option("-v", "--verbose", action="store_true", + dest="verbose") + + return parser.parse_args() + +class OppClient: + def __init__(self, session_path, verbose=False): + self.progress = 0 + self.transfer_path = None + self.verbose = verbose + bus = dbus.SessionBus() + obj = bus.get_object("org.bluez.obex.client", session_path) + self.session = dbus.Interface(obj, "org.bluez.obex.Session") + self.opp = dbus.Interface(obj, "org.bluez.obex.ObjectPush") + bus.add_signal_receiver(self.transfer_complete, + dbus_interface="org.bluez.obex.Transfer", + signal_name="Complete", + path_keyword="path") + bus.add_signal_receiver(self.transfer_error, + dbus_interface="org.bluez.obex.Transfer", + signal_name="Error", + path_keyword="path") + if self.verbose: + bus.add_signal_receiver(self.transfer_progress, + dbus_interface="org.bluez.obex.Transfer", + signal_name="PropertyChanged", + path_keyword="path") + + def create_transfer_reply(self, reply): + (path, properties) = reply + self.transfer_path = path + self.transfer_size = properties["Size"] + if self.verbose: + print "Transfer created: %s" % path + + def error(self, err): + print err + mainloop.quit() + + def transfer_complete(self, path): + if path != self.transfer_path: + return + if self.verbose: + print "Transfer finished" + mainloop.quit() + + def transfer_error(self, code, message, path): + if path != self.transfer_path: + return + print "Transfer finished with error %s: %s" % (code, message) + mainloop.quit() + + def transfer_progress(self, prop, value, path): + if path != self.transfer_path: + return + + if prop != "Progress": + return + + speed = (value - self.progress) / 1000 + print "Transfer progress %d/%d at %d kBps" % (value, + self.transfer_size, + speed) + self.progress = value + + def pull_business_card(self, filename): + self.opp.PullBusinessCard(os.path.abspath(filename), + reply_handler=self.create_transfer_reply, + error_handler=self.error) + + def send_file(self, filename): + self.opp.SendFile(os.path.abspath(filename), + reply_handler=self.create_transfer_reply, + error_handler=self.error) + +if __name__ == '__main__': + + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + + parser = OptionParser() + + (options, args) = parse_options() + + if not options.device: + parser.print_help() + sys.exit(0) + + bus = dbus.SessionBus() + mainloop = gobject.MainLoop() + + client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"), + "org.bluez.obex.Client") + + print "Creating Session" + path = client.CreateSession(options.device, { "Target": "OPP" }) + + opp_client = OppClient(path, options.verbose) + + if options.pull_to_file: + opp_client.pull_business_card(options.pull_to_file) + + if options.send_file: + opp_client.send_file(options.send_file) + + mainloop.run() diff --git a/test/pbap-client b/test/pbap-client new file mode 100755 index 000000000..fbe930c9b --- /dev/null +++ b/test/pbap-client @@ -0,0 +1,157 @@ +#!/usr/bin/python + +import gobject + +import sys +import os +import dbus +import dbus.service +import dbus.mainloop.glib + +class Transfer: + def __init__(self, callback_func): + self.callback_func = callback_func + self.path = None + self.filename = None + +class PbapClient: + def __init__(self, session_path): + self.transfers = 0 + self.props = dict() + self.flush_func = None + bus = dbus.SessionBus() + obj = bus.get_object("org.bluez.obex.client", session_path) + self.session = dbus.Interface(obj, "org.bluez.obex.Session") + self.pbap = dbus.Interface(obj, + "org.bluez.obex.PhonebookAccess") + bus.add_signal_receiver(self.transfer_complete, + dbus_interface="org.bluez.obex.Transfer", + signal_name="Complete", + path_keyword="path") + bus.add_signal_receiver(self.transfer_error, + dbus_interface="org.bluez.obex.Transfer", + signal_name="Error", + path_keyword="path") + + def register(self, reply, transfer): + (path, properties) = reply + transfer.path = path + transfer.filename = properties["Filename"] + self.props[path] = transfer + print "Transfer created: %s (file %s)" % (path, + transfer.filename) + + def error(self, err): + print err + mainloop.quit() + + def transfer_complete(self, path): + req = self.props.get(path) + if req == None: + return + self.transfers -= 1 + print "Transfer %s finished" % path + f = open(req.filename, "r") + os.remove(req.filename) + lines = f.readlines() + del self.props[path] + req.callback_func(lines) + + if (len(self.props) == 0) and (self.transfers == 0): + if self.flush_func != None: + f = self.flush_func + self.flush_func = None + f() + + def transfer_error(self, code, message, path): + req = self.props.get(path) + if req == None: + return + print "Transfer finished with error %s: %s" % (code, message) + mainloop.quit() + + def pull(self, vcard, params, func): + req = Transfer(func) + self.pbap.Pull(vcard, "", params, + reply_handler=lambda r: self.register(r, req), + error_handler=self.error) + self.transfers += 1 + + def pull_all(self, params, func): + req = Transfer(func) + self.pbap.PullAll("", params, + reply_handler=lambda r: self.register(r, req), + error_handler=self.error) + self.transfers += 1 + + def flush_transfers(self, func): + if (len(self.props) == 0) and (self.transfers == 0): + return + self.flush_func = func + + def interface(self): + return self.pbap + +if __name__ == '__main__': + + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + + bus = dbus.SessionBus() + mainloop = gobject.MainLoop() + + client = dbus.Interface(bus.get_object("org.bluez.obex.client", "/"), + "org.bluez.obex.Client") + + if (len(sys.argv) < 2): + print "Usage: %s " % (sys.argv[0]) + sys.exit(1) + + print "Creating Session" + session_path = client.CreateSession(sys.argv[1], { "Target": "PBAP" }) + + pbap_client = PbapClient(session_path) + + def process_result(lines, header): + if header != None: + print header + for line in lines: + print line, + print + + def test_paths(paths): + if len(paths) == 0: + print + print "FINISHED" + mainloop.quit() + return + + path = paths[0] + + print "\n--- Select Phonebook %s ---\n" % (path) + pbap_client.interface().Select("int", path) + + print "\n--- GetSize ---\n" + ret = pbap_client.interface().GetSize() + print "Size = %d\n" % (ret) + + print "\n--- List vCard ---\n" + try: + ret = pbap_client.interface().List(dbus.Dictionary()) + except: + ret = [] + + params = dbus.Dictionary({ "Format" : "vcard30", + "Fields" : [ "VERSION", "FN", "TEL"] }) + for item in ret: + print "%s : %s" % (item[0], item[1]) + pbap_client.pull(item[0], params, + lambda x: process_result(x, None)) + + pbap_client.pull_all(params, lambda x: process_result(x, + "\n--- PullAll ---\n")) + + pbap_client.flush_transfers(lambda: test_paths(paths[1:])) + + test_paths(["PB", "ICH", "OCH", "MCH", "CCH"]) + + mainloop.run() diff --git a/test/simple-obex-agent b/test/simple-obex-agent new file mode 100755 index 000000000..19020820d --- /dev/null +++ b/test/simple-obex-agent @@ -0,0 +1,132 @@ +#!/usr/bin/python + +import gobject + +import sys +import dbus +import dbus.service +import dbus.mainloop.glib + +class Agent(dbus.service.Object): + def __init__(self, conn=None, obj_path=None): + dbus.service.Object.__init__(self, conn, obj_path) + self.pending_auth = False + + @dbus.service.method("org.bluez.obex.Agent", in_signature="osssii", + out_signature="s") + def Authorize(self, dpath, device, filename, ftype, length, time): + global transfers + + self.pending_auth = True + print "Authorize (%s, %s, %s) Y/n" % (path, device, filename) + auth = raw_input().strip("\n ") + + if auth == "n" or auth == "N": + self.pending_auth = False + raise dbus.DBusException( + "org.bluez.obex.Error.Rejected: " + "Not Autorized") + + print "Full filename (including path):" + self.pending_auth = False + + transfers.append(Transfer(dpath, filename, 0, length)) + return raw_input().strip("\n ") + + @dbus.service.method("org.bluez.obex.Agent", in_signature="", + out_signature="") + def Cancel(self): + print "Authorization Canceled" + self.pending_auth = False + + +class Transfer(object): + def __init__(self, dpath, filename=None, transfered=-1, size=-1): + self.dpath = dpath + self.filename = filename + self.transfered = transfered + self.size = size + + def update(self, filename=None, transfered=-1, total=-1): + if filename: + self.filename = filename + self.transfered = transfered + self.size = total + + def cancel(self): + transfer_iface = dbus.Interface(bus.get_object( + "org.bluez.obex", self.dpath), + "org.bluez.obex.Transfer") + transfer_iface.Cancel() + + def __str__(self): + p = float(self.transfered) / float(self.size) * 100 + return "%s (%s) (%.2f%%)" % (self.filename, self.dpath, p) + + __repr__ = __str__ + + +if __name__ == '__main__': + global transfers + + def new_transfer(dpath): + print "new transfer" + bus.add_signal_receiver(progress, + dbus_interface="org.bluez.obex.Transfer", + signal_name="Progress", + path_keyword="dpath") + + def transfer_completed(dpath, success): + global transfers + + print "\ntransfer completed => %s" % (success and "Success" or "Fail") + transfers = [t for t in transfers if t.dpath != dpath] + + def progress(total, current, dpath): + s = [] + for t in transfers: + if t.dpath == dpath: + t.update(None, current, total) + s.append("%s" % (t)) + sys.stdout.write(" ".join(s) + "\r") + sys.stdout.flush() + + + transfers = [] + + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + + bus = dbus.SessionBus() + manager = dbus.Interface(bus.get_object("org.bluez.obex", "/"), + "org.bluez.obex.Manager") + bus.add_signal_receiver(new_transfer, + dbus_interface="org.bluez.obex.Manager", + signal_name="TransferStarted") + + bus.add_signal_receiver(transfer_completed, + dbus_interface="org.bluez.obex.Manager", + signal_name="TransferCompleted") + + path = "/test/agent" + agent = Agent(bus, path) + + mainloop = gobject.MainLoop() + + manager.RegisterAgent(path) + print "Agent registered" + + cont = True + while cont: + try: + mainloop.run() + except KeyboardInterrupt: + if agent.pending_auth: + agent.Cancel() + elif len(transfers) > 0: + for a in transfers: + a.cancel() + else: + cont = False + + # manager.UnregisterAgent(path) + # print "Agent unregistered" -- 2.47.3