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] == id

from 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'
		pass

class 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])