Ticket #8138: django_transaction_tests_3.diff
File django_transaction_tests_3.diff, 14.8 KB (added by , 16 years ago) |
---|
-
src/django/test/client.py
18 18 from django.utils.encoding import smart_str 19 19 from django.utils.http import urlencode 20 20 from django.utils.itercompat import is_iterable 21 from django.db import transaction 21 22 22 23 BOUNDARY = 'BoUnDaRyStRiNg' 23 24 MULTIPART_CONTENT = 'multipart/form-data; boundary=%s' % BOUNDARY … … 60 61 61 62 signals.request_started.send(sender=self.__class__) 62 63 try: 64 transaction.enter_transaction_management() 65 transaction.managed(True) 63 66 request = WSGIRequest(environ) 64 67 response = self.get_response(request) 68 transaction.commit() 69 transaction.leave_transaction_management() 65 70 66 71 # Apply response middleware. 67 72 for middleware_method in self._response_middleware: … … 134 139 '', 135 140 file.read() 136 141 ] 137 142 138 143 class Client: 139 144 """ 140 145 A class that can act as a client for testing purposes. … … 170 175 Obtains the current session variables. 171 176 """ 172 177 if 'django.contrib.sessions' in settings.INSTALLED_APPS: 178 # If a session db change hangs in a transaction, commit, 179 # just to be sure. 180 if transaction.is_dirty(): 181 transaction.commit() 173 182 engine = __import__(settings.SESSION_ENGINE, {}, {}, ['']) 174 183 cookie = self.cookies.get(settings.SESSION_COOKIE_NAME, None) 175 184 if cookie: -
src/django/test/testcases.py
1 1 2 import re 2 3 import unittest 3 4 from urlparse import urlsplit, urlunsplit … … 8 9 from django.core.management import call_command 9 10 from django.core.urlresolvers import clear_url_caches 10 11 from django.db import transaction 12 from django.db.models.signals import post_save 11 13 from django.http import QueryDict 12 14 from django.test import _doctest as doctest 13 15 from django.test.client import Client … … 55 57 """Tries to do a 'xml-comparision' of want and got. Plain string 56 58 comparision doesn't always work because, for example, attribute 57 59 ordering should not be important. 58 60 59 61 Based on http://codespeak.net/svn/lxml/trunk/src/lxml/doctestcompare.py 60 62 """ 61 63 _norm_whitespace_re = re.compile(r'[ \t\n][ \t\n]+') … … 102 104 wrapper = '<root>%s</root>' 103 105 want = wrapper % want 104 106 got = wrapper % got 105 107 106 108 # Parse the want and got strings, and compare the parsings. 107 109 try: 108 110 want_root = parseString(want).firstChild … … 169 171 # side effects on other tests. 170 172 transaction.rollback_unless_managed() 171 173 172 class TestCase(unittest.TestCase): 174 def run(self, test, compileflags=None, out=None, clear_globs=True): 175 """ 176 Wraps the parent run() and encloses it in a transaction. 177 """ 178 transaction.enter_transaction_management() 179 transaction.managed(True) 180 result = doctest.DocTestRunner.run(self, test, compileflags, out, clear_globs) 181 transaction.rollback() 182 transaction.leave_transaction_management() 183 return result 184 185 class TransactionTestCase(unittest.TestCase): 173 186 def _pre_setup(self): 174 187 """Performs any pre-test setup. This includes: 175 188 176 189 * Flushing the database. 177 * If the Test Case class has a 'fixtures' member, installing the 190 * If the Test Case class has a 'fixtures' member, installing the 178 191 named fixtures. 179 192 * If the Test Case class has a 'urls' member, replace the 180 193 ROOT_URLCONF with it. 181 194 * Clearing the mail test outbox. 182 195 """ 196 self._fixture_setup() 197 self._urlconf_setup() 198 mail.outbox = [] 199 200 def _fixture_setup(self): 183 201 call_command('flush', verbosity=0, interactive=False) 184 202 if hasattr(self, 'fixtures'): 185 203 # We have to use this slightly awkward syntax due to the fact 186 204 # that we're using *args and **kwargs together. 187 205 call_command('loaddata', *self.fixtures, **{'verbosity': 0}) 206 207 def _urlconf_setup(self): 188 208 if hasattr(self, 'urls'): 189 209 self._old_root_urlconf = settings.ROOT_URLCONF 190 210 settings.ROOT_URLCONF = self.urls 191 211 clear_url_caches() 192 mail.outbox = []193 212 194 213 def __call__(self, result=None): 195 214 """ … … 206 225 import sys 207 226 result.addError(self, sys.exc_info()) 208 227 return 209 super(T estCase, self).__call__(result)228 super(TransactionTestCase, self).__call__(result) 210 229 try: 211 230 self._post_teardown() 212 231 except (KeyboardInterrupt, SystemExit): … … 221 240 222 241 * Putting back the original ROOT_URLCONF if it was changed. 223 242 """ 243 self._fixture_teardown() 244 self._urlconf_teardown() 245 246 def _fixture_teardown(self): 247 pass 248 249 def _urlconf_teardown(self): 224 250 if hasattr(self, '_old_root_urlconf'): 225 251 settings.ROOT_URLCONF = self._old_root_urlconf 226 252 clear_url_caches() … … 354 380 self.failIf(template_name in template_names, 355 381 (u"Template '%s' was used unexpectedly in rendering the" 356 382 u" response") % template_name) 383 384 class TestCase(TransactionTestCase): 385 """ 386 Does basically the same as TransactionTestCase, but surrounds every test 387 with a transaction. You have to use TransactionTestCase, if you need 388 transaction management inside a test. 389 """ 390 391 # has the db been changed during the test 392 db_was_changed = False 393 394 def _fixture_setup(self): 395 transaction.enter_transaction_management() 396 transaction.managed(True) 397 398 # whenever a save occured, the db must be dirty 399 post_save.connect(self._set_db_was_changed) 400 # this seems more elegant than patching ClientHandler 401 #request_started.connect(self._do_commit) 402 #request_finished.connect(self._do_commit) 403 404 if hasattr(self, 'fixtures'): 405 call_command('loaddata', *self.fixtures, **{ 406 'verbosity': 0, 407 'no_commit': True 408 }) 409 # TODO: find out, if loaddata does emit a post_save signal 410 self._set_db_was_changed() 411 412 def _fixture_teardown(self): 413 # If the transaction is not dirty, but the DB was changed, 414 # a commit must have happened, so flush instead of rollback. 415 # This currently doesn't catch the following case: 416 # Inside a test a commit happens and after that more data is changed. 417 if not transaction.is_dirty() and self.db_was_changed: 418 transaction.leave_transaction_management() 419 call_command('flush', verbosity=0, interactive=False) 420 else: 421 transaction.rollback() 422 transaction.leave_transaction_management() 423 424 def _set_db_was_changed(self, *args, **kwargs): 425 self.db_was_changed = True 426 427 def _do_commit(self, *args, **kwargs): 428 transaction.commit() -
src/django/test/__init__.py
3 3 """ 4 4 5 5 from django.test.client import Client 6 from django.test.testcases import TestCase 6 from django.test.testcases import TestCase, TransactionTestCase -
src/django/core/management/commands/loaddata.py
28 28 29 29 verbosity = int(options.get('verbosity', 1)) 30 30 show_traceback = options.get('traceback', False) 31 no_commit = options.get('no_commit', False) 31 32 32 33 # Keep a count of the installed objects and fixtures 33 34 fixture_count = 0 … … 44 45 45 46 # Start transaction management. All fixtures are installed in a 46 47 # single transaction to ensure that all references are resolved. 47 transaction.commit_unless_managed() 48 transaction.enter_transaction_management() 49 transaction.managed(True) 48 if not no_commit: 49 transaction.commit_unless_managed() 50 transaction.enter_transaction_management() 51 transaction.managed(True) 50 52 51 53 app_fixtures = [os.path.join(os.path.dirname(app.__file__), 'fixtures') for app in get_apps()] 52 54 for fixture_label in fixture_labels: … … 133 135 (format, fixture_name, humanize(fixture_dir)) 134 136 135 137 136 # If any of the fixtures we loaded contain 0 objects, assume that an 138 # If any of the fixtures we loaded contain 0 objects, assume that an 137 139 # error was encountered during fixture loading. 138 140 if 0 in objects_per_fixture: 139 141 sys.stderr.write( … … 142 144 transaction.rollback() 143 145 transaction.leave_transaction_management() 144 146 return 145 146 # If we found even one object in a fixture, we need to reset the 147 148 # If we found even one object in a fixture, we need to reset the 147 149 # database sequences. 148 150 if object_count > 0: 149 151 sequence_sql = connection.ops.sequence_reset_sql(self.style, models) … … 152 154 print "Resetting sequences" 153 155 for line in sequence_sql: 154 156 cursor.execute(line) 155 156 transaction.commit()157 transaction.leave_transaction_management()158 157 158 if not no_commit: 159 transaction.commit() 160 transaction.leave_transaction_management() 161 159 162 if object_count == 0: 160 163 if verbosity > 1: 161 164 print "No fixtures found." 162 165 else: 163 166 if verbosity > 0: 164 167 print "Installed %d object(s) from %d fixture(s)" % (object_count, fixture_count) 165 168 166 169 # Close the DB connection. This is required as a workaround for an 167 170 # edge case in MySQL: if the same connection is used to 168 171 # create tables, load data, and query, the query can return 169 172 # incorrect results. See Django #7572, MySQL #37735. 170 connection.close() 173 if not no_commit: 174 connection.close() -
src/tests/regressiontests/admin_views/tests.py
1 1 # coding: utf-8 2 2 3 from django.test import TestCase 3 from django.test import TestCase, TransactionTestCase 4 4 from django.contrib.auth.models import User, Permission 5 5 from django.contrib.contenttypes.models import ContentType 6 6 from django.contrib.admin.models import LogEntry … … 13 13 14 14 class AdminViewBasicTest(TestCase): 15 15 fixtures = ['admin-views-users.xml'] 16 16 17 17 def setUp(self): 18 18 self.client.login(username='super', password='secret') 19 19 20 20 def tearDown(self): 21 21 self.client.logout() 22 22 23 23 def testTrailingSlashRequired(self): 24 24 """ 25 25 If you leave off the trailing slash, app should redirect and add it. … … 28 28 self.assertRedirects(request, 29 29 '/test_admin/admin/admin_views/article/add/' 30 30 ) 31 31 32 32 def testBasicAddGet(self): 33 33 """ 34 34 A smoke test to ensure GET on the add_view works. 35 35 """ 36 36 response = self.client.get('/test_admin/admin/admin_views/section/add/') 37 37 self.failUnlessEqual(response.status_code, 200) 38 38 39 39 def testBasicEditGet(self): 40 40 """ 41 41 A smoke test to ensureGET on the change_view works. 42 42 """ 43 43 response = self.client.get('/test_admin/admin/admin_views/section/1/') 44 44 self.failUnlessEqual(response.status_code, 200) 45 45 46 46 def testBasicAddPost(self): 47 47 """ 48 48 A smoke test to ensure POST on add_view works. … … 55 55 } 56 56 response = self.client.post('/test_admin/admin/admin_views/section/add/', post_data) 57 57 self.failUnlessEqual(response.status_code, 302) # redirect somewhere 58 58 59 59 def testBasicEditPost(self): 60 60 """ 61 61 A smoke test to ensure POST on edit_view works. … … 96 96 ct = ContentType.objects.get_for_model(Model) 97 97 return Permission.objects.get(content_type=ct, codename=perm) 98 98 99 class AdminViewPermissionsTest(TestCase): 100 """Tests for Admin Views Permissions.""" 99 class AdminViewPermissionsTest(TransactionTestCase): 100 """ 101 Tests for Admin Views Permissions. 101 102 103 We need TransactionTestCase here, because some data is lodaed manually 104 via the ORM, not via fixtures and test.Client is used. 105 """ 106 102 107 fixtures = ['admin-views-users.xml'] 103 108 104 109 def setUp(self): … … 438 443 should_contain = """<a href="../../%s/">%s</a>""" % (quote(self.pk), escape(self.pk)) 439 444 self.assertContains(response, should_contain) 440 445 441 class SecureViewTest(T estCase):446 class SecureViewTest(TransactionTestCase): 442 447 fixtures = ['admin-views-users.xml'] 443 448 444 449 def setUp(self): … … 471 476 LOGIN_FORM_KEY: 1, 472 477 'username': 'joepublic', 473 478 'password': 'secret'} 474 479 475 480 def tearDown(self): 476 481 self.client.logout() 477 482 478 483 def test_secure_view_shows_login_if_not_logged_in(self): 479 484 "Ensure that we see the login form" 480 485 response = self.client.get('/test_admin/admin/secure-view/' ) 481 486 self.assertTemplateUsed(response, 'admin/login.html') 482 487 483 488 def test_secure_view_login_successfully_redirects_to_original_url(self): 484 489 request = self.client.get('/test_admin/admin/secure-view/') 485 490 self.failUnlessEqual(request.status_code, 200) 486 491 query_string = "the-answer=42" 487 492 login = self.client.post('/test_admin/admin/secure-view/', self.super_login, QUERY_STRING = query_string ) 488 493 self.assertRedirects(login, '/test_admin/admin/secure-view/?%s' % query_string) 489 494 490 495 def test_staff_member_required_decorator_works_as_per_admin_login(self): 491 496 """ 492 497 Make sure only staff members can log in. 493 498 494 499 Successful posts to the login page will redirect to the orignal url. 495 Unsuccessfull attempts will continue to render the login page with 500 Unsuccessfull attempts will continue to render the login page with 496 501 a 200 status code. 497 502 """ 498 503 # Super User