Ticket #8138: django_transaction_tests_3.2.diff

django/test/client.py

@@ -18 +18 @@
 from django.utils.encoding import smart_str
 from django.utils.http import urlencode
 from django.utils.itercompat import is_iterable
+from django.db import transaction
 
 BOUNDARY = 'BoUnDaRyStRiNg'
 MULTIPART_CONTENT = 'multipart/form-data; boundary=%s' % BOUNDARY
@@ -60 +61 @@
 
         signals.request_started.send(sender=self.__class__)
         try:
+            transaction.enter_transaction_management()
+            transaction.managed(True)
             request = WSGIRequest(environ)
             response = self.get_response(request)
+            transaction.commit()
+            transaction.leave_transaction_management()
 
             # Apply response middleware.
             for middleware_method in self._response_middleware:
@@ -134 +139 @@
         '',
         file.read()
     ]
-    
+
 class Client:
     """
     A class that can act as a client for testing purposes.
@@ -170 +175 @@
         Obtains the current session variables.
         """
         if 'django.contrib.sessions' in settings.INSTALLED_APPS:
+            # If a session db change hangs in a transaction, commit,
+            # just to be sure.
+            if transaction.is_dirty():
+                transaction.commit()
             engine = __import__(settings.SESSION_ENGINE, {}, {}, [''])
             cookie = self.cookies.get(settings.SESSION_COOKIE_NAME, None)
             if cookie:

django/test/testcases.py

@@ +1 @@
+
 import re
 import unittest
 from urlparse import urlsplit, urlunsplit
@@ -8 +9 @@
 from django.core.management import call_command
 from django.core.urlresolvers import clear_url_caches
 from django.db import transaction
+from django.db.models.signals import post_save
 from django.http import QueryDict
 from django.test import _doctest as doctest
 from django.test.client import Client
@@ -55 +57 @@
         """Tries to do a 'xml-comparision' of want and got.  Plain string
         comparision doesn't always work because, for example, attribute
         ordering should not be important.
-        
+
         Based on http://codespeak.net/svn/lxml/trunk/src/lxml/doctestcompare.py
         """
         _norm_whitespace_re = re.compile(r'[ \t\n][ \t\n]+')
@@ -102 +104 @@
             wrapper = '<root>%s</root>'
             want = wrapper % want
             got = wrapper % got
-            
+
         # Parse the want and got strings, and compare the parsings.
         try:
             want_root = parseString(want).firstChild
@@ -169 +171 @@
         # side effects on other tests.
         transaction.rollback_unless_managed()
 
-class TestCase(unittest.TestCase):
+    def run(self, test, compileflags=None, out=None, clear_globs=True):
+        """
+        Wraps the parent run() and encloses it in a transaction.
+        """
+        transaction.enter_transaction_management()
+        transaction.managed(True)
+        result = doctest.DocTestRunner.run(self, test, compileflags, out, clear_globs)
+        transaction.rollback()
+        transaction.leave_transaction_management()
+        return result
+
+class TransactionTestCase(unittest.TestCase):
     def _pre_setup(self):
         """Performs any pre-test setup. This includes:
 
             * Flushing the database.
-            * If the Test Case class has a 'fixtures' member, installing the 
+            * If the Test Case class has a 'fixtures' member, installing the
               named fixtures.
             * If the Test Case class has a 'urls' member, replace the
               ROOT_URLCONF with it.
             * Clearing the mail test outbox.
         """
+        self._fixture_setup()
+        self._urlconf_setup()
+        mail.outbox = []
+
+    def _fixture_setup(self):
         call_command('flush', verbosity=0, interactive=False)
         if hasattr(self, 'fixtures'):
             # We have to use this slightly awkward syntax due to the fact
             # that we're using *args and **kwargs together.
             call_command('loaddata', *self.fixtures, **{'verbosity': 0})
+
+    def _urlconf_setup(self):
         if hasattr(self, 'urls'):
             self._old_root_urlconf = settings.ROOT_URLCONF
             settings.ROOT_URLCONF = self.urls
             clear_url_caches()
-        mail.outbox = []
 
     def __call__(self, result=None):
         """
@@ -206 +225 @@
             import sys
             result.addError(self, sys.exc_info())
             return
-        super(TestCase, self).__call__(result)
+        super(TransactionTestCase, self).__call__(result)
         try:
             self._post_teardown()
         except (KeyboardInterrupt, SystemExit):
@@ -221 +240 @@
 
             * Putting back the original ROOT_URLCONF if it was changed.
         """
