Ticket #11665: better_constraint_checks_during_testing.v5.diff

django/db/__init__.py

@@ -12 +12 @@
     raise ImproperlyConfigured("You must define a '%s' database" % DEFAULT_DB_ALIAS)
 
 connections = ConnectionHandler(settings.DATABASES)
+connections._ignore_num_queries = False
 
 router = ConnectionRouter(settings.DATABASE_ROUTERS)
 

django/db/backends/__init__.py

@@ -34 +34 @@
         self.transaction_state = []
         self.savepoint_state = 0
         self._dirty = None
+        
+        self.constraint_checking_disabled = False
 
     def __eq__(self, other):
         return self.alias == other.alias
@@ -238 +240 @@
         """
         if self.savepoint_state:
             self._savepoint_commit(sid)
-
+    
     @contextmanager
     def constraint_checks_disabled(self):
-        disabled = self.disable_constraint_checking()
+        self.disable_constraint_checking()
         try:
             yield
         finally:
-            if disabled:
+            if self.constraint_checking_disabled:
                 self.enable_constraint_checking()
-
+        
+    
     def disable_constraint_checking(self):
         """
         Backends can implement as needed to temporarily disable foreign key constraint
         checking.
         """
-        pass
+        self.constraint_checking_disabled = True
 
     def enable_constraint_checking(self):
         """
         Backends can implement as needed to re-enable foreign key constraint checking.
         """
-        pass
-
+        self.constraint_checking_disabled = False
+    
     def check_constraints(self, table_names=None):
         """
         Backends can override this method if they can apply constraint checking (e.g. via "SET CONSTRAINTS

django/db/backends/mysql/base.py

@@ -356 +356 @@
         to indicate constraint checks need to be re-enabled.
         """
         self.cursor().execute('SET foreign_key_checks=0')
-        return True
+        self.constraint_checking_disabled = True
 
     def enable_constraint_checking(self):
         """
         Re-enable foreign key checks after they have been disabled.
         """
         self.cursor().execute('SET foreign_key_checks=1')
-
+        self.constraint_checking_disabled = False
+    
     def check_constraints(self, table_names=None):
         """
         Checks each table name in `table_names` for rows with invalid foreign key references. This method is
@@ -389 +390 @@
                     SELECT REFERRING.`%s`, REFERRING.`%s` FROM `%s` as REFERRING
                     LEFT JOIN `%s` as REFERRED
                     ON (REFERRING.`%s` = REFERRED.`%s`)
-                    WHERE REFERRING.`%s` IS NOT NULL AND REFERRED.`%s` IS NULL"""
+                    WHERE REFERRING.`%s` IS NOT NULL
+                        AND REFERRED.`%s` IS NULL"""
                     % (primary_key_column_name, column_name, table_name, referenced_table_name,
-                    column_name, referenced_column_name, column_name, referenced_column_name))
+                       column_name, referenced_column_name, column_name, referenced_column_name))
                 for bad_row in cursor.fetchall():
-                    raise utils.IntegrityError("The row in table '%s' with primary key '%s' has an invalid "
-                        "foreign key: %s.%s contains a value '%s' that does not have a corresponding value in %s.%s."
-                        % (table_name, bad_row[0],
-                        table_name, column_name, bad_row[1],
-                        referenced_table_name, referenced_column_name))
+                    raise utils.IntegrityError("The row in table '%s' with primary key '%s' has an invalid \
+foreign key: %s.%s contains a value '%s' that does not have a corresponding value in %s.%s."
+                                         % (table_name, bad_row[0], table_name, column_name, bad_row[1],
+                                            referenced_table_name, referenced_column_name))

django/db/backends/postgresql_psycopg2/base.py

@@ -105 +105 @@
         self.introspection = DatabaseIntrospection(self)
         self.validation = BaseDatabaseValidation(self)
         self._pg_version = None
