Add apitest utility

2 files changed, 433 insertions, 1 deletions

diff --git a/test/Makefile.am b/test/Makefile.am
index ad596f50..dde68ed0 100644
--- a/test/Makefile.am
+++ b/test/Makefile.am
@@ -33,6 +33,6 @@ endif
 
 INCLUDES = -I$(top_srcdir)/common
 
-EXTRA_DIST = hsplay hsmicro bdaddr.8
+EXTRA_DIST = apitest hsplay hsmicro bdaddr.8
 
 MAINTAINERCLEANFILES = Makefile.in
diff --git a/test/apitest b/test/apitest
new file mode 100755
index 00000000..83988896
--- /dev/null
+++ b/test/apitest
@@ -0,0 +1,432 @@
+#!/usr/bin/env python
+
+import dbus
+import dbus.decorators
+import dbus.glib
+import gobject
+import sys
+import getopt
+from signal import *
+
+mgr_cmds = [ "InterfaceVersion", "ListAdapters", "DefaultAdapter" ]
+mgr_signals = [ "AdapterAdded", "AdapterRemoved" ]
+
+dev_cmds = [ "GetAddress",
+             "GetVersion",
+             "GetRevision",
+             "GetManufacturer",
+             "GetCompany",
+             "GetMode",
+             "SetMode",
+             "GetDiscoverableTimeout",
+             "SetDiscoverableTimeout",
+             "IsConnectable",
+             "IsDiscoverable",
+             "IsConnected",
+             "ListConnections",
+             "GetMajorClass",
+             "ListAvailableMinorClasses",
+             "GetMinorClass",
+             "SetMinorClass",
+             "GetServiceClasses",
+             "GetName",
+             "SetName",
+             "GetRemoteVersion",
+             "GetRemoteRevision",
+             "GetRemoteManufacturer",
+             "GetRemoteCompany",
+             "GetRemoteMajorClass",
+             "GetRemoteMinorClass",
+             "GetRemoteServiceClasses",
+             "GetRemoteClass",
+             "GetRemoteName",
+             "GetRemoteAlias",
+             "SetRemoteAlias",
+             "ClearRemoteAlias",
+             "LastSeen",
+             "LastUsed",
+             "DisconnectRemoteDevice",
+             "CreateBonding",
+             "CancelBondingProcess",
+             "RemoveBonding",
+             "HasBonding",
+             "ListBondings",
+             "GetPinCodeLength",
+             "GetEncryptionKeySize",
+             "DiscoverDevices",
+             "DiscoverDevicesWithoutNameResolving",
+             "CancelDiscovery" ]
+dev_signals = [ "ModeChanged",
+                "NameChanged",
+                "MinorClassChanged",
+                "DiscoveryStarted",
+                "DiscoveryCompleted",
+                "RemoteDeviceFound",
+                "RemoteNameUpdated",
+                "RemoteNameFailed",
+                "RemoteAliasChanged"
+                "RemoteAliasCleared",
+                "RemoteDeviceConnected",
+                "RemoteDeviceDisconnected",
+                "BondingCreated",
+                "BondingRemoved" ]
+
+dev_signals_filter = [ "/org/bluez/hci0", "/org/bluez/hci1",
+                       "/org/bluez/hci2", "/org/bluez/hci3",
+                       "/org/bluez/hci4", "/org/bluez/hci5",
+                       "/org/bluez/hci6", "/org/bluez/hci7" ]
+
+class Tester:
+    exit_events = []
+    dev_path = None
+    need_dev = False
+    listen = False
+    at_interrupt = None
+
+    def __init__(self, argv):
+        self.name = argv[0]
+
+        self.parse_args(argv[1:])
+
+        try:
+            self.dbus_setup()
+        except dbus.DBusException, e:
+            print 'Failed to do D-Bus setup: %s' % e
+            sys.exit(1)
+
+    def parse_args(self, argv):
+        try:
+            opts, args = getopt.getopt(argv, "hli:")
+        except getopt.GetoptError:
+            self.usage()
+            sys.exit(1)
+
+        for o, a in opts:
+            if o == "-h":
+                self.usage()
+                sys.exit()
+            elif o == "-l":
+                self.listen = True
+            elif o == "-i":
+                if a[0] == '/':
+                    self.dev_path = a
+                else:
+                    self.dev_path = '/org/bluez/%s' % a
+
+        if not (args or self.listen):
+            self.usage()
+            sys.exit(1)
+
+        if args:
+            self.cmd = args[0]
+            self.cmd_args = args[1:]
+
+    def dbus_dev_setup(self):
+        if not self.dev_path:
+            try:
+                self.dbus_mgr_setup()
+                self.dev_path = self.manager.DefaultAdapter()
+            except dbus.DBusException, e:
+                print 'Failed to get default device: %s' % e
+                sys.exit(1)
+        try:
+            obj = self.bus.get_object('org.bluez', self.dev_path)
+            self.device = dbus.Interface(obj, 'org.bluez.Adapter')
+        except dbus.DBusException, e:
+            print 'Failed to setup device path: %s' % e
+            sys.exit(1)
+
+    def dbus_dev_sig_setup(self):
+        try:
+           for signal in dev_signals:
+                for path in dev_signals_filter:
+                    self.bus.add_signal_receiver(self.dev_signal_handler,
+                                             signal, 'org.bluez.Adapter',
+                                             'org.bluez', path)
+        except dbus.DBusException, e:
+            print 'Failed to setup signal handler for device path: %s' % e
+            sys.exit(1)
+
+    def dbus_mgr_sig_setup(self):
+        try:
+            for signal in mgr_signals:
+                self.bus.add_signal_receiver(self.mgr_signal_handler,
+                                         signal,'org.bluez.Manager',
+                                         'org.bluez', '/org/bluez')
+        except dbus.DBusException, e:
+            print 'Failed to setup signal handler for manager path: %s' % e
+            sys.exit(1)
+
+    def dbus_mgr_setup(self):
+        self.manager_obj = self.bus.get_object('org.bluez', '/org/bluez')
+        self.manager = dbus.Interface(self.manager_obj, 'org.bluez.Manager')
+
+    def dbus_setup(self):
+        self.bus = dbus.SystemBus()
+
+    def usage(self):
+        print 'Usage: %s [-i <dev>] [-l] [-h] <cmd> [arg1..]' % self.name
+        print '  -i <dev>   Specify device (e.g. "hci0" or "/org/bluez/hci0")'
+        print '  -l         Listen for events (no command required)'
+        print '  -h         Show this help'
+        print 'Manager commands:'
+        for cmd in mgr_cmds:
+            print '\t%s' % cmd
+        print 'Adapter commands:'
+        for cmd in dev_cmds:
+            print '\t%s' % cmd
+
+    @dbus.decorators.explicitly_pass_message
+    def dev_signal_handler(*args, **keywords):
+        dbus_message = keywords["dbus_message"]
+        print '%s - %s: ' % (dbus_message.get_member(), dbus_message.get_path()),
+        for arg in args[1:]:
+            print '%s   ' % arg,
+        print
+
+    @dbus.decorators.explicitly_pass_message
+    def mgr_signal_handler(*args, **keywords):
+        dbus_message = keywords["dbus_message"]
+        print '%s: ' % dbus_message.get_member()
+        for arg in args[1:]:
+            print '%s   ' % arg,
+        print
+
+    def signal_cb(self, sig, frame):
+        print 'Caught signal, exiting'
+        if self.at_interrupt:
+            self.at_interrupt()
+        self.main_loop.quit()
+
+    def call_mgr_dbus_func(self):
+        if self.cmd == 'InterfaceVersion':
+            try:
+                print self.manager.InterfaceVersion()
+            except dbus.DBusException, e:
+                print 'Sending %s failed: %s' % (self.cmd, e)
+        if self.cmd == 'ListAdapters':
+            try:
+                devices = self.manager.ListAdapters()
+            except dbus.DBusException, e:
+                print 'Sending %s failed: %s' % (self.cmd, e)
+                sys.exit(1)
+            for device in devices:
+                print device
+        elif self.cmd == 'DefaultAdapter':
+            try:
+                print self.manager.DefaultAdapter()
+            except dbus.DBusException, e:
+                print 'Sending %s failed: %s' % (self.cmd, e)
+                sys.exit(1)
+
+    def call_dev_dbus_func(self):
+       try:
+           if self.cmd == 'GetAddress':
+               print self.device.GetAddress()
+           elif self.cmd == 'GetManufacturer':
+               print self.device.GetManufacturer()
+           elif self.cmd == 'GetVersion':
+               print self.device.GetVersion()
+           elif self.cmd == 'GetRevision':
+               print self.device.GetRevision()
+           elif self.cmd == 'GetCompany':
+               print self.device.GetCompany()
+           elif self.cmd == 'GetMode':
+               print self.device.GetMode()
+           elif self.cmd == 'SetMode':
+               if len(self.cmd_args) == 1:
+                   self.device.SetMode(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> SetMode scan_mode' % self.name
+           elif self.cmd == 'GetDiscoverableTimeout':
+               print '%u' % (self.device.GetDiscoverableTimeout())
+           elif self.cmd == 'SetDiscoverableTimeout':
+               if len(self.cmd_args) == 1:
+                   self.device.SetDiscoverableTimeout(dbus.UInt32(self.cmd_args[0]))
+               else:
+                   print 'Usage: %s -i <dev> SetDiscoverableTimeout timeout' % self.name
+           elif self.cmd == 'IsConnectable':
+               print self.device.IsConnectable()
+           elif self.cmd == 'IsDiscoverable':
+               print self.device.IsDiscoverable()
+           elif self.cmd == 'IsConnected':
+               if len(self.cmd_args) == 1:
+                   print self.device.IsConnected(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> IsConnected address' % self.name
+           elif self.cmd == 'ListConnections':
+               print self.device.ListConnections()
+           elif self.cmd == 'GetMajorClass':
+               print self.device.GetMajorClass()
+           elif self.cmd == 'ListAvailableMinorClasses':
+               print self.device.ListAvailableMinorClasses()
+           elif self.cmd == 'GetMinorClass':
+               print self.device.GetMinorClass()
+           elif self.cmd == 'SetMinorClass':
+               if len(self.cmd_args) == 1:
+                   self.device.SetMinorClass(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> SetMinorClass minor' % self.name
+           elif self.cmd == 'GetServiceClasses':
+               classes = self.device.GetServiceClasses()
+               for clas in classes: 
+                   print clas,
+           elif self.cmd == 'GetName':
+               print self.device.GetName()
+           elif self.cmd == 'SetName':
+               if len(self.cmd_args) == 1:
+                   self.device.SetName(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> SetName newname' % self.name
+           elif self.cmd == 'GetRemoteName':
+               if len(self.cmd_args) == 1:
+                   print self.device.GetRemoteName(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> GetRemoteName address' % self.name
+           elif self.cmd == 'GetRemoteVersion':
+               if len(self.cmd_args) == 1:
+                   print self.device.GetRemoteVersion(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> GetRemoteVersion address' % self.name
+           elif self.cmd == 'GetRemoteRevision':
+               if len(self.cmd_args) == 1:
+                   print self.device.GetRemoteRevision(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> GetRemoteRevision address' % self.name
+           elif self.cmd == 'GetRemoteManufacturer':
+               if len(self.cmd_args) == 1:
+                   print self.device.GetRemoteManufacturer(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> GetRemoteManufacturer address' % self.name
+           elif self.cmd == 'GetRemoteCompany':
+               if len(self.cmd_args) == 1:
+                   print self.device.GetRemoteCompany(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> GetRemoteCompany address' % self.name
+           elif self.cmd == 'GetRemoteAlias':
+               if len(self.cmd_args) == 1:
+                   print self.device.GetRemoteAlias(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> GetRemoteAlias address' % self.name
+           elif self.cmd == 'GetRemoteMajorClass':
+               if len(self.cmd_args) == 1:
+                   print self.device.GetRemoteMajorClass(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> GetRemoteMajorClass address' % self.name
+           elif self.cmd == 'GetRemoteMinorClass':
+               if len(self.cmd_args) == 1:
+                   print self.device.GetRemoteMinorClass(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> GetRemoteMinorClass address' % self.name
+           elif self.cmd == 'GetRemoteServiceClasses':
+               if len(self.cmd_args) == 1:
+                   print self.device.GetRemoteServiceClasses(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> GetRemoteServiceClasses address' % self.name
+           elif self.cmd == 'SetRemoteAlias':
+               if len(self.cmd_args) == 2:
+                   self.device.SetRemoteAlias(self.cmd_args[0], self.cmd_args[1])
+               else:
+                   print 'Usage: %s -i <dev> SetRemoteAlias address alias' % self.name
+           elif self.cmd == 'ClearRemoteAlias':
+               if len(self.cmd_args) == 1:
+                   print self.device.ClearRemoteAlias(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> ClearRemoteAlias address' % self.name
+           elif self.cmd == 'LastSeen':
+               if len(self.cmd_args) == 1:
+                   print self.device.LastSeen(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> LastSeen address' % self.name
+           elif self.cmd == 'LastUsed':
+               if len(self.cmd_args) == 1:
+                   print self.device.LastUsed(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> LastUsed address' % self.name
+           elif self.cmd == 'DisconnectRemoteDevice':
+               if len(self.cmd_args) == 1:
+                   print self.device.LastUsed(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> DisconnectRemoteDevice address' % self.name
+           elif self.cmd == 'CreateBonding':
+               if len(self.cmd_args) == 1:
+                   print self.device.CreateBonding(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> CreateBonding address' % self.name
+           elif self.cmd == 'RemoveBonding':
+               if len(self.cmd_args) == 1:
+                   print self.device.RemoveBonding(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> RemoveBonding address' % self.name
+           elif self.cmd == 'CancelBondingProcess':
+               if len(self.cmd_args) == 1:
+                   print self.device.CancelBondingProcess(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> CancelBondingProcess address' % self.name
+           elif self.cmd == 'HasBonding':
+               if len(self.cmd_args) == 1:
+                   print self.device.HasBonding(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> HasBonding address' % self.name
+           elif self.cmd == 'ListBondings':
+               bondings = self.device.ListBondings()
+               for bond in bondings: 
+                   print bond,
+           elif self.cmd == 'GetPinCodeLength':
+               if len(self.cmd_args) == 1:
+                   print self.device.GetPinCodeLength(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> GetPinCodeLength address' % self.name
+           elif self.cmd == 'GetEncryptionKeySize':
+               if len(self.cmd_args) == 1:
+                   print self.device.GetEncryptionKeySize(self.cmd_args[0])
+               else:
+                   print 'Usage: %s -i <dev> GetEncryptionKeySize address' % self.name
+           elif self.cmd == 'DiscoverDevices':
+               print self.device.DiscoverDevices()
+           elif self.cmd == 'DiscoverDevicesWithoutNameResolving':
+               print self.device.DiscoverDevicesWithoutNameResolving()
+           else:
+                # FIXME: remove at future version
+                print 'Script Error: Method %s not found. Maybe a mispelled word.' % (self.cmd_args)
+       except dbus.DBusException, e:
+           print '%s failed: %s' % (self.cmd, e)
+           sys.exit(1)
+
+    def run(self):
+        # Manager methods
+        if self.listen:
+            self.dbus_mgr_sig_setup()
+            self.dbus_dev_sig_setup()
+            print 'Listening for events...'
+        elif self.cmd in mgr_cmds:
+            try:
+                self.dbus_mgr_setup()
+            except dbus.DBusException, e:
+                print 'Failed to setup manager interface: %s' % e
+                sys.exit(1)
+            self.call_mgr_dbus_func()
+        elif self.cmd in dev_cmds:
+            try:
+                self.dbus_dev_setup()
+            except dbus.DBusException, e:
+                print 'Failed to setup device interface: %s' % e
+                sys.exit(1)
+            self.call_dev_dbus_func()
+        else:
+            print 'Unknown command: %s' % self.cmd
+            self.usage()
+            sys.exit(1)
+
+        if self.listen:
+            signal(SIGINT, self.signal_cb)
+            signal(SIGTERM, self.signal_cb)
+            self.main_loop = gobject.MainLoop()
+            self.main_loop.run()
+
+if __name__ == '__main__':
+    gobject.threads_init()
+    dbus.glib.init_threads()
+
+    tester = Tester(sys.argv)
+    tester.run()