1 import sys, time

     2 from django.core import management

     3 

     4 # This dictionary maps Field objects to their associated Oracle column

     5 # types, as strings. Column-type strings can contain format strings; they'll

     6 # be interpolated against the values of Field.__dict__ before being output.

     7 # If a column type is set to None, it won't be included in the output.

     8 DATA_TYPES = {

     9     'AutoField':                    'NUMBER(11)',

    10     'BooleanField':                 'NUMBER(1) CHECK (%(column)s IN (0,1))',

    11     'CharField':                    'NVARCHAR2(%(max_length)s)',

    12     'CommaSeparatedIntegerField':   'VARCHAR2(%(max_length)s)',

    13     'DateField':                    'DATE',

    14     'DateTimeField':                'TIMESTAMP',

    15     'DecimalField':                 'NUMBER(%(max_digits)s, %(decimal_places)s)',

    16     'FileField':                    'NVARCHAR2(%(max_length)s)',

    17     'FilePathField':                'NVARCHAR2(%(max_length)s)',

    18     'FloatField':                   'DOUBLE PRECISION',

    19     'ImageField':                   'NVARCHAR2(%(max_length)s)',

    20     'IntegerField':                 'NUMBER(11)',

    21     'IPAddressField':               'VARCHAR2(15)',

    22     'NullBooleanField':             'NUMBER(1) CHECK ((%(column)s IN (0,1)) OR (%(column)s IS NULL))',

    23     'OneToOneField':                'NUMBER(11)',

    24     'PhoneNumberField':             'VARCHAR2(20)',

    25     'PositiveIntegerField':         'NUMBER(11) CHECK (%(column)s >= 0)',

    26     'PositiveSmallIntegerField':    'NUMBER(11) CHECK (%(column)s >= 0)',

    27     'SlugField':                    'NVARCHAR2(50)',

    28     'SmallIntegerField':            'NUMBER(11)',

    29     'TextField':                    'NCLOB',

    30     'TimeField':                    'TIMESTAMP',

    31     'URLField':                     'VARCHAR2(%(max_length)s)',

    32     'USStateField':                 'CHAR(2)',

    33 }

    34 

    35 TEST_DATABASE_PREFIX = 'test_'

    36 PASSWORD = 'Im_a_lumberjack'

    37 REMEMBER = {}

    38 

    39 def create_test_db(settings, connection, verbosity=1, autoclobber=False):

    40     TEST_DATABASE_NAME = _test_database_name(settings)

    41     TEST_DATABASE_USER = _test_database_user(settings)

    42     TEST_DATABASE_PASSWD = _test_database_passwd(settings)

    43     TEST_DATABASE_TBLSPACE = _test_database_tblspace(settings)

    44     TEST_DATABASE_TBLSPACE_TMP = _test_database_tblspace_tmp(settings)

    45 

    46     parameters = {

    47         'dbname': TEST_DATABASE_NAME,

    48         'user': TEST_DATABASE_USER,

    49         'password': TEST_DATABASE_PASSWD,

    50         'tblspace': TEST_DATABASE_TBLSPACE,

    51         'tblspace_temp': TEST_DATABASE_TBLSPACE_TMP,

    52  	}

    53 

    54     REMEMBER['user'] = settings.DATABASE_USER

    55     REMEMBER['passwd'] = settings.DATABASE_PASSWORD

    56 

    57     cursor = connection.cursor()

    58     if _test_database_create(settings):

    59         if verbosity >= 1:

    60             print 'Creating test database...'

    61         try:

    62             _create_test_db(cursor, parameters, verbosity)

    63         except Exception, e:

    64             sys.stderr.write("Got an error creating the test database: %s\n" % e)

    65             if not autoclobber:

    66                 confirm = raw_input("It appears the test database, %s, already exists. Type 'yes' to delete it, or 'no' to cancel: " % TEST_DATABASE_NAME)

    67             if autoclobber or confirm == 'yes':

    68                 try:

    69                     if verbosity >= 1:

    70                         print "Destroying old test database..."

    71                     _destroy_test_db(cursor, parameters, verbosity)

    72                     if verbosity >= 1:

    73                         print "Creating test database..."

    74                     _create_test_db(cursor, parameters, verbosity)

    75                 except Exception, e:

    76                     sys.stderr.write("Got an error recreating the test database: %s\n" % e)

    77                     sys.exit(2)

    78             else:

    79                 print "Tests cancelled."

    80                 sys.exit(1)

    81 

    82     if _test_user_create(settings):

    83         if verbosity >= 1:

    84             print "Creating test user..."

    85         try:

    86             _create_test_user(cursor, parameters, verbosity)

    87         except Exception, e:

    88             sys.stderr.write("Got an error creating the test user: %s\n" % e)

    89             if not autoclobber:

    90                 confirm = raw_input("It appears the test user, %s, already exists. Type 'yes' to delete it, or 'no' to cancel: " % TEST_DATABASE_USER)

    91             if autoclobber or confirm == 'yes':

    92                 try:

    93                     if verbosity >= 1:

    94                         print "Destroying old test user..."

    95                     _destroy_test_user(cursor, parameters, verbosity)

    96                     if verbosity >= 1:

    97                         print "Creating test user..."

    98                     _create_test_user(cursor, parameters, verbosity)

    99                 except Exception, e:

   100                     sys.stderr.write("Got an error recreating the test user: %s\n" % e)

   101                     sys.exit(2)

   102             else:

   103                 print "Tests cancelled."

   104                 sys.exit(1)

   105 

   106     connection.close()

   107     settings.DATABASE_USER = TEST_DATABASE_USER

   108     settings.DATABASE_PASSWORD = TEST_DATABASE_PASSWD

   109 

   110     management.call_command('syncdb', verbosity=verbosity, interactive=False)

   111 

   112     # Get a cursor (even though we don't need one yet). This has

   113     # the side effect of initializing the test database.

   114     cursor = connection.cursor()

   115 

   116 def destroy_test_db(settings, connection, old_database_name, verbosity=1):

   117     connection.close()

   118 

   119     TEST_DATABASE_NAME = _test_database_name(settings)

   120     TEST_DATABASE_USER = _test_database_user(settings)

   121     TEST_DATABASE_PASSWD = _test_database_passwd(settings)

   122     TEST_DATABASE_TBLSPACE = _test_database_tblspace(settings)

   123     TEST_DATABASE_TBLSPACE_TMP = _test_database_tblspace_tmp(settings)

   124 

   125     settings.DATABASE_NAME = old_database_name

   126     settings.DATABASE_USER = REMEMBER['user']

   127     settings.DATABASE_PASSWORD = REMEMBER['passwd']

   128 

   129     parameters = {

   130         'dbname': TEST_DATABASE_NAME,

   131         'user': TEST_DATABASE_USER,

   132         'password': TEST_DATABASE_PASSWD,

   133         'tblspace': TEST_DATABASE_TBLSPACE,

   134         'tblspace_temp': TEST_DATABASE_TBLSPACE_TMP,

   135  	}

   136 

   137     REMEMBER['user'] = settings.DATABASE_USER

   138     REMEMBER['passwd'] = settings.DATABASE_PASSWORD

   139 

   140     cursor = connection.cursor()

   141     time.sleep(1) # To avoid "database is being accessed by other users" errors.

   142     if _test_user_create(settings):

   143         if verbosity >= 1:

   144             print 'Destroying test user...'

   145         _destroy_test_user(cursor, parameters, verbosity)

   146     if _test_database_create(settings):

   147         if verbosity >= 1:

   148             print 'Destroying test database...'

   149         _destroy_test_db(cursor, parameters, verbosity)

   150     connection.close()

   151 

   152 def _create_test_db(cursor, parameters, verbosity):

   153     if verbosity >= 2:

   154         print "_create_test_db(): dbname = %s" % parameters['dbname']

   155     statements = [

   156         """CREATE TABLESPACE %(tblspace)s

   157            DATAFILE '%(tblspace)s.dbf' SIZE 20M

   158            REUSE AUTOEXTEND ON NEXT 10M MAXSIZE 100M

   159         """,

   160         """CREATE TEMPORARY TABLESPACE %(tblspace_temp)s

   161            TEMPFILE '%(tblspace_temp)s.dbf' SIZE 20M

   162            REUSE AUTOEXTEND ON NEXT 10M MAXSIZE 100M

   163         """,

   164     ]

   165     _execute_statements(cursor, statements, parameters, verbosity)

   166 

   167 def _create_test_user(cursor, parameters, verbosity):

   168     if verbosity >= 2:

   169         print "_create_test_user(): username = %s" % parameters['user']

   170     statements = [

   171         """CREATE USER %(user)s

   172            IDENTIFIED BY %(password)s

   173            DEFAULT TABLESPACE %(tblspace)s

   174            TEMPORARY TABLESPACE %(tblspace_temp)s

   175         """,

   176         """GRANT CONNECT, RESOURCE TO %(user)s""",

   177     ]

   178     _execute_statements(cursor, statements, parameters, verbosity)

   179 

   180 def _destroy_test_db(cursor, parameters, verbosity):

   181     if verbosity >= 2:

   182         print "_destroy_test_db(): dbname=%s" % parameters['dbname']

   183     statements = [

   184         'DROP TABLESPACE %(tblspace)s INCLUDING CONTENTS AND DATAFILES CASCADE CONSTRAINTS',

   185         'DROP TABLESPACE %(tblspace_temp)s INCLUDING CONTENTS AND DATAFILES CASCADE CONSTRAINTS',

   186         ]

   187     _execute_statements(cursor, statements, parameters, verbosity)

   188 

   189 def _destroy_test_user(cursor, parameters, verbosity):

   190     if verbosity >= 2:

   191         print "_destroy_test_user(): user=%s" % parameters['user']

   192         print "Be patient.  This can take some time..."

   193     statements = [

   194         'DROP USER %(user)s CASCADE',

   195     ]

   196     _execute_statements(cursor, statements, parameters, verbosity)

   197 

   198 def _execute_statements(cursor, statements, parameters, verbosity):

   199     for template in statements:

   200         stmt = template % parameters

   201         if verbosity >= 2:

   202             print stmt

   203         try:

   204             cursor.execute(stmt)

   205         except Exception, err:

   206             sys.stderr.write("Failed (%s)\n" % (err))

   207             raise

   208 

   209 def _test_database_name(settings):

   210     name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME

   211     try:

   212         if settings.TEST_DATABASE_NAME:

   213             name = settings.TEST_DATABASE_NAME

   214     except AttributeError:

   215         pass

   216     except:

   217         raise

   218     return name

   219 

   220 def _test_database_create(settings):

   221     name = True

   222     try:

   223         if settings.TEST_DATABASE_CREATE:

   224             name = True

   225         else:

   226             name = False

   227     except AttributeError:

   228         pass

   229     except:

   230         raise

   231     return name

   232 

   233 def _test_user_create(settings):

   234     name = True

   235     try:

   236         if settings.TEST_USER_CREATE:

   237             name = True

   238         else:

   239             name = False

   240     except AttributeError:

   241         pass

   242     except:

   243         raise

   244     return name

   245 

   246 def _test_database_user(settings):

   247     name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME

   248     try:

   249         if settings.TEST_DATABASE_USER:

   250             name = settings.TEST_DATABASE_USER

   251     except AttributeError:

   252         pass

   253     except:

   254         raise

   255     return name

   256 

   257 def _test_database_passwd(settings):

   258     name = PASSWORD

   259     try:

   260         if settings.TEST_DATABASE_PASSWD:

   261             name = settings.TEST_DATABASE_PASSWD

   262     except AttributeError:

   263         pass

   264     except:

   265         raise

   266     return name

   267 

   268 def _test_database_tblspace(settings):

   269     name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME

   270     try:

   271         if settings.TEST_DATABASE_TBLSPACE:

   272             name = settings.TEST_DATABASE_TBLSPACE

   273     except AttributeError:

   274         pass

   275     except:

   276         raise

   277     return name

   278 

   279 def _test_database_tblspace_tmp(settings):

   280     name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME + '_temp'

   281     try:

   282         if settings.TEST_DATABASE_TBLSPACE_TMP:

   283             name = settings.TEST_DATABASE_TBLSPACE_TMP

   284     except AttributeError:

   285         pass

   286     except:

   287         raise

   288     return name