+        self._fixture_teardown()
+        self._urlconf_teardown()
+
+    def _fixture_teardown(self):
+        pass
+
+    def _urlconf_teardown(self):
         if hasattr(self, '_old_root_urlconf'):
             settings.ROOT_URLCONF = self._old_root_urlconf
             clear_url_caches()
@@ -354 +380 @@
         self.failIf(template_name in template_names,
             (u"Template '%s' was used unexpectedly in rendering the"
              u" response") % template_name)
+
+class TestCase(TransactionTestCase):
+    """
+    Does basically the same as TransactionTestCase, but surrounds every test
+    with a transaction. You have to use TransactionTestCase, if you need
+    transaction management inside a test.
+    """
+
+    # has the db been changed during the test
+    db_was_changed = False
+
+    def _fixture_setup(self):
+        transaction.enter_transaction_management()
+        transaction.managed(True)
+
+        # whenever a save occured, the db must be dirty
+        post_save.connect(self._set_db_was_changed)
+        # this seems more elegant than patching ClientHandler
+        #request_started.connect(self._do_commit)
+        #request_finished.connect(self._do_commit)
+
+        if hasattr(self, 'fixtures'):
+            call_command('loaddata', *self.fixtures, **{
+                                                        'verbosity': 0,
+                                                        'no_commit': True
+                                                        })
+            # TODO: find out, if loaddata does emit a post_save signal
+            self._set_db_was_changed()
+
+    def _fixture_teardown(self):
+        # If the transaction is not dirty, but the DB was changed,
+        # a commit must have happened, so flush instead of rollback.
+        # This currently doesn't catch the following case:
+        # Inside a test a commit happens and after that more data is changed.
+        if not transaction.is_dirty() and self.db_was_changed:
+            transaction.leave_transaction_management()
+            call_command('flush', verbosity=0, interactive=False)
+        else:
+            transaction.rollback()
+            transaction.leave_transaction_management()
+
+    def _set_db_was_changed(self, *args, **kwargs):
+        self.db_was_changed = True
+
+    def _do_commit(self, *args, **kwargs):
+        transaction.commit()

django/test/__init__.py

@@ -3 +3 @@
 """
 
 from django.test.client import Client
-from django.test.testcases import TestCase
+from django.test.testcases import TestCase, TransactionTestCase

django/core/management/commands/loaddata.py

@@ -28 +28 @@
 
         verbosity = int(options.get('verbosity', 1))
         show_traceback = options.get('traceback', False)
+        no_commit = options.get('no_commit', False)
 
         # Keep a count of the installed objects and fixtures
         fixture_count = 0
@@ -41 +42 @@
         # the side effect of initializing the test database (if
         # it isn't already initialized).
         cursor = connection.cursor()
-
+        print dir(connection)
         # Start transaction management. All fixtures are installed in a
         # single transaction to ensure that all references are resolved.
-        transaction.commit_unless_managed()
-        transaction.enter_transaction_management()
-        transaction.managed(True)
+        if not no_commit:
+            transaction.commit_unless_managed()
+            transaction.enter_transaction_management()
+            transaction.managed(True)
 
         app_fixtures = [os.path.join(os.path.dirname(app.__file__), 'fixtures') for app in get_apps()]
         for fixture_label in fixture_labels:
@@ -133 +135 @@
                                 (format, fixture_name, humanize(fixture_dir))
 
 
-        # If any of the fixtures we loaded contain 0 objects, assume that an 
+        # If any of the fixtures we loaded contain 0 objects, assume that an
         # error was encountered during fixture loading.
         if 0 in objects_per_fixture:
             sys.stderr.write(
@@ -142 +144 @@
             transaction.rollback()
             transaction.leave_transaction_management()
             return
-            
-        # If we found even one object in a fixture, we need to reset the 
+
+        # If we found even one object in a fixture, we need to reset the
         # database sequences.
         if object_count > 0:
             sequence_sql = connection.ops.sequence_reset_sql(self.style, models)
@@ -152 +154 @@
                     print "Resetting sequences"
                 for line in sequence_sql:
                     cursor.execute(line)
-            
-        transaction.commit()
-        transaction.leave_transaction_management()
 
+        if not no_commit:
+            transaction.commit()
+            transaction.leave_transaction_management()
+
         if object_count == 0:
             if verbosity > 1:
                 print "No fixtures found."
         else:
             if verbosity > 0:
                 print "Installed %d object(s) from %d fixture(s)" % (object_count, fixture_count)
-                
+
         # Close the DB connection. This is required as a workaround for an
         # edge case in MySQL: if the same connection is used to
         # create tables, load data, and query, the query can return
         # incorrect results. See Django #7572, MySQL #37735.
-        connection.close()
+        if not no_commit:
+            connection.close()

tests/regressiontests/admin_views/tests.py

@@ -1 +1 @@
 # coding: utf-8
 
-from django.test import TestCase
+from django.test import TestCase, TransactionTestCase
 from django.contrib.auth.models import User, Permission
 from django.contrib.contenttypes.models import ContentType
 from django.contrib.admin.models import LogEntry
@@ -13 +13 @@
 
 class AdminViewBasicTest(TestCase):
     fixtures = ['admin-views-users.xml']
-    
+
     def setUp(self):
         self.client.login(username='super', password='secret')
-    
+
     def tearDown(self):
         self.client.logout()
-    
+
     def testTrailingSlashRequired(self):
         """
         If you leave off the trailing slash, app should redirect and add it.
