boulder-oracle-sprint: Fixed #3174 with Andreas Mock's patch.

git-svn-id: http://code.djangoproject.com/svn/django/branches/boulder-oracle-sprint@4278 bcc190cf-cafb-0310-a4f2-bffc1f526a37
django · Jan 3, 2007 · 499bfc1 · 499bfc1
1 parent 68aaa5d
commit 499bfc1
Showing 1 changed file with 215 additions and 55 deletions.
diff --git a/django/db/backends/oracle/creation.py b/django/db/backends/oracle/creation.py
@@ -33,95 +33,174 @@
 
 TEST_DATABASE_PREFIX = 'test_'
 PASSWORD = 'Im_a_lumberjack'
-OLD_DATABASE_USER = None
+REMEMBER = {}
-OLD_DATABASE_PASSWORD = None
+
 
 def create_test_db(settings, connection, backend, verbosity=1, autoclobber=False):
-    if verbosity >= 1:
-        print "Creating test database..."
 
     TEST_DATABASE_NAME = _test_database_name(settings)
+    TEST_DATABASE_USER = _test_database_user(settings)
+    TEST_DATABASE_PASSWD = _test_database_passwd(settings)
+    TEST_DATABASE_TBLSPACE = _test_database_tblspace(settings)
+    TEST_DATABASE_TBLSPACE_TMP = _test_database_tblspace_tmp(settings)
+
+    parameters = {
+        'dbname': TEST_DATABASE_NAME,
+        'user': TEST_DATABASE_USER,
+        'password': TEST_DATABASE_PASSWD,
+        'tblspace': TEST_DATABASE_TBLSPACE,
+        'tblspace_temp': TEST_DATABASE_TBLSPACE_TMP,
+ 	}
+
+    REMEMBER['user'] = settings.DATABASE_USER
+    REMEMBER['passwd'] = settings.DATABASE_PASSWORD
 
     cursor = connection.cursor()
-    try:
+    if _test_database_create(settings):
-        _create_test_db(cursor, TEST_DATABASE_NAME, verbosity)
+        if verbosity >= 1:
-    except Exception, e:
+            print 'Creating test database...'
-        sys.stderr.write("Got an error creating the test database: %s\n" % e)
+        try:
-        if not autoclobber:
+            _create_test_db(cursor, parameters, verbosity)
-            confirm = raw_input("It appears the test database, %s, already exists. Type 'yes' to delete it, or 'no' to cancel: " % TEST_DATABASE_NAME)
+        except Exception, e:
-        if autoclobber or confirm == 'yes':
+            sys.stderr.write("Got an error creating the test database: %s\n" % e)
-            try:
+            if not autoclobber:
-                if verbosity >= 1:
+                confirm = raw_input("It appears the test database, %s, already exists. Type 'yes' to delete it, or 'no' to cancel: " % TEST_DATABASE_NAME)
-                    print "Destroying old test database..."
+            if autoclobber or confirm == 'yes':
-                _destroy_test_db(cursor, TEST_DATABASE_NAME, verbosity)
+                try:
-                if verbosity >= 1:
+                    if verbosity >= 1:
-                    print "Creating test database..."
+                        print "Destroying old test database..."
-                _create_test_db(cursor, TEST_DATABASE_NAME, verbosity)
+                    _destroy_test_db(cursor, parameters, verbosity)
-            except Exception, e:
+                    if verbosity >= 1:
-                sys.stderr.write("Got an error recreating the test database: %s\n" % e)
+                        print "Creating test database..."
-                sys.exit(2)
+                    _create_test_db(cursor, parameters, verbosity)
-        else:
+                except Exception, e:
-            print "Tests cancelled."
+                    sys.stderr.write("Got an error recreating the test database: %s\n" % e)
-            sys.exit(1)
+                    sys.exit(2)
+            else:
+                print "Tests cancelled."
+                sys.exit(1)
+
+    if _test_user_create(settings):
+        if verbosity >= 1:
+            print "Creating test user..."
+        try:
+            _create_test_user(cursor, parameters, verbosity)
+        except Exception, e:
+            sys.stderr.write("Got an error creating the test user: %s\n" % e)
+            if not autoclobber:
+                confirm = raw_input("It appears the test user, %s, already exists. Type 'yes' to delete it, or 'no' to cancel: " % TEST_DATABASE_USER)
+            if autoclobber or confirm == 'yes':
+                try:
+                    if verbosity >= 1:
+                        print "Destroying old test user..."
+                    _destroy_test_user(cursor, parameters, verbosity)
+                    if verbosity >= 1:
+                        print "Creating test user..."
+                    _create_test_user(cursor, parameters, verbosity)
+                except Exception, e:
+                    sys.stderr.write("Got an error recreating the test user: %s\n" % e)
+                    sys.exit(2)
+            else:
+                print "Tests cancelled."
+                sys.exit(1)
 
     connection.close()
