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
Dominik Bartenstein 76d4ab4c8d
Simplify annotation handling by using the flatten method: (#197) 
* Simplify annotation handling by using the flatten method:
f42ccdd835/django/db/models/expressions.py (L370)

Handle annotated cases when Subquery is part of the When.

* Before Django 3.2 flatten did not check for existence of flatten in the processed nodes.

* Add type BaseExpression to function "flatten".

* Add test case with annotated Coalesce.

* Add support for annotated raw SQL.

* Remove unnecessary code.

* Use as_sql instead of repr.

* Reorganize code

* Fix var name

* Improve naming: element -> expression

Co-authored-by: Dominik Bartenstein <db@zemtu.com>
Co-authored-by: Andrew Chen Wang <60190294+Andrew-Chen-Wang@users.noreply.github.com>
2021-08-21 20:55:01 -04:00

1090 lines
44 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.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
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.all()[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])
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])
def test_filter_empty(self):
qs = Test.objects.filter(public=True, name='user')
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [])
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])
def test_slicing(self):
qs = Test.objects.all()[:1]
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t1])
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])
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)
@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)
def test_reverse(self):
qs = Test.objects.reverse()
self.assert_tables(qs, Test)
self.assert_query_cached(qs, [self.t2, self.t1])
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})
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}])
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)
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)])
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)])
@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)])
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'))
qs = u.user_permissions.values_list('codename', flat=True)
self.assert_tables(qs, User, User.user_permissions.through, Permission)
self.assert_query_cached(qs, ['cuddle', 'discuss', 'touch'])
def test_subquery(self):
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)
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)
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, [])
def test_custom_subquery(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()))
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()))
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])
def test_aggregate(self):
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)
def test_annotate(self):
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])
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])
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])
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])
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])
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])
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)
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)
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)
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)
def test_filtered_relation(self):
from django.db.models import FilteredRelation
qs = TestChild.objects.annotate(
filtered_permissions=FilteredRelation(
'permissions', condition=Q(permissions__pk__gt=1)))
self.assert_tables(qs, TestChild)
self.assert_query_cached(qs)
values_qs = qs.values('filtered_permissions')
self.assert_tables(
values_qs, 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, TestChild, TestChild.permissions.through, Permission)
self.assert_query_cached(filtered_qs)
@skipUnlessDBFeature('supports_select_union')
def test_union(self):
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,
'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)
self.assert_tables(qs, Test, Permission)
with self.assertRaises((ProgrammingError, OperationalError)):
self.assert_query_cached(qs)
@skipUnlessDBFeature('supports_select_intersection')
def test_intersection(self):
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,
'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)
self.assert_tables(qs, Test, Permission)
with self.assertRaises((ProgrammingError, OperationalError)):
self.assert_query_cached(qs)
@skipUnlessDBFeature('supports_select_difference')
def test_difference(self):
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)
self.assert_tables(qs, Test, Permission)
with self.assertRaises((ProgrammingError, OperationalError)):
self.assert_query_cached(qs)
@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(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 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 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'])
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])
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])
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)
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 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 TABLE 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'
)
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(
2 if self.is_mysql and django_version[0] < 3
else 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,)])
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)
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')
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)
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):
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)
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)
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)
def test_decimal(self):
with self.assertNumQueries(1):
Test.objects.create(name='test1', a_decimal=Decimal('123.45'))
with self.assertNumQueries(1):
Test.objects.create(name='test1', 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'))
def test_ipv4_address(self):
with self.assertNumQueries(1):
Test.objects.create(name='test1', ip='127.0.0.1')
with self.assertNumQueries(1):
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')
def test_ipv6_address(self):
with self.assertNumQueries(1):
Test.objects.create(name='test1', ip='2001:db8:a0b:12f0::1/64')
with self.assertNumQueries(1):
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')
def test_duration(self):
with self.assertNumQueries(1):
Test.objects.create(name='test1', duration=datetime.timedelta(30))
with self.assertNumQueries(1):
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))
def test_uuid(self):
with self.assertNumQueries(1):
Test.objects.create(name='test1',
uuid='1cc401b7-09f4-4520-b8d0-c267576d196b')
with self.assertNumQueries(1):
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'))
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