--- /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()
--- /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()
--- 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