1 import unittest

     2 from sqlalchemy.orm.exc import MultipleResultsFound

     3 from datetime import datetime

     4 

     5 from iro.model.schema import User, Offer, Userright, Job, Message

     6 from decimal import Decimal

     7 

     8 from ..test_helpers.dbtestcase import DBTestCase

     9 

    10 class DBTests(DBTestCase):

    11     """tests for the db model"""

    12     

    13     def testRoutes(self):

    14         '''test routes'''

    15         with self.session() as session:

    16             u=User(name='test',apikey='abcdef123456789')

    17             o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

    18             u.rights.append(Userright(o)) 

    19             session.add(u)

    20         

    21         with self.session() as session:

    22             u=session.merge(u)

    23             self.assertEqual(u.routes('sms').all(),[('sipgate_basic',),])

    24 

    25     def testRoutesNoDefault(self):

    26         '''test default routes'''

    27         with self.session() as session:

    28             u=User(name='test',apikey='abcdef123456789')

    29             o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

    30             u.rights.append(Userright(o)) 

    31             session.add(u)

    32         

    33         with self.session() as session:

    34             u=session.merge(u)

    35             self.assertEqual(u.routes('sms',default=True).all(),[])

    36 

    37     def testRoutesDefault(self):

    38         '''test default routes'''

    39         with self.session() as session:

    40             u=User(name='test',apikey='abcdef123456789')

    41             o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

    42             u.rights.append(Userright(o,default=1)) 

    43             session.add(u)

    44 

    45         with self.session() as session:

    46             u=session.merge(u)

    47             self.assertEqual(u.routes('sms',default=True).all(),[('sipgate_basic',),])

    48 

    49     def testRoutesDefaultOrdering(self):

    50         '''test default routes ordering'''

    51         with self.session() as session:

    52             u=User(name='test',apikey='abcdef123456789')

    53             o=Offer(name="s1", provider="sipgate", route="basic", typ="sms")

    54             u.rights.append(Userright(o,default=3)) 

    55             o=Offer(name="s2", provider="sipgate", route="basic", typ="sms")

    56             u.rights.append(Userright(o,default=None)) 

    57             o=Offer(name="s3", provider="sipgate", route="basic", typ="sms")

    58             u.rights.append(Userright(o,default=1)) 

    59             session.add(u)

    60 

    61         with self.session() as session:

    62             u=session.merge(u)

    63             self.assertEqual(u.routes('sms',default=True).all(),[('s3',),('s1',)])

    64 

    65     def testProviders(self):

    66         '''test providers'''

    67         with self.session() as session:

    68             u=User(name='test',apikey='abcdef123456789')

    69             o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

    70             u.rights.append(Userright(o)) 

    71             session.add(u)

    72         

    73         with self.session() as session:

    74             u=session.merge(u)

    75             self.assertEqual(u.providers('sms').all(),[('sipgate',),])

    76 

    77     def testTyps(self):

    78         with self.session() as session:

    79             o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

    80             session.add(o)

    81         

    82         with self.session() as session:

    83             self.assertEqual(Offer.typs(session).all(),[('sms',),])

    84 

    85         with self.session() as session:

    86             o=Offer(name="s2", provider="sipgate", route="basic", typ="sms")

    87             session.add(o)

    88             o=Offer(name="s3", provider="sipgate", route="basic", typ="sms2")

    89             session.add(o)

    90         

    91         with self.session() as session:

    92             self.assertEqual(Offer.typs(session).order_by(Offer.typ).all(),[('sms',),('sms2',)])

    93 

    94     def testOfferRoutes(self):

    95         with self.session() as session:

    96             o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

    97             session.add(o)

    98         

    99         with self.session() as session:

   100             self.assertEqual([o.name for o in Offer.routes(session, "sms").order_by(Offer.name)],["sipgate_basic"])

   101 

   102         with self.session() as session:

   103             o=Offer(name="s2", provider="sipgate", route="basic", typ="sms")

   104             session.add(o)

   105             o=Offer(name="s3", provider="sipgate", route="basic", typ="sms2")

   106             session.add(o)

   107         

   108         with self.session() as session:

   109             self.assertEqual([o.name for o in Offer.routes(session, "sms2").order_by(Offer.name)],["s3"])

   110             self.assertEqual([o.name for o in Offer.routes(session, "sms").order_by(Offer.name)],["s2","sipgate_basic"])

   111 

   112     def testOfferProviders(self):

   113         with self.session() as session:

   114             o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

   115             session.add(o)

   116         

   117         with self.session() as session:

   118             self.assertEqual([o.provider for o in Offer.providers(session, "sms").order_by(Offer.provider)],["sipgate"])

   119 

   120         with self.session() as session:

   121             o=Offer(name="s2", provider="sipgate2", route="basic", typ="sms")

   122             session.add(o)

   123             o=Offer(name="s3", provider="sipgate", route="basic", typ="sms2")

   124             session.add(o)

   125         

   126         with self.session() as session:

   127             self.assertEqual([o.provider for o in Offer.providers(session, "sms2").order_by(Offer.provider)],["sipgate"])

   128             self.assertEqual([o.provider for o in Offer.providers(session, "sms").order_by(Offer.provider)],["sipgate","sipgate2"])

   129 

   130 class Has_RightTests(DBTestCase):

   131     '''test User.has_right'''

   132     def testSimple(self):

   133         '''test a very simple case'''

   134         with self.session() as session:

   135             u=User(name='test',apikey='abcdef123456789')

   136             o=Offer(name="sipgate_basic", provider="sipgate", route="basic", typ="sms")

   137             u.rights.append(Userright(o)) 

   138             session.add(u)

   139         

   140         with self.session() as session:

   141             u=session.merge(u)

   142             self.assertEqual(u.has_right("sms"),"sipgate_basic")

   143             self.assertEqual(u.has_right("sms", offer_name="sipgate_basic"),"sipgate_basic")

   144             self.assertEqual(u.has_right("sms", route="basic"),"sipgate_basic")

   145             self.assertEqual(u.has_right("sms", provider="sipgate"),"sipgate_basic")

   146 

   147     def testDouble(self):

   148         '''now we have different routes'''

   149         with self.session() as session:

   150             u=User(name='test',apikey='abcdef123456789')

   151             o=Offer(name="b", provider="sipgate", route="basic", typ="sms")

   152             u.rights.append(Userright(o)) 

   153             o=Offer(name="c", provider="sipgate", route="c", typ="sms")

   154             u.rights.append(Userright(o)) 

   155             session.add(u)

   156         

   157         with self.session() as session:

   158             u=session.merge(u)

   159             self.assertEqual(u.has_right("sms", offer_name="b"),"b")

   160             self.assertEqual(u.has_right("sms", route="basic"),"b")

   161             self.assertEqual(u.has_right("sms", offer_name="c"),"c")

   162             self.assertEqual(u.has_right("sms", route="c"),"c")

   163             self.assertRaises(MultipleResultsFound, u.has_right,"sms")

   164             self.assertRaises(MultipleResultsFound, u.has_right,"sms", provider="sipgate")

   165     

   166     def testUnknown(self):

   167         ''' a unknown typ'''

   168         with self.session() as session:

   169             u=User(name='test',apikey='abcdef123456789')

   170             o=Offer(name="b", provider="sipgate", route="basic", typ="sms")

   171             u.rights.append(Userright(o)) 

   172             o=Offer(name="c", provider="sipgate", route="c", typ="sms")

   173             u.rights.append(Userright(o)) 

   174             session.add(u)

   175         

   176         with self.session() as session:

   177             u=session.merge(u)

   178             self.assertEqual(u.has_right("fax", offer_name="b"), None)

   179             self.assertEqual(u.has_right("fax"), None)

   180 

   181 

   182 

   183 class BillTest(DBTestCase):

   184     """test the bill function"""

   185     

   186     def testBill(self):

   187         '''test bill function'''

   188         apikey='abcdef123456789'

   189         with self.session() as session:

   190             u=User(name='test',apikey=apikey)

   191             session.add(u)

   192             o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms")

   193             j = Job(info='i',status='sended')

   194             m = Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.30, offer=o, job=j)

   195             u.rights.append(Userright(o))

   196             u.jobs.append(j)

   197             session.add(m)

   198 

   199         with self.session() as session:

   200             u=session.merge(u)

   201             self.assertEqual(u.rights[0].bill.all(),[(1L,Decimal('0.3000'),'i')])

   202 

   203     def testBillAgreget(self):

   204         apikey='abcdef123456789'

   205         with self.session() as session:

   206             u=User(name='test',apikey=apikey)

   207             session.add(u)

   208             o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms")

   209             j = Job(info='i',status='sended')

   210             j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.30, offer=o))

   211             j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o))

   212             u.rights.append(Userright(o))

   213             u.jobs.append(j)

   214  

   215         with self.session() as session:

   216             u=session.merge(u)

   217             self.assertEqual(u.rights[0].bill.all(),[(2L,Decimal('0.7000'),'i')])

   218         

   219     def testBillIsBilled(self):

   220         apikey='abcdef123456789'

   221         with self.session() as session:

   222             u=User(name='test',apikey=apikey)

   223             session.add(u)

   224             o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms")

   225             j = Job(info='i',status='sended')

   226             j.messages.append(Message(recipient='0123456789', isBilled=True, date=datetime.now() , price=0.30, offer=o))

   227             j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o))

   228             u.rights.append(Userright(o))

   229             u.jobs.append(j)

   230 

   231         with self.session() as session:

   232             u=session.merge(u)

   233             self.assertEqual(u.rights[0].bill.all(),[(1L,Decimal('0.4000'),'i')])

   234 

   235     def testBillMultipleJobs(self):

   236         apikey='abcdef123456789'

   237         with self.session() as session:

   238             u=User(name='test',apikey=apikey)

   239             session.add(u)

   240             o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms")

   241             u.rights.append(Userright(o))

   242             

   243             j = Job(info='i',status='sended')

   244             j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now() , price=0.4, offer=o))

   245             u.jobs.append(j)

   246 

   247             j = Job(info='a',status='sended')

   248             j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.4, offer=o))

   249             u.jobs.append(j) 

   250         

   251         with self.session() as session:

   252             u=session.merge(u)

   253             self.assertEqual(u.rights[0].bill.order_by(Job.info).all(),

   254                     [(1L,Decimal('0.4000'),'a'),

   255                     (1L,Decimal('0.4000'),'i')

   256                     ])

   257 

   258     def testBillMultipleOffers(self):

   259         apikey='abcdef123456789'

   260         with self.session() as session:

   261             u=User(name='test',apikey=apikey)

   262             session.add(u)

   263             o = Offer(name='sipgate_basic',provider="sipgate",route="basic",typ="sms")

   264             u.rights.append(Userright(o))

   265             

   266             j = Job(info='a',status='sended')

   267             j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.4, offer=o))

   268             u.jobs.append(j) 

   269  

   270             o = Offer(name='sipgate_gold',provider="sipgate",route="gold",typ="sms")

   271             u.rights.append(Userright(offer=o))

   272             

   273             j = Job(info='a',status='sended')

   274             j.messages.append(Message(recipient='0123456789', isBilled=False, date=datetime.now(), price=0.5, offer=o))

   275             u.jobs.append(j) 

   276        

   277         with self.session() as session:

   278             u=session.merge(u)

   279             self.assertEqual(u.rights[0].bill.all(),

   280                     [(1L,Decimal('0.4000'),'a')])

   281         

   282             self.assertEqual(u.rights[1].bill.all(),[(1L,Decimal('0.5000'),'a')])

   283 

   284 

   285 if __name__ == '__main__':

   286         unittest.main()