diff -r eb04ac3a8327 -r 3f4bdea2abbf iro/test_helpers/dbtestcase.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/iro/test_helpers/dbtestcase.py Thu Sep 27 17:15:46 2012 +0200 @@ -0,0 +1,104 @@ +# Copyright (c) 2012 netzguerilla.net +# +# This file is part of Iro. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy of +# this software and associated documentation files (the "Software"), to deal in +# the Software without restriction, including without limitation the rights to use, +# copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the +# #Software, and to permit persons to whom the Software is furnished to do so, +# subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, +# INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +# PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + +from twisted.trial import unittest + +from sqlalchemy import create_engine, pool +from tempfile import mkdtemp + +from shutil import rmtree +import atexit +from ngdatabase.mysql import Server, createConfig, Database + +from iro.model import setEngine, setPool +from iro.model.utils import WithSession +from iro.model.schema import Base + +from iro.controller.pool import dbPool + +from .utils import DummyObserver + +class DBTestCase(unittest.TestCase): + '''a TestCase with DB connection + you have to set self.session manually in setUp''' + def __init__(self,*args,**kwargs): + unittest.TestCase.__init__(self,*args,**kwargs) + self.engine = md.engine + + def setUp(self): + md.setUp() + self.log = DummyObserver() + self.log.start() + + def tearDown(self): + self.log.stop() + self.__cleanDB() + + def session(self,autocommit=True): + '''returns a WithSession''' + return WithSession(self.engine,autocommit) + + def __cleanDB(self): + '''cleaning database''' + with self.session() as session: + for table in reversed(Base.metadata.sorted_tables): + session.execute(table.delete()) + + +class SampleDatabase(Database): + def createPassword(self): + self.password="test" + return self.password + +class ModuleData(object): + def __init__(self): + self.create() + + def close(self): + if self.valid: + self.server.stop() + rmtree(self.tdir) + self.valid= False + + def create(self): + self.tdir = mkdtemp(prefix='iro-mysql-') + self.server = Server('%s/my.cnf'%self.tdir) + self.db = SampleDatabase("iro","test",'%s/my.cnf'%self.tdir) + self.dburl = 'mysql://test:test@localhost/iro?unix_socket=%s/socket'%self.tdir + self.engine = create_engine(self.dburl, + poolclass = pool.SingletonThreadPool, pool_size=dbPool.maxthreads, )#echo=True) + self.valid = False + + def setUp(self): + if not self.valid: + 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) + setEngine(self.engine) + setPool(dbPool) + self.valid = True + + +md=ModuleData() +atexit.register(md.close)