db setUp/tearDown into dbtestcase devel
authorSandro Knauß <knauss@netzguerilla.net>
Mon, 06 Feb 2012 14:39:33 +0100
branchdevel
changeset 134 fae3fdfece65
parent 133 214fddcf7133
child 135 f8640c663e3e
db setUp/tearDown into dbtestcase
tests/db.py
tests/dbtestcase.py
tests/xmlrpc.py
--- a/tests/db.py	Mon Feb 06 12:32:14 2012 +0100
+++ b/tests/db.py	Mon Feb 06 14:39:33 2012 +0100
@@ -1,56 +1,14 @@
 import unittest
-from sqlalchemy import create_engine, pool
-from tempfile import mkdtemp
-import shutil
 
 from datetime import datetime
 
-from iro.model import POOL_SIZE as DB_POOL_SIZE
-from iro.model.schema import User, Base, Offer, Userright, Job, Message
+from iro.model.schema import User, Offer, Userright, Job, Message
 from decimal import Decimal
 
-from ngdatabase.mysql import Server, createConfig, Database
-from .dbtestcase import DBTestCase
-
-class SampleDatabase(Database):
-    def createPassword(self):
-        self.password="test"
-        return self.password
-
-class ModuleData:
-    def __init__(self):
-        self.tdir = mkdtemp(prefix='iro-mysql-')
-        self.server = Server('%s/my.cnf'%self.tdir)
-        self.db = SampleDatabase("test","test",'%s/my.cnf'%self.tdir)
-        self.engine = create_engine('mysql://test:test@localhost/test?unix_socket=%s/socket'%self.tdir,
-                poolclass = pool.SingletonThreadPool,  pool_size=DB_POOL_SIZE, )#echo=True)
-
-    def setUp(self):
-        with open('%s/my.cnf'%self.tdir,'w') as cnf:
-            cnf.write(createConfig(self.tdir))
-        self.server.create()
-        self.server.start()
-        self.db.create()
-        Base.metadata.create_all(self.engine)
-    
-    def tearDown(self):
-        self.server.stop()
-        shutil.rmtree(self.tdir)
- 
-
-md=ModuleData()
-
-def setUpModule():
-    md.setUp()
-
-def tearDownModule():
-    md.tearDown()
+from .dbtestcase import DBTestCase, setUpModule, tearDownModule
 
 class DBTests(DBTestCase):
     """tests for the db model"""
-    def setUp(self):
-        if not self.engine:
-            self.engine = md.engine
     
     def testRoutes(self):
         '''test routes'''
@@ -84,10 +42,7 @@
 
 class BillTest(DBTestCase):
     """test the bill function"""
-    def setUp(self):
-        if not self.engine:
-            self.engine = md.engine
-
+    
     def testBill(self):
         '''test bill function'''
         apikey='abcdef123456789'
@@ -186,5 +141,6 @@
         
             self.failUnlessEqual(u.rights[1].bill.all(),[(1L,Decimal('0.5000'),'a')])
 
+
 if __name__ == '__main__':
         unittest.main()
--- a/tests/dbtestcase.py	Mon Feb 06 12:32:14 2012 +0100
+++ b/tests/dbtestcase.py	Mon Feb 06 14:39:33 2012 +0100
@@ -1,13 +1,22 @@
 import unittest
 
+from sqlalchemy import create_engine, pool
+from tempfile import mkdtemp
+import shutil
+
+from ngdatabase.mysql import Server, createConfig, Database
+
 from iro.model import schema
 from iro.model.utils import WithSession
+from iro.model.schema import Base
+
+from iro.controller.pool import dbPool
 
 class DBTestCase(unittest.TestCase):
     '''a TestCase with DB connection
     you have to set self.session manually in setUp'''
     def __init__(self,*args,**kwargs):
-        self.engine=None
+        self.engine = md.engine
         unittest.TestCase.__init__(self,*args,**kwargs)
 
     def tearDown(self):
