# HG changeset patch # User Sandro Knauß # Date 1332075911 -3600 # Node ID 448dd8d36839505f65fe40ce85da2ca9138be1fa # Parent 1bd4c7f58b3f08d9d8c0f57a6cc84459c0ded84c moving to iro/test_helpers diff -r 1bd4c7f58b3f -r 448dd8d36839 iro/test_helpers/dbtestcase.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/iro/test_helpers/dbtestcase.py Sun Mar 18 14:05:11 2012 +0100 @@ -0,0 +1,78 @@ +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 + +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() + + def tearDown(self): + 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("test","test",'%s/my.cnf'%self.tdir) + self.dburl = 'mysql://test:test@localhost/test?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) diff -r 1bd4c7f58b3f -r 448dd8d36839 iro/test_helpers/smtp_helper.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/iro/test_helpers/smtp_helper.py Sun Mar 18 14:05:11 2012 +0100 @@ -0,0 +1,28 @@ +""" +A helper class which allows test code to intercept and store sent email. +(from: http://lakin.weckers.net/thoughts/twisted/part1/threaded/) +""" +import smtpd +import asyncore + +class TestSMTPServer(smtpd.SMTPServer): + """ + An SMTP Server used to allow integration tests which ensure email is sent. + """ + + def __init__(self, localaddr): + self.rcvd = [] + smtpd.SMTPServer.__init__(self, localaddr, None) + + def start(self): + import threading + self.poller = threading.Thread(target=asyncore.loop, + kwargs={'timeout':0.1, 'use_poll':True}) + self.poller.start() + + def process_message(self, peer, mailfrom, rcpttos, data): + self.rcvd.append((mailfrom, rcpttos, data)) + + def close(self): + smtpd.SMTPServer.close(self) + self.poller.join() diff -r 1bd4c7f58b3f -r 448dd8d36839 tests/dbtestcase.py --- a/tests/dbtestcase.py Fri Mar 16 12:57:21 2012 +0100 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,77 +0,0 @@ -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 - -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() - - def tearDown(self): - 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): - 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("test","test",'%s/my.cnf'%self.tdir) - self.dburl = 'mysql://test:test@localhost/test?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) diff -r 1bd4c7f58b3f -r 448dd8d36839 tests/smtp_helper.py --- a/tests/smtp_helper.py Fri Mar 16 12:57:21 2012 +0100 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,28 +0,0 @@ -""" -A helper class which allows test code to intercept and store sent email. -(from: http://lakin.weckers.net/thoughts/twisted/part1/threaded/) -""" -import smtpd -import asyncore - -class TestSMTPServer(smtpd.SMTPServer): - """ - An SMTP Server used to allow integration tests which ensure email is sent. - """ - - def __init__(self, localaddr): - self.rcvd = [] - smtpd.SMTPServer.__init__(self, localaddr, None) - - def start(self): - import threading - self.poller = threading.Thread(target=asyncore.loop, - kwargs={'timeout':0.1, 'use_poll':True}) - self.poller.start() - - def process_message(self, peer, mailfrom, rcpttos, data): - self.rcvd.append((mailfrom, rcpttos, data)) - - def close(self): - smtpd.SMTPServer.close(self) - self.poller.join()