-    settings.DATABASE_USER = TEST_DATABASE_NAME
+    settings.DATABASE_USER = TEST_DATABASE_USER
-    settings.DATABASE_PASSWORD = PASSWORD
+    settings.DATABASE_PASSWORD = TEST_DATABASE_PASSWD
 
     # Get a cursor (even though we don't need one yet). This has
     # the side effect of initializing the test database.
     cursor = connection.cursor()
 
+
 def destroy_test_db(settings, connection, backend, old_database_name, verbosity=1):
-    if verbosity >= 1:
-        print "Destroying test database..."
     connection.close()
 
     TEST_DATABASE_NAME = _test_database_name(settings)
+    TEST_DATABASE_USER = _test_database_user(settings)
+    TEST_DATABASE_PASSWD = _test_database_passwd(settings)
+    TEST_DATABASE_TBLSPACE = _test_database_tblspace(settings)
+    TEST_DATABASE_TBLSPACE_TMP = _test_database_tblspace_tmp(settings)
+
     settings.DATABASE_NAME = old_database_name
-    #settings.DATABASE_USER = 'old_user'
+    settings.DATABASE_USER = REMEMBER['user']
-    #settings.DATABASE_PASSWORD = 'old_password'
+    settings.DATABASE_PASSWORD = REMEMBER['passwd']
-    settings.DATABASE_USER = 'mboersma'
+
-    settings.DATABASE_PASSWORD = 'password'
+    parameters = {
+        'dbname': TEST_DATABASE_NAME,
+        'user': TEST_DATABASE_USER,
+        'password': TEST_DATABASE_PASSWD,
+        'tblspace': TEST_DATABASE_TBLSPACE,
+        'tblspace_temp': TEST_DATABASE_TBLSPACE_TMP,
+ 	}
+
+    REMEMBER['user'] = settings.DATABASE_USER
+    REMEMBER['passwd'] = settings.DATABASE_PASSWORD
 
     cursor = connection.cursor()
     time.sleep(1) # To avoid "database is being accessed by other users" errors.
-    _destroy_test_db(cursor, TEST_DATABASE_NAME, verbosity)
+    if _test_user_create(settings):
+        if verbosity >= 1:
+            print 'Destroying test user...'
+        _destroy_test_user(cursor, parameters, verbosity)
+    if _test_database_create(settings):
+        if verbosity >= 1:
+            print 'Destroying test database...'
+        _destroy_test_db(cursor, parameters, verbosity)
     connection.close()
 
-def _create_test_db(cursor, dbname, verbosity):
+
+def _create_test_db(cursor, parameters, verbosity):
     if verbosity >= 2:
-        print "_create_test_db(): dbname = %s" % dbname
+        print "_create_test_db(): dbname = %s" % parameters['dbname']
     statements = [
-        """CREATE TABLESPACE %(user)s
+        """CREATE TABLESPACE %(tblspace)s
-           DATAFILE '%(user)s.dbf' SIZE 10M AUTOEXTEND ON NEXT 10M MAXSIZE 20M
+           DATAFILE '%(tblspace)s.dbf' SIZE 10M AUTOEXTEND ON NEXT 10M MAXSIZE 20M
         """,
-        """CREATE TEMPORARY TABLESPACE %(user)s_temp
+        """CREATE TEMPORARY TABLESPACE %(tblspace_temp)s
-           TEMPFILE '%(user)s_temp.dbf' SIZE 10M AUTOEXTEND ON NEXT 10M MAXSIZE 20M
+           TEMPFILE '%(tblspace_temp)s.dbf' SIZE 10M AUTOEXTEND ON NEXT 10M MAXSIZE 20M
         """,
+    ]
+    _execute_statements(cursor, statements, parameters, verbosity)
+
+
+def _create_test_user(cursor, parameters, verbosity):
+    if verbosity >= 2:
+        print "_create_test_user(): username = %s" % parameters['user']
+    statements = [
         """CREATE USER %(user)s
            IDENTIFIED BY %(password)s
-           DEFAULT TABLESPACE %(user)s
+           DEFAULT TABLESPACE %(tblspace)s
-           TEMPORARY TABLESPACE %(user)s_temp
+           TEMPORARY TABLESPACE %(tblspace_temp)s
         """,
         """GRANT CONNECT, RESOURCE TO %(user)s""",
     ]
