--- /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)
--- /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()
--- 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)
--- 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()