summaryrefslogtreecommitdiff
path: root/django/db/backends/oracle/creation.py
diff options
context:
space:
mode:
authorShai Berger <shai@platonix.com>2015-06-18 00:10:52 +0300
committerShai Berger <shai@platonix.com>2015-06-20 23:48:18 +0300
commitabd7e48af78bbe6ec31359c7826757d7166e55d6 (patch)
tree4e6a791a50b964c58d812d4b79c5304de87cc2df /django/db/backends/oracle/creation.py
parent8b7bd62ae52313dbe96f27f21fbd63f0dd061744 (diff)
downloaddjango-abd7e48af78bbe6ec31359c7826757d7166e55d6.tar.gz
Changed Oracle's test-database-creation to use an explicit main-db-connection
rather than just relying on manipulation of settings to determine which invocation of connection.cursor() opens a connection to the test database and which opens a connection to the main database. Thanks Aymeric Augustin for motivation and Tim Graham for review.
Diffstat (limited to 'django/db/backends/oracle/creation.py')
-rw-r--r--django/db/backends/oracle/creation.py39
1 files changed, 33 insertions, 6 deletions
diff --git a/django/db/backends/oracle/creation.py b/django/db/backends/oracle/creation.py
index 5a5ff517b8..afe965d664 100644
--- a/django/db/backends/oracle/creation.py
+++ b/django/db/backends/oracle/creation.py
@@ -4,6 +4,7 @@ import time
from django.conf import settings
from django.db.backends.base.creation import BaseDatabaseCreation
from django.db.utils import DatabaseError
+from django.utils.functional import cached_property
from django.utils.six.moves import input
TEST_DATABASE_PREFIX = 'test_'
@@ -12,9 +13,26 @@ PASSWORD = 'Im_a_lumberjack'
class DatabaseCreation(BaseDatabaseCreation):
+ @cached_property
+ def _maindb_connection(self):
+ """
+ This is analogous to other backends' `_nodb_connection` property,
+ which allows access to an "administrative" connection which can
+ be used to manage the test databases.
+ For Oracle, the only connection that can be used for that purpose
+ is the main (non-test) connection.
+ """
+ settings_dict = settings.DATABASES[self.connection.alias]
+ user = settings_dict.get('SAVED_USER') or settings_dict['USER']
+ password = settings_dict.get('SAVED_PASSWORD') or settings_dict['PASSWORD']
+ settings_dict = settings_dict.copy()
+ settings_dict.update(USER=user, PASSWORD=password)
+ DatabaseWrapper = type(self.connection)
+ return DatabaseWrapper(settings_dict, alias=self.connection.alias)
+
def _create_test_db(self, verbosity=1, autoclobber=False, keepdb=False):
parameters = self._get_test_db_params()
- cursor = self.connection.cursor()
+ cursor = self._maindb_connection.cursor()
if self._test_database_create():
try:
self._execute_test_db_creation(cursor, parameters, verbosity)
@@ -79,8 +97,18 @@ class DatabaseCreation(BaseDatabaseCreation):
print("Tests cancelled.")
sys.exit(1)
- self.connection.close() # done with main user -- test user and tablespaces created
+ self._maindb_connection.close() # done with main user -- test user and tablespaces created
+ self._switch_to_test_user(parameters)
+ return self.connection.settings_dict['NAME']
+ def _switch_to_test_user(self, parameters):
+ """
+ Oracle doesn't have the concept of separate databases under the same user.
+ Thus, we use a separate user (see _create_test_db). This method is used
+ to switch to that user. We will need the main user again for clean-up when
+ we end testing, so we keep its credentials in SAVED_USER/SAVED_PASSWORD
+ entries in the settings dict.
+ """
real_settings = settings.DATABASES[self.connection.alias]
real_settings['SAVED_USER'] = self.connection.settings_dict['SAVED_USER'] = \
self.connection.settings_dict['USER']
@@ -92,8 +120,6 @@ class DatabaseCreation(BaseDatabaseCreation):
self.connection.settings_dict['USER'] = parameters['user']
real_settings['PASSWORD'] = self.connection.settings_dict['PASSWORD'] = parameters['password']
- return self.connection.settings_dict['NAME']
-
def set_as_test_mirror(self, primary_settings_dict):
"""
Set this database up to be used in testing as a mirror of a primary database
@@ -144,8 +170,9 @@ class DatabaseCreation(BaseDatabaseCreation):
"""
self.connection.settings_dict['USER'] = self.connection.settings_dict['SAVED_USER']
self.connection.settings_dict['PASSWORD'] = self.connection.settings_dict['SAVED_PASSWORD']
+ self.connection.close()
parameters = self._get_test_db_params()
- cursor = self.connection.cursor()
+ cursor = self._maindb_connection.cursor()
time.sleep(1) # To avoid "database is being accessed by other users" errors.
if self._test_user_create():
if verbosity >= 1:
@@ -155,7 +182,7 @@ class DatabaseCreation(BaseDatabaseCreation):
if verbosity >= 1:
print('Destroying test database tables...')
self._execute_test_db_destruction(cursor, parameters, verbosity)
- self.connection.close()
+ self._maindb_connection.close()
def _execute_test_db_creation(self, cursor, parameters, verbosity):
if verbosity >= 2: