Hopiu/django-cachalot
1
0
Fork 0
mirror of https://github.com/Hopiu/django-cachalot.git synced 2026-03-16 21:30:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
django-cachalot/cachalot/tests/read.py
Jack Linke 27eaaa57cc
Correct problems with tests (#234)
2023-05-24 21:35:03 -04:00

1292 lines
50 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import datetime
from unittest import skipIf
from uuid import UUID
from decimal import Decimal
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
from django.db import (
connection, transaction, DEFAULT_DB_ALIAS, ProgrammingError,
OperationalError)
from django.db.models import Case, Count, Q, Value, When
from django.db.models.expressions import RawSQL, Subquery, OuterRef, Exists
from django.db.models.functions import Coalesce, Now
from django.db.transaction import TransactionManagementError
from django.test import TransactionTestCase, skipUnlessDBFeature, override_settings
from pytz import UTC
from cachalot.cache import cachalot_caches
from ..settings import cachalot_settings
from ..utils import UncachableQuery
from .models import Test, TestChild, TestParent, UnmanagedModel
from .test_utils import TestUtilsMixin
from .tests_decorators import all_final_sql_checks, with_final_sql_check, no_final_sql_check
def is_field_available(name):
fields = []
try:
from django.db.models import JSONField
fields.append("JSONField")
except ImportError:
pass
return name in fields
class ReadTestCase(TestUtilsMixin, TransactionTestCase):
"""
Tests if every SQL request that only reads data is cached.
The only exception is for requests that don’t go through the ORM, using
``QuerySet.extra`` with ``select`` or ``where`` arguments,
``Model.objects.raw``, or ``cursor.execute``.
"""
def setUp(self):
super(ReadTestCase, self).setUp()
self.group = Group.objects.create(name='test_group')
self.group__permissions = list(Permission.objects.all()[:3])
self.group.permissions.add(*self.group__permissions)
self.user = User.objects.create_user('user')
self.user__permissions = list(Permission.objects.filter(content_type__app_label='auth')[3:6])
self.user.groups.add(self.group)
self.user.user_permissions.add(*self.user__permissions)
self.admin = User.objects.create_superuser('admin', 'admin@test.me',
'password')
self.t1__permission = (Permission.objects
.order_by('?')
.select_related('content_type')[0])
self.t1 = Test.objects.create(
name='test1', owner=self.user,
date='1789-07-14', datetime='1789-07-14T16:43:27',
permission=self.t1__permission)
self.t2 = Test.objects.create(
name='test2', owner=self.admin, public=True,
date='1944-06-06', datetime='1944-06-06T06:35:00')
def test_empty(self):
with self.assertNumQueries(0):
data1 = list(Test.objects.none())
with self.assertNumQueries(0):
data2 = list(Test.objects.none())
self.assertListEqual(data2, data1)
self.assertListEqual(data2, [])
def test_exists(self):
with self.assertNumQueries(1):
n1 = Test.objects.exists()
with self.assertNumQueries(0):
n2 = Test.objects.exists()
self.assertEqual(n2, n1)
self.assertTrue(n2)
def test_count(self):
with self.assertNumQueries(1):
n1 = Test.objects.count()
with self.assertNumQueries(0):
n2 = Test.objects.count()
self.assertEqual(n2, n1)
self.assertEqual(n2, 2)
def test_get(self):
with self.assertNumQueries(1):
data1 = Test.objects.get(name='test1')
with self.assertNumQueries(0):
data2 = Test.objects.get(name='test1')
self.assertEqual(data2, data1)
self.assertEqual(data2, self.t1)
def test_first(self):
with self.assertNumQueries(1):
self.assertEqual(Test.objects.filter(name='bad').first(), None)
with self.assertNumQueries(0):
self.assertEqual(Test.objects.filter(name='bad').first(), None)
with self.assertNumQueries(1):
data1 = Test.objects.first()
with self.assertNumQueries(0):
data2 = Test.objects.first()
self.assertEqual(data2, data1)
self.assertEqual(data2, self.t1)
def test_last(self):
with self.assertNumQueries(1):
data1 = Test.objects.last()
with self.assertNumQueries(0):
data2 = Test.objects.last()
self.assertEqual(data2, data1)
self.assertEqual(data2, self.t2)
def test_all(self):
with self.assertNumQueries(1):
data1 = list(Test.objects.all())
with self.assertNumQueries(0):
data2 = list(Test.objects.all())
self.assertListEqual(data2, data1)
self.assertListEqual(data2, [self.t1, self.t2])
@all_final_sql_checks
def test_filter(self):
qs = Test.objects.filter(public=True)
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t2])
qs = Test.objects.filter(name__in=['test2', 'test72'])
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t2])
qs = Test.objects.filter(date__gt=datetime.date(1900, 1, 1))
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t2])
qs = Test.objects.filter(datetime__lt=datetime.datetime(1900, 1, 1))
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t1])
@all_final_sql_checks
def test_filter_empty(self):
qs = Test.objects.filter(public=True, name='user')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [])
@all_final_sql_checks
def test_exclude(self):
qs = Test.objects.exclude(public=True)
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t1])
qs = Test.objects.exclude(name__in=['test2', 'test72'])
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t1])
@all_final_sql_checks
def test_slicing(self):
qs = Test.objects.all()[:1]
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t1])
@all_final_sql_checks
def test_order_by(self):
qs = Test.objects.order_by('pk')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t1, self.t2])
qs = Test.objects.order_by('-name')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t2, self.t1])
@all_final_sql_checks
def test_random_order_by(self):
qs = Test.objects.order_by('?')
with self.assertRaises(UncachableQuery):
self.assert_tables(qs, Test)
self.assert_query_cached(qs, after=1, compare_results=False)
@with_final_sql_check
def test_order_by_field_of_another_table_with_check(self):
qs = Test.objects.order_by('owner__username')
self.assert_tables(qs, Test, User)
self.assert_query_cached(qs, [self.t2, self.t1])
@no_final_sql_check
def test_order_by_field_of_another_table_no_check(self):
qs = Test.objects.order_by('owner__username')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t2, self.t1])
@with_final_sql_check
def test_order_by_field_of_another_table_with_expression_with_check(self):
qs = Test.objects.order_by(Coalesce('name', 'owner__username'))
self.assert_tables(qs, Test, User)
self.assert_query_cached(qs, [self.t1, self.t2])
@no_final_sql_check
def test_order_by_field_of_another_table_with_expression_no_check(self):
qs = Test.objects.order_by(Coalesce('name', 'owner__username'))
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t1, self.t2])
@all_final_sql_checks
@skipIf(connection.vendor == 'mysql',
'MySQL does not support limit/offset on a subquery. '
'Since Django only applies ordering in subqueries when they are '
'offset/limited, we can’t test it on MySQL.')
def test_random_order_by_subquery(self):
qs = Test.objects.filter(
pk__in=Test.objects.order_by('?')[:10])
with self.assertRaises(UncachableQuery):
self.assert_tables(qs, Test)
self.assert_query_cached(qs, after=1, compare_results=False)
@all_final_sql_checks
def test_reverse(self):
qs = Test.objects.reverse()
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t2, self.t1])
@all_final_sql_checks
def test_distinct(self):
# We ensure that the query without distinct should return duplicate
# objects, in order to have a real-world example.
qs = Test.objects.filter(
owner__user_permissions__content_type__app_label='auth')
self.assert_tables(qs, Test, User, User.user_permissions.through,
Permission, ContentType)
self.assert_query_cached(qs, [self.t1, self.t1, self.t1])
qs = qs.distinct()
self.assert_tables(qs, Test, User, User.user_permissions.through,
Permission, ContentType)
self.assert_query_cached(qs, [self.t1])
def test_iterator(self):
with self.assertNumQueries(1):
data1 = list(Test.objects.iterator())
with self.assertNumQueries(0):
data2 = list(Test.objects.iterator())
self.assertListEqual(data2, data1)
self.assertListEqual(data2, [self.t1, self.t2])
def test_in_bulk(self):
with self.assertNumQueries(1):
data1 = Test.objects.in_bulk((5432, self.t2.pk, 9200))
with self.assertNumQueries(0):
data2 = Test.objects.in_bulk((5432, self.t2.pk, 9200))
self.assertDictEqual(data2, data1)
self.assertDictEqual(data2, {self.t2.pk: self.t2})
@all_final_sql_checks
def test_values(self):
qs = Test.objects.values('name', 'public')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [{'name': 'test1', 'public': False},
{'name': 'test2', 'public': True}])
@all_final_sql_checks
def test_values_list(self):
qs = Test.objects.values_list('name', flat=True)
self.assert_tables(qs, Test)
self.assert_query_cached(qs, ['test1', 'test2'])
def test_earliest(self):
with self.assertNumQueries(1):
data1 = Test.objects.earliest('date')
with self.assertNumQueries(0):
data2 = Test.objects.earliest('date')
self.assertEqual(data2, data1)
self.assertEqual(data2, self.t1)
def test_latest(self):
with self.assertNumQueries(1):
data1 = Test.objects.latest('date')
with self.assertNumQueries(0):
data2 = Test.objects.latest('date')
self.assertEqual(data2, data1)
self.assertEqual(data2, self.t2)
@all_final_sql_checks
def test_dates(self):
qs = Test.objects.dates('date', 'year')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [datetime.date(1789, 1, 1),
datetime.date(1944, 1, 1)])
@all_final_sql_checks
def test_datetimes(self):
qs = Test.objects.datetimes('datetime', 'hour')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [datetime.datetime(1789, 7, 14, 16),
datetime.datetime(1944, 6, 6, 6)])
@all_final_sql_checks
@skipIf(connection.vendor == 'mysql',
'Time zones are not supported by MySQL.')
@override_settings(USE_TZ=True)
def test_datetimes_with_time_zones(self):
qs = Test.objects.datetimes('datetime', 'hour')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [
datetime.datetime(1789, 7, 14, 16, tzinfo=UTC),
datetime.datetime(1944, 6, 6, 6, tzinfo=UTC)])
@all_final_sql_checks
def test_foreign_key(self):
with self.assertNumQueries(3):
data1 = [t.owner for t in Test.objects.all()]
with self.assertNumQueries(0):
data2 = [t.owner for t in Test.objects.all()]
self.assertListEqual(data2, data1)
self.assertListEqual(data2, [self.user, self.admin])
qs = Test.objects.values_list('owner', flat=True)
self.assert_tables(qs, Test, User)
self.assert_query_cached(qs, [self.user.pk, self.admin.pk])
def _test_many_to_many(self):
u = User.objects.create_user('test_user')
ct = ContentType.objects.get_for_model(User)
u.user_permissions.add(
Permission.objects.create(
name='Can discuss', content_type=ct, codename='discuss'),
Permission.objects.create(
name='Can touch', content_type=ct, codename='touch'),
Permission.objects.create(
name='Can cuddle', content_type=ct, codename='cuddle'))
return u.user_permissions.values_list('codename', flat=True)
@with_final_sql_check
def test_many_to_many_when_sql_check(self):
qs = self._test_many_to_many()
self.assert_tables(qs, User, User.user_permissions.through, Permission, ContentType)
self.assert_query_cached(qs, ['cuddle', 'discuss', 'touch'])
@no_final_sql_check
def test_many_to_many_when_no_sql_check(self):
qs = self._test_many_to_many()
self.assert_tables(qs, User, User.user_permissions.through, Permission)
self.assert_query_cached(qs, ['cuddle', 'discuss', 'touch'])
@all_final_sql_checks
def test_subquery(self):
additional_tables = []
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())
self.assert_tables(qs, Test, User)
self.assert_query_cached(qs, [self.t1, self.t2])
qs = Test.objects.filter(
owner__groups__permissions__in=Permission.objects.all()
)
self.assert_tables(
qs, Test, User, User.groups.through, Group,
Group.permissions.through, Permission,
*additional_tables
)
self.assert_query_cached(qs, [self.t1, self.t1, self.t1])
qs = Test.objects.filter(
owner__groups__permissions__in=Permission.objects.all()
).distinct()
self.assert_tables(
qs, Test, User, User.groups.through, Group,
Group.permissions.through, Permission,
*additional_tables
)
self.assert_query_cached(qs, [self.t1])
qs = TestChild.objects.exclude(permissions__isnull=True)
self.assert_tables(
qs, TestParent, TestChild,
TestChild.permissions.through, Permission
)
self.assert_query_cached(qs, [])
qs = TestChild.objects.exclude(permissions__name='')
self.assert_tables(
qs, TestParent, TestChild,
TestChild.permissions.through, Permission
)
self.assert_query_cached(qs, [])
@with_final_sql_check
def test_custom_subquery_with_check(self):
tests = Test.objects.filter(permission=OuterRef('pk')).values('name')
qs = Permission.objects.annotate(first_permission=Subquery(tests[:1]))
self.assert_tables(qs, Permission, Test, ContentType)
self.assert_query_cached(qs, list(Permission.objects.all()))
@no_final_sql_check
def test_custom_subquery_no_check(self):
tests = Test.objects.filter(permission=OuterRef('pk')).values('name')
qs = Permission.objects.annotate(first_permission=Subquery(tests[:1]))
self.assert_tables(qs, Permission, Test)
self.assert_query_cached(qs, list(Permission.objects.all()))
@with_final_sql_check
def test_custom_subquery_exists(self):
tests = Test.objects.filter(permission=OuterRef('pk'))
qs = Permission.objects.annotate(has_tests=Exists(tests))
self.assert_tables(qs, Permission, Test, ContentType)
self.assert_query_cached(qs, list(Permission.objects.all()))
@no_final_sql_check
def test_custom_subquery_exists(self):
tests = Test.objects.filter(permission=OuterRef('pk'))
qs = Permission.objects.annotate(has_tests=Exists(tests))
self.assert_tables(qs, Permission, Test)
self.assert_query_cached(qs, list(Permission.objects.all()))
@all_final_sql_checks
def test_raw_subquery(self):
with self.assertNumQueries(0):
raw_sql = RawSQL('SELECT id FROM auth_permission WHERE id = %s',
(self.t1__permission.pk,))
qs = Test.objects.filter(permission=raw_sql)
self.assert_tables(qs, Test, Permission)
self.assert_query_cached(qs, [self.t1])
qs = Test.objects.filter(
pk__in=Test.objects.filter(permission=raw_sql))
self.assert_tables(qs, Test, Permission)
self.assert_query_cached(qs, [self.t1])
@all_final_sql_checks
def test_aggregate(self):
test3 = Test.objects.create(name='test3', owner=self.user)
with self.assertNumQueries(1):
n1 = User.objects.aggregate(n=Count('test'))['n']
with self.assertNumQueries(0):
n2 = User.objects.aggregate(n=Count('test'))['n']
self.assertEqual(n2, n1)
self.assertEqual(n2, 3)
test3.delete()
@all_final_sql_checks
def test_annotate(self):
test3 = Test.objects.create(name='test3', owner=self.user)
qs = (User.objects.annotate(n=Count('test')).order_by('pk')
.values_list('n', flat=True))
self.assert_tables(qs, User, Test)
self.assert_query_cached(qs, [2, 1])
test3.delete()
@all_final_sql_checks
def test_annotate_subquery(self):
tests = Test.objects.filter(owner=OuterRef('pk')).values('name')
qs = User.objects.annotate(first_test=Subquery(tests[:1]))
self.assert_tables(qs, User, Test)
self.assert_query_cached(qs, [self.user, self.admin])
@all_final_sql_checks
def test_annotate_case_with_when_and_query_in_default(self):
tests = Test.objects.filter(owner=OuterRef('pk')).values('name')
qs = User.objects.annotate(
first_test=Case(
When(Q(pk=1), then=Value('noname')),
default=Subquery(tests[:1])
)
)
self.assert_tables(qs, User, Test)
self.assert_query_cached(qs, [self.user, self.admin])
@all_final_sql_checks
def test_annotate_case_with_when(self):
tests = Test.objects.filter(owner=OuterRef('pk')).values('name')
qs = User.objects.annotate(
first_test=Case(
When(Q(pk=1), then=Subquery(tests[:1])),
default=Value('noname')
)
)
self.assert_tables(qs, User, Test)
self.assert_query_cached(qs, [self.user, self.admin])
@all_final_sql_checks
def test_annotate_coalesce(self):
tests = Test.objects.filter(owner=OuterRef('pk')).values('name')
qs = User.objects.annotate(
name=Coalesce(
Subquery(tests[:1]),
Value('notest')
)
)
self.assert_tables(qs, User, Test)
self.assert_query_cached(qs, [self.user, self.admin])
@all_final_sql_checks
def test_annotate_raw(self):
qs = User.objects.annotate(
perm_id=RawSQL('SELECT id FROM auth_permission WHERE id = %s',
(self.t1__permission.pk,))
)
self.assert_tables(qs, User, Permission)
self.assert_query_cached(qs, [self.user, self.admin])
@all_final_sql_checks
def test_only(self):
with self.assertNumQueries(1):
t1 = Test.objects.only('name').first()
t1.name
with self.assertNumQueries(0):
t2 = Test.objects.only('name').first()
t2.name
with self.assertNumQueries(1):
t1.public
with self.assertNumQueries(0):
t2.public
self.assertEqual(t2, t1)
self.assertEqual(t2.name, t1.name)
self.assertEqual(t2.public, t1.public)
@all_final_sql_checks
def test_defer(self):
with self.assertNumQueries(1):
t1 = Test.objects.defer('name').first()
t1.public
with self.assertNumQueries(0):
t2 = Test.objects.defer('name').first()
t2.public
with self.assertNumQueries(1):
t1.name
with self.assertNumQueries(0):
t2.name
self.assertEqual(t2, t1)
self.assertEqual(t2.name, t1.name)
self.assertEqual(t2.public, t1.public)
@all_final_sql_checks
def test_select_related(self):
# Simple select_related
with self.assertNumQueries(1):
t1 = Test.objects.select_related('owner').get(name='test1')
self.assertEqual(t1.owner, self.user)
with self.assertNumQueries(0):
t2 = Test.objects.select_related('owner').get(name='test1')
self.assertEqual(t2.owner, self.user)
self.assertEqual(t2, t1)
self.assertEqual(t2, self.t1)
# Select_related through a foreign key
with self.assertNumQueries(1):
t3 = Test.objects.select_related('permission__content_type')[0]
self.assertEqual(t3.permission, self.t1.permission)
self.assertEqual(t3.permission.content_type,
self.t1__permission.content_type)
with self.assertNumQueries(0):
t4 = Test.objects.select_related('permission__content_type')[0]
self.assertEqual(t4.permission, self.t1.permission)
self.assertEqual(t4.permission.content_type,
self.t1__permission.content_type)
self.assertEqual(t4, t3)
self.assertEqual(t4, self.t1)
@all_final_sql_checks
def test_prefetch_related(self):
# Simple prefetch_related
with self.assertNumQueries(2):
data1 = list(User.objects.prefetch_related('user_permissions'))
with self.assertNumQueries(0):
permissions1 = [p for u in data1 for p in u.user_permissions.all()]
with self.assertNumQueries(0):
data2 = list(User.objects.prefetch_related('user_permissions'))
permissions2 = [p for u in data2 for p in u.user_permissions.all()]
self.assertListEqual(permissions2, permissions1)
self.assertListEqual(permissions2, self.user__permissions)
# Prefetch_related through a foreign key where exactly
# the same prefetch_related SQL request was executed before
with self.assertNumQueries(1):
data3 = list(Test.objects.select_related('owner')
.prefetch_related('owner__user_permissions'))
with self.assertNumQueries(0):
permissions3 = [p for t in data3
for p in t.owner.user_permissions.all()]
with self.assertNumQueries(0):
data4 = list(Test.objects.select_related('owner')
.prefetch_related('owner__user_permissions'))
permissions4 = [p for t in data4
for p in t.owner.user_permissions.all()]
self.assertListEqual(permissions4, permissions3)
self.assertListEqual(permissions4, self.user__permissions)
# Prefetch_related through a foreign key where exactly
# the same prefetch_related SQL request was not fetched before
with self.assertNumQueries(2):
data5 = list(Test.objects
.select_related('owner')
.prefetch_related('owner__user_permissions')[:1])
with self.assertNumQueries(0):
permissions5 = [p for t in data5
for p in t.owner.user_permissions.all()]
with self.assertNumQueries(0):
data6 = list(Test.objects.select_related('owner')
.prefetch_related('owner__user_permissions')[:1])
permissions6 = [p for t in data6
for p in t.owner.user_permissions.all()]
self.assertListEqual(permissions6, permissions5)
self.assertListEqual(permissions6, self.user__permissions)
# Prefetch_related through a many to many
with self.assertNumQueries(2):
data7 = list(Test.objects.select_related('owner')
.prefetch_related('owner__groups__permissions'))
with self.assertNumQueries(0):
permissions7 = [p for t in data7
for g in t.owner.groups.all()
for p in g.permissions.all()]
with self.assertNumQueries(0):
data8 = list(Test.objects.select_related('owner')
.prefetch_related('owner__groups__permissions'))
permissions8 = [p for t in data8
for g in t.owner.groups.all()
for p in g.permissions.all()]
self.assertListEqual(permissions8, permissions7)
self.assertListEqual(permissions8, self.group__permissions)
@all_final_sql_checks
def test_test_parent(self):
child = TestChild.objects.create(name='child')
qs = TestChild.objects.filter(name='child')
self.assert_query_cached(qs)
parent = TestParent.objects.all().first()
parent.name = 'another name'
parent.save()
child = TestChild.objects.all().first()
self.assertEqual(child.name, 'another name')
def _filtered_relation(self):
"""
Resulting query:
SELECT "cachalot_testparent"."id", "cachalot_testparent"."name",
"cachalot_testchild"."testparent_ptr_id", "cachalot_testchild"."public"
FROM "cachalot_testchild" INNER JOIN "cachalot_testparent" ON
("cachalot_testchild"."testparent_ptr_id" = "cachalot_testparent"."id")
"""
from django.db.models import FilteredRelation
qs = TestChild.objects.annotate(
filtered_permissions=FilteredRelation(
'permissions', condition=Q(permissions__pk__gt=1))
)
return qs
def _filtered_relation_common_asserts(self, qs):
self.assert_query_cached(qs)
values_qs = qs.values('filtered_permissions')
self.assert_tables(
values_qs, TestParent, TestChild, TestChild.permissions.through, Permission
)
self.assert_query_cached(values_qs)
filtered_qs = qs.filter(filtered_permissions__pk__gt=2)
self.assert_tables(
values_qs, TestParent, TestChild, TestChild.permissions.through, Permission
)
self.assert_query_cached(filtered_qs)
@with_final_sql_check
def test_filtered_relation_with_check(self):
qs = self._filtered_relation()
self.assert_tables(qs, TestParent, TestChild)
self._filtered_relation_common_asserts(qs)
@no_final_sql_check
def test_filtered_relation_no_check(self):
qs = self._filtered_relation()
self.assert_tables(qs, TestChild)
self._filtered_relation_common_asserts(qs)
def _test_union(self, check: bool):
qs = (
Test.objects.filter(pk__lt=5)
| Test.objects.filter(permission__name__contains='a')
)
self.assert_tables(qs, Test, Permission)
self.assert_query_cached(qs)
with self.assertRaisesMessage(
AssertionError if DJANGO_VERSION < (4, 0) else TypeError,
'Cannot combine queries on two different base models.'
):
Test.objects.all() | Permission.objects.all()
qs = Test.objects.filter(pk__lt=5)
sub_qs = Test.objects.filter(permission__name__contains='a')
if self.is_sqlite:
qs = qs.order_by()
sub_qs = sub_qs.order_by()
qs = qs.union(sub_qs)
self.assert_tables(qs, Test, Permission)
self.assert_query_cached(qs)
qs = Test.objects.all()
sub_qs = Permission.objects.all()
if self.is_sqlite:
qs = qs.order_by()
sub_qs = sub_qs.order_by()
qs = qs.union(sub_qs)
tables = {Test, Permission}
# Sqlite does not do an ORDER BY django_content_type
if not self.is_sqlite and check:
tables.add(ContentType)
self.assert_tables(qs, *tables)
with self.assertRaises((ProgrammingError, OperationalError)):
self.assert_query_cached(qs)
@with_final_sql_check
@skipUnlessDBFeature('supports_select_union')
def test_union_with_sql_check(self):
self._test_union(check=True)
@no_final_sql_check
@skipUnlessDBFeature('supports_select_union')
def test_union_with_sql_check(self):
self._test_union(check=False)
def _test_intersection(self, check: bool):
qs = (Test.objects.filter(pk__lt=5)
& Test.objects.filter(permission__name__contains='a'))
self.assert_tables(qs, Test, Permission)
self.assert_query_cached(qs)
with self.assertRaisesMessage(
AssertionError if DJANGO_VERSION < (4, 0) else TypeError,
'Cannot combine queries on two different base models.'):
Test.objects.all() & Permission.objects.all()
qs = Test.objects.filter(pk__lt=5)
sub_qs = Test.objects.filter(permission__name__contains='a')
if self.is_sqlite:
qs = qs.order_by()
sub_qs = sub_qs.order_by()
qs = qs.intersection(sub_qs)
self.assert_tables(qs, Test, Permission)
self.assert_query_cached(qs)
qs = Test.objects.all()
sub_qs = Permission.objects.all()
if self.is_sqlite:
qs = qs.order_by()
sub_qs = sub_qs.order_by()
qs = qs.intersection(sub_qs)
tables = {Test, Permission}
if not self.is_sqlite and check:
tables.add(ContentType)
self.assert_tables(qs, *tables)
with self.assertRaises((ProgrammingError, OperationalError)):
self.assert_query_cached(qs)
@with_final_sql_check
@skipUnlessDBFeature('supports_select_intersection')
def test_intersection_with_check(self):
self._test_intersection(check=True)
@no_final_sql_check
@skipUnlessDBFeature('supports_select_intersection')
def test_intersection_with_check(self):
self._test_intersection(check=False)
def _test_difference(self, check: bool):
qs = Test.objects.filter(pk__lt=5)
sub_qs = Test.objects.filter(permission__name__contains='a')
if self.is_sqlite:
qs = qs.order_by()
sub_qs = sub_qs.order_by()
qs = qs.difference(sub_qs)
self.assert_tables(qs, Test, Permission)
self.assert_query_cached(qs)
qs = Test.objects.all()
sub_qs = Permission.objects.all()
if self.is_sqlite:
qs = qs.order_by()
sub_qs = sub_qs.order_by()
qs = qs.difference(sub_qs)
tables = {Test, Permission}
if not self.is_sqlite and check:
tables.add(ContentType)
self.assert_tables(qs, *tables)
with self.assertRaises((ProgrammingError, OperationalError)):
self.assert_query_cached(qs)
@with_final_sql_check
@skipUnlessDBFeature('supports_select_difference')
def test_difference_with_check(self):
self._test_difference(check=True)
@no_final_sql_check
@skipUnlessDBFeature('supports_select_difference')
def test_difference_with_check(self):
self._test_difference(check=False)
@skipUnlessDBFeature('has_select_for_update')
def test_select_for_update(self):
"""
Tests if ``select_for_update`` queries are not cached.
"""
with self.assertRaises(TransactionManagementError):
list(Test.objects.select_for_update())
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(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(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())
self.assertListEqual(data3, [self.t1, self.t2])
self.assertListEqual(data4, [self.t1, self.t2])
self.assertListEqual([t.name for t in data3],
['test1', 'test2'])
self.assertListEqual([t.name for t in data4],
['test1', 'test2'])
@all_final_sql_checks
def test_having(self):
qs = (User.objects.annotate(n=Count('user_permissions')).filter(n__gte=1))
self.assert_tables(qs, User, User.user_permissions.through, Permission)
self.assert_query_cached(qs, [self.user])
with self.assertNumQueries(1):
self.assertEqual(User.objects.annotate(n=Count('user_permissions'))
.filter(n__gte=1).count(), 1)
with self.assertNumQueries(0):
self.assertEqual(User.objects.annotate(n=Count('user_permissions'))
.filter(n__gte=1).count(), 1)
def test_extra_select(self):
username_length_sql = """
SELECT LENGTH(%(user_table)s.username)
FROM %(user_table)s
WHERE %(user_table)s.id = %(test_table)s.owner_id
""" % {'user_table': User._meta.db_table,
'test_table': Test._meta.db_table}
with self.assertNumQueries(1):
data1 = list(Test.objects.extra(
select={'username_length': username_length_sql}))
self.assertListEqual(data1, [self.t1, self.t2])
self.assertListEqual([o.username_length for o in data1], [4, 5])
with self.assertNumQueries(0):
data2 = list(Test.objects.extra(
select={'username_length': username_length_sql}))
self.assertListEqual(data2, [self.t1, self.t2])
self.assertListEqual([o.username_length for o in data2], [4, 5])
@all_final_sql_checks
def test_extra_where(self):
sql_condition = ("owner_id IN "
"(SELECT id FROM auth_user WHERE username = 'admin')")
qs = Test.objects.extra(where=[sql_condition])
self.assert_tables(qs, Test, User)
self.assert_query_cached(qs, [self.t2])
@all_final_sql_checks
def test_extra_tables(self):
qs = Test.objects.extra(tables=['auth_user'],
select={'extra_id': 'auth_user.id'})
self.assert_tables(qs, Test, User)
self.assert_query_cached(qs)
@all_final_sql_checks
def test_extra_order_by(self):
qs = Test.objects.extra(order_by=['-cachalot_test.name'])
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t2, self.t1])
def test_table_inheritance(self):
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):
self.assertEqual(TestChild.objects.get(), t_child)
with self.assertNumQueries(0):
self.assertEqual(TestChild.objects.get(), t_child)
def test_explain(self):
explain_kwargs = {}
if self.is_sqlite:
expected = (r'\d+ 0 0 SCAN cachalot_test\n'
r'\d+ 0 0 USE TEMP B-TREE FOR ORDER BY')
elif self.is_mysql:
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,
costs=False,
)
operation_detail = (r'\(actual time=[\d\.]+..[\d\.]+\ '
r'rows=\d+ loops=\d+\)')
expected = (
r'^Sort %s\n'
r' Sort Key: name\n'
r' Sort Method: quicksort Memory: \d+kB\n'
r' -> Seq Scan on cachalot_test %s\n'
r'Planning Time: [\d\.]+ ms\n'
r'Execution Time: [\d\.]+ ms$') % (operation_detail,
operation_detail)
with self.assertNumQueries(1):
explanation1 = Test.objects.explain(**explain_kwargs)
self.assertRegex(explanation1, expected)
with self.assertNumQueries(0):
explanation2 = Test.objects.explain(**explain_kwargs)
self.assertEqual(explanation2, explanation1)
def test_raw(self):
"""
Tests if ``Model.objects.raw`` queries are not cached.
"""
sql = 'SELECT * FROM %s;' % Test._meta.db_table
with self.assertNumQueries(1):
data1 = list(Test.objects.raw(sql))
with self.assertNumQueries(1):
data2 = list(Test.objects.raw(sql))
self.assertListEqual(data2, data1)
self.assertListEqual(data2, [self.t1, self.t2])
def test_raw_no_table(self):
sql = 'SELECT * FROM (SELECT 1 AS id UNION ALL SELECT 2) AS t;'
with self.assertNumQueries(1):
data1 = list(Test.objects.raw(sql))
with self.assertNumQueries(1):
data2 = list(Test.objects.raw(sql))
self.assertListEqual(data2, data1)
self.assertListEqual(data2, [Test(pk=1), Test(pk=2)])
def test_cursor_execute_unicode(self):
"""
Tests if queries executed from a DB cursor are not cached.
"""
attname_column_list = [f.get_attname_column()
for f in Test._meta.fields]
attnames = [t[0] for t in attname_column_list]
columns = [t[1] for t in attname_column_list]
sql = "SELECT CAST('é' AS CHAR), %s FROM %s;" % (', '.join(columns),
Test._meta.db_table)
with self.assertNumQueries(1):
with connection.cursor() as cursor:
cursor.execute(sql)
data1 = list(cursor.fetchall())
with self.assertNumQueries(1):
with connection.cursor() as cursor:
cursor.execute(sql)
data2 = list(cursor.fetchall())
self.assertListEqual(data2, data1)
self.assertListEqual(
data2,
[('é',) + l for l in Test.objects.values_list(*attnames)])
@skipIf(connection.vendor == 'sqlite',
'SQLite doesn’t accept bytes as raw query.')
def test_cursor_execute_bytes(self):
attname_column_list = [f.get_attname_column()
for f in Test._meta.fields]
attnames = [t[0] for t in attname_column_list]
columns = [t[1] for t in attname_column_list]
sql = "SELECT CAST('é' AS CHAR), %s FROM %s;" % (', '.join(columns),
Test._meta.db_table)
sql = sql.encode('utf-8')
with self.assertNumQueries(1):
with connection.cursor() as cursor:
cursor.execute(sql)
data1 = list(cursor.fetchall())
with self.assertNumQueries(1):
with connection.cursor() as cursor:
cursor.execute(sql)
data2 = list(cursor.fetchall())
self.assertListEqual(data2, data1)
self.assertListEqual(
data2,
[('é',) + l for l in Test.objects.values_list(*attnames)])
def test_cursor_execute_no_table(self):
sql = 'SELECT * FROM (SELECT 1 AS id UNION ALL SELECT 2) AS t;'
with self.assertNumQueries(1):
with connection.cursor() as cursor:
cursor.execute(sql)
data1 = list(cursor.fetchall())
with self.assertNumQueries(1):
with connection.cursor() as cursor:
cursor.execute(sql)
data2 = list(cursor.fetchall())
self.assertListEqual(data2, data1)
self.assertListEqual(data2, [(1,), (2,)])
@all_final_sql_checks
def test_missing_table_cache_key(self):
qs = Test.objects.all()
self.assert_tables(qs, Test)
self.assert_query_cached(qs)
table_cache_key = cachalot_settings.CACHALOT_TABLE_KEYGEN(
connection.alias, Test._meta.db_table)
cachalot_caches.get_cache().delete(table_cache_key)
self.assert_query_cached(qs)
@all_final_sql_checks
def test_broken_query_cache_value(self):
"""
In some undetermined cases, cache.get_many return wrong values such
as `None` or other invalid values. They should be gracefully handled.
See https://github.com/noripyt/django-cachalot/issues/110
This test artificially creates a wrong value, but it’s usually
a cache backend bug that leads to these wrong values.
"""
qs = Test.objects.all()
self.assert_tables(qs, Test)
self.assert_query_cached(qs)
query_cache_key = cachalot_settings.CACHALOT_QUERY_KEYGEN(
qs.query.get_compiler(DEFAULT_DB_ALIAS))
cachalot_caches.get_cache().set(query_cache_key, (),
cachalot_settings.CACHALOT_TIMEOUT)
self.assert_query_cached(qs)
def test_unicode_get(self):
with self.assertNumQueries(1):
with self.assertRaises(Test.DoesNotExist):
Test.objects.get(name='Clémentine')
with self.assertNumQueries(0):
with self.assertRaises(Test.DoesNotExist):
Test.objects.get(name='Clémentine')
@all_final_sql_checks
def test_unicode_table_name(self):
"""
Tests if using unicode in table names does not break caching.
"""
table_name = 'Clémentine'
if self.is_postgresql:
table_name = '"%s"' % table_name
with connection.cursor() as cursor:
cursor.execute('CREATE TABLE %s (taste VARCHAR(20));' % table_name)
qs = Test.objects.extra(tables=['Clémentine'],
select={'taste': '%s.taste' % table_name})
# Here the table `Clémentine` is not detected because it is not
# registered by Django, and we only check for registered tables
# to avoid creating an extra SQL query fetching table names.
self.assert_tables(qs, Test)
self.assert_query_cached(qs)
with connection.cursor() as cursor:
cursor.execute('DROP TABLE %s;' % table_name)
@all_final_sql_checks
def test_unmanaged_model(self):
qs = UnmanagedModel.objects.all()
self.assert_tables(qs, UnmanagedModel)
self.assert_query_cached(qs)
def test_now_annotate(self):
"""Check that queries with a Now() annotation are not cached #193"""
qs = Test.objects.annotate(now=Now())
self.assert_query_cached(qs, after=1)
class ParameterTypeTestCase(TestUtilsMixin, TransactionTestCase):
@all_final_sql_checks
def test_tuple(self):
qs = Test.objects.filter(pk__in=(1, 2, 3))
self.assert_tables(qs, Test)
self.assert_query_cached(qs)
qs = Test.objects.filter(pk__in=(4, 5, 6))
self.assert_tables(qs, Test)
self.assert_query_cached(qs)
@all_final_sql_checks
def test_list(self):
qs = Test.objects.filter(pk__in=[1, 2, 3])
self.assert_tables(qs, Test)
self.assert_query_cached(qs)
l = [4, 5, 6]
qs = Test.objects.filter(pk__in=l)
self.assert_tables(qs, Test)
self.assert_query_cached(qs)
l.append(7)
self.assert_tables(qs, Test)
# The queryset is not taking the new element into account because
# the list was copied during `.filter()`.
self.assert_query_cached(qs, before=0)
qs = Test.objects.filter(pk__in=l)
self.assert_tables(qs, Test)
self.assert_query_cached(qs)
@all_final_sql_checks
def test_binary(self):
"""
Binary data should be cached on PostgreSQL & MySQL, but not on SQLite,
because SQLite uses a type making it hard to access data itself.
So this also tests how django-cachalot handles unknown params, in this
case the `memory` object passed to SQLite.
"""
qs = Test.objects.filter(bin=None)
self.assert_tables(qs, Test)
self.assert_query_cached(qs)
qs = Test.objects.filter(bin=b'abc')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, after=1 if self.is_sqlite else 0)
qs = Test.objects.filter(bin=b'def')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, after=1 if self.is_sqlite else 0)
def test_float(self):
with self.assertNumQueries(1):
Test.objects.create(name='test1', a_float=0.123456789)
with self.assertNumQueries(1):
Test.objects.create(name='test2', a_float=12345.6789)
with self.assertNumQueries(1):
data1 = list(Test.objects.values_list('a_float', flat=True).filter(
a_float__isnull=False).order_by('a_float'))
with self.assertNumQueries(0):
data2 = list(Test.objects.values_list('a_float', flat=True).filter(
a_float__isnull=False).order_by('a_float'))
self.assertListEqual(data2, data1)
self.assertEqual(len(data2), 2)
self.assertAlmostEqual(data2[0], 0.123456789, delta=0.0001)
self.assertAlmostEqual(data2[1], 12345.6789, delta=0.0001)
with self.assertNumQueries(1):
Test.objects.get(a_float=0.123456789)
with self.assertNumQueries(0):
Test.objects.get(a_float=0.123456789)
@all_final_sql_checks
def test_decimal(self):
with self.assertNumQueries(1):
test1 = Test.objects.create(name='test1', a_decimal=Decimal('123.45'))
with self.assertNumQueries(1):
test2 = Test.objects.create(name='test2', a_decimal=Decimal('12.3'))
qs = Test.objects.values_list('a_decimal', flat=True).filter(
a_decimal__isnull=False).order_by('a_decimal')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [Decimal('12.3'), Decimal('123.45')])
with self.assertNumQueries(1):
Test.objects.get(a_decimal=Decimal('123.45'))
with self.assertNumQueries(0):
Test.objects.get(a_decimal=Decimal('123.45'))
test1.delete()
test2.delete()
@all_final_sql_checks
def test_ipv4_address(self):
with self.assertNumQueries(1):
test1 = Test.objects.create(name='test1', ip='127.0.0.1')
with self.assertNumQueries(1):
test2 = Test.objects.create(name='test2', ip='192.168.0.1')
qs = Test.objects.values_list('ip', flat=True).filter(
ip__isnull=False).order_by('ip')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, ['127.0.0.1', '192.168.0.1'])
with self.assertNumQueries(1):
Test.objects.get(ip='127.0.0.1')
with self.assertNumQueries(0):
Test.objects.get(ip='127.0.0.1')
test1.delete()
test2.delete()
@all_final_sql_checks
def test_ipv6_address(self):
with self.assertNumQueries(1):
test1 = Test.objects.create(name='test1', ip='2001:db8:a0b:12f0::1/64')
with self.assertNumQueries(1):
test2 = Test.objects.create(name='test2', ip='2001:db8:0:85a3::ac1f:8001')
qs = Test.objects.values_list('ip', flat=True).filter(
ip__isnull=False).order_by('ip')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, ['2001:db8:0:85a3::ac1f:8001',
'2001:db8:a0b:12f0::1/64'])
with self.assertNumQueries(1):
Test.objects.get(ip='2001:db8:0:85a3::ac1f:8001')
with self.assertNumQueries(0):
Test.objects.get(ip='2001:db8:0:85a3::ac1f:8001')
test1.delete()
test2.delete()
@all_final_sql_checks
def test_duration(self):
with self.assertNumQueries(1):
test1 = Test.objects.create(name='test1', duration=datetime.timedelta(30))
with self.assertNumQueries(1):
test2 = Test.objects.create(name='test2', duration=datetime.timedelta(60))
qs = Test.objects.values_list('duration', flat=True).filter(
duration__isnull=False).order_by('duration')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [datetime.timedelta(30),
datetime.timedelta(60)])
with self.assertNumQueries(1):
Test.objects.get(duration=datetime.timedelta(30))
with self.assertNumQueries(0):
Test.objects.get(duration=datetime.timedelta(30))
test1.delete()
test2.delete()
@all_final_sql_checks
def test_uuid(self):
with self.assertNumQueries(1):
test1 = Test.objects.create(name='test1',
uuid='1cc401b7-09f4-4520-b8d0-c267576d196b')
with self.assertNumQueries(1):
test2 = Test.objects.create(name='test2',
uuid='ebb3b6e1-1737-4321-93e3-4c35d61ff491')
qs = Test.objects.values_list('uuid', flat=True).filter(
uuid__isnull=False).order_by('uuid')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [
UUID('1cc401b7-09f4-4520-b8d0-c267576d196b'),
UUID('ebb3b6e1-1737-4321-93e3-4c35d61ff491')])
with self.assertNumQueries(1):
Test.objects.get(uuid=UUID('1cc401b7-09f4-4520-b8d0-c267576d196b'))
with self.assertNumQueries(0):
Test.objects.get(uuid=UUID('1cc401b7-09f4-4520-b8d0-c267576d196b'))
test1.delete()
test2.delete()
def test_now(self):
"""
Checks that queries with a Now() parameter are not cached.
"""
obj = Test.objects.create(datetime='1992-07-02T12:00:00')
qs = Test.objects.filter(datetime__lte=Now())
with self.assertNumQueries(1):
obj1 = qs.get()
with self.assertNumQueries(1):
obj2 = qs.get()
self.assertEqual(obj1, obj2)
self.assertEqual(obj1, obj)
Reference in a new issue View git blame Copy permalink