Skip to content

Commit

Permalink
[1.2.X] Fixed #14415 -- Corrected the process of creating and destroy…
Browse files Browse the repository at this point in the history
…ing test databases to prevent accidental deletion of the source database. Also improves the mirroring process to prevent some cross-connection effects. Thanks to Shai Berger for the help diagnosing and testing.

Backport of r14696 from trunk.

git-svn-id: http://code.djangoproject.com/svn/django/branches/releases/1.2.X@14697 bcc190cf-cafb-0310-a4f2-bffc1f526a37
  • Loading branch information
freakboy3742 committed Nov 25, 2010
1 parent 73ff0f1 commit 7704bd9
Showing 1 changed file with 49 additions and 13 deletions.
62 changes: 49 additions & 13 deletions django/test/simple.py
Original file line number Original file line Diff line number Diff line change
Expand Up @@ -255,19 +255,52 @@ def build_suite(self, test_labels, extra_tests=None, **kwargs):


def setup_databases(self, **kwargs): def setup_databases(self, **kwargs):
from django.db import connections from django.db import connections
old_names = []
mirrors = [] # First pass -- work out which databases actually need to be created,
# and which ones are test mirrors or duplicate entries in DATABASES
mirrored_aliases = {}
test_databases = {}
for alias in connections: for alias in connections:
connection = connections[alias] connection = connections[alias]
# If the database is a test mirror, redirect it's connection
# instead of creating a test database.
if connection.settings_dict['TEST_MIRROR']: if connection.settings_dict['TEST_MIRROR']:
mirrors.append((alias, connection)) # If the database is marked as a test mirror, save
mirror_alias = connection.settings_dict['TEST_MIRROR'] # the alias.
connections._connections[alias] = connections[mirror_alias] mirrored_aliases[alias] = connection.settings_dict['TEST_MIRROR']
else: else:
old_names.append((connection, connection.settings_dict['NAME'])) # Store the (engine, name) pair. If we have two aliases
connection.creation.create_test_db(self.verbosity, autoclobber=not self.interactive) # with the same pair, we only need to create the test database
# once.
test_databases.setdefault((
connection.settings_dict['HOST'],
connection.settings_dict['PORT'],
connection.settings_dict['ENGINE'],
connection.settings_dict['NAME'],
), []).append(alias)

# Second pass -- actually create the databases.
old_names = []
mirrors = []
for (host, port, engine, db_name), aliases in test_databases.items():
# Actually create the database for the first connection
connection = connections[aliases[0]]
old_names.append((connection, db_name, True))
test_db_name = connection.creation.create_test_db(self.verbosity, 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:
# If settings_dict['NAME'] isn't defined, we have a backend where
# the name isn't important -- e.g., SQLite, which uses :memory:.
# Force create the database instead of assuming it's a duplicate.
old_names.append((connection, db_name, True))
connection.creation.create_test_db(self.verbosity, 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 return old_names, mirrors


def run_suite(self, suite, **kwargs): def run_suite(self, suite, **kwargs):
Expand All @@ -277,11 +310,14 @@ def teardown_databases(self, old_config, **kwargs):
from django.db import connections from django.db import connections
old_names, mirrors = old_config old_names, mirrors = old_config
# Point all the mirrors back to the originals # Point all the mirrors back to the originals
for alias, connection in mirrors: for alias, old_name in mirrors:
connections._connections[alias] = connection connections[alias].settings_dict['NAME'] = old_name
# Destroy all the non-mirror databases # Destroy all the non-mirror databases
for connection, old_name in old_names: for connection, old_name, destroy in old_names:
connection.creation.destroy_test_db(old_name, self.verbosity) if destroy:
connection.creation.destroy_test_db(old_name, self.verbosity)
else:
connection.settings_dict['NAME'] = old_name


def teardown_test_environment(self, **kwargs): def teardown_test_environment(self, **kwargs):
teardown_test_environment() teardown_test_environment()
Expand Down

0 comments on commit 7704bd9

Please sign in to comment.