1 import unittest |
1 import unittest |
2 from sqlalchemy import create_engine, pool |
|
3 from tempfile import mkdtemp |
|
4 import shutil |
|
5 |
2 |
6 from datetime import datetime |
3 from datetime import datetime |
7 |
4 |
8 from iro.model import POOL_SIZE as DB_POOL_SIZE |
5 from iro.model.schema import User, Offer, Userright, Job, Message |
9 from iro.model.schema import User, Base, Offer, Userright, Job, Message |
|
10 from decimal import Decimal |
6 from decimal import Decimal |
11 |
7 |
12 from ngdatabase.mysql import Server, createConfig, Database |
8 from .dbtestcase import DBTestCase, setUpModule, tearDownModule |
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 |
9 |
49 class DBTests(DBTestCase): |
10 class DBTests(DBTestCase): |
50 """tests for the db model""" |
11 """tests for the db model""" |
51 def setUp(self): |
|
52 if not self.engine: |
|
53 self.engine = md.engine |
|
54 |
12 |
55 def testRoutes(self): |
13 def testRoutes(self): |
56 '''test routes''' |
14 '''test routes''' |
57 with self.session() as session: |
15 with self.session() as session: |
58 u=User(name='test',apikey='abcdef123456789') |
16 u=User(name='test',apikey='abcdef123456789') |
82 self.failUnlessEqual(session.typs.order_by(Offer.typ).all(),[('sms',),('sms2',)]) |
40 self.failUnlessEqual(session.typs.order_by(Offer.typ).all(),[('sms',),('sms2',)]) |
83 |
41 |
84 |
42 |
85 class BillTest(DBTestCase): |
43 class BillTest(DBTestCase): |
86 """test the bill function""" |
44 """test the bill function""" |
87 def setUp(self): |
45 |
88 if not self.engine: |
|
89 self.engine = md.engine |
|
90 |
|
91 def testBill(self): |
46 def testBill(self): |
92 '''test bill function''' |
47 '''test bill function''' |
93 apikey='abcdef123456789' |
48 apikey='abcdef123456789' |
94 with self.session() as session: |
49 with self.session() as session: |
95 u=User(name='test',apikey=apikey) |
50 u=User(name='test',apikey=apikey) |
184 self.failUnlessEqual(u.rights[0].bill.all(), |
139 self.failUnlessEqual(u.rights[0].bill.all(), |
185 [(1L,Decimal('0.4000'),'a')]) |
140 [(1L,Decimal('0.4000'),'a')]) |
186 |
141 |
187 self.failUnlessEqual(u.rights[1].bill.all(),[(1L,Decimal('0.5000'),'a')]) |
142 self.failUnlessEqual(u.rights[1].bill.all(),[(1L,Decimal('0.5000'),'a')]) |
188 |
143 |
|
144 |
189 if __name__ == '__main__': |
145 if __name__ == '__main__': |
190 unittest.main() |
146 unittest.main() |