x/sqlalchemy-migrate
Code Issues Proposed changes

use unittest.TestCase for tests

Browse Source
This commit is contained in:
iElectric 2009-06-05 00:41:05 +00:00
parent 31bd33bcff
commit 4356e8b582
4 changed files with 77 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
test/fixture/__init__.py
View File

@ -1,14 +1,19 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import unittest
import sys
## Append test method name,etc. to descriptions automatically.
## Yes, this is ugly, but it's the simplest way...
#def getDescription(self,test):
# ret = str(test)
# if self.descriptions:
# ret += "\n\t"+(test.shortDescription() or '')
# return ret
#unittest._TextTestResult.getDescription = getDescription
# Append test method name,etc. to descriptions automatically.
# Yes, this is ugly, but it's the simplest way...
def getDescription(self, test):
ret = str(test)
if self.descriptions:
return test.shortDescription() or ret
return ret
unittest._TextTestResult.getDescription = getDescription
class Result(unittest._TextTestResult):
# test description may be changed as we go; store the description at
@ -59,4 +64,3 @@ from base import Base
from pathed import Pathed
from shell import Shell
from database import DB,usedb

48
test/fixture/base.py
View File

@ -1,57 +1,33 @@
#import unittest
#from py.test import raises
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import unittest
from nose.tools import raises, eq_
class FakeTestCase(object):
"""Mimics unittest.testcase methods
Minimize changes needed in migration to py.test
"""
def setUp(self):
pass
class Base(unittest.TestCase):
def setup_method(self,func=None):
self.setUp()
def tearDown(self):
pass
def teardown_method(self,func=None):
self.tearDown()
def assert_(self,x,doc=None):
assert x
def assertEquals(self,x,y,doc=None):
eq_(x, y)
def assertNotEquals(self,x,y,doc=None):
assert x != y
def assertRaises(self, exceptions, func, *arg, **kw):
if not hasattr(exceptions, '__iter__'):
exceptions = (exceptions, )
valid = ' or '.join([e.__name__ for e in exceptions])
try:
func(*arg, **kw)
except exceptions:
pass
except:
raise
else:
message = "%s() did not raise %s" % (func.__name__, valid)
raise AssertionError(message)
def assertEqualsIgnoreWhitespace(self, v1, v2):
"""Compares two strings that should be\
identical except for whitespace
"""
def createLines(s):
s = s.replace(' ', '')
lines = s.split('\n')
return [ line for line in lines if line ]
return [line for line in lines if line]
lines1 = createLines(v1)
lines2 = createLines(v2)
self.assertEquals(len(lines1), len(lines2))
for line1, line2 in zip(lines1, lines2):
self.assertEquals(line1, line2)
class Base(FakeTestCase):
"""Base class for other test cases"""
def ignoreErrors(self, func, *p,**k):
"""Call a function, ignoring any exceptions"""
try:

101
test/fixture/database.py
View File

