diff -r 85fc1e5dea79 -r 67ecea2d9b67 iro/tests/testXMLRPCServer.py --- a/iro/tests/testXMLRPCServer.py Thu Nov 03 22:47:30 2011 +0100 +++ b/iro/tests/testXMLRPCServer.py Fri Nov 18 22:36:50 2011 +0100 @@ -1,115 +1,16 @@ # -*- coding: utf-8 -*- -import unittest, ConfigParser - -import threading, xmlrpclib -from multiprocessing import Queue -from multiprocessing.managers import BaseManager - -from iro import xmlrpc,anbieter -from iro.iro import MySMTP,MySmstrade,MyUserDB -from iro.job import SMSJob, FAXJob, MailJob -from iro.joblist import Joblist -from iro.providerlist import Providerlist -from iro.user import User, Admin - - -class StoppableXMLRPCServer(xmlrpc.SecureUserDBXMLRPCServer, threading.Thread): - running=False - def __init__(self, *args, **kwargs): - xmlrpc.SecureUserDBXMLRPCServer.__init__(self, *args, **kwargs) - threading.Thread.__init__(self) - self.timeout=.5 - - def start(self): - self.running=True - threading.Thread.start(self) - - - def run(self): - # *serve_forever* muss in einem eigenen Thread laufen, damit man es - # unterbrochen werden kann! - while (self.running): - self.handle_request() - - def stop(self): - if (self.running): - self.running=False - self.server_close() - self.join() - - def __del__(self): - self.stop() - - def __enter__(self): - self.start() - return self - - def __exit__(self,type,value,traceback): - self.stop() - - +import unittest -def init_server(): - userlist=[{"name":"test","password":"test", "class":User}, - {"name":"test2","password":"test2", "class": User}, - {"name":"admin","password":"admin", "class": Admin}] - - - - class MyManager(BaseManager): - pass - - MyManager.register('SMSJob', SMSJob) - MyManager.register('FAXob', FAXJob) - MyManager.register('MailJob',MailJob) - MyManager.register('Providerlist',Providerlist) - manager = MyManager() - manager.start() - - - #anbieter erzeugen und konfigurieren - sip=anbieter.sipgate() - sip.read_basic_config("iro.conf") - - localhost=MySMTP() - localhost.read_basic_config("iro.conf") - - smstrade=MySmstrade() - smstrade.read_basic_config("iro.conf") +import xmlrpclib +from stopableServer import init_server - #Benutzerdatenbank erstellen - queue = Queue() - provider=Providerlist() - provider.add("sipgate", sip, ["sms", "fax", ]) - provider.add("smstrade", smstrade, ["sms", ]) - provider.add("geonet", None, ["sms", "fax", ]) - provider.add("fax.de", None, ["sms", "fax", ]) - provider.add("localhost", localhost, ["mail", ]) - provider.setDefault("sms","smstrade") - provider.setDefault("fax","sipgate") - provider.setDefault("mail","localhost") - jobqueue=Joblist(manager, queue, provider) - userdb=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=cert,privatekey=key, - logRequests=False) - serv.relam="xmlrpc" - return serv - - 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):