diff options
Diffstat (limited to 'test')
| -rwxr-xr-x | test/ftp-client | 178 | ||||
| -rwxr-xr-x | test/map-client | 232 | ||||
| -rwxr-xr-x | test/monitor-bluetooth | 8 | ||||
| -rwxr-xr-x | test/opp-client | 119 | ||||
| -rwxr-xr-x | test/pbap-client | 175 | ||||
| -rwxr-xr-x | test/simple-agent | 8 | ||||
| -rwxr-xr-x | test/simple-endpoint | 7 | ||||
| -rwxr-xr-x | test/simple-player | 10 | ||||
| -rwxr-xr-x | test/simple-service | 128 | ||||
| -rwxr-xr-x | test/test-adapter | 4 | ||||
| -rwxr-xr-x | test/test-alert | 15 | ||||
| -rwxr-xr-x | test/test-cyclingspeed | 13 | ||||
| -rwxr-xr-x | test/test-device | 10 | ||||
| -rwxr-xr-x | test/test-discovery | 8 | ||||
| -rwxr-xr-x | test/test-health | 11 | ||||
| -rwxr-xr-x | test/test-health-sink | 9 | ||||
| -rwxr-xr-x | test/test-heartrate | 10 | ||||
| -rwxr-xr-x | test/test-hfp | 12 | ||||
| -rwxr-xr-x | test/test-manager | 6 | ||||
| -rwxr-xr-x | test/test-nap | 2 | ||||
| -rwxr-xr-x | test/test-network | 2 | ||||
| -rwxr-xr-x | test/test-profile | 8 | ||||
| -rwxr-xr-x | test/test-proximity | 10 | ||||
| -rwxr-xr-x | test/test-thermometer | 10 |
24 files changed, 806 insertions, 189 deletions
diff --git a/test/ftp-client b/test/ftp-client new file mode 100755 index 00000000..78c32b3f --- /dev/null +++ b/test/ftp-client @@ -0,0 +1,178 @@ +#!/usr/bin/python + +from __future__ import absolute_import, print_function, unicode_literals + +from optparse import OptionParser +import os.path +import sys +import dbus +import dbus.service +import dbus.mainloop.glib +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject + +BUS_NAME='org.bluez.obex' +PATH = '/org/bluez/obex' +CLIENT_INTERFACE='org.bluez.obex.Client1' +SESSION_INTERFACE='org.bluez.obex.Session1' +FILE_TRASNFER_INTERFACE='org.bluez.obex.FileTransfer1' +TRANSFER_INTERFACE='org.bluez.obex.Transfer1' + +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.transferred = 0 + self.transfer_path = None + self.transfer_size = 0 + self.verbose = verbose + bus = dbus.SessionBus() + obj = bus.get_object(BUS_NAME, session_path) + self.session = dbus.Interface(obj, SESSION_INTERFACE) + self.ftp = dbus.Interface(obj, FILE_TRASNFER_INTERFACE) + bus.add_signal_receiver(self.properties_changed, + dbus_interface="org.freedesktop.DBus.Properties", + signal_name="PropertiesChanged", + path_keyword="path") + + def create_transfer_reply(self, path, properties): + 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 properties_changed(self, interface, properties, invalidated, path): + if path != self.transfer_path: + return + + if properties['Status'] == 'complete' or \ + properties['Status'] == 'error': + if self.verbose: + print("Transfer %s" % properties['Status']) + mainloop.quit() + return + + if properties["Transferred"] == None: + return + + speed = (value - self.transferred) / 1000 + print("Transfer progress %d/%d at %d kBps" % (value, + self.transfer_size, + speed)) + self.transferred = 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(BUS_NAME, PATH,), + CLIENT_INTERFACE) + + 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/map-client b/test/map-client new file mode 100755 index 00000000..b9695da6 --- /dev/null +++ b/test/map-client @@ -0,0 +1,232 @@ +#!/usr/bin/python + +from __future__ import absolute_import, print_function, unicode_literals + +from optparse import OptionParser +import os +from pprint import pformat +import sys +import dbus +import dbus.mainloop.glib +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject + +BUS_NAME='org.bluez.obex' +PATH = '/org/bluez/obex' +CLIENT_INTERFACE = 'org.bluez.obex.Client1' +SESSION_INTERFACE = 'org.bluez.obex.Session1' +MESSAGE_ACCESS_INTERFACE = 'org.bluez.obex.MessageAccess1' +MESSAGE_INTERFACE = 'org.bluez.obex.Message1' +TRANSFER_INTERFACE = 'org.bluez.obex.Transfer1' + +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("-p", "--push", action="store", dest="push_msg", + help="Push message") + 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(BUS_NAME, session_path) + self.session = dbus.Interface(obj, SESSION_INTERFACE) + self.map = dbus.Interface(obj, MESSAGE_ACCESS_INTERFACE) + bus.add_signal_receiver(self.properties_changed, + dbus_interface="org.freedesktop.DBus.Properties", + signal_name="PropertiesChanged", + path_keyword="path") + + def create_transfer_reply(self, path, properties): + 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 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, path): + print("Transfer %s error" % path) + mainloop.quit() + + def properties_changed(self, interface, properties, invalidated, path): + req = self.props.get(path) + if req == None: + return + + if properties['Status'] == 'complete': + self.transfer_complete(path) + return + + if properties['Status'] == 'error': + self.transfer_error(path) + return + + 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(BUS_NAME, path) + msg = dbus.Interface(obj, MESSAGE_INTERFACE) + msg.Get("", True, reply_handler=self.create_transfer_reply, + error_handler=self.error) + + def push_message(self, filename): + self.map.PushMessage(filename, "telecom/msg/outbox", dict(), + 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(BUS_NAME, path) + msg = dbus.Interface(obj, "org.freedesktop.DBus.Properties") + ret = msg.GetAll(MESSAGE_INTERFACE) + 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(BUS_NAME, path) + msg = dbus.Interface(obj, MESSAGE_INTERFACE) + 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(BUS_NAME, PATH), + CLIENT_INTERFACE) + + 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.push_msg is not None: + map_client.push_message(options.push_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/monitor-bluetooth b/test/monitor-bluetooth index bc5ddaf9..d9b5472f 100755 --- a/test/monitor-bluetooth +++ b/test/monitor-bluetooth @@ -2,10 +2,12 @@ from __future__ import absolute_import, print_function, unicode_literals -import gobject - import dbus import dbus.mainloop.glib +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject relevant_ifaces = [ "org.bluez.Adapter1", "org.bluez.Device1" ] @@ -48,5 +50,5 @@ if __name__ == '__main__': dbus_interface="org.freedesktop.DBus.ObjectManager", signal_name="InterfacesRemoved") - mainloop = gobject.MainLoop() + mainloop = GObject.MainLoop() mainloop.run() diff --git a/test/opp-client b/test/opp-client new file mode 100755 index 00000000..62d5b845 --- /dev/null +++ b/test/opp-client @@ -0,0 +1,119 @@ +#!/usr/bin/python + +from __future__ import absolute_import, print_function, unicode_literals + +from optparse import OptionParser +import os.path +import sys +import dbus +import dbus.mainloop.glib +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject + +BUS_NAME='org.bluez.obex' +PATH = '/org/bluez/obex' +CLIENT_INTERFACE='org.bluez.obex.Client1' +SESSION_INTERFACE='org.bluez.obex.Session1' +OBJECT_PUSH_INTERFACE='org.bluez.obex.ObjectPush1' +TRANSFER_INTERFACE='org.bluez.obex.Transfer1' + +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.transferred = 0 + self.transfer_path = None + self.verbose = verbose + bus = dbus.SessionBus() + obj = bus.get_object(BUS_NAME, session_path) + self.session = dbus.Interface(obj, SESSION_INTERFACE) + self.opp = dbus.Interface(obj, OBJECT_PUSH_INTERFACE) + bus.add_signal_receiver(self.properties_changed, + dbus_interface="org.freedesktop.DBus.Properties", + signal_name="PropertiesChanged", + path_keyword="path") + + def create_transfer_reply(self, path, properties): + 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 properties_changed(self, interface, properties, invalidated, path): + if path != self.transfer_path: + return + + if "Status" in properties and \ + (properties["Status"] == "complete" or \ + properties["Status"] == "error"): + if self.verbose: + print("Transfer %s" % properties["Status"]) + mainloop.quit() + return + + if "Transferred" not in properties: + return + + value = properties["Transferred"] + speed = (value - self.transferred) / 1000 + print("Transfer progress %d/%d at %d kBps" % (value, + self.transfer_size, + speed)) + self.transferred = 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(BUS_NAME, PATH), + CLIENT_INTERFACE) + + 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 00000000..51e26ebd --- /dev/null +++ b/test/pbap-client @@ -0,0 +1,175 @@ +#!/usr/bin/python + +from __future__ import absolute_import, print_function, unicode_literals + +import os +import sys +import dbus +import dbus.service +import dbus.mainloop.glib +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject + +BUS_NAME='org.bluez.obex' +PATH = '/org/bluez/obex' +CLIENT_INTERFACE = 'org.bluez.obex.Client1' +SESSION_INTERFACE = 'org.bluez.obex.Session1' +PHONEBOOK_ACCESS_INTERFACE = 'org.bluez.obex.PhonebookAccess1' +TRANSFER_INTERFACE = 'org.bluez.obex.Transfer1' + +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(BUS_NAME, session_path) + self.session = dbus.Interface(obj, SESSION_INTERFACE) + self.pbap = dbus.Interface(obj, PHONEBOOK_ACCESS_INTERFACE) + bus.add_signal_receiver(self.properties_changed, + dbus_interface="org.freedesktop.DBus.Properties", + signal_name="PropertiesChanged", + path_keyword="path") + + def register(self, path, properties, transfer): + 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 complete" % path) + try: + f = open(req.filename, "r") + os.remove(req.filename) + lines = f.readlines() + del self.props[path] + req.callback_func(lines) + except: + pass + + 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, path): + print("Transfer %s error" % path) + mainloop.quit() + + def properties_changed(self, interface, properties, invalidated, path): + req = self.props.get(path) + if req == None: + return + + if properties['Status'] == 'complete': + self.transfer_complete(path) + return + + if properties['Status'] == 'error': + self.transfer_error(path) + return + + def pull(self, vcard, params, func): + req = Transfer(func) + self.pbap.Pull(vcard, "", params, + reply_handler=lambda o, p: self.register(o, p, req), + error_handler=self.error) + self.transfers += 1 + + def pull_all(self, params, func): + req = Transfer(func) + self.pbap.PullAll("", params, + reply_handler=lambda o, p: self.register(o, p, 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(BUS_NAME, PATH), + CLIENT_INTERFACE) + + if (len(sys.argv) < 2): + print("Usage: %s <device>" % (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" : ["PHOTO"] }) + 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-agent b/test/simple-agent index 854e1af4..a69299a3 100755 --- a/test/simple-agent +++ b/test/simple-agent @@ -2,13 +2,15 @@ from __future__ import absolute_import, print_function, unicode_literals -from gi.repository import GObject - +from optparse import OptionParser import sys import dbus import dbus.service import dbus.mainloop.glib -from optparse import OptionParser +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject import bluezutils BUS_NAME = 'org.bluez' diff --git a/test/simple-endpoint b/test/simple-endpoint index 590f83a9..0164cff9 100755 --- a/test/simple-endpoint +++ b/test/simple-endpoint @@ -6,7 +6,10 @@ import sys import dbus import dbus.service import dbus.mainloop.glib -import gobject +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject import bluezutils A2DP_SOURCE_UUID = "0000110A-0000-1000-8000-00805F9B34FB" @@ -94,7 +97,7 @@ if __name__ == '__main__': path = "/test/endpoint" endpoint = Endpoint(bus, path) - mainloop = gobject.MainLoop() + mainloop = GObject.MainLoop() properties = dbus.Dictionary({ "UUID" : A2DP_SOURCE_UUID, "Codec" : SBC_CODEC, diff --git a/test/simple-player b/test/simple-player index 01bec068..23e78add 100755 --- a/test/simple-player +++ b/test/simple-player @@ -1,12 +1,16 @@ #!/usr/bin/python from __future__ import print_function + import os import sys import dbus import dbus.service import dbus.mainloop.glib -import gobject +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject import bluezutils class Player(dbus.service.Object): @@ -56,7 +60,7 @@ class Player(dbus.service.Object): signature="sv") handler = InputHandler(self) - gobject.io_add_watch(sys.stdin, gobject.IO_IN, + GObject.io_add_watch(sys.stdin, GObject.IO_IN, handler.handle) @dbus.service.method("org.freedesktop.DBus.Properties", @@ -136,7 +140,7 @@ if __name__ == '__main__': path = "/test/player" player = Player(bus, path) - mainloop = gobject.MainLoop() + mainloop = GObject.MainLoop() if len(sys.argv) > 2: player.set_object(sys.argv[2]) diff --git a/test/simple-service b/test/simple-service deleted file mode 100755 index 02d76489..00000000 --- a/test/simple-service +++ /dev/null @@ -1,128 +0,0 @@ -#!/usr/bin/python - -from __future__ import absolute_import, print_function, unicode_literals - -import sys -import time -import dbus -import bluezutils - -xml = ' \ -<?xml version="1.0" encoding="UTF-8" ?> \ -<record> \ - <attribute id="0x0001"> \ - <sequence> \ - <uuid value="0x1101"/> \ - </sequence> \ - </attribute> \ - \ - <attribute id="0x0002"> \ - <uint32 value="0"/> \ - </attribute> \ - \ - <attribute id="0x0003"> \ - <uuid value="00001101-0000-1000-8000-00805f9b34fb"/> \ - </attribute> \ - \ - <attribute id="0x0004"> \ - <sequence> \ - <sequence> \ - <uuid value="0x0100"/> \ - </sequence> \ - <sequence> \ - <uuid value="0x0003"/> \ - <uint8 value="23"/> \ - </sequence> \ - </sequence> \ - </attribute> \ - \ - <attribute id="0x0005"> \ - <sequence> \ - <uuid value="0x1002"/> \ - </sequence> \ - </attribute> \ - \ - <attribute id="0x0006"> \ - <sequence> \ - <uint16 value="0x656e"/> \ - <uint16 value="0x006a"/> \ - <uint16 value="0x0100"/> \ - </sequence> \ - </attribute> \ - \ - <attribute id="0x0007"> \ - <uint32 value="0"/> \ - </attribute> \ - \ - <attribute id="0x0008"> \ - <uint8 value="0xff"/> \ - </attribute> \ - \ - <attribute id="0x0009"> \ - <sequence> \ - <sequence> \ - <uuid value="0x1101"/> \ - <uint16 value="0x0100"/> \ - </sequence> \ - </sequence> \ - </attribute> \ - \ - <attribute id="0x000a"> \ - <url value="http://www.bluez.org/"/> \ - </attribute> \ - \ - <attribute id="0x000b"> \ - <url value="http://www.bluez.org/"/> \ - </attribute> \ - \ - <attribute id="0x000c"> \ - <url value="http://www.bluez.org/"/> \ - </attribute> \ - \ - <attribute id="0x0100"> \ - <text value="Serial Port"/> \ - </attribute> \ - \ - <attribute id="0x0101"> \ - <text value="Serial Port Service"/> \ - </attribute> \ - \ - <attribute id="0x0102"> \ - <text value="BlueZ"/> \ - </attribute> \ - \ - <attribute id="0x0200"> \ - <sequence> \ - <uint16 value="0x0100"/> \ - </sequence> \ - </attribute> \ - \ - <attribute id="0x0201"> \ - <uint32 value="0"/> \ - </attribute> \ -</record> \ -' - -bus = dbus.SystemBus() - -if len(sys.argv) > 1: - path = bluezutils.find_adapter(sys.argv[1]).object_path -else: - path = bluezutils.find_adapter().object_path - -service = dbus.Interface(bus.get_object("org.bluez", path), - "org.bluez.Service") - -handle = service.AddRecord(xml) - -print("Service record with handle 0x%04x added" % (handle)) - -print("Press CTRL-C to remove service record") - -try: - time.sleep(1000) - print("Terminating session") -except: - pass - -service.RemoveRecord(dbus.UInt32(handle)) diff --git a/test/test-adapter b/test/test-adapter index 5deeda48..959a4370 100755 --- a/test/test-adapter +++ b/test/test-adapter @@ -2,10 +2,10 @@ from __future__ import absolute_import, print_function, unicode_literals +from optparse import OptionParser, make_option import sys -import dbus import time -from optparse import OptionParser, make_option +import dbus import bluezutils bus = dbus.SystemBus() diff --git a/test/test-alert b/test/test-alert index 066e5378..43b3cf36 100755 --- a/test/test-alert +++ b/test/test-alert @@ -1,12 +1,17 @@ #!/usr/bin/python + from __future__ import absolute_import, print_function, unicode_literals + +import optparse +import os +import sys import dbus import dbus.service import dbus.mainloop.glib -import gobject -import optparse -import sys -import os +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject BUS_NAME = 'org.bluez' ALERT_INTERFACE = 'org.bluez.Alert1' @@ -149,7 +154,7 @@ parser.disable_interspersed_args() dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) bus = dbus.SystemBus() -mainloop = gobject.MainLoop() +mainloop = GObject.MainLoop() alert = dbus.Interface(bus.get_object(BUS_NAME, BLUEZ_OBJECT_PATH), ALERT_INTERFACE) alert_agent = AlertAgent(bus, TEST_OBJECT_PATH, alert, mainloop) diff --git a/test/test-cyclingspeed b/test/test-cyclingspeed index 75bd7d74..393f79c7 100755 --- a/test/test-cyclingspeed +++ b/test/test-cyclingspeed @@ -6,13 +6,16 @@ from __future__ import absolute_import, print_function, unicode_literals Cycling Speed and Cadence test script ''' -import gobject - +from optparse import OptionParser, make_option import sys import dbus import dbus.service import dbus.mainloop.glib -from optparse import OptionParser, make_option +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject +import bluezutils BUS_NAME = 'org.bluez' CYCLINGSPEED_MANAGER_INTERFACE = 'org.bluez.CyclingSpeedManager1' @@ -141,7 +144,7 @@ if __name__ == "__main__": device_path = device.object_path cscmanager = dbus.Interface(bus.get_object(BUS_NAME, adapter_path), - CYCLINGSPEED_WATCHER_INTERFACE) + CYCLINGSPEED_MANAGER_INTERFACE) watcher_path = "/test/watcher" watcher = Watcher(bus, watcher_path) @@ -190,5 +193,5 @@ if __name__ == "__main__": print("Unknown command") sys.exit(1) - mainloop = gobject.MainLoop() + mainloop = GObject.MainLoop() mainloop.run() diff --git a/test/test-device b/test/test-device index 3d7b8527..b490d53f 100755 --- a/test/test-device +++ b/test/test-device @@ -2,13 +2,15 @@ from __future__ import absolute_import, print_function, unicode_literals -from gi.repository import GObject - +from optparse import OptionParser, make_option +import re import sys import dbus import dbus.mainloop.glib -import re -from optparse import OptionParser, make_option +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject import bluezutils dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) diff --git a/test/test-discovery b/test/test-discovery index c13bfac7..73b81611 100755 --- a/test/test-discovery +++ b/test/test-discovery @@ -2,11 +2,13 @@ from __future__ import absolute_import, print_function, unicode_literals -from gi.repository import GObject - +from optparse import OptionParser, make_option import dbus import dbus.mainloop.glib -from optparse import OptionParser, make_option +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject import bluezutils compact = False diff --git a/test/test-health b/test/test-health index 052a6024..343f29c0 100755 --- a/test/test-health +++ b/test/test-health @@ -3,11 +3,14 @@ from __future__ import absolute_import, print_function, unicode_literals # -*- coding: utf-8 -*- +import sys import dbus import dbus.service -import gobject from dbus.mainloop.glib import DBusGMainLoop -import sys +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject BUS_NAME = 'org.bluez' PATH = '/org/bluez' @@ -16,7 +19,7 @@ HEALTH_MANAGER_INTERFACE = 'org.bluez.HealthManager1' HEALTH_DEVICE_INTERFACE = 'org.bluez.HealthDevice1' DBusGMainLoop(set_as_default=True) -loop = gobject.MainLoop() +loop = GObject.MainLoop() bus = dbus.SystemBus() @@ -48,7 +51,7 @@ def enter_mainloop(): try: print("Entering main lopp, push Ctrl+C for finish") - mainloop = gobject.MainLoop() + mainloop = GObject.MainLoop() mainloop.run() except KeyboardInterrupt: pass diff --git a/test/test-health-sink b/test/test-health-sink index 32afd711..52be5351 100755 --- a/test/test-health-sink +++ b/test/test-health-sink @@ -3,11 +3,14 @@ from __future__ import absolute_import, print_function, unicode_literals # -*- coding: utf-8 -*- +import sys import dbus import dbus.service -import gobject from dbus.mainloop.glib import DBusGMainLoop -import sys +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject BUS_NAME = 'org.bluez' PATH = '/org/bluez' @@ -16,7 +19,7 @@ HEALTH_MANAGER_INTERFACE = 'org.bluez.HealthManager1' HEALTH_DEVICE_INTERFACE = 'org.bluez.HealthDevice1' DBusGMainLoop(set_as_default=True) -loop = gobject.MainLoop() +loop = GObject.MainLoop() bus = dbus.SystemBus() diff --git a/test/test-heartrate b/test/test-heartrate index f26b3dba..5e4e7e5c 100755 --- a/test/test-heartrate +++ b/test/test-heartrate @@ -6,13 +6,15 @@ from __future__ import absolute_import, print_function, unicode_literals Heart Rate Monitor test script ''' -import gobject - +from optparse import OptionParser, make_option import sys import dbus import dbus.service import dbus.mainloop.glib -from optparse import OptionParser, make_option +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject import bluezutils BUS_NAME = 'org.bluez' @@ -102,5 +104,5 @@ if __name__ == "__main__": print("unknown command") sys.exit(1) - mainloop = gobject.MainLoop() + mainloop = GObject.MainLoop() mainloop.run() diff --git a/test/test-hfp b/test/test-hfp index 873de0a2..a8060439 100755 --- a/test/test-hfp +++ b/test/test-hfp @@ -2,16 +2,18 @@ from __future__ import absolute_import, print_function, unicode_literals -from gi.repository import GObject - +from optparse import OptionParser, make_option import os +from socket import SOCK_SEQPACKET, socket import sys import dbus -import glib import dbus.service import dbus.mainloop.glib -from optparse import OptionParser, make_option -from socket import SOCK_SEQPACKET, socket +import glib +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject mainloop = None audio_supported = True diff --git a/test/test-manager b/test/test-manager index 1e3882f8..4f5994f6 100755 --- a/test/test-manager +++ b/test/test-manager @@ -2,10 +2,12 @@ from __future__ import absolute_import, print_function, unicode_literals -from gi.repository import GObject - import dbus import dbus.mainloop.glib +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject import bluezutils def interfaces_added(path, interfaces): diff --git a/test/test-nap b/test/test-nap index 197e3c25..00a2585a 100755 --- a/test/test-nap +++ b/test/test-nap @@ -2,10 +2,10 @@ from __future__ import absolute_import, print_function, unicode_literals +from optparse import OptionParser, make_option import sys import time import dbus -from optparse import OptionParser, make_option import bluezutils bus = dbus.SystemBus() diff --git a/test/test-network b/test/test-network index 3e8713f6..6f094864 100755 --- a/test/test-network +++ b/test/test-network @@ -2,10 +2,10 @@ from __future__ import absolute_import, print_function, unicode_literals +from optparse import OptionParser, make_option import sys import time import dbus -from optparse import OptionParser, make_option import bluezutils bus = dbus.SystemBus() diff --git a/test/test-profile b/test/test-profile index b78d00c9..27915806 100755 --- a/test/test-profile +++ b/test/test-profile @@ -2,15 +2,17 @@ from __future__ import absolute_import, print_function, unicode_literals -from gi.repository import GObject - +from optparse import OptionParser, make_option import os import sys import uuid import dbus import dbus.service import dbus.mainloop.glib -from optparse import OptionParser, make_option +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject class Profile(dbus.service.Object): fd = -1 diff --git a/test/test-proximity b/test/test-proximity index 2f47824e..66b7bc24 100755 --- a/test/test-proximity +++ b/test/test-proximity @@ -6,12 +6,14 @@ from __future__ import absolute_import, print_function, unicode_literals Proximity Monitor test script ''' -import gobject - +from optparse import OptionParser, make_option import sys import dbus import dbus.mainloop.glib -from optparse import OptionParser, make_option +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject import bluezutils BUS_NAME = 'org.bluez' @@ -64,5 +66,5 @@ if __name__ == "__main__": print("Proximity SetProperty('%s', '%s')" % (args[0], args[1])) device_prop.Set(PROXIMITY_MONITOR_INTERFACE, args[0], args[1]) - mainloop = gobject.MainLoop() + mainloop = GObject.MainLoop() mainloop.run() diff --git a/test/test-thermometer b/test/test-thermometer index 6c143bef..7e67c234 100755 --- a/test/test-thermometer +++ b/test/test-thermometer @@ -6,13 +6,15 @@ from __future__ import absolute_import, print_function, unicode_literals Thermometer test script ''' -import gobject - +from optparse import OptionParser, make_option import sys import dbus import dbus.service import dbus.mainloop.glib -from optparse import OptionParser, make_option +try: + from gi.repository import GObject +except ImportError: + import gobject as GObject import bluezutils BUS_NAME = 'org.bluez' @@ -93,5 +95,5 @@ if __name__ == "__main__": print("unknown command") sys.exit(1) - mainloop = gobject.MainLoop() + mainloop = GObject.MainLoop() mainloop.run() |