+    
+    def check_constraints(self, table_names=None):
+        """
+        To check constraints, we set constraints to immediate. Then, when, we're done we must ensure they
+        are returned to deferred.
+        """
+        self.cursor().execute('SET CONSTRAINTS ALL IMMEDIATE')
+        self.cursor().execute('SET CONSTRAINTS ALL DEFERRED')
 
     def check_constraints(self, table_names=None):
         """

django/db/backends/sqlite3/introspection.py

@@ -156 +156 @@
             name = info[0][2] # seqno, cid, name
             indexes[name]['unique'] = True
         return indexes
+    
+    def get_primary_key_column(self, cursor, table_name):
+        """
+        Get the column name of the primary key for the given table.
+        """
+        # Don't use PRAGMA because that causes issues with some transactions
+        cursor.execute("SELECT sql FROM sqlite_master WHERE tbl_name = %s AND type = %s", [table_name, "table"])
+        results = cursor.fetchone()[0].strip()
+        results = results[results.index('(')+1:results.rindex(')')]
+        for field_desc in results.split(','):
+            field_desc = field_desc.strip()
+            m = re.search('"(.*)".*PRIMARY KEY$', field_desc)
+            if m:
+                return m.groups()[0]
+        return None
 
     def get_primary_key_column(self, cursor, table_name):
         """

django/db/backends/util.py

@@ -29 +29 @@
 class CursorDebugWrapper(CursorWrapper):
 
     def execute(self, sql, params=()):
+        from django.db import connections
         start = time()
         try:
             return self.cursor.execute(sql, params)
@@ -36 +37 @@
             stop = time()
             duration = stop - start
             sql = self.db.ops.last_executed_query(self.cursor, sql, params)
-            self.db.queries.append({
-                'sql': sql,
-                'time': "%.3f" % duration,
-            })
+            if not connections._ignore_num_queries:
+                self.db.queries.append({
+                    'sql': sql,
+                    'time': "%.3f" % duration,
+                })
             logger.debug('(%.3f) %s; args=%s' % (duration, sql, params),
                 extra={'duration':duration, 'sql':sql, 'params':params}
             )
@@ -143 +145 @@
         context.prec = max_digits
         return u'%s' % str(value.quantize(decimal.Decimal(".1") ** decimal_places, context=context))
     else:
-        return u"%.*f" % (decimal_places, value)
+        return u"%.*f" % (decimal_places, value)
+ No newline at end of file

django/test/testcases.py

@@ -23 +23 @@
 __all__ = ('DocTestRunner', 'OutputChecker', 'TestCase', 'TransactionTestCase',
            'SimpleTestCase', 'skipIfDBFeature', 'skipUnlessDBFeature')
 
+
+def ignore_num_queries(fn):
+    @wraps(fn)
+    def num_queries_ignored(*args, **kwargs):
+        connections._ignore_num_queries = True
+        try:
+            return fn(*args, **kwargs)
+        finally:
+            connections._ignore_num_queries = False
+    return num_queries_ignored
+
 normalize_long_ints = lambda s: re.sub(r'(?<![\w])(\d+)L(?![\w])', '\\1', s)
 normalize_decimals = lambda s: re.sub(r"Decimal\('(\d+(\.\d*)?)'\)", lambda m: "Decimal(\"%s\")" % m.groups()[0], s)
 
@@ -39 +50 @@
 
 real_commit = transaction.commit
 real_rollback = transaction.rollback
+real_commit_unless_managed = transaction.commit_unless_managed
+real_rollback_unless_managed = transaction.rollback_unless_managed
 real_enter_transaction_management = transaction.enter_transaction_management
 real_leave_transaction_management = transaction.leave_transaction_management
 real_managed = transaction.managed
@@ -46 +59 @@
 def nop(*args, **kwargs):
     return
 
