25 'DecimalField': 'NUMBER(%(max_digits)s, %(decimal_places)s)', |
24 'DecimalField': 'NUMBER(%(max_digits)s, %(decimal_places)s)', |
26 'FileField': 'NVARCHAR2(%(max_length)s)', |
25 'FileField': 'NVARCHAR2(%(max_length)s)', |
27 'FilePathField': 'NVARCHAR2(%(max_length)s)', |
26 'FilePathField': 'NVARCHAR2(%(max_length)s)', |
28 'FloatField': 'DOUBLE PRECISION', |
27 'FloatField': 'DOUBLE PRECISION', |
29 'IntegerField': 'NUMBER(11)', |
28 'IntegerField': 'NUMBER(11)', |
|
29 'BigIntegerField': 'NUMBER(19)', |
30 'IPAddressField': 'VARCHAR2(15)', |
30 'IPAddressField': 'VARCHAR2(15)', |
31 'NullBooleanField': 'NUMBER(1) CHECK ((%(qn_column)s IN (0,1)) OR (%(qn_column)s IS NULL))', |
31 'NullBooleanField': 'NUMBER(1) CHECK ((%(qn_column)s IN (0,1)) OR (%(qn_column)s IS NULL))', |
32 'OneToOneField': 'NUMBER(11)', |
32 'OneToOneField': 'NUMBER(11)', |
33 'PositiveIntegerField': 'NUMBER(11) CHECK (%(qn_column)s >= 0)', |
33 'PositiveIntegerField': 'NUMBER(11) CHECK (%(qn_column)s >= 0)', |
34 'PositiveSmallIntegerField': 'NUMBER(11) CHECK (%(qn_column)s >= 0)', |
34 'PositiveSmallIntegerField': 'NUMBER(11) CHECK (%(qn_column)s >= 0)', |
40 } |
40 } |
41 |
41 |
42 remember = {} |
42 remember = {} |
43 |
43 |
44 def _create_test_db(self, verbosity=1, autoclobber=False): |
44 def _create_test_db(self, verbosity=1, autoclobber=False): |
45 TEST_DATABASE_NAME = self._test_database_name(settings) |
45 TEST_NAME = self._test_database_name() |
46 TEST_DATABASE_USER = self._test_database_user(settings) |
46 TEST_USER = self._test_database_user() |
47 TEST_DATABASE_PASSWD = self._test_database_passwd(settings) |
47 TEST_PASSWD = self._test_database_passwd() |
48 TEST_DATABASE_TBLSPACE = self._test_database_tblspace(settings) |
48 TEST_TBLSPACE = self._test_database_tblspace() |
49 TEST_DATABASE_TBLSPACE_TMP = self._test_database_tblspace_tmp(settings) |
49 TEST_TBLSPACE_TMP = self._test_database_tblspace_tmp() |
50 |
50 |
51 parameters = { |
51 parameters = { |
52 'dbname': TEST_DATABASE_NAME, |
52 'dbname': TEST_NAME, |
53 'user': TEST_DATABASE_USER, |
53 'user': TEST_USER, |
54 'password': TEST_DATABASE_PASSWD, |
54 'password': TEST_PASSWD, |
55 'tblspace': TEST_DATABASE_TBLSPACE, |
55 'tblspace': TEST_TBLSPACE, |
56 'tblspace_temp': TEST_DATABASE_TBLSPACE_TMP, |
56 'tblspace_temp': TEST_TBLSPACE_TMP, |
57 } |
57 } |
58 |
58 |
59 self.remember['user'] = settings.DATABASE_USER |
59 self.remember['user'] = self.connection.settings_dict['USER'] |
60 self.remember['passwd'] = settings.DATABASE_PASSWORD |
60 self.remember['passwd'] = self.connection.settings_dict['PASSWORD'] |
61 |
61 |
62 cursor = self.connection.cursor() |
62 cursor = self.connection.cursor() |
63 if self._test_database_create(settings): |
63 if self._test_database_create(): |
64 if verbosity >= 1: |
64 if verbosity >= 1: |
65 print 'Creating test database...' |
65 print 'Creating test database...' |
66 try: |
66 try: |
67 self._execute_test_db_creation(cursor, parameters, verbosity) |
67 self._execute_test_db_creation(cursor, parameters, verbosity) |
68 except Exception, e: |
68 except Exception, e: |
69 sys.stderr.write("Got an error creating the test database: %s\n" % e) |
69 sys.stderr.write("Got an error creating the test database: %s\n" % e) |
70 if not autoclobber: |
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) |
71 confirm = raw_input("It appears the test database, %s, already exists. Type 'yes' to delete it, or 'no' to cancel: " % TEST_NAME) |
72 if autoclobber or confirm == 'yes': |
72 if autoclobber or confirm == 'yes': |
73 try: |
73 try: |
74 if verbosity >= 1: |
74 if verbosity >= 1: |
75 print "Destroying old test database..." |
75 print "Destroying old test database..." |
76 self._execute_test_db_destruction(cursor, parameters, verbosity) |
76 self._execute_test_db_destruction(cursor, parameters, verbosity) |
82 sys.exit(2) |
82 sys.exit(2) |
83 else: |
83 else: |
84 print "Tests cancelled." |
84 print "Tests cancelled." |
85 sys.exit(1) |
85 sys.exit(1) |
86 |
86 |
87 if self._test_user_create(settings): |
87 if self._test_user_create(): |
88 if verbosity >= 1: |
88 if verbosity >= 1: |
89 print "Creating test user..." |
89 print "Creating test user..." |
90 try: |
90 try: |
91 self._create_test_user(cursor, parameters, verbosity) |
91 self._create_test_user(cursor, parameters, verbosity) |
92 except Exception, e: |
92 except Exception, e: |
93 sys.stderr.write("Got an error creating the test user: %s\n" % e) |
93 sys.stderr.write("Got an error creating the test user: %s\n" % e) |
94 if not autoclobber: |
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) |
95 confirm = raw_input("It appears the test user, %s, already exists. Type 'yes' to delete it, or 'no' to cancel: " % TEST_USER) |
96 if autoclobber or confirm == 'yes': |
96 if autoclobber or confirm == 'yes': |
97 try: |
97 try: |
98 if verbosity >= 1: |
98 if verbosity >= 1: |
99 print "Destroying old test user..." |
99 print "Destroying old test user..." |
100 self._destroy_test_user(cursor, parameters, verbosity) |
100 self._destroy_test_user(cursor, parameters, verbosity) |
106 sys.exit(2) |
106 sys.exit(2) |
107 else: |
107 else: |
108 print "Tests cancelled." |
108 print "Tests cancelled." |
109 sys.exit(1) |
109 sys.exit(1) |
110 |
110 |
111 settings.TEST_DATABASE_USER = settings.DATABASE_USER = self.connection.settings_dict["DATABASE_USER"] = TEST_DATABASE_USER |
111 self.connection.settings_dict['TEST_USER'] = self.connection.settings_dict["USER"] = TEST_USER |
112 settings.DATABASE_PASSWORD = self.connection.settings_dict["DATABASE_PASSWORD"] = TEST_DATABASE_PASSWD |
112 self.connection.settings_dict["PASSWORD"] = TEST_PASSWD |
113 |
113 |
114 return settings.DATABASE_NAME |
114 return self.connection.settings_dict['NAME'] |
115 |
115 |
116 def _destroy_test_db(self, test_database_name, verbosity=1): |
116 def _destroy_test_db(self, test_database_name, verbosity=1): |
117 """ |
117 """ |
118 Destroy a test database, prompting the user for confirmation if the |
118 Destroy a test database, prompting the user for confirmation if the |
119 database already exists. Returns the name of the test database created. |
119 database already exists. Returns the name of the test database created. |
120 """ |
120 """ |
121 TEST_DATABASE_NAME = self._test_database_name(settings) |
121 TEST_NAME = self._test_database_name() |
122 TEST_DATABASE_USER = self._test_database_user(settings) |
122 TEST_USER = self._test_database_user() |
123 TEST_DATABASE_PASSWD = self._test_database_passwd(settings) |
123 TEST_PASSWD = self._test_database_passwd() |
124 TEST_DATABASE_TBLSPACE = self._test_database_tblspace(settings) |
124 TEST_TBLSPACE = self._test_database_tblspace() |
125 TEST_DATABASE_TBLSPACE_TMP = self._test_database_tblspace_tmp(settings) |
125 TEST_TBLSPACE_TMP = self._test_database_tblspace_tmp() |
126 |
126 |
127 settings.DATABASE_USER = self.connection.settings_dict["DATABASE_USER"] = self.remember['user'] |
127 self.connection.settings_dict["USER"] = self.remember['user'] |
128 settings.DATABASE_PASSWORD = self.connection.settings_dict["DATABASE_PASSWORD"] = self.remember['passwd'] |
128 self.connection.settings_dict["PASSWORD"] = self.remember['passwd'] |
129 |
129 |
130 parameters = { |
130 parameters = { |
131 'dbname': TEST_DATABASE_NAME, |
131 'dbname': TEST_NAME, |
132 'user': TEST_DATABASE_USER, |
132 'user': TEST_USER, |
133 'password': TEST_DATABASE_PASSWD, |
133 'password': TEST_PASSWD, |
134 'tblspace': TEST_DATABASE_TBLSPACE, |
134 'tblspace': TEST_TBLSPACE, |
135 'tblspace_temp': TEST_DATABASE_TBLSPACE_TMP, |
135 'tblspace_temp': TEST_TBLSPACE_TMP, |
136 } |
136 } |
137 |
137 |
138 self.remember['user'] = settings.DATABASE_USER |
138 self.remember['user'] = self.connection.settings_dict['USER'] |
139 self.remember['passwd'] = settings.DATABASE_PASSWORD |
139 self.remember['passwd'] = self.connection.settings_dict['PASSWORD'] |
140 |
140 |
141 cursor = self.connection.cursor() |
141 cursor = self.connection.cursor() |
142 time.sleep(1) # To avoid "database is being accessed by other users" errors. |
142 time.sleep(1) # To avoid "database is being accessed by other users" errors. |
143 if self._test_user_create(settings): |
143 if self._test_user_create(): |
144 if verbosity >= 1: |
144 if verbosity >= 1: |
145 print 'Destroying test user...' |
145 print 'Destroying test user...' |
146 self._destroy_test_user(cursor, parameters, verbosity) |
146 self._destroy_test_user(cursor, parameters, verbosity) |
147 if self._test_database_create(settings): |
147 if self._test_database_create(): |
148 if verbosity >= 1: |
148 if verbosity >= 1: |
149 print 'Destroying test database tables...' |
149 print 'Destroying test database tables...' |
150 self._execute_test_db_destruction(cursor, parameters, verbosity) |
150 self._execute_test_db_destruction(cursor, parameters, verbosity) |
151 self.connection.close() |
151 self.connection.close() |
152 |
152 |
205 cursor.execute(stmt) |
205 cursor.execute(stmt) |
206 except Exception, err: |
206 except Exception, err: |
207 sys.stderr.write("Failed (%s)\n" % (err)) |
207 sys.stderr.write("Failed (%s)\n" % (err)) |
208 raise |
208 raise |
209 |
209 |
210 def _test_database_name(self, settings): |
210 def _test_database_name(self): |
211 name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME |
211 name = TEST_DATABASE_PREFIX + self.connection.settings_dict['NAME'] |
212 try: |
212 try: |
213 if settings.TEST_DATABASE_NAME: |
213 if self.connection.settings_dict['TEST_NAME']: |
214 name = settings.TEST_DATABASE_NAME |
214 name = self.connection.settings_dict['TEST_NAME'] |
215 except AttributeError: |
215 except AttributeError: |
216 pass |
216 pass |
217 except: |
217 except: |
218 raise |
218 raise |
219 return name |
219 return name |
220 |
220 |
221 def _test_database_create(self, settings): |
221 def _test_database_create(self): |
222 name = True |
222 name = True |
223 try: |
223 try: |
224 if settings.TEST_DATABASE_CREATE: |
224 if self.connection.settings_dict['TEST_CREATE']: |
225 name = True |
225 name = True |
226 else: |
226 else: |
227 name = False |
227 name = False |
228 except AttributeError: |
228 except KeyError: |
229 pass |
229 pass |
230 except: |
230 except: |
231 raise |
231 raise |
232 return name |
232 return name |
233 |
233 |
234 def _test_user_create(self, settings): |
234 def _test_user_create(self): |
235 name = True |
235 name = True |
236 try: |
236 try: |
237 if settings.TEST_USER_CREATE: |
237 if self.connection.settings_dict['TEST_USER_CREATE']: |
238 name = True |
238 name = True |
239 else: |
239 else: |
240 name = False |
240 name = False |
241 except AttributeError: |
241 except KeyError: |
242 pass |
242 pass |
243 except: |
243 except: |
244 raise |
244 raise |
245 return name |
245 return name |
246 |
246 |
247 def _test_database_user(self, settings): |
247 def _test_database_user(self): |
248 name = TEST_DATABASE_PREFIX + settings.DATABASE_USER |
248 name = TEST_DATABASE_PREFIX + self.connection.settings_dict['USER'] |
249 try: |
249 try: |
250 if settings.TEST_DATABASE_USER: |
250 if self.connection.settings_dict['TEST_USER']: |
251 name = settings.TEST_DATABASE_USER |
251 name = self.connection.settings_dict['TEST_USER'] |
252 except AttributeError: |
252 except KeyError: |
253 pass |
253 pass |
254 except: |
254 except: |
255 raise |
255 raise |
256 return name |
256 return name |
257 |
257 |
258 def _test_database_passwd(self, settings): |
258 def _test_database_passwd(self): |
259 name = PASSWORD |
259 name = PASSWORD |
260 try: |
260 try: |
261 if settings.TEST_DATABASE_PASSWD: |
261 if self.connection.settings_dict['TEST_PASSWD']: |
262 name = settings.TEST_DATABASE_PASSWD |
262 name = self.connection.settings_dict['TEST_PASSWD'] |
263 except AttributeError: |
263 except KeyError: |
264 pass |
264 pass |
265 except: |
265 except: |
266 raise |
266 raise |
267 return name |
267 return name |
268 |
268 |
269 def _test_database_tblspace(self, settings): |
269 def _test_database_tblspace(self): |
270 name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME |
270 name = TEST_DATABASE_PREFIX + self.connection.settings_dict['NAME'] |
271 try: |
271 try: |
272 if settings.TEST_DATABASE_TBLSPACE: |
272 if self.connection.settings_dict['TEST_TBLSPACE']: |
273 name = settings.TEST_DATABASE_TBLSPACE |
273 name = self.connection.settings_dict['TEST_TBLSPACE'] |
274 except AttributeError: |
274 except KeyError: |
275 pass |
275 pass |
276 except: |
276 except: |
277 raise |
277 raise |
278 return name |
278 return name |
279 |
279 |
280 def _test_database_tblspace_tmp(self, settings): |
280 def _test_database_tblspace_tmp(self): |
281 name = TEST_DATABASE_PREFIX + settings.DATABASE_NAME + '_temp' |
281 name = TEST_DATABASE_PREFIX + self.connection.settings_dict['NAME'] + '_temp' |
282 try: |
282 try: |
283 if settings.TEST_DATABASE_TBLSPACE_TMP: |
283 if self.connection.settings_dict['TEST_TBLSPACE_TMP']: |
284 name = settings.TEST_DATABASE_TBLSPACE_TMP |
284 name = self.connection.settings_dict['TEST_TBLSPACE_TMP'] |
285 except AttributeError: |
285 except KeyError: |
286 pass |
286 pass |
287 except: |
287 except: |
288 raise |
288 raise |
289 return name |
289 return name |