Ticket #9479: 9479-r10858.diff

new file tests/regressiontests/delete_regress/__init__.py

@@ +1 @@
+

new file tests/regressiontests/delete_regress/models.py

@@ +1 @@
+from django.conf import settings
+from django.db import models, backend, connection, transaction
+from django.test import TransactionTestCase
+
+class Child(models.Model):
+    data = models.IntegerField()
+
+class Parent(models.Model):
+    ref = models.ForeignKey(Child)
+
+if settings.DATABASE_ENGINE != 'sqlite3':
+    class DeleteLockingTest(TransactionTestCase):
+        def setUp(self):
+            # Create a second connection to the database
+            self.conn2 = backend.DatabaseWrapper({
+                'DATABASE_HOST': settings.DATABASE_HOST,
+                'DATABASE_NAME': settings.DATABASE_NAME,
+                'DATABASE_OPTIONS': settings.DATABASE_OPTIONS,
+                'DATABASE_PASSWORD': settings.DATABASE_PASSWORD,
+                'DATABASE_PORT': settings.DATABASE_PORT,
+                'DATABASE_USER': settings.DATABASE_USER,
+                'TIME_ZONE': settings.TIME_ZONE,
+            })
+
+            # Put both DB connections into managed transaction mode
+            transaction.enter_transaction_management(True)
+            self.conn2._enter_transaction_management(True)
+
+        def tearDown(self):
+            # Close down the second connection.
+            self.conn2.close()
+
+        def test_concurrent_delete(self):
+            "Deletes on concurrent transactions don't collide and lock the database"
+
+            # Create some dummy data
+            c1 = Child(id=1, data=1)
+            c2 = Child(id=2, data=2)
+            c1.save()
+            c2.save()
+
+            p1 = Parent(id=1)
+            p2 = Parent(id=2)
+
+            p1.ref = c1
+            p2.ref = c2
+
+            p1.save()
+            p2.save()
+            transaction.commit()
+
+            self.assertEquals(2, Child.objects.count())
+            self.assertEquals(2, Parent.objects.count())
+
+            # Now with connection 1 committed, delete something using connection 2
+            # ... but don't commit the deletion.
+
+            cursor2 = self.conn2.cursor()
+
+            cursor2.execute('DELETE from delete_regress_parent WHERE id=1')
+            cursor2.execute('DELETE from delete_regress_child WHERE id=1')
+
+            # NOTE! If you uncomment this line, the test will pass.
+            # However, the test _should_ pass without the need for this commit
+            # self.conn2._commit()
+
+            # Back on connection 1, try to delete some objects that will result
+            # in the bulk deletion collect operation finding the same objects
+            # that have just been deleted in the second connection.
+
+            # This command will not return lock
+            Child.objects.filter(pk=1).delete()
+
+            self.conn2._commit()
+            transaction.commit()
+
+            self.assertEquals(1, Child.objects.count())
+            self.assertEquals(1, Parent.objects.count())
+ No newline at end of file