@@ -22,3 +31,38 @@
         with self.session() as session:
             for table in schema.__tables__:
                 session.query(getattr(schema,table)).delete()
+
+class SampleDatabase(Database):
+    def createPassword(self):
+        self.password="test"
+        return self.password
+
+class ModuleData:
+    def __init__(self):
+        self.tdir = mkdtemp(prefix='iro-mysql-')
+        self.server = Server('%s/my.cnf'%self.tdir)
+        self.db = SampleDatabase("test","test",'%s/my.cnf'%self.tdir)
+        self.engine = create_engine('mysql://test:test@localhost/test?unix_socket=%s/socket'%self.tdir,
+                poolclass = pool.SingletonThreadPool,  pool_size=dbPool.maxthreads, )#echo=True)
+
+    def setUp(self):
+        with open('%s/my.cnf'%self.tdir,'w') as cnf:
+            cnf.write(createConfig(self.tdir))
+        self.server.create()
+        self.server.start()
+        self.db.create()
+        Base.metadata.create_all(self.engine)
+    
+    def tearDown(self):
+        self.server.stop()
+        shutil.rmtree(self.tdir)
+ 
+
+md=ModuleData()
+
+def setUpModule():
+    md.setUp()
+
+def tearDownModule():
+    md.tearDown()
+
--- a/tests/xmlrpc.py	Mon Feb 06 12:32:14 2012 +0100
+++ b/tests/xmlrpc.py	Mon Feb 06 14:39:33 2012 +0100
@@ -1,32 +1,19 @@
 from multiprocessing import Process
-from sqlalchemy import create_engine, pool
 import unittest
 
-from tempfile import mkdtemp
-import shutil
-
 from datetime import datetime
 
 import time
 
 from xmlrpclib import Server as xServer, ServerProxy, Fault
 
-from iro.controller.pool import dbPool
-
-from iro.model.schema import User, Base, Offer, Userright, Job, Message
+from iro.model.schema import User, Offer, Userright, Job, Message
 
 from iro.main import runReactor
 
 import iro.error as IroError
 
-from ngdatabase.mysql import Server, createConfig, Database
-
-from .dbtestcase import DBTestCase
-
-class SampleDatabase(Database):
-    def createPassword(self):
-        self.password="test"
-        return self.password
+from .dbtestcase import DBTestCase, setUpModule, tearDownModule
 
 
 #activates all logging we can get.
@@ -40,9 +27,7 @@
 class XMLRPCTest(DBTestCase):
     """tests for the xmlrpc interface"""
     def setUp(self):
-        if not self.engine:
-            self.engine = md.engine
-        self.s = Process(target=startReactor, args=(md.engine,))
+        self.s = Process(target=startReactor, args=(self.engine,))
         self.s.start()
         #the new process needs time to get stated, so this process has to sleep
         time.sleep(.2)
@@ -247,36 +232,5 @@
     root.putChild('debug', XMLRPCDebug())
     runReactor(reactor, engine, root)
 
-
-class ModuleData:
-    def __init__(self):
-        self.tdir = mkdtemp(prefix='iro-mysql-')
-        self.server = Server('%s/my.cnf'%self.tdir)
-        self.db = SampleDatabase("test","test",'%s/my.cnf'%self.tdir)
-        self.engine = create_engine('mysql://test:test@localhost/test?unix_socket=%s/socket'%self.tdir,
-                poolclass = pool.SingletonThreadPool,  pool_size=dbPool.maxthreads, )#echo=True)
-
-    def setUp(self):
-        with open('%s/my.cnf'%self.tdir,'w') as cnf:
-            cnf.write(createConfig(self.tdir))
-        self.server.create()
-        self.server.start()
-        self.db.create()
-        Base.metadata.create_all(self.engine)
-    
-    def tearDown(self):
-        self.server.stop()
-        shutil.rmtree(self.tdir)
- 
-
-md=ModuleData()
-
-def setUpModule():
-    md.setUp()
-
-def tearDownModule():
-    md.tearDown()
-
-    
 if __name__ == '__main__':
         unittest.main()