Created
April 3, 2011 20:36
-
-
Save Kami/900775 to your computer and use it in GitHub Desktop.
Django parallel test runner setup_databases function
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| ... | |
| 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