# HG changeset patch # User Sandro Knauß # Date 1327962494 -3600 # Node ID c51c3e8c3ec0764db05f7910df19671d30af16af # Parent 05e599aa83c3295d183c9234848079e1e04109ab schema tests without xmlrpc interface diff -r 05e599aa83c3 -r c51c3e8c3ec0 tests/db.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/tests/db.py Mon Jan 30 23:28:14 2012 +0100 @@ -0,0 +1,190 @@ +import unittest +from sqlalchemy import create_engine, pool +from tempfile import mkdtemp +import shutil + +from datetime import datetime + +from iro.model import POOL_SIZE as DB_POOL_SIZE +from iro.model.schema import User, Base, Offer, Userright, Job, Message +from decimal import Decimal + +from ngdatabase.mysql import Server, createConfig, Database +from .dbtestcase import DBTestCase + +class SampleDatabase(Database): + def createPassword(self): + self.password="test" + return self.password + +class ModuleData: + def __init__(self): + self.tdir = mkdtemp(prefix='iro-mysql-') + self.server = Server('%s/my.cnf'%self.tdir) + self.db = SampleDatabase("test","test",'%s/my.cnf'%self.tdir) + self.engine = create_engine('mysql://test:test@localhost/test?unix_socket=%s/socket'%self.tdir, + poolclass = pool.SingletonThreadPool, pool_size=DB_POOL_SIZE, )#echo=True) + + def setUp(self): + with open('%s/my.cnf'%self.tdir,'w') as cnf: + cnf.write(createConfig(self.tdir)) + self.server.create() + self.server.start() + self.db.create() + Base.metadata.create_all(self.engine) + + def tearDown(self): + self.server.stop() + shutil.rmtree(self.tdir) + + +md=ModuleData() + +def setUpModule(): + md.setUp() + +def tearDownModule(): + md.tearDown() + +class DBTests(DBTestCase): + """tests for the db model""" + def setUp(self): + if not self.engine: + self.engine = md.engine + + def testRoutes(self): + '''test routes''' + with self.session() as session: + u=User(name='test',apikey='abcdef123456789') + o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms") + u.rights.append(Userright(o)) + session.add(u) + + with self.session() as session: + u=session.merge(u) + self.failUnlessEqual(u.routes('sms').all(),[('sipgate_basic',),]) + + + def testTyps(self): + with self.session() as session: + o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms") + session.add(o) + + with self.session() as session: + self.failUnlessEqual(session.typs.all(),[('sms',),]) + + with self.session() as session: + o=Offer(name="s2", provider="sipgate", route="basic", typ="sms") + o=Offer(name="s3", provider="sipgate", route="basic", typ="sms2") + session.add(o) + + with self.session() as session: + self.failUnlessEqual(session.typs.order_by(Offer.typ).all(),[('sms',),('sms2',)]) + + +class BillTest(DBTestCase): + """test the bill function""" + def setUp(self): + if not self.engine: + self.engine = md.engine + + def testBill(self): + '''test bill function''' + apikey='abcdef123456789' + with self.session() as session: + u=User(name='test',apikey=apikey) + session.add(u) + o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms") + j = Job(hash='a1',info='i',status='sended') + m = Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.30, offer=o, job=j) + u.rights.append(Userright(o)) + u.jobs.append(j) + session.add(m) + + with self.session() as session: + u=session.merge(u) + self.failUnlessEqual(u.rights[0].bill.all(),[(1L,Decimal('0.3000'),'i')]) + + def testBillAgreget(self): + apikey='abcdef123456789' + with self.session() as session: + u=User(name='test',apikey=apikey) + session.add(u) + o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms") + j = Job(hash='a1',info='i',status='sended') + j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.30, offer=o)) + j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o)) + u.rights.append(Userright(o)) + u.jobs.append(j) + + with self.session() as session: + u=session.merge(u) + self.failUnlessEqual(u.rights[0].bill.all(),[(2L,Decimal('0.7000'),'i')]) + + def testBillIsBilled(self): + apikey='abcdef123456789' + with self.session() as session: + u=User(name='test',apikey=apikey) + session.add(u) + o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms") + j = Job(hash='a1',info='i',status='sended') + j.messages.append(Message(recipient='0123456789', isBilled=True, date=datetime.now() , price=0.30, offer=o)) + j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o)) + u.rights.append(Userright(o)) + u.jobs.append(j) + + with self.session() as session: + u=session.merge(u) + self.failUnlessEqual(u.rights[0].bill.all(),[(1L,Decimal('0.4000'),'i')]) + + def testBillMultipleJobs(self): + apikey='abcdef123456789' + with self.session() as session: + u=User(name='test',apikey=apikey) + session.add(u) + o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms") + u.rights.append(Userright(o)) + + j = Job(hash='a1',info='i',status='sended') + j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o)) + u.jobs.append(j) + + j = Job(hash='a2',info='a',status='sended') + j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.4, offer=o)) + u.jobs.append(j) + + with self.session() as session: + u=session.merge(u) + self.failUnlessEqual(u.rights[0].bill.order_by(Job.info).all(), + [(1L,Decimal('0.4000'),'a'), + (1L,Decimal('0.4000'),'i') + ]) + + def testBillMultipleOffers(self): + apikey='abcdef123456789' + with self.session() as session: + u=User(name='test',apikey=apikey) + session.add(u) + o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms") + u.rights.append(Userright(o)) + + j = Job(hash='a2',info='a',status='sended') + j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.4, offer=o)) + u.jobs.append(j) + + o = Offer(name='sipgate_gold',provider="sipgate",route="gold",typ="sms") + u.rights.append(Userright(offer=o)) + + j = Job(hash='a3',info='a',status='sended') + j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.5, offer=o)) + u.jobs.append(j) + + with self.session() as session: + u=session.merge(u) + self.failUnlessEqual(u.rights[0].bill.all(), + [(1L,Decimal('0.4000'),'a')]) + + self.failUnlessEqual(u.rights[1].bill.all(),[(1L,Decimal('0.5000'),'a')]) + +if __name__ == '__main__': + unittest.main() diff -r 05e599aa83c3 -r c51c3e8c3ec0 tests/dbtestcase.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/tests/dbtestcase.py Mon Jan 30 23:28:14 2012 +0100 @@ -0,0 +1,24 @@ +import unittest + +from iro.model import schema +from iro.model.utils import WithSession + +class DBTestCase(unittest.TestCase): + '''a TestCase with DB connection + you have to set self.session manually in setUp''' + def __init__(self,*args,**kwargs): + self.engine=None + unittest.TestCase.__init__(self,*args,**kwargs) + + def tearDown(self): + self.__cleanDB() + + def session(self,autocommit=True): + '''returns a WithSession''' + return WithSession(self.engine,autocommit) + + def __cleanDB(self): + '''cleaning database''' + with self.session() as session: + for table in schema.__tables__: + session.query(getattr(schema,table)).delete() diff -r 05e599aa83c3 -r c51c3e8c3ec0 tests/xmlrpc.py --- a/tests/xmlrpc.py Mon Jan 30 22:15:21 2012 +0100 +++ b/tests/xmlrpc.py Mon Jan 30 23:28:14 2012 +0100 @@ -11,11 +11,9 @@ from xmlrpclib import Server as xServer, ServerProxy, Fault -from iro.model.utils import WithSession from iro.model import POOL_SIZE as DB_POOL_SIZE from iro.model.schema import User, Base, Offer, Userright, Job, Message -import iro.model.schema as schema from iro.main import runReactor @@ -23,6 +21,8 @@ from ngdatabase.mysql import Server, createConfig, Database +from .dbtestcase import DBTestCase + class SampleDatabase(Database): def createPassword(self): self.password="test" @@ -37,9 +37,11 @@ observer = log.PythonLoggingObserver() observer.start() -class XMLRPCTest(unittest.TestCase): +class XMLRPCTest(DBTestCase): """tests for the xmlrpc interface""" def setUp(self): + if not self.engine: + self.engine = md.engine self.s = Process(target=startReactor, args=(md.engine,)) self.s.start() #the new process needs time to get stated, so this process has to sleep @@ -49,12 +51,7 @@ self.__debug().stop() time.sleep(.2) self.s.join() - self.__cleanDB() - - def __cleanDB(self): - with WithSession(md.engine, autocommit=True) as session: - for table in schema.__tables__: - session.query(getattr(schema,table)).delete() + DBTestCase.tearDown(self) def __debug(self): return xServer('http://localhost:7080/debug') @@ -74,12 +71,12 @@ def testStatus(self): ''' test the status function''' - with WithSession(md.engine, autocommit=True) as session: + with self.session() as session: u = User(name='test',apikey='abcdef123456789') session.add(User(name='test',apikey='abcdef123456789')) self.failUnlessEqual(self.__rpc2().status('abcdef123456789'), {}) - with WithSession(md.engine, autocommit=True) as session: + with self.session() as session: u = session.merge(u) j = Job(hash="a1", info='info', status="started") j.user=u @@ -131,7 +128,7 @@ def testRoutes(self): '''test the route function''' - with WithSession(md.engine, autocommit=True) as session: + with self.session() as session: u=User(name='test',apikey='abcdef123456789') o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms") u.rights.append(Userright(o)) @@ -144,7 +141,7 @@ self.failUnlessEqual(exc.faultCode, 700) self.failUnlessEqual(exc.faultString, "Typ is not valid.") - with WithSession(md.engine, autocommit=True) as session: + with self.session() as session: o=Offer(name="sipgate_plus", provider="sipgate", route="plus", typ="sms") u = session.query(User).filter_by(name="test").first() u.rights.append(Userright(o)) @@ -154,7 +151,7 @@ self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','sms'),['sipgate_basic','sipgate_plus']) self.failUnlessEqual(self.__rpc2().routes('abcdef123456789','fax'),[]) - with WithSession(md.engine, autocommit=True) as session: + with self.session() as session: u = session.query(User).filter_by(name="test").first() u.rights.append(Userright(o)) @@ -163,7 +160,7 @@ def testDefaultRoutes(self): '''test the defaultRoute function''' - with WithSession(md.engine, autocommit=True) as session: + with self.session() as session: u=User(name='test',apikey='abcdef123456789') o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms") u.rights.append(Userright(o,True)) @@ -202,42 +199,24 @@ def testBill(self): '''test bill function''' apikey='abcdef123456789' - with WithSession(md.engine, autocommit=True) as session: + with self.session() as session: u=User(name='test',apikey=apikey) session.add(u) + self.failUnlessEqual(self.__rpc2().bill(apikey),{'total':{'price':0.0,'anz':0}}) - with WithSession(md.engine, autocommit=True) as session: + with self.session() as session: u = session.merge(u) o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms") + u.rights.append(Userright(o)) j = Job(hash='a1',info='i',status='sended') - m = Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.30, offer=o, job=j) - u.rights.append(Userright(o)) + j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o)) u.jobs.append(j) - session.add(m) - - self.failUnlessEqual(self.__rpc2().bill(apikey),{'total':{'price':0.3,'anz':1}, - 'sipgate_basic':{'price':0.3,'anz':1,'info':{'i':{'price':0.3,'anz':1}}}}) - - with WithSession(md.engine, autocommit=True) as session: - j = session.merge(j) - j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o)) - self.failUnlessEqual(self.__rpc2().bill(apikey),{'total':{'price':0.7,'anz':2}, - 'sipgate_basic':{'price':0.7,'anz':2,'info':{'i':{'price':0.7,'anz':2}}}}) - - with WithSession(md.engine, autocommit=True) as session: - m = session.merge(m) - m.isBilled=True - - self.failUnlessEqual(self.__rpc2().bill(apikey),{'total':{'price':0.4,'anz':1}, - 'sipgate_basic':{'price':0.4,'anz':1,'info':{'i':{'price':0.4,'anz':1}}}}) - - with WithSession(md.engine, autocommit=True) as session: - u = session.merge(u) j = Job(hash='a2',info='a',status='sended') j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.4, offer=o)) u.jobs.append(j) + ret=self.__rpc2().bill(apikey) self.failUnlessEqual(ret['total'],{'price':0.8,'anz':2}) self.failUnlessEqual(ret['sipgate_basic'], @@ -247,23 +226,7 @@ } }) - with WithSession(md.engine, autocommit=True) as session: - u = session.merge(u) - j = Job(hash='a3',info='a',status='sended') - o = Offer(name='sipgate_gold',provider="sipgate",route="gold",typ="sms") - j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.5, offer=o)) - u.rights.append(Userright(offer=o)) - u.jobs.append(j) - - ret=self.__rpc2().bill(apikey) - self.failUnlessEqual(ret['total'],{'price':1.3,'anz':3}) - self.failUnlessEqual(ret['sipgate_gold'], - {'price':0.5,'anz':1, - 'info':{ - 'a':{'price':0.5,'anz':1}, - } - }) - + def startReactor(engine): """starts the Rector with a special debug Clild, so that the reactor can be stopped remotly. """ from twisted.internet import reactor