1 import unittest |
1 from twisted.trial import unittest |
2 |
2 |
3 from sqlalchemy import create_engine, pool |
3 from sqlalchemy import create_engine, pool |
4 from tempfile import mkdtemp |
4 from tempfile import mkdtemp |
5 import shutil |
|
6 |
5 |
|
6 from shutil import rmtree |
|
7 import atexit |
7 from ngdatabase.mysql import Server, createConfig, Database |
8 from ngdatabase.mysql import Server, createConfig, Database |
8 |
9 |
9 from iro.model import schema, setEngine |
10 from iro.model import schema, setEngine |
10 from iro.model.utils import WithSession |
11 from iro.model.utils import WithSession |
11 from iro.model.schema import Base |
12 from iro.model.schema import Base |
14 |
15 |
15 class DBTestCase(unittest.TestCase): |
16 class DBTestCase(unittest.TestCase): |
16 '''a TestCase with DB connection |
17 '''a TestCase with DB connection |
17 you have to set self.session manually in setUp''' |
18 you have to set self.session manually in setUp''' |
18 def __init__(self,*args,**kwargs): |
19 def __init__(self,*args,**kwargs): |
|
20 unittest.TestCase.__init__(self,*args,**kwargs) |
19 self.engine = md.engine |
21 self.engine = md.engine |
20 unittest.TestCase.__init__(self,*args,**kwargs) |
22 |
|
23 def setUp(self): |
|
24 md.setUp() |
21 |
25 |
22 def tearDown(self): |
26 def tearDown(self): |
23 self.__cleanDB() |
27 self.__cleanDB() |
24 |
28 |
25 def session(self,autocommit=True): |
29 def session(self,autocommit=True): |
30 '''cleaning database''' |
34 '''cleaning database''' |
31 with self.session() as session: |
35 with self.session() as session: |
32 for table in schema.__tables__: |
36 for table in schema.__tables__: |
33 session.query(getattr(schema,table)).delete() |
37 session.query(getattr(schema,table)).delete() |
34 |
38 |
|
39 |
35 class SampleDatabase(Database): |
40 class SampleDatabase(Database): |
36 def createPassword(self): |
41 def createPassword(self): |
37 self.password="test" |
42 self.password="test" |
38 return self.password |
43 return self.password |
39 |
44 |
40 class ModuleData: |
45 class ModuleData(object): |
41 def __init__(self): |
46 def __init__(self): |
|
47 self.create() |
|
48 |
|
49 def close(self): |
|
50 self.server.stop() |
|
51 rmtree(self.tdir) |
|
52 self.valid= False |
|
53 |
|
54 def create(self): |
42 self.tdir = mkdtemp(prefix='iro-mysql-') |
55 self.tdir = mkdtemp(prefix='iro-mysql-') |
43 self.server = Server('%s/my.cnf'%self.tdir) |
56 self.server = Server('%s/my.cnf'%self.tdir) |
44 self.db = SampleDatabase("test","test",'%s/my.cnf'%self.tdir) |
57 self.db = SampleDatabase("test","test",'%s/my.cnf'%self.tdir) |
45 self.engine = create_engine('mysql://test:test@localhost/test?unix_socket=%s/socket'%self.tdir, |
58 self.engine = create_engine('mysql://test:test@localhost/test?unix_socket=%s/socket'%self.tdir, |
46 poolclass = pool.SingletonThreadPool, pool_size=dbPool.maxthreads, )#echo=True) |
59 poolclass = pool.SingletonThreadPool, pool_size=dbPool.maxthreads, )#echo=True) |
|
60 self.vaild = False |
47 |
61 |
48 def setUp(self): |
62 def setUp(self): |
49 with open('%s/my.cnf'%self.tdir,'w') as cnf: |
63 if not self.vaild: |
50 cnf.write(createConfig(self.tdir)) |
64 with open('%s/my.cnf'%self.tdir,'w') as cnf: |
51 self.server.create() |
65 cnf.write(createConfig(self.tdir)) |
52 self.server.start() |
66 self.server.create() |
53 self.db.create() |
67 self.server.start() |
54 Base.metadata.create_all(self.engine) |
68 self.db.create() |
55 setEngine(self.engine) |
69 Base.metadata.create_all(self.engine) |
|
70 setEngine(self.engine) |
|
71 self.vaild = True |
56 |
72 |
57 def tearDown(self): |
|
58 self.server.stop() |
|
59 shutil.rmtree(self.tdir) |
|
60 |
|
61 |
73 |
62 md=ModuleData() |
74 md=ModuleData() |
63 |
75 atexit.register(md.close) |
64 def setUpModule(): |
|
65 md.setUp() |
|
66 |
|
67 def tearDownModule(): |
|
68 md.tearDown() |
|
69 |
|