@@ -28 +28 @@
         self.assertRedirects(request,
             '/test_admin/admin/admin_views/article/add/'
         )
-    
+
     def testBasicAddGet(self):
         """
         A smoke test to ensure GET on the add_view works.
         """
         response = self.client.get('/test_admin/admin/admin_views/section/add/')
         self.failUnlessEqual(response.status_code, 200)
-    
+
     def testBasicEditGet(self):
         """
         A smoke test to ensureGET on the change_view works.
         """
         response = self.client.get('/test_admin/admin/admin_views/section/1/')
         self.failUnlessEqual(response.status_code, 200)
-    
+
     def testBasicAddPost(self):
         """
         A smoke test to ensure POST on add_view works.
@@ -55 +55 @@
         }
         response = self.client.post('/test_admin/admin/admin_views/section/add/', post_data)
         self.failUnlessEqual(response.status_code, 302) # redirect somewhere
-    
+
     def testBasicEditPost(self):
         """
         A smoke test to ensure POST on edit_view works.
@@ -96 +96 @@
     ct = ContentType.objects.get_for_model(Model)
     return Permission.objects.get(content_type=ct, codename=perm)
 
-class AdminViewPermissionsTest(TestCase):
-    """Tests for Admin Views Permissions."""
+class AdminViewPermissionsTest(TransactionTestCase):
+    """
+    Tests for Admin Views Permissions.
 
+    We need TransactionTestCase here, because some data is lodaed manually
+    via the ORM, not via fixtures and test.Client is used.
+    """
+
     fixtures = ['admin-views-users.xml']
 
     def setUp(self):
@@ -438 +443 @@
         should_contain = """<a href="../../%s/">%s</a>""" % (quote(self.pk), escape(self.pk))
         self.assertContains(response, should_contain)
 
-class SecureViewTest(TestCase):
+class SecureViewTest(TransactionTestCase):
     fixtures = ['admin-views-users.xml']
 
     def setUp(self):
@@ -471 +476 @@
                      LOGIN_FORM_KEY: 1,
                      'username': 'joepublic',
                      'password': 'secret'}
-    
+
     def tearDown(self):
         self.client.logout()
-    
+
     def test_secure_view_shows_login_if_not_logged_in(self):
         "Ensure that we see the login form"
         response = self.client.get('/test_admin/admin/secure-view/' )
         self.assertTemplateUsed(response, 'admin/login.html')
-    
+
     def test_secure_view_login_successfully_redirects_to_original_url(self):
         request = self.client.get('/test_admin/admin/secure-view/')
         self.failUnlessEqual(request.status_code, 200)
         query_string = "the-answer=42"
         login = self.client.post('/test_admin/admin/secure-view/', self.super_login, QUERY_STRING = query_string )
         self.assertRedirects(login, '/test_admin/admin/secure-view/?%s' % query_string)
-    
+
     def test_staff_member_required_decorator_works_as_per_admin_login(self):
         """
         Make sure only staff members can log in.
 
         Successful posts to the login page will redirect to the orignal url.
-        Unsuccessfull attempts will continue to render the login page with 
+        Unsuccessfull attempts will continue to render the login page with
         a 200 status code.
         """
         # Super User