1 from twisted.trial import unittest

     2 

     3 from sqlalchemy import create_engine, pool

     4 from tempfile import mkdtemp

     5 

     6 from shutil import rmtree

     7 import atexit 

     8 from ngdatabase.mysql import Server, createConfig, Database

     9 

    10 from iro.model import setEngine, setPool

    11 from iro.model.utils import WithSession

    12 from iro.model.schema import Base

    13 

    14 from iro.controller.pool import dbPool

    15 

    16 class DBTestCase(unittest.TestCase):

    17     '''a TestCase with DB connection

    18     you have to set self.session manually in setUp'''

    19     def __init__(self,*args,**kwargs):

    20         unittest.TestCase.__init__(self,*args,**kwargs)

    21         self.engine = md.engine

    22 

    23     def setUp(self):

    24         md.setUp()

    25 

    26     def tearDown(self):

    27         self.__cleanDB()

    28     

    29     def session(self,autocommit=True):

    30         '''returns a WithSession'''

    31         return WithSession(self.engine,autocommit)

    32     

    33     def __cleanDB(self):

    34         '''cleaning database'''

    35         with self.session() as session:

    36             for table in reversed(Base.metadata.sorted_tables):

    37                 session.execute(table.delete())

    38 

    39 

    40 class SampleDatabase(Database):

    41     def createPassword(self):

    42         self.password="test"

    43         return self.password

    44 

    45 class ModuleData(object):

    46     def __init__(self):

    47         self.create()

    48 

    49     def close(self):

    50         if self.valid:

    51             self.server.stop()

    52             rmtree(self.tdir)

    53             self.valid= False

    54 

    55     def create(self):

    56         self.tdir = mkdtemp(prefix='iro-mysql-')

    57         self.server = Server('%s/my.cnf'%self.tdir)

    58         self.db = SampleDatabase("test","test",'%s/my.cnf'%self.tdir)

    59         self.dburl = 'mysql://test:test@localhost/test?unix_socket=%s/socket'%self.tdir

    60         self.engine = create_engine(self.dburl, 

    61                 poolclass = pool.SingletonThreadPool,  pool_size=dbPool.maxthreads, )#echo=True)

    62         self.valid = False

    63 

    64     def setUp(self):

    65         if not self.valid:

    66             with open('%s/my.cnf'%self.tdir,'w') as cnf:

    67                 cnf.write(createConfig(self.tdir))

    68             self.server.create()

    69             self.server.start()

    70             self.db.create()

    71             Base.metadata.create_all(self.engine)

    72             setEngine(self.engine)

    73             setPool(dbPool)

    74             self.valid = True

    75     

    76 

    77 md=ModuleData()

    78 atexit.register(md.close)