Skip to content

Instantly share code, notes, and snippets.

@Kami
Created April 3, 2011 20:36
Show Gist options
  • Select an option

  • Save Kami/900775 to your computer and use it in GitHub Desktop.

Select an option

Save Kami/900775 to your computer and use it in GitHub Desktop.
Django parallel test runner setup_databases function
...
def setup_databases(self, **kwargs):
if VERSION[0] == 1:
if VERSION[1] == 2 and VERSION[2] < 4:
return self.setup_databases_12(**kwargs)
elif VERSION[2] >= 4 or VERSION[1] == 3:
return self.setup_databases_13(**kwargs)
raise Exception('Unsupported Django Version: %s' % (str(VERSION)))
def setup_databases_12(self, **kwargs):
# Taken from django.test.simple
old_names = []
mirrors = []
worker_index = kwargs.get('worker_index', None)
for alias in connections:
connection = connections[alias]
database_name = 'test_%d_%s' % (worker_index, connection.settings_dict['NAME'])
connection.settings_dict['TEST_NAME'] = database_name
if connection.settings_dict['TEST_MIRROR']:
mirrors.append((alias, connection))
mirror_alias = connection.settings_dict['TEST_MIRROR']
connections._connections[alias] = connections[mirror_alias]
else:
old_names.append((connection, connection.settings_dict['NAME']))
connection.creation.create_test_db(verbosity=0, autoclobber=not self.interactive)
return old_names, mirrors
def setup_databases_13(self, **kwargs):
# Taken from django.test.simple
from django.test.simple import dependency_ordered
mirrored_aliases = {}
test_databases = {}
dependencies = {}
worker_index = kwargs.get('worker_index', None)
for alias in connections:
connection = connections[alias]
database_name = 'test_%d_%s' % (worker_index, connection.settings_dict['NAME'])
connection.settings_dict['TEST_NAME'] = database_name
item = test_databases.setdefault(
connection.creation.test_db_signature(),
(connection.settings_dict['NAME'], [])
)
item[1].append(alias)
if alias != DEFAULT_DB_ALIAS:
dependencies[alias] = connection.settings_dict.get('TEST_DEPENDENCIES', [DEFAULT_DB_ALIAS])
old_names = []
mirrors = []
for signature, (db_name, aliases) in dependency_ordered(test_databases.items(), dependencies):
connection = connections[aliases[0]]
old_names.append((connection, db_name, True))
test_db_name = connection.creation.create_test_db(verbosity=0, autoclobber=not self.interactive)
for alias in aliases[1:]:
connection = connections[alias]
if db_name:
old_names.append((connection, db_name, False))
connection.settings_dict['NAME'] = test_db_name
else:
old_names.append((connection, db_name, True))
connection.creation.create_test_db(verbosity=0, autoclobber=not self.interactive)
for alias, mirror_alias in mirrored_aliases.items():
mirrors.append((alias, connections[alias].settings_dict['NAME']))
connections[alias].settings_dict['NAME'] = connections[mirror_alias].settings_dict['NAME']
return old_names, mirrors
...
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment