Adds tests for select_for_update.

cachalot/tests/read.py

@@ -8,9 +8,10 @@
     from unittest2 import skip
 
 from django.contrib.auth.models import Group, Permission, User
-from django.db import connection
+from django.db import connection, transaction
 from django.db.models import Count
-from django.test import TransactionTestCase
+from django.db.transaction import TransactionManagementError
+from django.test import TransactionTestCase, skipUnlessDBFeature
 
 from .models import Test
 
@@ -439,9 +440,24 @@ def test_prefetch_related(self):
     def test_using(self):
         pass
 
-    @skip(NotImplementedError)
+    @skipUnlessDBFeature('has_select_for_update')
     def test_select_for_update(self):
-        pass
+        with self.assertRaises(TransactionManagementError):
+            list(Test.objects.select_for_update())
+
+        with self.assertNumQueries(1):
+            with transaction.atomic():
+                data1 = list(Test.objects.select_for_update())
+                self.assertListEqual(data1, [self.t1, self.t2])
+                self.assertListEqual([t.name for t in data1],
+                                     ['test1', 'test2'])
+
+        with self.assertNumQueries(0):
+            with transaction.atomic():
+                data2 = list(Test.objects.select_for_update())
+                self.assertListEqual(data2, [self.t1, self.t2])
+                self.assertListEqual([t.name for t in data2],
+                                     ['test1', 'test2'])
 
     def test_extra_select(self):
         """

cachalot/tests/write.py

@@ -9,9 +9,9 @@
 from django import VERSION as django_version
 from django.contrib.auth.models import User, Permission, Group
 from django.core.exceptions import MultipleObjectsReturned
-from django.db import connection
+from django.db import connection, transaction
 from django.db.models import Count
-from django.test import TransactionTestCase
+from django.test import TransactionTestCase, skipUnlessDBFeature
 
 from .models import Test
 
@@ -481,6 +481,27 @@ def test_invalidate_prefetch_related(self):
                          .prefetch_related('owner__groups__permissions'))
             self.assertEqual(data7[0].owner.username, 'modified_user')
 
+    @skipUnlessDBFeature('has_select_for_update')
+    def test_invalidate_select_for_update(self):
+        with self.assertNumQueries(1):
+            Test.objects.bulk_create([Test(name='test1'), Test(name='test2')])
+
+        with self.assertNumQueries(1):
+            with transaction.atomic():
+                data1 = list(Test.objects.select_for_update())
+                self.assertListEqual([t.name for t in data1],
+                                     ['test1', 'test2'])
+
+        with self.assertNumQueries(1):
+            with transaction.atomic():
+                qs = Test.objects.select_for_update()
+                qs.update(name='test3')
+
+        with self.assertNumQueries(1):
+            with transaction.atomic():
+                data2 = list(Test.objects.select_for_update())
+                self.assertListEqual([t.name for t in data2], ['test3'] * 2)
+
     @skip(NotImplementedError)
     def test_invalidate_extra_select(self):
         pass