3
3
# Unit tests for cache framework
4
4
# Uses whatever cache backend is set in the test settings file.
5
from __future__ import with_statement, absolute_import
5
from __future__ import absolute_import, unicode_literals
15
16
from django.conf import settings
16
17
from django.core import management
17
from django.core.cache import get_cache, DEFAULT_CACHE_ALIAS
18
from django.core.cache import get_cache
18
19
from django.core.cache.backends.base import (CacheKeyWarning,
19
20
InvalidCacheBackendError)
20
21
from django.db import router
21
from django.http import HttpResponse, HttpRequest, QueryDict
22
from django.http import (HttpResponse, HttpRequest, StreamingHttpResponse,
22
24
from django.middleware.cache import (FetchFromCacheMiddleware,
23
25
UpdateCacheMiddleware, CacheMiddleware)
24
26
from django.template import Template
25
27
from django.template.response import TemplateResponse
26
28
from django.test import TestCase, TransactionTestCase, RequestFactory
27
from django.test.utils import (get_warnings_state, restore_warnings_state,
29
from django.test.utils import override_settings, six
29
30
from django.utils import timezone, translation, unittest
30
31
from django.utils.cache import (patch_vary_headers, get_cache_key,
31
32
learn_cache_key, patch_cache_control, patch_response_headers)
32
from django.utils.encoding import force_unicode
33
from django.utils.encoding import force_text
33
34
from django.views.decorators.cache import cache_page
35
36
from .models import Poll, expensive_calculation
140
141
def test_unicode(self):
141
142
"Unicode values are ignored by the dummy cache"
143
u'ascii': u'ascii_value',
144
u'unicode_ascii': u'Iñtërnâtiônàlizætiøn1',
145
u'Iñtërnâtiônàlizætiøn': u'Iñtërnâtiônàlizætiøn2',
146
u'ascii2': {u'x' : 1 }
144
'ascii': 'ascii_value',
145
'unicode_ascii': 'Iñtërnâtiônàlizætiøn1',
146
'Iñtërnâtiônàlizætiøn': 'Iñtërnâtiônàlizætiøn2',
148
149
for (key, value) in stuff.items():
149
150
self.cache.set(key, value)
265
267
self.assertEqual(self.cache.get('answer'), 42)
266
268
self.assertEqual(self.cache.decr('answer', 10), 32)
267
269
self.assertEqual(self.cache.get('answer'), 32)
270
self.assertEqual(self.cache.decr('answer', -10), 42)
268
271
self.assertRaises(ValueError, self.cache.decr, 'does_not_exist')
270
273
def test_data_types(self):
338
341
def test_unicode(self):
339
342
# Unicode values can be cached
341
u'ascii': u'ascii_value',
342
u'unicode_ascii': u'Iñtërnâtiônàlizætiøn1',
343
u'Iñtërnâtiônàlizætiøn': u'Iñtërnâtiônàlizætiøn2',
344
u'ascii2': {u'x' : 1 }
344
'ascii': 'ascii_value',
345
'unicode_ascii': 'Iñtërnâtiônàlizætiøn1',
346
'Iñtërnâtiônàlizætiøn': 'Iñtërnâtiônàlizætiøn2',
347
350
for (key, value) in stuff.items():
365
368
# Binary strings should be cacheable
366
369
from zlib import compress, decompress
367
370
value = 'value_to_be_compressed'
368
compressed_value = compress(value)
371
compressed_value = compress(value.encode())
371
374
self.cache.set('binary1', compressed_value)
372
375
compressed_result = self.cache.get('binary1')
373
376
self.assertEqual(compressed_value, compressed_result)
374
self.assertEqual(value, decompress(compressed_result))
377
self.assertEqual(value, decompress(compressed_result).decode())
377
380
self.cache.add('binary1-add', compressed_value)
378
381
compressed_result = self.cache.get('binary1-add')
379
382
self.assertEqual(compressed_value, compressed_result)
380
self.assertEqual(value, decompress(compressed_result))
383
self.assertEqual(value, decompress(compressed_result).decode())
383
386
self.cache.set_many({'binary1-set_many': compressed_value})
384
387
compressed_result = self.cache.get('binary1-set_many')
385
388
self.assertEqual(compressed_value, compressed_result)
386
self.assertEqual(value, decompress(compressed_result))
389
self.assertEqual(value, decompress(compressed_result).decode())
388
391
def test_set_many(self):
389
392
# Multiple keys can be set using set_many
467
470
old_func = self.cache.key_func
468
471
self.cache.key_func = func
469
# On Python 2.6+ we could use the catch_warnings context
470
# manager to test this warning nicely. Since we can't do that
471
# yet, the cleanest option is to temporarily ask for
472
# CacheKeyWarning to be raised as an exception.
473
_warnings_state = get_warnings_state()
474
warnings.simplefilter("error", CacheKeyWarning)
477
# memcached does not allow whitespace or control characters in keys
478
self.assertRaises(CacheKeyWarning, self.cache.set, 'key with spaces', 'value')
479
# memcached limits key length to 250
480
self.assertRaises(CacheKeyWarning, self.cache.set, 'a' * 251, 'value')
474
with warnings.catch_warnings(record=True) as w:
475
warnings.simplefilter("always")
476
# memcached does not allow whitespace or control characters in keys
477
self.cache.set('key with spaces', 'value')
478
self.assertEqual(len(w), 2)
479
self.assertTrue(isinstance(w[0].message, CacheKeyWarning))
480
with warnings.catch_warnings(record=True) as w:
481
warnings.simplefilter("always")
482
# memcached limits key length to 250
483
self.cache.set('a' * 251, 'value')
484
self.assertEqual(len(w), 1)
485
self.assertTrue(isinstance(w[0].message, CacheKeyWarning))
482
restore_warnings_state(_warnings_state)
483
487
self.cache.key_func = old_func
485
489
def test_cache_versioning_get_set(self):
775
779
get_cache_data = fetch_middleware.process_request(request)
776
780
self.assertNotEqual(get_cache_data, None)
777
self.assertEqual(get_cache_data.content, content)
781
self.assertEqual(get_cache_data.content, content.encode('utf-8'))
778
782
self.assertEqual(get_cache_data.cookies, response.cookies)
780
784
update_middleware.process_response(request, get_cache_data)
781
785
get_cache_data = fetch_middleware.process_request(request)
782
786
self.assertNotEqual(get_cache_data, None)
783
self.assertEqual(get_cache_data.content, content)
787
self.assertEqual(get_cache_data.content, content.encode('utf-8'))
784
788
self.assertEqual(get_cache_data.cookies, response.cookies)
786
790
def custom_key_func(key, key_prefix, version):
819
823
self.perform_cull_test(50, 18)
821
825
def test_second_call_doesnt_crash(self):
822
err = StringIO.StringIO()
823
management.call_command('createcachetable', self._table_name, verbosity=0, interactive=False, stderr=err)
824
self.assertTrue("Cache table 'test cache table' could not be created" in err.getvalue())
827
DBCacheWithTimeZoneTests = override_settings(USE_TZ=True)(DBCacheTests)
826
with six.assertRaisesRegex(self, management.CommandError,
827
"Cache table 'test cache table' could not be created"):
828
management.call_command(
836
@override_settings(USE_TZ=True)
837
class DBCacheWithTimeZoneTests(DBCacheTests):
830
841
class DBCacheRouter(object):
926
937
# memcached backend isn't guaranteed to be available.
927
938
# To check the memcached backend, the test settings file will
928
# need to contain a cache backend setting that points at
939
# need to contain at least one cache backend setting that points at
929
940
# your memcache server.
941
@unittest.skipUnless(
942
any(cache['BACKEND'].startswith('django.core.cache.backends.memcached.')
943
for cache in settings.CACHES.values()),
944
"memcached not available")
930
945
class MemcachedCacheTests(unittest.TestCase, BaseCacheTests):
931
backend_name = 'django.core.cache.backends.memcached.MemcachedCache'
934
name = settings.CACHES[DEFAULT_CACHE_ALIAS]['LOCATION']
935
self.cache = get_cache(self.backend_name, LOCATION=name)
936
self.prefix_cache = get_cache(self.backend_name, LOCATION=name, KEY_PREFIX='cacheprefix')
937
self.v2_cache = get_cache(self.backend_name, LOCATION=name, VERSION=2)
938
self.custom_key_cache = get_cache(self.backend_name, LOCATION=name, KEY_FUNCTION=custom_key_func)
939
self.custom_key_cache2 = get_cache(self.backend_name, LOCATION=name, KEY_FUNCTION='regressiontests.cache.tests.custom_key_func')
948
for cache_key, cache in settings.CACHES.items():
949
if cache['BACKEND'].startswith('django.core.cache.backends.memcached.'):
951
random_prefix = ''.join(random.choice(string.ascii_letters) for x in range(10))
952
self.cache = get_cache(cache_key)
953
self.prefix_cache = get_cache(cache_key, KEY_PREFIX=random_prefix)
954
self.v2_cache = get_cache(cache_key, VERSION=2)
955
self.custom_key_cache = get_cache(cache_key, KEY_FUNCTION=custom_key_func)
956
self.custom_key_cache2 = get_cache(cache_key, KEY_FUNCTION='regressiontests.cache.tests.custom_key_func')
941
958
def tearDown(self):
942
959
self.cache.clear()
980
995
"""Test that keys are hashed into subdirectories correctly"""
981
996
self.cache.set("foo", "bar")
982
997
key = self.cache.make_key("foo")
983
keyhash = hashlib.md5(key).hexdigest()
998
keyhash = hashlib.md5(key.encode()).hexdigest()
984
999
keypath = os.path.join(self.dirname, keyhash[:2], keyhash[2:4], keyhash[4:])
985
1000
self.assertTrue(os.path.exists(keypath))
1144
1169
parts = set(cc_delim_re.split(response['Cache-Control']))
1145
1170
self.assertEqual(parts, expected_cc)
1147
CacheUtils = override_settings(
1148
CACHE_MIDDLEWARE_KEY_PREFIX='settingsprefix',
1149
CACHE_MIDDLEWARE_SECONDS=1,
1152
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
1158
PrefixedCacheUtils = override_settings(
1161
1176
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
1162
1177
'KEY_PREFIX': 'cacheprefix',
1181
class PrefixedCacheUtils(CacheUtils):
1186
CACHE_MIDDLEWARE_SECONDS=60,
1187
CACHE_MIDDLEWARE_KEY_PREFIX='test',
1190
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
1168
1194
class CacheHEADTest(TestCase):
1170
1196
def setUp(self):
1214
1240
request = self._get_request('HEAD')
1215
1241
get_cache_data = FetchFromCacheMiddleware().process_request(request)
1216
1242
self.assertNotEqual(get_cache_data, None)
1217
self.assertEqual(test_content, get_cache_data.content)
1219
CacheHEADTest = override_settings(
1220
CACHE_MIDDLEWARE_SECONDS=60,
1221
CACHE_MIDDLEWARE_KEY_PREFIX='test',
1243
self.assertEqual(test_content.encode(), get_cache_data.content)
1247
CACHE_MIDDLEWARE_KEY_PREFIX='settingsprefix',
1224
1250
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
1230
1258
class CacheI18nTest(TestCase):
1232
1260
def setUp(self):
1286
1314
request = self._get_request()
1287
1315
# This is tightly coupled to the implementation,
1288
1316
# but it's the most straightforward way to test the key.
1289
tz = force_unicode(timezone.get_current_timezone_name(), errors='ignore')
1290
tz = tz.encode('ascii', 'ignore').replace(' ', '_')
1317
tz = force_text(timezone.get_current_timezone_name(), errors='ignore')
1318
tz = tz.encode('ascii', 'ignore').decode('ascii').replace(' ', '_')
1291
1319
response = HttpResponse()
1292
1320
key = learn_cache_key(request, response)
1293
1321
self.assertIn(tz, key, "Cache keys should include the time zone name when time zones are active")
1298
1326
def test_cache_key_no_i18n (self):
1299
1327
request = self._get_request()
1300
1328
lang = translation.get_language()
1301
tz = force_unicode(timezone.get_current_timezone_name(), errors='ignore')
1302
tz = tz.encode('ascii', 'ignore').replace(' ', '_')
1329
tz = force_text(timezone.get_current_timezone_name(), errors='ignore')
1330
tz = tz.encode('ascii', 'ignore').decode('ascii').replace(' ', '_')
1303
1331
response = HttpResponse()
1304
1332
key = learn_cache_key(request, response)
1305
1333
self.assertNotIn(lang, key, "Cache keys shouldn't include the language name when i18n isn't active")
1316
1344
request = self._get_request()
1317
1345
response = HttpResponse()
1318
1346
with timezone.override(CustomTzName()):
1319
CustomTzName.name = 'Hora estándar de Argentina' # UTF-8 string
1347
CustomTzName.name = 'Hora estándar de Argentina'.encode('UTF-8') # UTF-8 string
1320
1348
sanitized_name = 'Hora_estndar_de_Argentina'
1321
1349
self.assertIn(sanitized_name, learn_cache_key(request, response),
1322
1350
"Cache keys should include the time zone name when time zones are active")
1324
CustomTzName.name = u'Hora estándar de Argentina' # unicode
1352
CustomTzName.name = 'Hora estándar de Argentina' # unicode
1325
1353
sanitized_name = 'Hora_estndar_de_Argentina'
1326
1354
self.assertIn(sanitized_name, learn_cache_key(request, response),
1327
1355
"Cache keys should include the time zone name when time zones are active")
1352
1380
get_cache_data = FetchFromCacheMiddleware().process_request(request)
1353
1381
# cache must return content
1354
1382
self.assertNotEqual(get_cache_data, None)
1355
self.assertEqual(get_cache_data.content, content)
1383
self.assertEqual(get_cache_data.content, content.encode())
1356
1384
# different QUERY_STRING, cache must be empty
1357
1385
request = self._get_request_cache(query_string='foo=bar&somethingelse=true')
1358
1386
get_cache_data = FetchFromCacheMiddleware().process_request(request)
1367
1395
get_cache_data = FetchFromCacheMiddleware().process_request(request)
1368
1396
# Check that we can recover the cache
1369
1397
self.assertNotEqual(get_cache_data, None)
1370
self.assertEqual(get_cache_data.content, en_message)
1398
self.assertEqual(get_cache_data.content, en_message.encode())
1371
1399
# Check that we use etags
1372
1400
self.assertTrue(get_cache_data.has_header('ETag'))
1373
1401
# Check that we can disable etags
1383
1411
translation.activate('en')
1384
1412
# retrieve the content from cache
1385
1413
get_cache_data = FetchFromCacheMiddleware().process_request(request)
1386
self.assertEqual(get_cache_data.content, en_message)
1414
self.assertEqual(get_cache_data.content, en_message.encode())
1387
1415
# change again the language
1388
1416
translation.activate('es')
1389
1417
get_cache_data = FetchFromCacheMiddleware().process_request(request)
1390
self.assertEqual(get_cache_data.content, es_message)
1418
self.assertEqual(get_cache_data.content, es_message.encode())
1391
1419
# reset the language
1392
1420
translation.deactivate()
1394
CacheI18nTest = override_settings(
1395
CACHE_MIDDLEWARE_KEY_PREFIX='settingsprefix',
1398
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
1407
PrefixedCacheI18nTest = override_settings(
1423
CACHE_MIDDLEWARE_KEY_PREFIX="test",
1424
CACHE_MIDDLEWARE_SECONDS=60,
1427
def test_middleware_doesnt_cache_streaming_response(self):
1428
request = self._get_request()
1429
get_cache_data = FetchFromCacheMiddleware().process_request(request)
1430
self.assertIsNone(get_cache_data)
1432
# This test passes on Python < 3.3 even without the corresponding code
1433
# in UpdateCacheMiddleware, because pickling a StreamingHttpResponse
1434
# fails (http://bugs.python.org/issue14288). LocMemCache silently
1435
# swallows the exception and doesn't store the response in cache.
1436
content = ['Check for cache with streaming content.']
1437
response = StreamingHttpResponse(content)
1438
UpdateCacheMiddleware().process_response(request, response)
1440
get_cache_data = FetchFromCacheMiddleware().process_request(request)
1441
self.assertIsNone(get_cache_data)
1410
1446
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
1411
1447
'KEY_PREFIX': 'cacheprefix'
1451
class PrefixedCacheI18nTest(CacheI18nTest):
1417
1455
def hello_world_view(request, value):
1418
1456
return HttpResponse('Hello World %s' % value)
1460
CACHE_MIDDLEWARE_ALIAS='other',
1461
CACHE_MIDDLEWARE_KEY_PREFIX='middlewareprefix',
1462
CACHE_MIDDLEWARE_SECONDS=30,
1463
CACHE_MIDDLEWARE_ANONYMOUS_ONLY=False,
1466
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
1469
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
1470
'LOCATION': 'other',
1421
1475
class CacheMiddlewareTest(TestCase):
1477
# The following tests will need to be modified in Django 1.6 to not use
1478
# deprecated ways of using the cache_page decorator that will be removed in
1423
1480
def setUp(self):
1424
1481
self.factory = RequestFactory()
1425
1482
self.default_cache = get_cache('default')
1426
1483
self.other_cache = get_cache('other')
1484
self.save_warnings_state()
1485
warnings.filterwarnings('ignore', category=DeprecationWarning,
1486
module='django.views.decorators.cache')
1428
1488
def tearDown(self):
1489
self.restore_warnings_state()
1429
1490
self.default_cache.clear()
1430
1491
self.other_cache.clear()
1480
1541
# Repeating the request should result in a cache hit
1481
1542
result = middleware.process_request(request)
1482
self.assertNotEquals(result, None)
1483
self.assertEqual(result.content, 'Hello World 1')
1543
self.assertNotEqual(result, None)
1544
self.assertEqual(result.content, b'Hello World 1')
1485
1546
# The same request through a different middleware won't hit
1486
1547
result = prefix_middleware.process_request(request)
1489
1550
# The same request with a timeout _will_ hit
1490
1551
result = timeout_middleware.process_request(request)
1491
self.assertNotEquals(result, None)
1492
self.assertEqual(result.content, 'Hello World 1')
1552
self.assertNotEqual(result, None)
1553
self.assertEqual(result.content, b'Hello World 1')
1494
1555
@override_settings(CACHE_MIDDLEWARE_ANONYMOUS_ONLY=True)
1495
1556
def test_cache_middleware_anonymous_only_wont_cause_session_access(self):
1559
1620
# Request the view once
1560
1621
response = default_view(request, '1')
1561
self.assertEqual(response.content, 'Hello World 1')
1622
self.assertEqual(response.content, b'Hello World 1')
1563
1624
# Request again -- hit the cache
1564
1625
response = default_view(request, '2')
1565
self.assertEqual(response.content, 'Hello World 1')
1626
self.assertEqual(response.content, b'Hello World 1')
1567
1628
# Requesting the same view with the explicit cache should yield the same result
1568
1629
response = explicit_default_view(request, '3')
1569
self.assertEqual(response.content, 'Hello World 1')
1630
self.assertEqual(response.content, b'Hello World 1')
1571
1632
# Requesting with a prefix will hit a different cache key
1572
1633
response = explicit_default_with_prefix_view(request, '4')
1573
self.assertEqual(response.content, 'Hello World 4')
1634
self.assertEqual(response.content, b'Hello World 4')
1575
1636
# Hitting the same view again gives a cache hit
1576
1637
response = explicit_default_with_prefix_view(request, '5')
1577
self.assertEqual(response.content, 'Hello World 4')
1638
self.assertEqual(response.content, b'Hello World 4')
1579
1640
# And going back to the implicit cache will hit the same cache
1580
1641
response = default_with_prefix_view(request, '6')
1581
self.assertEqual(response.content, 'Hello World 4')
1642
self.assertEqual(response.content, b'Hello World 4')
1583
1644
# Requesting from an alternate cache won't hit cache
1584
1645
response = other_view(request, '7')
1585
self.assertEqual(response.content, 'Hello World 7')
1646
self.assertEqual(response.content, b'Hello World 7')
1587
1648
# But a repeated hit will hit cache
1588
1649
response = other_view(request, '8')
1589
self.assertEqual(response.content, 'Hello World 7')
1650
self.assertEqual(response.content, b'Hello World 7')
1591
1652
# And prefixing the alternate cache yields yet another cache entry
1592
1653
response = other_with_prefix_view(request, '9')
1593
self.assertEqual(response.content, 'Hello World 9')
1654
self.assertEqual(response.content, b'Hello World 9')
1595
1656
# Request from the alternate cache with a new prefix and a custom timeout
1596
1657
response = other_with_timeout_view(request, '10')
1597
self.assertEqual(response.content, 'Hello World 10')
1658
self.assertEqual(response.content, b'Hello World 10')
1599
1660
# But if we wait a couple of seconds...
1602
1663
# ... the default cache will still hit
1603
1664
cache = get_cache('default')
1604
1665
response = default_view(request, '11')
1605
self.assertEqual(response.content, 'Hello World 1')
1666
self.assertEqual(response.content, b'Hello World 1')
1607
1668
# ... the default cache with a prefix will still hit
1608
1669
response = default_with_prefix_view(request, '12')
1609
self.assertEqual(response.content, 'Hello World 4')
1670
self.assertEqual(response.content, b'Hello World 4')
1611
1672
# ... the explicit default cache will still hit
1612
1673
response = explicit_default_view(request, '13')
1613
self.assertEqual(response.content, 'Hello World 1')
1674
self.assertEqual(response.content, b'Hello World 1')
1615
1676
# ... the explicit default cache with a prefix will still hit
1616
1677
response = explicit_default_with_prefix_view(request, '14')
1617
self.assertEqual(response.content, 'Hello World 4')
1678
self.assertEqual(response.content, b'Hello World 4')
1619
1680
# .. but a rapidly expiring cache won't hit
1620
1681
response = other_view(request, '15')
1621
self.assertEqual(response.content, 'Hello World 15')
1682
self.assertEqual(response.content, b'Hello World 15')
1623
1684
# .. even if it has a prefix
1624
1685
response = other_with_prefix_view(request, '16')
1625
self.assertEqual(response.content, 'Hello World 16')
1686
self.assertEqual(response.content, b'Hello World 16')
1627
1688
# ... but a view with a custom timeout will still hit
1628
1689
response = other_with_timeout_view(request, '17')
1629
self.assertEqual(response.content, 'Hello World 10')
1690
self.assertEqual(response.content, b'Hello World 10')
1631
1692
# And if we wait a few more seconds
1634
# the custom timeouot cache will miss
1695
# the custom timeout cache will miss
1635
1696
response = other_with_timeout_view(request, '18')
1636
self.assertEqual(response.content, 'Hello World 18')
1638
CacheMiddlewareTest = override_settings(
1639
CACHE_MIDDLEWARE_ALIAS='other',
1640
CACHE_MIDDLEWARE_KEY_PREFIX='middlewareprefix',
1641
CACHE_MIDDLEWARE_SECONDS=30,
1642
CACHE_MIDDLEWARE_ANONYMOUS_ONLY=False,
1697
self.assertEqual(response.content, b'Hello World 18')
1701
CACHE_MIDDLEWARE_KEY_PREFIX='settingsprefix',
1702
CACHE_MIDDLEWARE_SECONDS=1,
1645
1705
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
1648
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
1649
'LOCATION': 'other',
1653
)(CacheMiddlewareTest)
1656
1710
class TestWithTemplateResponse(TestCase):
1658
1712
Tests various headers w/ TemplateResponse.
1740
1794
response = response.render()
1741
1795
self.assertTrue(response.has_header('ETag'))
1743
TestWithTemplateResponse = override_settings(
1744
CACHE_MIDDLEWARE_KEY_PREFIX='settingsprefix',
1745
CACHE_MIDDLEWARE_SECONDS=1,
1748
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
1752
)(TestWithTemplateResponse)
1755
1798
class TestEtagWithAdmin(TestCase):
1756
1799
# See https://code.djangoproject.com/ticket/16003