1 import sys, time

     2 from django.conf import settings

     3 from django.core import management

     4 from django.db.backends.creation import BaseDatabaseCreation

     5 

     6 TEST_DATABASE_PREFIX = 'test_'

     7 PASSWORD = 'Im_a_lumberjack'

     8 

     9 class DatabaseCreation(BaseDatabaseCreation):

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

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

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

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

    14     #

    15     # Any format strings starting with "qn_" are quoted before being used in the

    16     # output (the "qn_" prefix is stripped before the lookup is performed.

    17 

    18     data_types = {

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

    20         'BooleanField':                 'NUMBER(1) CHECK (%(qn_column)s IN (0,1))',

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

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

    23         'DateField':                    'DATE',

    24         'DateTimeField':                'TIMESTAMP',

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

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

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

    28         'FloatField':                   'DOUBLE PRECISION',

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

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

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

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

    33         'PositiveIntegerField':         'NUMBER(11) CHECK (%(qn_column)s >= 0)',

    34         'PositiveSmallIntegerField':    'NUMBER(11) CHECK (%(qn_column)s >= 0)',

    35         'SlugField':                    'NVARCHAR2(%(max_length)s)',

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

    37         'TextField':                    'NCLOB',

    38         'TimeField':                    'TIMESTAMP',

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

    40     }

    41 

    42     remember = {}

    43 

    44     def _create_test_db(self, verbosity=1, autoclobber=False):

    45         TEST_DATABASE_NAME = self._test_database_name(settings)

    46         TEST_DATABASE_USER = self._test_database_user(settings)

    47         TEST_DATABASE_PASSWD = self._test_database_passwd(settings)

    48         TEST_DATABASE_TBLSPACE = self._test_database_tblspace(settings)

    49         TEST_DATABASE_TBLSPACE_TMP = self._test_database_tblspace_tmp(settings)

    50 

    51         parameters = {

    52             'dbname': TEST_DATABASE_NAME,

    53             'user': TEST_DATABASE_USER,

    54             'password': TEST_DATABASE_PASSWD,

    55             'tblspace': TEST_DATABASE_TBLSPACE,

    56             'tblspace_temp': TEST_DATABASE_TBLSPACE_TMP,

    57         }

    58 

    59         self.remember['user'] = settings.DATABASE_USER

    60         self.remember['passwd'] = settings.DATABASE_PASSWORD

    61 

    62         cursor = self.connection.cursor()

    63         if self._test_database_create(settings):

    64             if verbosity >= 1:

    65                 print 'Creating test database...'

    66             try:

    67                 self._execute_test_db_creation(cursor, parameters, verbosity)

    68             except Exception, e:

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

    70                 if not autoclobber:

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

    72                 if autoclobber or confirm == 'yes':

    73                     try:

    74                         if verbosity >= 1:

    75                             print "Destroying old test database..."

    76                         self._execute_test_db_destruction(cursor, parameters, verbosity)

    77                         if verbosity >= 1:

    78                             print "Creating test database..."

    79                         self._execute_test_db_creation(cursor, parameters, verbosity)

    80                     except Exception, e:

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

    82                         sys.exit(2)

    83                 else:

    84                     print "Tests cancelled."

    85                     sys.exit(1)

    86 

    87         if self._test_user_create(settings):

    88             if verbosity >= 1:

    89                 print "Creating test user..."

    90             try:

    91                 self._create_test_user(cursor, parameters, verbosity)

    92             except Exception, e:

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

    94                 if not autoclobber:

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

    96                 if autoclobber or confirm == 'yes':

    97                     try:

    98                         if verbosity >= 1:

    99                             print "Destroying old test user..."

   100                         self._destroy_test_user(cursor, parameters, verbosity)

   101                         if verbosity >= 1:

   102                             print "Creating test user..."

   103                         self._create_test_user(cursor, parameters, verbosity)

   104                     except Exception, e:

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

   106                         sys.exit(2)

   107                 else:

   108                     print "Tests cancelled."

   109                     sys.exit(1)

   110 

   111         settings.TEST_DATABASE_USER = settings.DATABASE_USER = self.connection.settings_dict["DATABASE_USER"] = TEST_DATABASE_USER

   112         settings.DATABASE_PASSWORD = self.connection.settings_dict["DATABASE_PASSWORD"] = TEST_DATABASE_PASSWD

   113 

   114         return settings.DATABASE_NAME

   115 

   116     def _destroy_test_db(self, test_database_name, verbosity=1):

   117         """

   118         Destroy a test database, prompting the user for confirmation if the

   119         database already exists. Returns the name of the test database created.

   120         """

   121         TEST_DATABASE_NAME = self._test_database_name(settings)

   122         TEST_DATABASE_USER = self._test_database_user(settings)

   123         TEST_DATABASE_PASSWD = self._test_database_passwd(settings)

   124         TEST_DATABASE_TBLSPACE = self._test_database_tblspace(settings)

   125         TEST_DATABASE_TBLSPACE_TMP = self._test_database_tblspace_tmp(settings)

   126 

   127         settings.DATABASE_USER = self.connection.settings_dict["DATABASE_USER"] = self.remember['user']

   128         settings.DATABASE_PASSWORD = self.connection.settings_dict["DATABASE_PASSWORD"] = self.remember['passwd']

   129 

   130         parameters = {

   131             'dbname': TEST_DATABASE_NAME,

   132             'user': TEST_DATABASE_USER,

   133             'password': TEST_DATABASE_PASSWD,

   134             'tblspace': TEST_DATABASE_TBLSPACE,

   135             'tblspace_temp': TEST_DATABASE_TBLSPACE_TMP,

   136         }

   137 

   138         self.remember['user'] = settings.DATABASE_USER

   139         self.remember['passwd'] = settings.DATABASE_PASSWORD

   140 

   141         cursor = self.connection.cursor()

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

   143         if self._test_user_create(settings):

   144             if verbosity >= 1:

   145                 print 'Destroying test user...'

   146             self._destroy_test_user(cursor, parameters, verbosity)

   147         if self._test_database_create(settings):

   148             if verbosity >= 1:

   149                 print 'Destroying test database tables...'

   150             self._execute_test_db_destruction(cursor, parameters, verbosity)

   151         self.connection.close()

   152 

   153     def _execute_test_db_creation(self, cursor, parameters, verbosity):

   154         if verbosity >= 2:

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

   156         statements = [

   157             """CREATE TABLESPACE %(tblspace)s

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

   159                REUSE AUTOEXTEND ON NEXT 10M MAXSIZE 100M

   160             """,

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

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

   163                REUSE AUTOEXTEND ON NEXT 10M MAXSIZE 100M

   164             """,

   165         ]

   166         self._execute_statements(cursor, statements, parameters, verbosity)

   167 

   168     def _create_test_user(self, cursor, parameters, verbosity):

   169         if verbosity >= 2:

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

   171         statements = [

   172             """CREATE USER %(user)s

   173                IDENTIFIED BY %(password)s

   174                DEFAULT TABLESPACE %(tblspace)s

   175                TEMPORARY TABLESPACE %(tblspace_temp)s

   176             """,

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

   178         ]

   179         self._execute_statements(cursor, statements, parameters, verbosity)

   180 

   181     def _execute_test_db_destruction(self, cursor, parameters, verbosity):

   182         if verbosity >= 2:

   183             print "_execute_test_db_destruction(): dbname=%s" % parameters['dbname']

   184         statements = [

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

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

   187             ]

   188         self._execute_statements(cursor, statements, parameters, verbosity)

   189 

   190     def _destroy_test_user(self, cursor, parameters, verbosity):

   191         if verbosity >= 2:

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

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

   194         statements = [

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

   196         ]

   197         self._execute_statements(cursor, statements, parameters, verbosity)

   198 

   199     def _execute_statements(self, cursor, statements, parameters, verbosity):

   200         for template in statements:

   201             stmt = template % parameters

   202             if verbosity >= 2:

   203                 print stmt

   204             try:

   205                 cursor.execute(stmt)

   206             except Exception, err:

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

   208                 raise

   209 

   210     def _test_database_name(self, settings):

   211         name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME

   212         try:

   213             if settings.TEST_DATABASE_NAME:

   214                 name = settings.TEST_DATABASE_NAME

   215         except AttributeError:

   216             pass

   217         except:

   218             raise

   219         return name

   220 

   221     def _test_database_create(self, settings):

   222         name = True

   223         try:

   224             if settings.TEST_DATABASE_CREATE:

   225                 name = True

   226             else:

   227                 name = False

   228         except AttributeError:

   229             pass

   230         except:

   231             raise

   232         return name

   233 

   234     def _test_user_create(self, settings):

   235         name = True

   236         try:

   237             if settings.TEST_USER_CREATE:

   238                 name = True

   239             else:

   240                 name = False

   241         except AttributeError:

   242             pass

   243         except:

   244             raise

   245         return name

   246 

   247     def _test_database_user(self, settings):

   248         name = TEST_DATABASE_PREFIX + settings.DATABASE_USER

   249         try:

   250             if settings.TEST_DATABASE_USER:

   251                 name = settings.TEST_DATABASE_USER

   252         except AttributeError:

   253             pass

   254         except:

   255             raise

   256         return name

   257 

   258     def _test_database_passwd(self, settings):

   259         name = PASSWORD

   260         try:

   261             if settings.TEST_DATABASE_PASSWD:

   262                 name = settings.TEST_DATABASE_PASSWD

   263         except AttributeError:

   264             pass

   265         except:

   266             raise

   267         return name

   268 

   269     def _test_database_tblspace(self, settings):

   270         name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME

   271         try:

   272             if settings.TEST_DATABASE_TBLSPACE:

   273                 name = settings.TEST_DATABASE_TBLSPACE

   274         except AttributeError:

   275             pass

   276         except:

   277             raise

   278         return name

   279 

   280     def _test_database_tblspace_tmp(self, settings):

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

   282         try:

   283             if settings.TEST_DATABASE_TBLSPACE_TMP:

   284                 name = settings.TEST_DATABASE_TBLSPACE_TMP

   285         except AttributeError:

   286             pass

   287         except:

   288             raise

   289         return name