Correct problems with tests to get django-cachalot to work with 4.2

CHANGELOG.rst

@@ -4,14 +4,20 @@ What’s new in django-cachalot?
 2.6.0
 -----
 
-- Dropped Django 2.2 and 4.0 support. Added Django 4.2 and Python 3.11 support. Added psycopg support (#229)
+- Dropped Django 2.2 and 4.0 support
+- Added Django 4.2 and Python 3.11 support
+- Added psycopg support (#229)
+- Updated tests to account for the `BEGIN` and `COMMIT` query changes in Django 4.2
+- Standardized django version comparisons in tests
 
 2.5.3
 -----
+
 - Verify get_meta isn't none before requesting db_table (#225 #226)
 
 2.5.2
 -----
+
 - Added Django 4.1 support (#217)
 
 2.5.1

cachalot/tests/models.py

@@ -58,7 +58,7 @@ class PostgresModel(Model):
                            null=True, blank=True)
 
     hstore = HStoreField(null=True, blank=True)
-    if DJANGO_VERSION[0] < 4:
+    if DJANGO_VERSION < (4, 0):
         from django.contrib.postgres.fields import JSONField
         json = JSONField(null=True, blank=True)
 

cachalot/tests/multi_db.py

@@ -1,6 +1,5 @@
 from unittest import skipIf
 
-from django import VERSION as DJANGO_VERSION
 from django.conf import settings
 from django.db import DEFAULT_DB_ALIAS, connections, transaction
 from django.test import TransactionTestCase
@@ -27,24 +26,6 @@ def setUp(self):
             # will execute an extra SQL request below.
             connection2.cursor()
 
-    def is_django_21_below_and_sqlite2(self):
-        """
-        Note: See test_utils.py with this function name
-        Checks if Django 2.1 or below and SQLite2
-        """
-        django_version = DJANGO_VERSION
-        if not self.is_sqlite2:
-            # Immediately know if SQLite
-            return False
-        if django_version[0] < 2:
-            # Takes Django 0 and 1 out of the picture
-            return True
-        else:
-            if django_version[0] == 2 and django_version[1] < 2:
-                # Takes Django 2.0-2.1 out
-                return True
-            return False
-
     def test_read(self):
         with self.assertNumQueries(1):
             data1 = list(Test.objects.all())
@@ -66,8 +47,7 @@ def test_invalidate_other_db(self):
             data1 = list(Test.objects.using(self.db_alias2))
             self.assertListEqual(data1, [])
 
-        with self.assertNumQueries(2 if self.is_django_21_below_and_sqlite2() else 1,
-                                   using=self.db_alias2):
+        with self.assertNumQueries(1, using=self.db_alias2):
             t3 = Test.objects.using(self.db_alias2).create(name='test3')
 
         with self.assertNumQueries(1, using=self.db_alias2):
@@ -82,8 +62,7 @@ def test_invalidation_independence(self):
             data1 = list(Test.objects.all())
             self.assertListEqual(data1, [self.t1, self.t2])
 
-        with self.assertNumQueries(2 if self.is_django_21_below_and_sqlite2() else 1,
-                                   using=self.db_alias2):
+        with self.assertNumQueries(1, using=self.db_alias2):
             Test.objects.using(self.db_alias2).create(name='test3')
 
         with self.assertNumQueries(0):

cachalot/tests/read.py

@@ -3,7 +3,7 @@
 from uuid import UUID
 from decimal import Decimal
 
-from django import VERSION as django_version
+from django import VERSION as DJANGO_VERSION
 from django.conf import settings
 from django.contrib.auth.models import Group, Permission, User
 from django.contrib.contenttypes.models import ContentType
@@ -353,7 +353,7 @@ def test_many_to_many_when_no_sql_check(self):
     @all_final_sql_checks
     def test_subquery(self):
         additional_tables = []
-        if django_version[0] == 4 and django_version[1] < 1 and settings.CACHALOT_FINAL_SQL_CHECK:
+        if DJANGO_VERSION >= (4, 0) and DJANGO_VERSION < (4, 1) and settings.CACHALOT_FINAL_SQL_CHECK:
             # with Django 4.0 comes some query optimalizations that do selects little differently.
             additional_tables.append('django_content_type')
         qs = Test.objects.filter(owner__in=User.objects.all())
@@ -694,7 +694,7 @@ def _test_union(self, check: bool):
         self.assert_query_cached(qs)
 
         with self.assertRaisesMessage(
-            AssertionError if django_version[0] < 4 else TypeError,
+            AssertionError if DJANGO_VERSION < (4, 0) else TypeError,
             'Cannot combine queries on two different base models.'
         ):
             Test.objects.all() | Permission.objects.all()
@@ -739,7 +739,7 @@ def _test_intersection(self, check: bool):
         self.assert_query_cached(qs)
 
         with self.assertRaisesMessage(
-                AssertionError if django_version[0] < 4 else TypeError,
+                AssertionError if DJANGO_VERSION < (4, 0) else TypeError,
                 'Cannot combine queries on two different base models.'):
             Test.objects.all() & Permission.objects.all()
 
@@ -816,21 +816,21 @@ def test_select_for_update(self):
         with self.assertRaises(TransactionManagementError):
             list(Test.objects.select_for_update())
 
-        with self.assertNumQueries(1):
+        with self.assertNumQueries(3 if DJANGO_VERSION >= (4, 2) else 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(1):
+        with self.assertNumQueries(3 if DJANGO_VERSION >= (4, 2) else 1):
             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'])
 
-        with self.assertNumQueries(2):
+        with self.assertNumQueries(4 if DJANGO_VERSION >= (4, 2) else 2):
             with transaction.atomic():
                 data3 = list(Test.objects.select_for_update())
                 data4 = list(Test.objects.select_for_update())
@@ -896,7 +896,9 @@ def test_extra_order_by(self):
         self.assert_query_cached(qs, [self.t2, self.t1])
 
     def test_table_inheritance(self):
-        with self.assertNumQueries(3 if self.is_sqlite else 2):
+        with self.assertNumQueries(
+            3 if self.is_sqlite else (4 if DJANGO_VERSION >= (4, 2) else 2)
+        ):
             t_child = TestChild.objects.create(name='test_child')
 
         with self.assertNumQueries(1):
@@ -911,15 +913,10 @@ def test_explain(self):
             expected = (r'\d+ 0 0 SCAN cachalot_test\n'
                         r'\d+ 0 0 USE TEMP B-TREE FOR ORDER BY')
         elif self.is_mysql:
-            if self.django_version < (3, 1):
-                expected = (
-                    r'1 SIMPLE cachalot_test '
-                    r'(?:None )?ALL None None None None 2 100\.0 Using filesort')
-            else:
-                expected = (
-                    r'-> Sort row IDs: cachalot_test.`name`  \(cost=[\d\.]+ rows=\d\)\n    '
-                    r'-> Table scan on cachalot_test  \(cost=[\d\.]+ rows=\d\)\n'
-                )
+            expected = (
+                r'-> Sort row IDs: cachalot_test.`name`  \(cost=[\d\.]+ rows=\d\)\n    '
+                r'-> Table scan on cachalot_test  \(cost=[\d\.]+ rows=\d\)\n'
+            )
         else:
             explain_kwargs.update(
                 analyze=True,
@@ -935,9 +932,7 @@ def test_explain(self):
                 r'Planning Time: [\d\.]+ ms\n'
                 r'Execution Time: [\d\.]+ ms$') % (operation_detail,
                                                    operation_detail)
-        with self.assertNumQueries(
-                2 if self.is_mysql and django_version[0] < 3
-                else 1):
+        with self.assertNumQueries(1):
             explanation1 = Test.objects.explain(**explain_kwargs)
         self.assertRegex(explanation1, expected)
         with self.assertNumQueries(0):

cachalot/tests/test_utils.py

@@ -1,4 +1,3 @@
-from django import VERSION as DJANGO_VERSION
 from django.core.management.color import no_style
 from django.db import connection, transaction
 
@@ -11,16 +10,13 @@ def setUp(self):
         self.is_sqlite = connection.vendor == 'sqlite'
         self.is_mysql = connection.vendor == 'mysql'
         self.is_postgresql = connection.vendor == 'postgresql'
-        self.django_version = DJANGO_VERSION
         self.force_reopen_connection()
 
     # TODO: Remove this workaround when this issue is fixed:
     #       https://code.djangoproject.com/ticket/29494
     def tearDown(self):
         if connection.vendor == 'postgresql':
             flush_args = [no_style(), (PostgresModel._meta.db_table,),]
-            if float(".".join(map(str, DJANGO_VERSION[:2]))) < 3.1:
-                flush_args.append(())
             flush_sql_list = connection.ops.sql_flush(*flush_args)
             with transaction.atomic():
                 for sql in flush_sql_list:

cachalot/tests/thread_safety.py

@@ -1,5 +1,6 @@
 from threading import Thread
 
+from django import VERSION as DJANGO_VERSION
 from django.db import connection, transaction
 from django.test import TransactionTestCase, skipUnlessDBFeature
 
@@ -29,7 +30,7 @@ def test_concurrent_caching(self):
         self.assertEqual(t2, t)
 
     def test_concurrent_caching_during_atomic(self):
-        with self.assertNumQueries(1):
+        with self.assertNumQueries(3 if DJANGO_VERSION >= (4, 2) else 1):
             with transaction.atomic():
                 t1 = TestThread().start_and_join()
                 t = Test.objects.create(name='test')
@@ -45,7 +46,7 @@ def test_concurrent_caching_during_atomic(self):
     def test_concurrent_caching_before_and_during_atomic_1(self):
         t1 = TestThread().start_and_join()
 
-        with self.assertNumQueries(1):
+        with self.assertNumQueries(3 if DJANGO_VERSION >= (4, 2) else 1):
             with transaction.atomic():
                 t2 = TestThread().start_and_join()
                 t = Test.objects.create(name='test')
@@ -60,7 +61,7 @@ def test_concurrent_caching_before_and_during_atomic_1(self):
     def test_concurrent_caching_before_and_during_atomic_2(self):
         t1 = TestThread().start_and_join()
 
-        with self.assertNumQueries(1):
+        with self.assertNumQueries(3 if DJANGO_VERSION >= (4, 2) else 1):
             with transaction.atomic():
                 t = Test.objects.create(name='test')
                 t2 = TestThread().start_and_join()
@@ -73,7 +74,7 @@ def test_concurrent_caching_before_and_during_atomic_2(self):
         self.assertEqual(data, t)
 
     def test_concurrent_caching_during_and_after_atomic_1(self):
-        with self.assertNumQueries(1):
+        with self.assertNumQueries(3 if DJANGO_VERSION >= (4, 2) else 1):
             with transaction.atomic():
                 t1 = TestThread().start_and_join()
                 t = Test.objects.create(name='test')
@@ -88,7 +89,7 @@ def test_concurrent_caching_during_and_after_atomic_1(self):
         self.assertEqual(data, t)
 
     def test_concurrent_caching_during_and_after_atomic_2(self):
-        with self.assertNumQueries(1):
+        with self.assertNumQueries(3 if DJANGO_VERSION >= (4, 2) else 1):
             with transaction.atomic():
                 t = Test.objects.create(name='test')
                 t1 = TestThread().start_and_join()
@@ -103,7 +104,7 @@ def test_concurrent_caching_during_and_after_atomic_2(self):
         self.assertEqual(data, t)
 
     def test_concurrent_caching_during_and_after_atomic_3(self):
-        with self.assertNumQueries(1):
+        with self.assertNumQueries(3 if DJANGO_VERSION >= (4, 2) else 1):
             with transaction.atomic():
                 t1 = TestThread().start_and_join()
                 t = Test.objects.create(name='test')