Skip to content

Commit

Permalink
Browse files Browse the repository at this point in the history
Better control over database isolation of tests
 * reworked internal implementation of how (or rather when) database creation, transaction begin/rollback, flush and fixtures loading are made so that minimum of these time-consuming operations are made, and developer now have better control when those operations are (or are not) done.
 * database isolation control now works for packages and modules too (i.e. if you want db flush after each TestCase or test function in the module, you just put database_flush = True at the beginning of the module)
 * added tests for testing features mentioned above
  • Loading branch information
tdivis committed Apr 6, 2012
1 parent b9460d5 commit 580fce7
Show file tree
Hide file tree
Showing 12 changed files with 396 additions and 71 deletions.
4 changes: 2 additions & 2 deletions djangosanetesting/cases.py
Expand Up @@ -187,7 +187,7 @@ class NonIsolatedDatabaseTestCase(DatabaseTestCase):
"""
database_single_transaction = False
database_flush = False
database_single_transaction_alfter_case = True
database_single_transaction_at_end = True


class NonIsolatedDestructiveDatabaseTestCase(DestructiveDatabaseTestCase):
Expand All @@ -197,7 +197,7 @@ class NonIsolatedDestructiveDatabaseTestCase(DestructiveDatabaseTestCase):
"""
database_single_transaction = False
database_flush = False
database_flush_alfter_case = True
database_flush_at_end = True


class HttpTestCase(DestructiveDatabaseTestCase):
Expand Down
95 changes: 95 additions & 0 deletions djangosanetesting/contextstack.py
@@ -0,0 +1,95 @@
"""
NoseContextStack to hande important information while nosetests traverses (DFS) through
packages, modules, test_cases classess, test methods and test functions.
"""

class ContextInfo(object):
""" Information about one node in traversed tree """
def __init__(self, fixtures=None, no_database_interaction=None,
database_single_transaction=None, database_flush=None,
database_single_transaction_at_end=None, database_flush_at_end=None):
super(ContextInfo, self).__init__()
self.fixtures = fixtures # list of fixtures
self.fixtures_loaded = False # flag - fixtures are loaded in test db?
self.no_database_interaction = no_database_interaction
self.database_single_transaction = database_single_transaction
self.database_flush = database_flush
self.database_single_transaction_at_end = database_single_transaction_at_end
self.database_flush_at_end = database_flush_at_end

# Sanity check:
if self.no_database_interaction:
for attr_name in ['fixtures', 'database_single_transaction', 'database_flush',
'database_single_transaction_at_end', 'database_flush_at_end']:
if getattr(self, attr_name):
raise RuntimeError('You cannot have "%s" without database' % attr_name)
if self.database_flush:
for attr_name in ['database_single_transaction', 'database_flush_at_end', 'database_single_transaction_at_end']:
if getattr(self, attr_name):
raise RuntimeError('Having "database_flush" and "%s" does not make sence.' % attr_name)
if self.database_flush_at_end and self.database_single_transaction_at_end:
raise RuntimeError('Having "database_flush_at_end" and "database_single_transaction_at_end" does not make sence.')
if self.database_single_transaction and self.database_single_transaction_at_end:
raise RuntimeError('Having "database_single_transactoin_at_end" and "database_single_transaction" does not make sence.')


def __repr__(self):
return '<nodeinfo %s, %s, %s, %s, %s, %s, %s>' % (self.fixtures, self.fixtures_loaded, self.no_database_interaction,
self.database_single_transaction, self.database_flush,
self.database_single_transaction_at_end, self.database_flush_at_end)


class ContextStack(list):
def __init__(self):
super(ContextStack, self).__init__()


def push(self, node):
self.append(node)

def push_context(self, nose_context):
''' Get NodeInfo object from nose context, push it to stack and return it as return value '''
node = ContextInfo(
fixtures=getattr(nose_context, 'fixtures', None),
no_database_interaction=getattr(nose_context, 'no_database_interaction', None),
database_single_transaction=getattr(nose_context, 'database_single_transaction', None),
database_flush=getattr(nose_context, 'database_flush', None),
database_single_transaction_at_end=getattr(nose_context, 'database_single_transaction_at_end', None),
database_flush_at_end=getattr(nose_context, 'database_flush_at_end', None)
)
self.push(node)
return node

def top(self):
return self[-1]

def set_attr_whole_stack(self, attr, value):
''' Sets attribute attr to value for all nodes in the stack '''
for node in self:
setattr(node, attr, value)

def get_any_attr(self, attr):
''' Search the whole stack for attr, returns True if any of the nodes has attr == True '''
for node in self:
if getattr(node, attr, None):
return True
return False

