test: Add session test script

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()