+@ignore_num_queries
+def check_constraints(using=None):
+    """
+    Emulate the constraint check behavior that normally occurs when a transaction is rolled back or committed.
+    """
+    if using is None:
+        using = DEFAULT_DB_ALIAS
+    connection = connections[using]
+    # Don't check constraints if they have been manually disabled
+    if not connection.constraint_checking_disabled:
+        connection.check_constraints()
+
 def disable_transaction_methods():
-    transaction.commit = nop
+    transaction.commit = check_constraints
     transaction.rollback = nop
+    transaction.commit_unless_managed = check_constraints
+    transaction.rollback_unless_managed = nop
     transaction.enter_transaction_management = nop
     transaction.leave_transaction_management = nop
     transaction.managed = nop
@@ -56 +83 @@
 def restore_transaction_methods():
     transaction.commit = real_commit
     transaction.rollback = real_rollback
+    transaction.commit_unless_managed = real_commit_unless_managed
+    transaction.rollback_unless_managed = real_rollback_unless_managed
     transaction.enter_transaction_management = real_enter_transaction_management
     transaction.leave_transaction_management = real_leave_transaction_management
     transaction.managed = real_managed
@@ -593 +622 @@
 
         from django.contrib.sites.models import Site
         Site.objects.clear_cache()
+        
+        from django.contrib.contenttypes.models import ContentType
+        ContentType.objects.clear_cache()
 
         for db in databases:
             if hasattr(self, 'fixtures'):

tests/regressiontests/backends/tests.py

@@ -347 +347 @@
         except IntegrityError:
             return
         self.skipTest("This backend does not support integrity checks.")
-
+    
     def test_disable_constraint_checks_manually(self):
         """
-        When constraint checks are disabled, should be able to write bad data without IntegrityErrors.
+        When constraint checks are disabled, should be able to write bad data without IntegrityErrors. Also,
+        should set disabled flag.
         """
         with transaction.commit_manually():
             # Create an Article.
@@ -360 +361 @@
             a.reporter_id = 30
             try:
                 connection.disable_constraint_checking()
+                self.assertTrue(connection.constraint_checking_disabled)
                 a.save()
                 connection.enable_constraint_checking()
             except IntegrityError:
                 self.fail("IntegrityError should not have occurred.")
             finally:
                 transaction.rollback()
-
+    
     def test_disable_constraint_checks_context_manager(self):
         """
         When constraint checks are disabled (using context manager), should be able to write bad data without IntegrityErrors.
@@ -379 +381 @@
             a.reporter_id = 30
             try:
                 with connection.constraint_checks_disabled():
+                    self.assertTrue(connection.constraint_checking_disabled)
                     a.save()
             except IntegrityError:
                 self.fail("IntegrityError should not have occurred.")
             finally:
                 transaction.rollback()
-
+    
     def test_check_constraints(self):
         """
         Constraint checks should raise an IntegrityError when bad data is in the DB.

tests/regressiontests/comment_tests/tests/templatetag_tests.py

@@ -42 +42 @@
     def testRenderCommentFormFromObjectWithQueryCount(self):
         def test():
             self.testRenderCommentFormFromObject()
-        self.assertNumQueries(1, test)
+        # 1 to select object
+        # 1 to get the contenttype
+        self.assertNumQueries(2, test)
 
     def testGetCommentCount(self, tag=None):
         self.createSomeComments()

tests/regressiontests/fixtures_regress/tests.py

@@ -361 +361 @@
             """[{"pk": %d, "model": "fixtures_regress.widget", "fields": {"name": "grommet"}}]"""
             % widget.pk
             )
