Python

from dbustesting import *class HelloService(MockService):	@dbus.service.method("org.freedesktop.HelloWorldIFace")	def Hello(self):		self.called("Hello")		return "olleH"	@dbus.service.method("org.freedesktop.HelloWorldIFace")	def Hallo(self, param):		self.called("Hallo", param)		self.HalloSig(param)		return param		@dbus.service.signal("org.freedesktop.HelloWorldIFace")	def HalloSig(self, s):		pass		class HelloServiceTest(ServiceTestCase):	def run(self):		self["hello"] = (HelloService, "org.freedesktop.HelloWorld","/org/freedesktop/HelloWorldObject")					self["hello"].WaitCall("Hello")		self["hello"].WaitCall("Hallo")		self["hello"].HalloSig("test")class HelloClientTest(ClientTestCase):	def run(self):		self["hello"] = ("org.freedesktop.HelloWorld","/org/freedesktop/HelloWorldObject", "org.freedesktop.HelloWorldIFace")					echo = self["hello"].Hello()		assert echo == "olleH"				self["hello|HalloSig"].Listen()		echo = self["hello"].Hallo("echomethis")		assert echo == "echomethis"		sigs = self["hello|HalloSig"].Wait()		assert sigs[0] == "echomethis"				self["hello|HalloSig"].Listen()		sigs = self["hello|HalloSig"].Wait()		assert sigs[0] == "test"class HelloMixedTest(MixedTestCase):	def run(self):		self["hello"] = (HelloService, "org.freedesktop.HelloWorld","/org/freedesktop/HelloWorldObject", "org.freedesktop.HelloWorldIFace")		echo = self["hello"].Hello()		assert echo == "olleH"				self["hello|HalloSig"].Listen()		echo = self["hello"].Hallo("echomethis")		assert echo == "echomethis"				sigs = self["hello|HalloSig"].Wait()		assert sigs[0] == "echomethis"				self["hello|HalloSig"].Listen()		self["hello|HalloSig"].Emit("test")		sigs = self["hello|HalloSig"].Wait()		assert sigs[0] == "test"		class NotifyClientTest(ClientTestCase):	def run(self):		self["notif"] = ("org.freedesktop.Notifications","/org/freedesktop/Notifications","org.freedesktop.Notifications")					id = self["notif"].Notify("notifname", 0, "", "title", "content",[],{}, 0)		self["notif|NotificationClosed"].Listen()		self["notif"].CloseNotification(id)		results = self["notif|NotificationClosed"].Wait()		assert results[0] == idfrom telepathytesting import *class TelepathyClientTest(GenericTelepathyClientTest):	def run(self):		# Test Setup		self.start_manager("client1")		self.start_manager("client2")				# Connect the two accounts		conn1 = self.connect("client1", "jabber",			{"account": "kikidonk@amessage.be", "password": "kikidonk"})		conn2 = self.connect("client2", "google-talk",			{ "account": "tryggve.tryggvason@gmail.com","password": "badger"})				# Start listening to NewChannel signal emitted when client2 starts to talk to client1		self["client1-conn|NewChannel"].Listen()				# Create a text channel		self["client2-chan"] = (			# The connection service			conn2,			# A new text channel to the client 1			self["client2-conn"].RequestChannel(				"org.freedesktop.Telepathy.Channel.Type.Text", 1,				self["client2-conn"].RequestHandle (1, "kikidonk@amessage.be"),	False), 			# The interfaces for this text channel			{"text": "org.freedesktop.Telepathy.Channel.Type.Text", "chan": "org.freedesktop.Telepathy.Channel"})				# Send a test message to client 1 from client 2		self["client2-chan"]["text"].Send (0, "Test")				# We got a newchannel from client 1		obj, channel_type, handle_type, handle, suppress_handler = self["client1-conn|NewChannel"].Wait()				assert channel_type == "org.freedesktop.Telepathy.Channel.Type.Text"		assert handle_type == 1		assert suppress_handler == False				# Create the channel structure for client 1		self["client1-chan"] = (			conn1,			obj, 			{"text": "org.freedesktop.Telepathy.Channel.Type.Text", "chan": "org.freedesktop.Telepathy.Channel"})				# Retreive pending messages		messages = self["client1-chan"]["text"].ListPendingMessages()				id, stamp, from_handle, msg_type, string = messages[0]		assert len(messages) == 1		assert from_handle == handle		assert msg_type == 0		assert string == "Test"				# Close/Disconnect stuff		self["client2-chan"]["chan"].Close()		self["client1-chan"]["chan"].Close()		self["client2-conn"].Disconnect()		self["client1-conn"].Disconnect()class TelepathyConnection(MockService):	@dbus.service.method("org.freedesktop.Telepathy.Connection")	def RequestHandle(self, handle_type, name):		self.called("RequestHandle", handle_type, name)		return 3	@dbus.service.method("org.freedesktop.Telepathy.Connection")	def RequestChannel(self, chan_type, handle_type, handle, suppress_handler):		self.called("RequestChannel", chan_type, handle_type, handle, suppress_handler)		self.NewChannel("/org/telepathy/chan_obj2", chan_type, handle_type, handle, suppress_handler)		return "/org/telepathy/chan_obj2"		@dbus.service.method("org.freedesktop.Telepathy.Connection")	def Disconnect(self):		self.called("Disconnect")		pass			@dbus.service.signal("org.freedesktop.Telepathy.Connection")	def NewChannel(self, obj, channel_type, handle_type, handle, suppress_handler):		print 'Emitting NewChannel'		passclass TelepathyChannel(MockService):	def __init__(self, *args):		MockService.__init__(self, *args)		self.messages = []			@dbus.service.method("org.freedesktop.Telepathy.Channel.Type.Text")	def Send(self, text_type, msg):		self.called("Send", text_type, msg)		self.messages.append((len(self.messages), 123, 3, text_type, msg))			@dbus.service.method("org.freedesktop.Telepathy.Channel.Type.Text")	def ListPendingMessages(self):		self.called("ListPendingMessages")		#id, stamp, from_handle, msg_type, string		return self.messages		@dbus.service.method("org.freedesktop.Telepathy.Channel")	def Close(self):		self.called("Close")	class TelepathyMixedTest(MixedTestCase):	def run(self):		self["client1-conn"] = (TelepathyConnection, "org.freedesktop.Telepathy.Connection.testconn", "/org/telepathy/testconn", "org.freedesktop.Telepathy.Connection")		self["client2-conn"] = (TelepathyConnection, "org.freedesktop.Telepathy.Connection.testconn2", "/org/telepathy/testconn2", "org.freedesktop.Telepathy.Connection")				# Start listening to NewChannel signal emitted when client2 starts to talk to client1		self["client1-conn|NewChannel"].Listen()				# Create a text channel		self["client2-chan"] = (			# The service class			TelepathyChannel,			# The connection service			"org.freedesktop.Telepathy.Connection.testconn2",			# A new text channel to the client 1			self["client2-conn"].RequestChannel(				"org.freedesktop.Telepathy.Channel.Type.Text", 1,				self["client2-conn"].RequestHandle (1, "kikidonk@amessage.be"),	False), 			# The interfaces for this text channel			{"text": "org.freedesktop.Telepathy.Channel.Type.Text", "chan": "org.freedesktop.Telepathy.Channel"})				# Send a test message to client 1 from client 2		self["client2-chan"]["text"].Send (0, "Test")				self["client1-conn|NewChannel"].Emit("/org/telepathy/chan_obj", "org.freedesktop.Telepathy.Channel.Type.Text", 1, 3, False)				# We got a newchannel from client 1		obj, channel_type, handle_type, handle, suppress_handler = self["client1-conn|NewChannel"].Wait()				assert channel_type == "org.freedesktop.Telepathy.Channel.Type.Text"		assert handle_type == 1		assert suppress_handler == False				# Create the channel structure for client 1		self["client1-chan"] = (			TelepathyChannel,			"org.freedesktop.Telepathy.Connection.testconn",			obj, 			{"text": "org.freedesktop.Telepathy.Channel.Type.Text", "chan": "org.freedesktop.Telepathy.Channel"})				# Retreive pending messages		messages = self["client1-chan"]["text"].ListPendingMessages()				id, stamp, from_handle, msg_type, string = messages[0]		assert len(messages) == 1		assert from_handle == handle		assert msg_type == 0		assert string == "Test"				# Close/Disconnect stuff		self["client2-chan"]["chan"].Close()		self["client1-chan"]["chan"].Close()		self["client2-conn"].Disconnect()		self["client1-conn"].Disconnect()		run([HelloServiceTest,HelloClientTest,HelloMixedTest,NotifyClientTest,TelepathyClientTest, TelepathyMixedTest])