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