diff options
Diffstat (limited to 'test/test-session')
-rwxr-xr-x | test/test-session | 244 |
1 files changed, 244 insertions, 0 deletions
diff --git a/test/test-session b/test/test-session new file mode 100755 index 00000000..1a425e3e --- /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() |