def get_unloaded_fixtures(self):
''' Return set of all fixtures, which are not loaded (so not including items with fixtures_loaded == True) '''
return set([fixture
for context in self if not context.fixtures_loaded and context.fixtures
for fixture in context.fixtures])

def is_transaction(self):
''' Returns whether current state should be in transaction (this is used to check, whether transaction
should be started and whether the commit=False should be used for fixtures. (of course first priority have
test mathod/function attributes, only if unspecified there, this stack method is used).
'''
for node in reversed(self):
if node.database_single_transaction or node.database_single_transaction_at_end:
return True
elif node.database_flush or node.database_flush_at_end:
# flush is stronger than transaction so we don't care what is deeper in the stack anymore:
return False
return False
143 changes: 74 additions & 69 deletions djangosanetesting/noseplugins.py
Expand Up @@ -22,7 +22,7 @@
import djangosanetesting
from djangosanetesting import MULTIDB_SUPPORT, DEFAULT_DB_ALIAS
from djangosanetesting.cache import flush_django_cache

from djangosanetesting.contextstack import ContextStack
from djangosanetesting.utils import (
get_databases, get_live_server_path,
get_server_handler,
Expand Down Expand Up @@ -97,6 +97,13 @@ def getattr_test(nose_test, attr_name, default=False):
else:
return getattr(get_test_case_instance(nose_test), attr_name, default)

def getattr_test_meth(nose_test, attr, default=None):
""" Return attribute of test method/function """
test_meth = get_test_case_method(nose_test)
if test_meth is None:
raise RuntimeError('%s is not test method/function' % nose_test)
return getattr(test_meth, attr, default)

def enable_test(test_case, plugin_attribute):
if not getattr(test_case, plugin_attribute, False):
setattr(test_case, plugin_attribute, True)
Expand Down Expand Up @@ -330,51 +337,30 @@ def __init__(self):
self.persist_test_database = None
self.test_database_created = False
self.old_config = None
self.stack = ContextStack()
self.db_rollback_done = True
self.db_flush_done = True

def startContext(self, context):
if ismodule(context) or is_test_case_class(context):
if ismodule(context):
attr_suffix = ''
else:
attr_suffix = '_after_all_tests'
if getattr(context, 'database_single_transaction' + attr_suffix, False):
#TODO: When no test case in this module needing database is run (for example
# user selected only one unitTestCase), database should not be initialized.
# So it would be best if db is initialized when first test case needing
# database is run.

# create test database if not already created
if not self.test_database_created:
self._create_test_databases()

if getattr(context, 'database_single_transaction' + attr_suffix, False):
from django.db import transaction
transaction.enter_transaction_management()
transaction.managed(True)

# when used from startTest, nose-wrapped testcase is provided -- while now,
# we have 'bare' test case.

self._prepare_tests_fixtures(context)
#print '>>>>', context
self.stack.push_context(context)

def stopContext(self, context):
if ismodule(context) or is_test_case_class(context):
from django.conf import settings
from django.db import transaction

if ismodule(context):
attr_suffix = ''
else:
attr_suffix = '_after_all_tests'

if self.test_database_created:
if getattr(context, 'database_single_transaction' + attr_suffix, False):
transaction.rollback()
transaction.leave_transaction_management()

if getattr(context, "database_flush" + attr_suffix, None):
for db in self._get_tests_databases(getattr(context, 'multidb', False)):
getattr(settings, "TEST_DATABASE_FLUSH_COMMAND", flush_database)(self, database=db)
#print '<<<<', context
node = self.stack.pop()

if self.test_database_created:
if not self.db_rollback_done and (
node.database_single_transaction_at_end or
(self.stack and self.stack.top().database_single_transaction)
):
self._do_rollback(context)

if not self.db_flush_done and (
node.database_flush_at_end or
(self.stack and self.stack.top().database_flush)
):
self._do_flush(context)

def options(self, parser, env=os.environ): # pylint: disable=W0102
Plugin.options(self, parser, env)
Expand Down Expand Up @@ -450,8 +436,6 @@ def finalize(self, result): # pylint: disable=W0613

if not self.persist_test_database and getattr(self, 'test_database_created', None):
self.teardown_databases(self.old_config, verbosity=False)
# from django.db import connection
# connection.creation.destroy_test_db(self.old_name, verbosity=False)

def beforeTest(self, test):
# enabling test must be in beforeTest so test can be checked for is_skipped() where is also required_sane_plugin
Expand All @@ -466,7 +450,6 @@ def startTest(self, test):
"""
When preparing test, check whether to make our database fresh
"""

test_case = get_test_case_class(test)
if issubclass(test_case, DjangoTestCase):
return
Expand Down Expand Up @@ -497,23 +480,33 @@ def startTest(self, test):
#####
### Database handling follows
#####
if getattr_test(test, 'no_database_interaction', False):
# for true unittests, we can leave database handling for later,
if getattr_test_meth(test, 'no_database_interaction', None) or self.stack.top().no_database_interaction:
# for true unittests, we don't need database handling,
# as unittests by definition do not interacts with database
return

# create test database if not already created
if not self.test_database_created:
self._create_test_databases()

# make self.transaction available
test_case.transaction = transaction

if getattr_test(test, 'database_single_transaction'):
# detect if we are in single transaction mode:
if (getattr_test_meth(test, "database_single_transaction", False)
or (not getattr_test_meth(test, "database_flush", False) and self.stack.is_transaction())):
is_transaction = True
else:
is_transaction = False
# start transaction if needed:
if self.db_rollback_done and is_transaction:
transaction.enter_transaction_management()
transaction.managed(True)
self.db_rollback_done = False
# we are in database test, so we need to reset this flag:
self.db_flush_done = False

self._prepare_tests_fixtures(test)
# don't use commits if we are in single_transaction mode
self._prepare_tests_fixtures(test, commit=not is_transaction)

def stopTest(self, test):
"""
Expand All @@ -524,7 +517,6 @@ def stopTest(self, test):
if issubclass(test_case, DjangoTestCase):
return

from django.db import transaction
from django.conf import settings

test_case = get_test_case_class(test)
Expand All @@ -543,13 +535,17 @@ def stopTest(self, test):
# as unittests by definition do not interacts with database
return

if getattr_test(test, 'database_single_transaction'):
transaction.rollback()
transaction.leave_transaction_management()
if not self.db_rollback_done and (
getattr_test_meth(test, 'database_single_transaction') or
(getattr_test_meth(test, 'database_single_transaction') is None and self.stack.top().database_single_transaction)
):
self._do_rollback(test)

if getattr_test(test, "database_flush", True):
for db in self._get_tests_databases(getattr_test(test, 'multi_db')):
getattr(settings, "TEST_DATABASE_FLUSH_COMMAND", flush_database)(self, database=db)
if not self.db_flush_done and (
getattr_test_meth(test, "database_flush") or
(getattr_test_meth(test, 'database_flush') is None and self.stack.top().database_flush)
):
self._do_flush(test)

def _get_databases(self):
try:
Expand All @@ -576,18 +572,12 @@ def _get_tests_databases(self, multi_db):
databases = connections
return databases

def _prepare_tests_fixtures(self, test):
# fixtures are loaded inside transaction, thus we don't need to flush
# between database_single_transaction tests when their fixtures differ
if hasattr_test(test, 'fixtures'):
if getattr_test(test, "database_flush", True):
# commits are allowed during tests
commit = True
else:
commit = False
for db in self._get_tests_databases(getattr_test(test, 'multi_db')):
call_command('loaddata', *getattr_test(test, 'fixtures'),
**{'verbosity': 0, 'commit' : commit, 'database' : db})
def _prepare_tests_fixtures(self, test, commit):
fixtures = self.stack.get_unloaded_fixtures().union(getattr_test_meth(test, 'fixtures', []))
for db in self._get_tests_databases(getattr_test(test, 'multi_db')):
call_command('loaddata', *fixtures,
**{'verbosity': 0, 'commit' : commit, 'database' : db})
self.stack.set_attr_whole_stack('fixtures_loaded', True)

def _create_test_databases(self):
from django.conf import settings
Expand Down Expand Up @@ -622,6 +612,21 @@ def _create_test_databases(self):
if getattr(settings, "FLUSH_TEST_DATABASE_AFTER_INITIAL_SYNCDB", False):
getattr(settings, "TEST_DATABASE_FLUSH_COMMAND", flush_database)(self, database=db)

def _do_flush(self, test):
from django.conf import settings
for db in self._get_tests_databases(getattr_test(test, 'multi_db')):
getattr(settings, "TEST_DATABASE_FLUSH_COMMAND", flush_database)(self, database=db)
self.db_flush_done = True
self.db_rollback_done = True # flush is stronger than rollback
self.stack.set_attr_whole_stack('fixtures_loaded', False)

def _do_rollback(self, test):
from django.db import transaction
transaction.rollback()
transaction.leave_transaction_management()
self.db_rollback_done = True
self.stack.set_attr_whole_stack('fixtures_loaded', False)


class DjangoTranslationPlugin(Plugin):
"""
Expand Down
Empty file.

0 comments on commit 580fce7

Please sign in to comment.