+    
+    def test_loaddata_works_when_fixture_has_forward_refs(self):
+        """
+        Regression for #3615 - Forward references cause fixtures not to load in MySQL (InnoDB)
+        """
+        management.call_command(
+            'loaddata',
+            'forward_ref.json',
+            verbosity=0,
+            commit=False
+        )
+        self.assertEqual(Book.objects.all()[0].id, 1)
+        self.assertEqual(Person.objects.all()[0].id, 4)
+    
+    def test_loaddata_raises_error_when_fixture_has_invalid_foreign_key(self):
+        """
+        Regression for #3615 - Ensure data with nonexistent child key references raises error
+        """
+        stderr = StringIO()
+        management.call_command(
+            'loaddata',
+            'forward_ref_bad_data.json',
+            verbosity=0,
+            commit=False,
+            stderr=stderr,
+        )
+        self.assertTrue(
+            stderr.getvalue().startswith('Problem installing fixture')
+        )
 
     def test_loaddata_works_when_fixture_has_forward_refs(self):
         """

tests/regressiontests/multiple_database/tests.py

@@ -973 +973 @@
         # Make the 'other' database appear to be a slave of the 'default'
         self.old_routers = router.routers
         router.routers = [TestRouter()]
+        
+        # Disable constraint checking, for now
+        for connection_name in connections:
+            connection = connections[connection_name]
+            connection.disable_constraint_checking()
 
     def tearDown(self):
         # Restore the 'other' database as an independent database
         router.routers = self.old_routers
+        
+        # Re-enable constraint checking
+        for connection_name in connections:
+            connection = connections[connection_name]
+            connection.enable_constraint_checking()
 
     def test_db_selection(self):
         "Check that querysets obey the router for db suggestions"
@@ -1692 +1702 @@
 
     def setUp(self):
         self.old_routers = router.routers
+        
+        # Disable constraint checking, for now
+        for connection_name in connections:
+            connection = connections[connection_name]
+            connection.disable_constraint_checking()
 
     def tearDown(self):
         router.routers = self.old_routers
+        
+        # Re-enable constraint checking
+        for connection_name in connections:
+            connection = connections[connection_name]
+            connection.enable_constraint_checking()
 
     def _write_to_other(self):
         "Sends all writes to 'other'."

tests/regressiontests/test_utils/models.py

@@ -3 +3 @@
 
 class Person(models.Model):
     name = models.CharField(max_length=100)
+
+class Pet(models.Model):
+    name = models.CharField(max_length=100)
+    owner = models.ForeignKey(Person)
+
+    def __unicode__(self):
+        return self.name
+
+    class Meta:
+        ordering = ('name',)

tests/regressiontests/test_utils/tests.py

@@ -1 +1 @@
 from __future__ import with_statement
 
-from django.test import SimpleTestCase, TestCase, skipUnlessDBFeature
+from django.test import TestCase, SimpleTestCase, skipUnlessDBFeature
+from django.test.testcases import ignore_num_queries
 from django.utils.unittest import skip
+from django.db import IntegrityError
 
-from models import Person
+from models import Person, Pet
 
 
 class SkippingTestCase(TestCase):
@@ -47 +49 @@
             self.client.get("/test_utils/get_person/%s/" % person.pk)
             self.client.get("/test_utils/get_person/%s/" % person.pk)
         self.assertNumQueries(2, test_func)
+    
+    def test_assert_num_queries_ignore_decorator(self):
+        person = Person.objects.create(name='test')
+        
+        @ignore_num_queries
+        def test_func():
+            self.client.get("/test_utils/get_person/%s/" % person.pk)
+            self.client.get("/test_utils/get_person/%s/" % person.pk)
+        self.assertNumQueries(0, test_func)
 
 
 class AssertNumQueriesContextManagerTests(TestCase):
@@ -87 +98 @@
             self.client.get("/test_utils/get_person/%s/" % person.pk)
 
 
+class TransactionPatchingTests(TestCase):
+    def test_bad_data_should_raise_data_integrity_error(self):
+        """
+        Ensure bad data cannot be saved to DB during tests.
+        """
+        bill = Person.objects.create(name="Bill")
+        dog = Pet.objects.create(name="Spot", owner=bill)
+        dog.owner_id = 20 # Does not exist
+        with self.assertRaises(IntegrityError):
+            dog.save()
+
 class SaveRestoreWarningState(TestCase):
     def test_save_restore_warnings_state(self):
         """