@ -1,49 +1,58 @@
from base import Base
from pathed import Pathed
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.orm import create_session
from pkg_resources import resource_stream
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.orm import create_session
from test.fixture.base import Base
from test.fixture.pathed import Pathed
def readurls():
filename='test_db.cfg'
fullpath = os.path.join(os.curdir,filename)
ret=[]
tmpfile=Pathed.tmp()
"""read URLs from config file return a list"""
filename = 'test_db.cfg'
ret = list()
# TODO: remove tmpfile since sqlite can store db in memory
tmpfile = Pathed.tmp()
fullpath = os.path.join(os.curdir, filename)
try:
fd=open(fullpath)
fd = open(fullpath)
except IOError:
raise IOError("""You must specify the databases to use for testing!
Copy %(filename)s.tmpl to %(filename)s and edit your database URLs.""" % locals())
#fd = resource_stream('__main__',filename)
for line in fd:
if line.startswith('#'):
continue
line=line.replace('__tmp__',tmpfile).strip()
line = line.replace('__tmp__', tmpfile).strip()
ret.append(line)
fd.close()
return ret
def is_supported(url,supported,not_supported):
db = url.split(':',1)[0]
def is_supported(url, supported, not_supported):
db = url.split(':', 1)[0]
if supported is not None:
if isinstance(supported,basestring):
supported = (supported,)
ret = db in supported
if isinstance(supported, basestring):
return supported == db
else:
return db in supported
elif not_supported is not None:
if isinstance(not_supported,basestring):
not_supported = (not_supported,)
ret = not (db in not_supported)
else:
ret = True
return ret
if isinstance(not_supported, basestring):
return not_supported != db
else:
return not (db in not_supported)
return True
#we make the engines global, which should make the tests run a bit faster
# we make the engines global, which should make the tests run a bit faster
urls = readurls()
engines=dict([(url,create_engine(url, echo=True)) for url in urls])
engines = dict([(url, create_engine(url, echo=True)) for url in urls])
def usedb(supported=None,not_supported=None):
def usedb(supported=None, not_supported=None):
"""Decorates tests to be run with a database connection
These tests are run once for each available database
@ -54,10 +63,10 @@ def usedb(supported=None,not_supported=None):
to be supported
"""
if supported is not None and not_supported is not None:
msg = "Can't specify both supported and not_supported in fixture.db()"
assert False, msg
raise AssertionError("Can't specify both supported and not_supported in fixture.db()")
my_urls = [url for url in urls if is_supported(url, supported, not_supported)]
my_urls = [url for url in urls if is_supported(url,supported,not_supported)]
def dec(func):
def entangle(self):
for url in my_urls:
@ -71,23 +80,15 @@ def usedb(supported=None,not_supported=None):
class DB(Base):
# Constants: connection level
NONE=0 # No connection; just set self.url
CONNECT=1 # Connect; no transaction
TXN=2 # Everything in a transaction
NONE = 0 # No connection; just set self.url
CONNECT = 1 # Connect; no transaction
TXN = 2 # Everything in a transaction
level=TXN
def shortDescription(self,*p,**k):
"""List database connection info with description of the test"""
ret = super(DB,self).shortDescription(*p,**k) or str(self)
engine = self._engineInfo()
if engine is not None:
ret = "(%s) %s"%(engine,ret)
return ret
level = TXN
def _engineInfo(self,url=None):
if url is None:
url=self.url
url = self.url
return url
def _setup(self, url):
@ -111,23 +112,23 @@ class DB(Base):
#self.txn.add(self.engine)
def _disconnect(self):
if hasattr(self,'txn'):
if hasattr(self, 'txn'):
self.txn.rollback()
if hasattr(self,'session'):
if hasattr(self, 'session'):
self.session.close()
#if hasattr(self,'conn'):
# self.conn.close()
def _supported(self,url):
def _supported(self, url):
db = url.split(':',1)[0]
func = getattr(self,self._TestCase__testMethodName)
if hasattr(func,'supported'):
func = getattr(self, self._TestCase__testMethodName)
if hasattr(func, 'supported'):
return db in func.supported
if hasattr(func,'not_supported'):
if hasattr(func, 'not_supported'):
return not (db in func.not_supported)
# Neither list assigned; assume all are supported
return True
def _not_supported(self,url):
def _not_supported(self, url):
return not self._supported(url)
def refresh_table(self,name=None):
@ -141,4 +142,4 @@ class DB(Base):
if name is None:
name = self.table.name
self.meta.clear()
self.table = Table(name,self.meta,autoload=True)
self.table = Table(name, self.meta, autoload=True)

2
test/versioning/test_repository.py
View File

@ -51,7 +51,7 @@ class TestRepository(fixture.Pathed):
class TestVersionedRepository(fixture.Pathed):
"""Tests on an existing repository with a single python script"""
script_cls = script.PythonScript
def setup(self):
def setUp(self):
Repository.clear()
self.path_repos=self.tmp_repos()
# Create repository, script