--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/iro/tests/db.py Sun Mar 18 14:06:27 2012 +0100
@@ -0,0 +1,286 @@
+import unittest
+from sqlalchemy.orm.exc import MultipleResultsFound
+from datetime import datetime
+
+from iro.model.schema import User, Offer, Userright, Job, Message
+from decimal import Decimal
+
+from ..test_helpers.dbtestcase import DBTestCase
+
+class DBTests(DBTestCase):
+ """tests for the db model"""
+
+ 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.assertEqual(u.routes('sms').all(),[('sipgate_basic',),])
+
+ def testRoutesNoDefault(self):
+ '''test default 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.assertEqual(u.routes('sms',default=True).all(),[])
+
+ def testRoutesDefault(self):
+ '''test default 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,default=1))
+ session.add(u)
+
+ with self.session() as session:
+ u=session.merge(u)
+ self.assertEqual(u.routes('sms',default=True).all(),[('sipgate_basic',),])
+
+ def testRoutesDefaultOrdering(self):
+ '''test default routes ordering'''
+ with self.session() as session:
+ u=User(name='test',apikey='abcdef123456789')
+ o=Offer(name="s1", provider="sipgate", route="basic", typ="sms")
+ u.rights.append(Userright(o,default=3))
+ o=Offer(name="s2", provider="sipgate", route="basic", typ="sms")
+ u.rights.append(Userright(o,default=None))
+ o=Offer(name="s3", provider="sipgate", route="basic", typ="sms")
+ u.rights.append(Userright(o,default=1))
+ session.add(u)
+
+ with self.session() as session:
+ u=session.merge(u)
+ self.assertEqual(u.routes('sms',default=True).all(),[('s3',),('s1',)])
+
+ def testProviders(self):
+ '''test providers'''
+ 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.assertEqual(u.providers('sms').all(),[('sipgate',),])
+
+ 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.assertEqual(Offer.typs(session).all(),[('sms',),])
+
+ with self.session() as session:
+ o=Offer(name="s2", provider="sipgate", route="basic", typ="sms")
+ session.add(o)
+ o=Offer(name="s3", provider="sipgate", route="basic", typ="sms2")
+ session.add(o)
+
+ with self.session() as session:
+ self.assertEqual(Offer.typs(session).order_by(Offer.typ).all(),[('sms',),('sms2',)])
+
+ def testOfferRoutes(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.assertEqual([o.name for o in Offer.routes(session, "sms").order_by(Offer.name)],["sipgate_basic"])
+
+ with self.session() as session:
+ o=Offer(name="s2", provider="sipgate", route="basic", typ="sms")
+ session.add(o)
+ o=Offer(name="s3", provider="sipgate", route="basic", typ="sms2")
+ session.add(o)
+
+ with self.session() as session:
+ self.assertEqual([o.name for o in Offer.routes(session, "sms2").order_by(Offer.name)],["s3"])
+ self.assertEqual([o.name for o in Offer.routes(session, "sms").order_by(Offer.name)],["s2","sipgate_basic"])
+
+ def testOfferProviders(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.assertEqual([o.provider for o in Offer.providers(session, "sms").order_by(Offer.provider)],["sipgate"])
+
+ with self.session() as session:
+ o=Offer(name="s2", provider="sipgate2", route="basic", typ="sms")
+ session.add(o)
+ o=Offer(name="s3", provider="sipgate", route="basic", typ="sms2")
+ session.add(o)
+
+ with self.session() as session:
+ self.assertEqual([o.provider for o in Offer.providers(session, "sms2").order_by(Offer.provider)],["sipgate"])
+ self.assertEqual([o.provider for o in Offer.providers(session, "sms").order_by(Offer.provider)],["sipgate","sipgate2"])
+
+class Has_RightTests(DBTestCase):
+ '''test User.has_right'''
+ def testSimple(self):
+ '''test a very simple case'''
+ 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.assertEqual(u.has_right("sms"),"sipgate_basic")
+ self.assertEqual(u.has_right("sms", offer_name="sipgate_basic"),"sipgate_basic")
+ self.assertEqual(u.has_right("sms", route="basic"),"sipgate_basic")
+ self.assertEqual(u.has_right("sms", provider="sipgate"),"sipgate_basic")
+
+ def testDouble(self):
+ '''now we have different routes'''
+ with self.session() as session:
+ u=User(name='test',apikey='abcdef123456789')
+ o=Offer(name="b", provider="sipgate", route="basic", typ="sms")
+ u.rights.append(Userright(o))
+ o=Offer(name="c", provider="sipgate", route="c", typ="sms")
+ u.rights.append(Userright(o))
+ session.add(u)
+
+ with self.session() as session:
+ u=session.merge(u)
+ self.assertEqual(u.has_right("sms", offer_name="b"),"b")
+ self.assertEqual(u.has_right("sms", route="basic"),"b")
+ self.assertEqual(u.has_right("sms", offer_name="c"),"c")
+ self.assertEqual(u.has_right("sms", route="c"),"c")
+ self.assertRaises(MultipleResultsFound, u.has_right,"sms")
+ self.assertRaises(MultipleResultsFound, u.has_right,"sms", provider="sipgate")
+
+ def testUnknown(self):
+ ''' a unknown typ'''
+ with self.session() as session:
+ u=User(name='test',apikey='abcdef123456789')
+ o=Offer(name="b", provider="sipgate", route="basic", typ="sms")
+ u.rights.append(Userright(o))
+ o=Offer(name="c", provider="sipgate", route="c", typ="sms")
+ u.rights.append(Userright(o))
+ session.add(u)
+
+ with self.session() as session:
+ u=session.merge(u)
+ self.assertEqual(u.has_right("fax", offer_name="b"), None)
+ self.assertEqual(u.has_right("fax"), None)
+
+
+
+class BillTest(DBTestCase):
+ """test the bill function"""
+
+ 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(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.assertEqual(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(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.assertEqual(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(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.assertEqual(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(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(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.assertEqual(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(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(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.assertEqual(u.rights[0].bill.all(),
+ [(1L,Decimal('0.4000'),'a')])
+
+ self.assertEqual(u.rights[1].bill.all(),[(1L,Decimal('0.5000'),'a')])
+
+
+if __name__ == '__main__':
+ unittest.main()