# HG changeset patch # User Sandro Knauß # Date 1328535573 -3600 # Node ID fae3fdfece65cdb4995231342c0e31df962a2f69 # Parent 214fddcf71335819948e2adb0029fe0620d65414 db setUp/tearDown into dbtestcase diff -r 214fddcf7133 -r fae3fdfece65 tests/db.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() diff -r 214fddcf7133 -r fae3fdfece65 tests/dbtestcase.py --- 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() + diff -r 214fddcf7133 -r fae3fdfece65 tests/xmlrpc.py --- 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()