1 import unittest

     2 from sqlalchemy import create_engine, pool

     3 from tempfile import mkdtemp

     4 import shutil

     5 

     6 from datetime import datetime

     7 

     8 from iro.model import POOL_SIZE as DB_POOL_SIZE

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

    10 from decimal import Decimal

    11 

    12 from ngdatabase.mysql import Server, createConfig, Database

    13 from .dbtestcase import DBTestCase

    14 

    15 class SampleDatabase(Database):

    16     def createPassword(self):

    17         self.password="test"

    18         return self.password

    19 

    20 class ModuleData:

    21     def __init__(self):

    22         self.tdir = mkdtemp(prefix='iro-mysql-')

    23         self.server = Server('%s/my.cnf'%self.tdir)

    24         self.db = SampleDatabase("test","test",'%s/my.cnf'%self.tdir)

    25         self.engine = create_engine('mysql://test:test@localhost/test?unix_socket=%s/socket'%self.tdir,

    26                 poolclass = pool.SingletonThreadPool,  pool_size=DB_POOL_SIZE, )#echo=True)

    27 

    28     def setUp(self):

    29         with open('%s/my.cnf'%self.tdir,'w') as cnf:

    30             cnf.write(createConfig(self.tdir))

    31         self.server.create()

    32         self.server.start()

    33         self.db.create()

    34         Base.metadata.create_all(self.engine)

    35     

    36     def tearDown(self):

    37         self.server.stop()

    38         shutil.rmtree(self.tdir)

    39  

    40 

    41 md=ModuleData()

    42 

    43 def setUpModule():

    44     md.setUp()

    45 

    46 def tearDownModule():

    47     md.tearDown()

    48 

    49 class DBTests(DBTestCase):

    50     """tests for the db model"""

    51     def setUp(self):

    52         if not self.engine:

    53             self.engine = md.engine

    54     

    55     def testRoutes(self):

    56         '''test routes'''

    57         with self.session() as session:

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

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

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

    61             session.add(u)

    62         

    63         with self.session() as session:

    64             u=session.merge(u)

    65             self.failUnlessEqual(u.routes('sms').all(),[('sipgate_basic',),])

    66 

    67 

    68     def testTyps(self):

    69         with self.session() as session:

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

    71             session.add(o)

    72         

    73         with self.session() as session:

    74             self.failUnlessEqual(session.typs.all(),[('sms',),])

    75 

    76         with self.session() as session:

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

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

    79             session.add(o)

    80         

    81         with self.session() as session:

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

    83 

    84 

    85 class BillTest(DBTestCase):

    86     """test the bill function"""

    87     def setUp(self):

    88         if not self.engine:

    89             self.engine = md.engine

    90 

    91     def testBill(self):

    92         '''test bill function'''

    93         apikey='abcdef123456789'

    94         with self.session() as session:

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

    96             session.add(u)

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

    98             j = Job(hash='a1',info='i',status='sended')

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

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

   101             u.jobs.append(j)

   102             session.add(m)

   103 

   104         with self.session() as session:

   105             u=session.merge(u)

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

   107 

   108     def testBillAgreget(self):

   109         apikey='abcdef123456789'

   110         with self.session() as session:

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

   112             session.add(u)

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

   114             j = Job(hash='a1',info='i',status='sended')

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

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

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

   118             u.jobs.append(j)

   119  

   120         with self.session() as session:

   121             u=session.merge(u)

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

   123         

   124     def testBillIsBilled(self):

   125         apikey='abcdef123456789'

   126         with self.session() as session:

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

   128             session.add(u)

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

   130             j = Job(hash='a1',info='i',status='sended')

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

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

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

   134             u.jobs.append(j)

   135 

   136         with self.session() as session:

   137             u=session.merge(u)

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

   139 

   140     def testBillMultipleJobs(self):

   141         apikey='abcdef123456789'

   142         with self.session() as session:

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

   144             session.add(u)

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

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

   147             

   148             j = Job(hash='a1',info='i',status='sended')

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

   150             u.jobs.append(j)

   151 

   152             j = Job(hash='a2',info='a',status='sended')

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

   154             u.jobs.append(j) 

   155         

   156         with self.session() as session:

   157             u=session.merge(u)

   158             self.failUnlessEqual(u.rights[0].bill.order_by(Job.info).all(),

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

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

   161                     ])

   162 

   163     def testBillMultipleOffers(self):

   164         apikey='abcdef123456789'

   165         with self.session() as session:

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

   167             session.add(u)

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

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

   170             

   171             j = Job(hash='a2',info='a',status='sended')

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

   173             u.jobs.append(j) 

   174  

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

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

   177             

   178             j = Job(hash='a3',info='a',status='sended')

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

   180             u.jobs.append(j) 

   181        

   182         with self.session() as session:

   183             u=session.merge(u)

   184             self.failUnlessEqual(u.rights[0].bill.all(),

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

   186         

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

   188 

   189 if __name__ == '__main__':

   190         unittest.main()