-    _execute_statements(cursor, statements, dbname, verbosity)
+    _execute_statements(cursor, statements, parameters, verbosity)
 
-def _destroy_test_db(cursor, dbname, verbosity):
+
+def _destroy_test_db(cursor, parameters, verbosity):
     if verbosity >= 2:
-        print "_destroy_test_db(): dbname=%s" % dbname
+        print "_destroy_test_db(): dbname=%s" % parameters['dbname']
     statements = [
-        'DROP USER %(user)s CASCADE',
+        'DROP TABLESPACE %(tblspace)s INCLUDING CONTENTS AND DATAFILES CASCADE CONSTRAINTS',
-        'DROP TABLESPACE %(user)s INCLUDING CONTENTS AND DATAFILES CASCADE CONSTRAINTS',
+        'DROP TABLESPACE %(tblspace_temp)s INCLUDING CONTENTS AND DATAFILES CASCADE CONSTRAINTS',
-        'DROP TABLESPACE %(user)s_TEMP INCLUDING CONTENTS AND DATAFILES CASCADE CONSTRAINTS',
         ]
-    _execute_statements(cursor, statements, dbname, verbosity)
+    _execute_statements(cursor, statements, parameters, verbosity)
+
 
-def _execute_statements(cursor, statements, dbname, verbosity):
+def _destroy_test_user(cursor, parameters, verbosity):
+    if verbosity >= 2:
+        print "_destroy_test_user(): user=%s" % parameters['user']
+        print "Be patient.  This can take some time..."
+    statements = [
+        'DROP USER %(user)s CASCADE',
+    ]
+    _execute_statements(cursor, statements, parameters, verbosity)
+
+
+def _execute_statements(cursor, statements, parameters, verbosity):
     for template in statements:
-        stmt = template % {'user': dbname,
+        stmt = template % parameters
-			   'password': PASSWORD}
         if verbosity >= 2:
             print stmt
         try:
@@ -130,9 +209,90 @@ def _execute_statements(cursor, statements, dbname, verbosity):
             sys.stderr.write("Failed (%s)\n" % (err))
             raise
 
+
 def _test_database_name(settings):
-    if settings.TEST_DATABASE_NAME:
+    name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME
-        name = settings.TEST_DATABASE_NAME
+    try:
-    else:
+        if settings.TEST_DATABASE_NAME:
-        name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME
+            name = settings.TEST_DATABASE_NAME
+    except AttributeError:
+        pass
+    except:
+        raise
+    return name
+
+
+def _test_database_create(settings):
+    name = True
+    try:
+        if settings.TEST_DATABASE_CREATE:
+            name = True
+        else:
+            name = False
+    except AttributeError:
+        pass
+    except:
+        raise
+    return name
+
+
+def _test_user_create(settings):
+    name = True
+    try:
+        if settings.TEST_USER_CREATE:
+            name = True
+        else:
+            name = False
+    except AttributeError:
+        pass
+    except:
+        raise
+    return name
+
+
+def _test_database_user(settings):
+    name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME
+    try:
+        if settings.TEST_DATABASE_USER:
+            name = settings.TEST_DATABASE_USER
+    except AttributeError:
+        pass
+    except:
+        raise
+    return name
+
+
+def _test_database_passwd(settings):
+    name = PASSWORD
+    try:
+        if settings.TEST_DATABASE_PASSWD:
+            name = settings.TEST_DATABASE_PASSWD
+    except AttributeError:
+        pass
+    except:
+        raise
+    return name
+
+
+def _test_database_tblspace(settings):
+    name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME
+    try:
+        if settings.TEST_DATABASE_TBLSPACE:
+            name = settings.TEST_DATABASE_TBLSPACE
+    except AttributeError:
+        pass
+    except:
+        raise
+    return name
+
+
+def _test_database_tblspace_tmp(settings):
+    name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME + '_temp'
+    try:
+        if settings.TEST_DATABASE_TBLSPACE_TMP:
+            name = settings.TEST_DATABASE_TBLSPACE_TMP
+    except AttributeError:
+        pass
+    except:
+        raise
     return name