testJob - prüft, ob SMS, Fax und MailJobs richtig in der Queue landen
anpassungen bei user.py:
fax kann nun als liste übergeben werden, oder als reiner string
--- a/iro/tests/stopableServer.py Thu Nov 18 02:58:28 2010 +0100
+++ b/iro/tests/stopableServer.py Thu Nov 18 03:00:47 2010 +0100
@@ -50,6 +50,8 @@
self.stop()
+class Internal:
+ pass
def init_server():
userlist=[{"name":"test","password":"test", "class":User},
@@ -60,14 +62,17 @@
class MyManager(BaseManager):
pass
+
+ internal=Internal()
MyManager.register('SMSJob', SMSJob)
- MyManager.register('FAXob', FAXJob)
+ MyManager.register('FaxJob', FAXJob)
MyManager.register('MailJob',MailJob)
MyManager.register('Providerlist',Providerlist)
manager = MyManager()
manager.start()
+ internal.manager=manager
#anbieter erzeugen und konfigurieren
sip=anbieter.sipgate()
@@ -81,7 +86,9 @@
#Benutzerdatenbank erstellen
queue = Queue()
+ internal.queue=queue
provider=Providerlist()
+ internal.provider=provider
provider.add("sipgate", sip, ["sms", "fax", ])
provider.add("smstrade", smstrade, ["sms", ])
provider.add("geonet", None, ["sms", "fax", ])
@@ -91,7 +98,9 @@
provider.setDefault("fax","sipgate")
provider.setDefault("mail","localhost")
jobqueue=Joblist(manager, queue, provider)
+ internal.jobqueue=jobqueue
userdb=MyUserDB(userlist,jobqueue)
+ internal.userdb=userdb
#Server starten
@@ -104,5 +113,6 @@
certificate=cert,privatekey=key,
logRequests=False)
serv.relam="xmlrpc"
- return serv
+ internal.serv=serv
+ return internal
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/iro/tests/testJob.py Thu Nov 18 03:00:47 2010 +0100
@@ -0,0 +1,82 @@
+# -*- coding: utf-8 -*-
+
+import unittest
+
+import xmlrpclib
+from stopableServer import init_server
+from iro.anbieter.content import SMS,FAX,Mail
+
+class TestServer(unittest.TestCase):
+
+ def setUp(self):
+ self.i = init_server()
+ self.serv=self.i.serv
+ self.serv.start()
+
+ def tearDown(self):
+ self.serv.stop()
+
+
+ def SendSMS(self,msg):
+ servstr="https://test:test@localhost:8000"
+ client=xmlrpclib.Server(servstr)
+ id=client.startSMS(msg,["01234", ] )
+ self.assertEqual(client.status(id),{id: {'status': ['init',{}], 'name': unicode(msg)}} )
+ ele=self.i.queue.get(.1)
+ self.assertEqual(ele.getRecipients(),["01234", ] )
+ self.assertNotEqual(ele.getMessage(),SMS('') )
+ self.assertEqual(ele.getMessage(),SMS(msg) )
+
+ def testSimpleSMS(self):
+ self.SendSMS("test")
+
+ def testSpecialCharacters(self):
+ self.SendSMS(u"!\"§$%&/()=?\'")
+ self.SendSMS(u"@ł€ł€¶ŧł¼¼½¬¬↓ŧ←ĸ↓→øđŋħ“”µ·…–|")
+
+ def testSendFAX(self):
+ servstr="https://test:test@localhost:8000"
+ client=xmlrpclib.Server(servstr)
+ msg="2134wergsdfg4w56q34134æſðđæðſđŋ³@¼ŧæðđŋł€¶ŧ€¶ŧ"
+ id=client.startFAX("test",xmlrpclib.Binary(msg),["01234", ] )
+ self.assertEqual(client.status(id),{id: {'status': ['init',{}], 'name': 'test'}} )
+ ele=self.i.queue.get(.1)
+ self.assertEqual(ele.getRecipients(),["01234", ] )
+ self.assertEqual(ele.getMessage(),FAX('test','',[msg]))
+
+ def testDoubleFAX(self):
+ servstr="https://test:test@localhost:8000"
+ client=xmlrpclib.Server(servstr)
+ msg="2134wergsdfg4w56q34134æſðđæðſđŋ³@¼ŧæðđŋł€¶ŧ€¶ŧ"
+ pdf=open('tests/test.pdf').read()
+ id=client.startFAX("test",[xmlrpclib.Binary(msg),xmlrpclib.Binary(pdf)],["01234", ] )
+ self.assertEqual(client.status(id),{id: {'status': ['init',{}], 'name': 'test'}} )
+ ele=self.i.queue.get(.1)
+ self.assertEqual(ele.getRecipients(),["01234", ] )
+ self.assertEqual(ele.getMessage(),FAX('test','',[msg, pdf]))
+
+ def testSendMail(self):
+ servstr="https://test:test@localhost:8000"
+ client=xmlrpclib.Server(servstr)
+ msg=u"2134wergsdfg4w56q34134æſðđæðſđŋ³@¼ŧæðđŋł€¶ŧ€¶ŧ"
+ id=client.startMail("test",msg,["test@test.de", ],'absender@test.de' )
+ self.assertEqual(client.status(id),{id: {'status': ['init',{}], 'name': 'test'}} )
+ ele=self.i.queue.get(.1)
+ self.assertEqual(ele.getRecipients(),["test@test.de", ] )
+ self.assertEqual(ele.getMessage(),Mail('test',msg,'absender@test.de'))
+ self.assertEqual(ele.getMessage().as_string(),"""Content-Type: text/plain; charset="utf-8"
+MIME-Version: 1.0
+Content-Transfer-Encoding: base64
+Subject: =?utf-8?q?test?=
+
+MjEzNHdlcmdzZGZnNHc1NnEzNDEzNMOmxb/DsMSRw6bDsMW/xJHFi8KzQMK8xafDpsOwxJHFi8WC
+4oKswrbFp+KCrMK2xac=
+""")
+ sub=u"³¼½ſðđŋſ€¼½ÖÄÜß"
+ id=client.startMail(sub,msg,["test@test.de", ],'absender@test.de' )
+ self.assertEqual(client.status(id),{id: {'status': ['init',{}], 'name': sub}})
+ ele=self.i.queue.get(.1)
+ self.assertEqual(ele.getMessage(),Mail(sub, msg, 'absender@test.de'))
+
+if __name__ == "__main__":
+ unittest.main()
--- a/iro/tests/testXMLRPCServer.py Thu Nov 18 02:58:28 2010 +0100
+++ b/iro/tests/testXMLRPCServer.py Thu Nov 18 03:00:47 2010 +0100
@@ -8,7 +8,9 @@
class TestServer(unittest.TestCase):
def setUp(self):
- self.serv = init_server()
+ self.i = init_server()
+ self.serv=self.i.serv
+
self.serv.start()
def tearDown(self):
--- a/iro/user.py Thu Nov 18 02:58:28 2010 +0100
+++ b/iro/user.py Thu Nov 18 03:00:47 2010 +0100
@@ -94,7 +94,12 @@
logger.debug("startFAX(%s,%s,%s,%s)"%(subject, fax, recipients, provider))
if not "fax" in self.features:
raise NotSupportedFeature("fax")
- id = self.jobqueue.newFAX(subject, fax,recipients,provider)
+
+ if type(fax) != list:
+ fax=[fax]
+ f=[i.data for i in fax]
+
+ id = self.jobqueue.newFAX(subject, f,recipients,provider)
self.jobs[id]=self.jobqueue[id]
return id