Merge pull request #4877 from shaib/oracle-maindb-connection

Changed Oracle test-db-creation to use an explicit main-db-connection

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: