From 499bfc1f4a8dd6eb99f1c171bd8ef1c252fad0c5 Mon Sep 17 00:00:00 2001 From: Boulder Sprinters Date: Wed, 3 Jan 2007 20:23:07 +0000 Subject: [PATCH] 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/db/backends/oracle/creation.py | 270 ++++++++++++++++++++------ 1 file changed, 215 insertions(+), 55 deletions(-) diff --git a/django/db/backends/oracle/creation.py b/django/db/backends/oracle/creation.py index c3326cf84371d..8d62dcb2a1189 100644 --- 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 -OLD_DATABASE_PASSWORD = None +REMEMBER = {} + 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: - _create_test_db(cursor, TEST_DATABASE_NAME, verbosity) - except Exception, e: - sys.stderr.write("Got an error creating the test database: %s\n" % e) - if not autoclobber: - confirm = raw_input("It appears the test database, %s, already exists. Type 'yes' to delete it, or 'no' to cancel: " % TEST_DATABASE_NAME) - if autoclobber or confirm == 'yes': - try: - if verbosity >= 1: - print "Destroying old test database..." - _destroy_test_db(cursor, TEST_DATABASE_NAME, verbosity) - if verbosity >= 1: - print "Creating test database..." - _create_test_db(cursor, TEST_DATABASE_NAME, verbosity) - except Exception, e: - sys.stderr.write("Got an error recreating the test database: %s\n" % e) - sys.exit(2) - else: - print "Tests cancelled." - sys.exit(1) + if _test_database_create(settings): + if verbosity >= 1: + print 'Creating test database...' + try: + _create_test_db(cursor, parameters, verbosity) + except Exception, e: + sys.stderr.write("Got an error creating the test database: %s\n" % e) + if not autoclobber: + confirm = raw_input("It appears the test database, %s, already exists. Type 'yes' to delete it, or 'no' to cancel: " % TEST_DATABASE_NAME) + if autoclobber or confirm == 'yes': + try: + if verbosity >= 1: + print "Destroying old test database..." + _destroy_test_db(cursor, parameters, verbosity) + if verbosity >= 1: + print "Creating test database..." + _create_test_db(cursor, parameters, verbosity) + except Exception, e: + sys.stderr.write("Got an error recreating the test database: %s\n" % e) + 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_PASSWORD = PASSWORD + settings.DATABASE_USER = TEST_DATABASE_USER + 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_PASSWORD = 'old_password' - settings.DATABASE_USER = 'mboersma' - settings.DATABASE_PASSWORD = 'password' + settings.DATABASE_USER = REMEMBER['user'] + settings.DATABASE_PASSWORD = REMEMBER['passwd'] + + 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 - DATAFILE '%(user)s.dbf' SIZE 10M AUTOEXTEND ON NEXT 10M MAXSIZE 20M + """CREATE TABLESPACE %(tblspace)s + DATAFILE '%(tblspace)s.dbf' SIZE 10M AUTOEXTEND ON NEXT 10M MAXSIZE 20M """, - """CREATE TEMPORARY TABLESPACE %(user)s_temp - TEMPFILE '%(user)s_temp.dbf' SIZE 10M AUTOEXTEND ON NEXT 10M MAXSIZE 20M + """CREATE TEMPORARY TABLESPACE %(tblspace_temp)s + 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 - TEMPORARY TABLESPACE %(user)s_temp + DEFAULT TABLESPACE %(tblspace)s + 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 %(user)s INCLUDING CONTENTS AND DATAFILES CASCADE CONSTRAINTS', - 'DROP TABLESPACE %(user)s_TEMP INCLUDING CONTENTS AND DATAFILES CASCADE CONSTRAINTS', + 'DROP TABLESPACE %(tblspace)s INCLUDING CONTENTS AND DATAFILES CASCADE CONSTRAINTS', + 'DROP TABLESPACE %(tblspace_temp)s 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, - 'password': PASSWORD} + stmt = template % parameters 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 = settings.TEST_DATABASE_NAME - else: - name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME + name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME + try: + if settings.TEST_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