--- a/iro/test.py Fri Oct 23 01:15:06 2009 +0200
+++ b/iro/test.py Fri Oct 23 01:15:23 2009 +0200
@@ -1,15 +1,15 @@
# -*- coding: utf-8 -*-
-import unittest
-import server
+import unittest, ConfigParser
+import iro
import threading, xmlrpclib
from multiprocessing import Queue
from multiprocessing.managers import BaseManager
-class StoppableXMLRPCServer(server.SecureUserDBXMLRPCServer, threading.Thread):
+class StoppableXMLRPCServer(iro.SecureUserDBXMLRPCServer, threading.Thread):
running=True
def __init__(self, *args, **kwargs):
- server.SecureUserDBXMLRPCServer.__init__(self, *args, **kwargs)
+ iro.SecureUserDBXMLRPCServer.__init__(self, *args, **kwargs)
threading.Thread.__init__(self)
@@ -26,16 +26,16 @@
def init_server():
- userlist=[{"name":"test","password":"test", "class":server.User},
- {"name":"test2","password":"test2", "class": server.User},
- {"name":"admin","password":"admin", "class": server.Admin}]
+ userlist=[{"name":"test","password":"test", "class":iro.User},
+ {"name":"test2","password":"test2", "class": iro.User},
+ {"name":"admin","password":"admin", "class": iro.Admin}]
class MyManager(BaseManager):
pass
- MyManager.register('MessageJob', server.MessageJob)
+ MyManager.register('MessageJob', iro.MessageJob)
manager = MyManager()
manager.start()
@@ -44,19 +44,28 @@
import anbieter
sip=anbieter.sipgate()
sip.read_basic_config("iro.conf")
- localhost=""
+
+ localhost=iro.MySMTP()
+ localhost.read_basic_config("iro.conf")
+
+ smstrade=iro.MySmstrade()
+ smstrade.read_basic_config("iro.conf")
#Benutzerdatenbank erstellen
queue = Queue()
provider={"sms":sip, "fax":sip, "mail":localhost}
- jobqueue=server.Jobs(manager, queue, provider)
- userdb=server.MyUserDB(userlist,jobqueue)
+ jobqueue=iro.Jobs(manager, queue, provider)
+ userdb=iro.MyUserDB(userlist,jobqueue)
#Server starten
+ cp = ConfigParser.ConfigParser()
+ cp.read(["iro.conf"])
+ cert=cp.get('server', 'cert')
+ key=cp.get('server', 'key')
serv = StoppableXMLRPCServer(addr=("localhost", 8000),
userdb=userdb,
- certificate="./certs/test.cert.pem",privatekey="./certs/test.key.pem")
+ certificate=cert,privatekey=key)
serv.relam="xmlrpc"
return serv
@@ -78,30 +87,37 @@
self.assertRaises(xmlrpclib.ProtocolError,xmlrpclib.Server ("https://test:test2@localhost:8000").status)
def testsendSMS(self):
- servstr="https://test:test@localhost:8000"
- client=xmlrpclib.Server(servstr)
- id=client.startSMS("test",["01234"] )
- self.assertEqual(client.status(id),{id: {'status': ['init',{}], 'name': {'content': 'test'}}} )
+ servstr="https://test:test@localhost:8000"
+ client=xmlrpclib.Server(servstr)
+ id=client.startSMS("test",["01234"] )
+ self.assertEqual(client.status(id),{id: {'status': ['init',{}], 'name': 'test'}} )
def testTwoUser(self):
- u1="https://test:test@localhost:8000"
- u2="https://test2:test2@localhost:8000"
- admin="https://admin:admin@localhost:8000"
- client1=xmlrpclib.Server(u1)
- client2=xmlrpclib.Server(u2)
- admin=xmlrpclib.Server(admin)
- id1=client1.startSMS("test",["01234"] )
- self.assertEqual(client2.status(),{} )
- self.assertEqual(admin.status(id1),{id1: {'status': ['init',{}], 'name': {'content': 'test'}}} )
- id2=client2.startSMS("test2",["01234"] )
- self.assertNotEqual(id1, id2)
- self.assertEqual(client1.status(),{id1: {'status': ['init',{}], 'name': {'content': 'test'}}})
- self.assertEqual(client2.status(),{id2: {'status': ['init',{}], 'name': {'content': 'test2'}}})
- self.assertEqual(admin.status(),{id1: {'status': ['init',{}], 'name': {'content': 'test'}},
- id2: {'status': ['init',{}], 'name': {'content': 'test2'}}} )
-
- self.assertRaises(xmlrpclib.Fault, client2.status, id1)
+ u1="https://test:test@localhost:8000"
+ u2="https://test2:test2@localhost:8000"
+ admin="https://admin:admin@localhost:8000"
+ client1=xmlrpclib.Server(u1)
+ client2=xmlrpclib.Server(u2)
+ admin=xmlrpclib.Server(admin)
+ id1=client1.startSMS("test",["01234"] )
+ self.assertEqual(client2.status(),{} )
+ self.assertEqual(admin.status(id1),{id1: {'status': ['init',{}], 'name': 'test'}} )
+ id2=client2.startSMS("test2",["01234"] )
+ self.assertNotEqual(id1, id2)
+ self.assertEqual(client1.status(),{id1: {'status': ['init',{}], 'name': 'test'}})
+ self.assertEqual(client2.status(),{id2: {'status': ['init',{}], 'name': 'test2'}})
+ self.assertEqual(admin.status(),{id1: {'status': ['init',{}], 'name': 'test'},
+ id2: {'status': ['init',{}], 'name': 'test2'}} )
+
+ self.assertRaises(xmlrpclib.Fault, client2.status, id1)
self.assertRaises(xmlrpclib.Fault, client1.status, id2)
+
+ def testGetProvider(self):
+ servstr="https://test:test@localhost:8000"
+ client=xmlrpclib.Server(servstr)
+ self.assertEqual(client.getProvider("sms").sort(), ["fax.de","geonet", "sipgate", "smstrade"])
+ self.assertEqual(client.getProvider("fax").sort(), ["fax.de","geonet", "sipgate"])
+ self.assertEqual(client.getProvider("mail").sort(), ["localhost"])
if __name__ == "__main__":
unittest.main()