# HG changeset patch # User Sandro Knauß # Date 1286235932 -7200 # Node ID 31a2b1cd4981b33749e80504d49b4e97bd55ac23 # Parent 254c00f3cbe024aad0a1ce383290d47a4dc0b48f anpassung der tests auf ein eigenes Verzeichnis diff -r 254c00f3cbe0 -r 31a2b1cd4981 iro/anbieter/test.py --- a/iro/anbieter/test.py Mon Oct 04 04:28:28 2010 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,59 +0,0 @@ -# -*- coding: utf-8 -*- -#Copyright (C) 2009 Sandro Knauß - -#This program is free software; you can redistribute it and/or modify it under the terms -#of the GNU General Public License as published by the Free Software Foundation; -#either version 3 of the License, or any later version. -#This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; -#without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. -#See the GNU General Public License for more details. - -#You should have received a copy of the GNU General Public License -#along with this program; if not, see . - -class anbieter: - default_conf = '' # override this - -import unittest -import telnumber as tn - -class TestTelnumber(unittest.TestCase): - def equalNumber(self, tel1, tel2): - self.assertEqual(tel1.number, tel2.number) - self.assertEqual(tel1.land, tel2.land) - - def testWrongNumber(self): - telnum=tn.telnumber() - self.assertRaises(tn.NotATelNumber, telnum.createNumber, "hallo") - self.assertRaises(tn.NotATelNumber, telnum.createNumber, "0?242") - - - def testNumber(self): - telnum=tn.telnumber("0551-28293640") - telnum2=tn.telnumber("+49551/28293640") - telnum3=tn.telnumber("00495512829364-0") - telnum4=tn.telnumber("+49(0)551-28293640") - - self.assertEqual(telnum.land, "49") - self.assertEqual(telnum.number, "55128293640") - - self.equalNumber(telnum, telnum2) - self.equalNumber(telnum, telnum3) - self.equalNumber(telnum, telnum4) - - def testEqual(self): - telnum=tn.telnumber("0551-28293640") - telnum2=tn.telnumber("+49551/28293640") - li=[] - self.assertEqual(telnum == telnum2, True) - self.assertEqual(telnum <> telnum2, False) - self.assertEqual(telnum, telnum2) - self.assertEqual(telnum in li,False) - li.append(telnum) - self.assertEqual(telnum in li,True) - self.assertEqual(telnum in li,True) - -if __name__ == "__main__": - unittest.main() - - diff -r 254c00f3cbe0 -r 31a2b1cd4981 iro/anbieter/tests/testTelnumber.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/iro/anbieter/tests/testTelnumber.py Tue Oct 05 01:45:32 2010 +0200 @@ -0,0 +1,59 @@ +# -*- coding: utf-8 -*- +#Copyright (C) 2009 Sandro Knauß + +#This program is free software; you can redistribute it and/or modify it under the terms +#of the GNU General Public License as published by the Free Software Foundation; +#either version 3 of the License, or any later version. +#This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; +#without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +#See the GNU General Public License for more details. + +#You should have received a copy of the GNU General Public License +#along with this program; if not, see . + +class anbieter: + default_conf = '' # override this + +import unittest +import iro.anbieter.telnumber as tn + +class TestTelnumber(unittest.TestCase): + def equalNumber(self, tel1, tel2): + self.assertEqual(tel1.number, tel2.number) + self.assertEqual(tel1.land, tel2.land) + + def testWrongNumber(self): + telnum=tn.telnumber() + self.assertRaises(tn.NotATelNumber, telnum.createNumber, "hallo") + self.assertRaises(tn.NotATelNumber, telnum.createNumber, "0?242") + + + def testNumber(self): + telnum=tn.telnumber("0551-28293640") + telnum2=tn.telnumber("+49551/28293640") + telnum3=tn.telnumber("00495512829364-0") + telnum4=tn.telnumber("+49(0)551-28293640") + + self.assertEqual(telnum.land, "49") + self.assertEqual(telnum.number, "55128293640") + + self.equalNumber(telnum, telnum2) + self.equalNumber(telnum, telnum3) + self.equalNumber(telnum, telnum4) + + def testEqual(self): + telnum=tn.telnumber("0551-28293640") + telnum2=tn.telnumber("+49551/28293640") + li=[] + self.assertEqual(telnum == telnum2, True) + self.assertEqual(telnum <> telnum2, False) + self.assertEqual(telnum, telnum2) + self.assertEqual(telnum in li,False) + li.append(telnum) + self.assertEqual(telnum in li,True) + self.assertEqual(telnum in li,True) + +if __name__ == "__main__": + unittest.main() + + diff -r 254c00f3cbe0 -r 31a2b1cd4981 iro/test.py --- a/iro/test.py Mon Oct 04 04:28:28 2010 +0200 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,166 +0,0 @@ -# -*- coding: utf-8 -*- - -import unittest, ConfigParser -import iro -from job import SMSJob, FAXJob, MailJob -from joblist import Joblist -from providerlist import Providerlist -import threading, xmlrpclib -from multiprocessing import Queue -from multiprocessing.managers import BaseManager - -class StoppableXMLRPCServer(iro.SecureUserDBXMLRPCServer, threading.Thread): - running=False - def __init__(self, *args, **kwargs): - iro.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() - - - -def init_server(): - 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('SMSJob', SMSJob) - MyManager.register('FAXob', FAXJob) - MyManager.register('MailJob',MailJob) - MyManager.register('Providerlist',Providerlist) - manager = MyManager() - manager.start() - - - #anbieter erzeugen und konfigurieren - import anbieter - sip=anbieter.sipgate() - sip.read_basic_config("iro.conf") - - localhost=iro.MySMTP() - localhost.read_basic_config("iro.conf") - - smstrade=iro.MySmstrade() - smstrade.read_basic_config("iro.conf") - - #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=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=cert,privatekey=key, - logRequests=False) - serv.relam="xmlrpc" - return serv - - -class TestServer(unittest.TestCase): - - def setUp(self): - self.serv = init_server() - self.serv.start() - - def tearDown(self): - self.serv.stop() - - def testLogin(self): - self.assertEqual(xmlrpclib.Server("https://test:test@localhost:8000").status(), {}) - self.assertEqual(xmlrpclib.Server("https://test2:test2@localhost:8000").status(), {}) - self.assertRaises(xmlrpclib.ProtocolError, xmlrpclib.Server("https://test2:test@localhost:8000").status) - 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': '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': '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.assertEqual(client2.status(id1), {}) - self.assertEqual(client1.status(id2), {}) - - def testGetProvider(self): - servstr="https://test:test@localhost:8000" - client=xmlrpclib.Server(servstr) - self.assertEqual(client.getProvider("sms"), ["fax.de","geonet", "sipgate", "smstrade"]) - self.assertEqual(client.getProvider("fax"), ["fax.de","geonet", "sipgate"]) - self.assertEqual(client.getProvider("mail"), ["localhost"]) - - self.assertRaises(xmlrpclib.ProtocolError,client.getProvider, "temp") - - def testGetDefault(self): - servstr="https://test:test@localhost:8000" - client=xmlrpclib.Server(servstr) - self.assertEqual(client.getDefaultProvider("sms"), "smstrade") - self.assertEqual(client.getDefaultProvider("fax"),"sipgate") - self.assertEqual(client.getDefaultProvider("mail"), "localhost") - - self.assertRaises(xmlrpclib.ProtocolError,client.getDefaultProvider, "temp") - - -if __name__ == "__main__": - unittest.main() diff -r 254c00f3cbe0 -r 31a2b1cd4981 iro/tests/testXMLRPCServer.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/iro/tests/testXMLRPCServer.py Tue Oct 05 01:45:32 2010 +0200 @@ -0,0 +1,170 @@ +# -*- 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() + + + +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") + + #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.serv.start() + + def tearDown(self): + self.serv.stop() + + def testLogin(self): + self.assertEqual(xmlrpclib.Server("https://test:test@localhost:8000").status(), {}) + self.assertEqual(xmlrpclib.Server("https://test2:test2@localhost:8000").status(), {}) + self.assertRaises(xmlrpclib.ProtocolError, xmlrpclib.Server("https://test2:test@localhost:8000").status) + 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': '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': '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.assertEqual(client2.status(id1), {}) + self.assertEqual(client1.status(id2), {}) + + def testGetProvider(self): + servstr="https://test:test@localhost:8000" + client=xmlrpclib.Server(servstr) + self.assertEqual(client.getProvider("sms"), ["fax.de","geonet", "sipgate", "smstrade"]) + self.assertEqual(client.getProvider("fax"), ["fax.de","geonet", "sipgate"]) + self.assertEqual(client.getProvider("mail"), ["localhost"]) + + self.assertRaises(xmlrpclib.ProtocolError,client.getProvider, "temp") + + def testGetDefault(self): + servstr="https://test:test@localhost:8000" + client=xmlrpclib.Server(servstr) + self.assertEqual(client.getDefaultProvider("sms"), "smstrade") + self.assertEqual(client.getDefaultProvider("fax"),"sipgate") + self.assertEqual(client.getDefaultProvider("mail"), "localhost") + + self.assertRaises(xmlrpclib.ProtocolError,client.getDefaultProvider, "temp") + + +if __name__ == "__main__": + unittest.main()