test: Add session test script

diff --git a/test/test-session b/test/test-session
new file mode 100755 (executable)
index 0000000..1a425e3
--- /dev/null
+++ b/test/test-session
@@ -0,0 +1,244 @@
+#!/usr/bin/python
+
+import sys
+import gobject
+
+import dbus
+import dbus.service
+import dbus.mainloop.glib
+
+import glib
+
+import traceback
+
+def extract_list(list):
+       val = "["
+       for i in list:
+               val += " " + str(i)
+       val += " ]"
+       return val
+
+def extract_values(values):
+       val = "{"
+       for key in values.keys():
+               val += " " + key + "="
+               if key in ["PrefixLength"]:
+                       val += "%s" % (int(values[key]))
+               else:
+                       if key in ["Servers", "Excludes"]:
+                               val += extract_list(values[key])
+                       else:
+                               val += str(values[key])
+       val += " }"
+       return val
+
+class Notification(dbus.service.Object):
+       def __init__(self, bus, app, notify_path):
+               dbus.service.Object.__init__(self)
+               self.app = app
+
+       @dbus.service.method("net.connman.Notification",
+                               in_signature='', out_signature='')
+       def Release(self):
+               print("Release")
+               self.app.release()
+
+       @dbus.service.method("net.connman.Notification",
+                               in_signature='a{sv}', out_signature='')
+       def Update(self, settings):
+               print "Update called"
+
+               try:
+                       for key in settings.keys():
+                               if key in ["IPv4", "IPv6"]:
+                                       val = extract_values(settings[key])
+                               elif key in  ["AllowedBearers"]:
+                                       val = extract_list(settings[key])
+                               else:
+                                       val = settings[key]
+                               print "    %s = %s" % (key, val)
+               except:
+                       print "Exception:"
+                       traceback.print_exc()
+
+class SessionApplication(dbus.service.Object):
+       def __init__(self, bus, object_path, mainloop):
+               dbus.service.Object.__init__(self, bus, object_path)
+
+               self.manager = None
+               self.notify_path = object_path + "/notify"
+               self.notify = None
+               self.session_path = None
+               self.mainloop = mainloop
+               self.session = None
+
+               try:
+                       bus = dbus.SystemBus()
+                       bus.watch_name_owner('net.connman', self.connman_name_owner_changed)
+               except dbus.DBusException:
+                       traceback.print_exc()
+                       exit(1)
+
+       def connman_name_owner_changed(self, proxy):
+               try:
+                       if proxy:
+                               print "connman appeared on D-Bus ", str(proxy)
+
+                               bus = dbus.SystemBus()
+                               self.manager = dbus.Interface(bus.get_object("net.connman", "/"),
+                                                             "net.connman.Manager")
+                       else:
+                               print "connman disappeared on D-Bus"
+                               self.manager = None
+                               self.session = None
+                               if self.notify:
+                                       self.notify.remove_from_connection()
+                                       self.notify = None
+
+               except dbus.DBusException:
+                       traceback.print_exc()
+                       exit(1)
+
+       def release(self):
+               if self.session:
+                       self.manager.DestroySession(self.session_path)
+                       self.session = None
+               if self.notify:
+                       self.notify.remove_from_connection()
+                       self.notify = None
+
+       @dbus.service.method("com.example.TestSession",
+                               in_signature='', out_signature='')
+       def CreateSession(self):
+               print "Create session"
+
+               if self.session:
+                       print "already running a session -> drop reqest"
+                       return
+
+               try:
+                       bus = dbus.SystemBus()
+
+                       self.notify = Notification(bus, self, self.notify_path)
+                       self.notify.add_to_connection(bus, self.notify_path)
+
+                       #settings = { "AllowedBearers" : [ "ethernet", "*" ] }
+                       settings = { "AllowedBearers" : [ "ethernet" ] }
+                       self.session_path = self.manager.CreateSession(settings, self.notify_path)
+
+                       print "notify path %s" % (self.notify_path)
+                       print "session path %s" % (self.session_path)
+
+                       self.session = dbus.Interface(bus.get_object("net.connman", self.session_path),
+                                                     "net.connman.Session")
+
+               except dbus.DBusException:
+                       traceback.print_exc()
+                       exit(1)
+
+       @dbus.service.method("com.example.TestSession",
+                               in_signature='', out_signature='')
+       def DestroySession(self):
+               print "Destroy session"
+
+               if self.session == None:
+                       print "no session running -> drop request"
+                       return
+
+               try:
+                       self.release()
+               except dbus.DBusException:
+                       traceback.print_exc()
+                       exit(1)
+
+       @dbus.service.method("com.example.TestSession",
+                               in_signature='', out_signature='')
+       def Connect(self):
+               print "Connect session"
+               try:
+                       self.session.Connect()
+               except dbus.DBusException, e:
+                       if e.get_dbus_name() in ['net.connman.Error.InProgress',
+                                                'net.connman.Error.AlreadyConnected']:
+                               print e.get_dbus_message()
+                               return
+                       traceback.print_exc()
+                       exit(1)
+
+       @dbus.service.method("com.example.TestSession",
+                               in_signature='', out_signature='')
+       def Disconnect(self):
+               print "Disconnect session"
+               try:
+                       self.session.Disconnect()
+               except dbus.DBusException, e:
+                       if e.get_dbus_name() in ['net.connman.Error.AlreadyDisabled',
+                                                'net.connman.Error.NotConnected']:
+                               print e.get_dbus_message()
+                               return
+                       traceback.print_exc()
+                       exit(1)
+
+def main():
+       if len(sys.argv) < 2:
+               print "Usage: %s <command>" % (sys.argv[0])
+               print ""
+               print "  enable"
+               print "  disable"
+               print "  create <app_path>"
+               print "  destroy <app_path>"
+               print "  connect <app_path>"
+               print "  disconnect <app_path>"
+               print ""
+               print "  run <app_path>"
+               sys.exit(1)
+
+       dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
+
+       if sys.argv[1] == "enable":
+               bus = dbus.SystemBus()
+               manager = dbus.Interface(bus.get_object("net.connman", "/"),
+                                        "net.connman.Manager")
+               manager.SetProperty("SessionMode", True)
+               return
+
+       elif sys.argv[1] == "disable":
+               bus = dbus.SystemBus()
+               manager = dbus.Interface(bus.get_object("net.connman", "/"),
+                                        "net.connman.Manager")
+               manager.SetProperty("SessionMode", False)
+               return
+
+       if (len(sys.argv) < 3):
+               print "Need test application path"
+               sys.exit(1)
+
+       app_path = sys.argv[2]
+       bus = dbus.SessionBus()
+
+       if sys.argv[1] == "run":
+               name = dbus.service.BusName("com.example.SessionApplication", bus)
+               mainloop = gobject.MainLoop()
+
+               app = SessionApplication(bus, app_path, mainloop)
+
+               mainloop.run()
+               return
+
+       app = dbus.Interface(bus.get_object("com.example.SessionApplication", app_path),
+                            "com.example.TestSession")
+
+       if sys.argv[1] == "create":
+               app.CreateSession()
+
+       elif sys.argv[1] == "destroy":
+               app.DestroySession()
+
+       elif sys.argv[1] == "connect":
+               app.Connect()
+
+       elif sys.argv[1] == "disconnect":
+               app.Disconnect()
+
+if __name__ == '__main__':
+       main()