summaryrefslogtreecommitdiff
path: root/tests/cache
diff options
context:
space:
mode:
authordjango-bot <ops@djangoproject.com>2022-02-03 20:24:19 +0100
committerMariusz Felisiak <felisiak.mariusz@gmail.com>2022-02-07 20:37:05 +0100
commit9c19aff7c7561e3a82978a272ecdaad40dda5c00 (patch)
treef0506b668a013d0063e5fba3dbf4863b466713ba /tests/cache
parentf68fa8b45dfac545cfc4111d4e52804c86db68d3 (diff)
downloaddjango-9c19aff7c7561e3a82978a272ecdaad40dda5c00.tar.gz
Refs #33476 -- Reformatted code with Black.
Diffstat (limited to 'tests/cache')
-rw-r--r--tests/cache/models.py2
-rw-r--r--tests/cache/tests.py2041
-rw-r--r--tests/cache/tests_async.py149
3 files changed, 1195 insertions, 997 deletions
diff --git a/tests/cache/models.py b/tests/cache/models.py
index 349fd87473..e0aa6ab4df 100644
--- a/tests/cache/models.py
+++ b/tests/cache/models.py
@@ -10,4 +10,4 @@ def expensive_calculation():
class Poll(models.Model):
question = models.CharField(max_length=200)
answer = models.CharField(max_length=200)
- pub_date = models.DateTimeField('date published', default=expensive_calculation)
+ pub_date = models.DateTimeField("date published", default=expensive_calculation)
diff --git a/tests/cache/tests.py b/tests/cache/tests.py
index 6327698dfa..d010bc0708 100644
--- a/tests/cache/tests.py
+++ b/tests/cache/tests.py
@@ -17,7 +17,11 @@ from unittest import mock, skipIf
from django.conf import settings
from django.core import management, signals
from django.core.cache import (
- DEFAULT_CACHE_ALIAS, CacheHandler, CacheKeyWarning, InvalidCacheKey, cache,
+ DEFAULT_CACHE_ALIAS,
+ CacheHandler,
+ CacheKeyWarning,
+ InvalidCacheKey,
+ cache,
caches,
)
from django.core.cache.backends.base import InvalidCacheBackendError
@@ -26,24 +30,35 @@ from django.core.cache.utils import make_template_fragment_key
from django.db import close_old_connections, connection, connections
from django.db.backends.utils import CursorWrapper
from django.http import (
- HttpRequest, HttpResponse, HttpResponseNotModified, StreamingHttpResponse,
+ HttpRequest,
+ HttpResponse,
+ HttpResponseNotModified,
+ StreamingHttpResponse,
)
from django.middleware.cache import (
- CacheMiddleware, FetchFromCacheMiddleware, UpdateCacheMiddleware,
+ CacheMiddleware,
+ FetchFromCacheMiddleware,
+ UpdateCacheMiddleware,
)
from django.middleware.csrf import CsrfViewMiddleware
from django.template import engines
from django.template.context_processors import csrf
from django.template.response import TemplateResponse
from django.test import (
- RequestFactory, SimpleTestCase, TestCase, TransactionTestCase,
+ RequestFactory,
+ SimpleTestCase,
+ TestCase,
+ TransactionTestCase,
override_settings,
)
from django.test.signals import setting_changed
from django.test.utils import CaptureQueriesContext
from django.utils import timezone, translation
from django.utils.cache import (
- get_cache_key, learn_cache_key, patch_cache_control, patch_vary_headers,
+ get_cache_key,
+ learn_cache_key,
+ patch_cache_control,
+ patch_vary_headers,
)
from django.views.decorators.cache import cache_control, cache_page
@@ -70,15 +85,17 @@ def empty_response(request):
KEY_ERRORS_WITH_MEMCACHED_MSG = (
- 'Cache key contains characters that will cause errors if used with memcached: %r'
+ "Cache key contains characters that will cause errors if used with memcached: %r"
)
-@override_settings(CACHES={
- 'default': {
- 'BACKEND': 'django.core.cache.backends.dummy.DummyCache',
+@override_settings(
+ CACHES={
+ "default": {
+ "BACKEND": "django.core.cache.backends.dummy.DummyCache",
+ }
}
-})
+)
class DummyCacheTests(SimpleTestCase):
# The Dummy cache backend doesn't really behave like a test backend,
# so it has its own test case.
@@ -101,18 +118,18 @@ class DummyCacheTests(SimpleTestCase):
def test_get_many(self):
"get_many returns nothing for the dummy cache backend"
- cache.set_many({'a': 'a', 'b': 'b', 'c': 'c', 'd': 'd'})
- self.assertEqual(cache.get_many(['a', 'c', 'd']), {})
- self.assertEqual(cache.get_many(['a', 'b', 'e']), {})
+ cache.set_many({"a": "a", "b": "b", "c": "c", "d": "d"})
+ self.assertEqual(cache.get_many(["a", "c", "d"]), {})
+ self.assertEqual(cache.get_many(["a", "b", "e"]), {})
def test_get_many_invalid_key(self):
- msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ':1:key with spaces'
+ msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ":1:key with spaces"
with self.assertWarnsMessage(CacheKeyWarning, msg):
- cache.get_many(['key with spaces'])
+ cache.get_many(["key with spaces"])
def test_delete(self):
"Cache deletion is transparently ignored on the dummy cache backend"
- cache.set_many({'key1': 'spam', 'key2': 'eggs'})
+ cache.set_many({"key1": "spam", "key2": "eggs"})
self.assertIsNone(cache.get("key1"))
self.assertIs(cache.delete("key1"), False)
self.assertIsNone(cache.get("key1"))
@@ -132,39 +149,39 @@ class DummyCacheTests(SimpleTestCase):
def test_incr(self):
"Dummy cache values can't be incremented"
- cache.set('answer', 42)
+ cache.set("answer", 42)
with self.assertRaises(ValueError):
- cache.incr('answer')
+ cache.incr("answer")
with self.assertRaises(ValueError):
- cache.incr('does_not_exist')
+ cache.incr("does_not_exist")
with self.assertRaises(ValueError):
- cache.incr('does_not_exist', -1)
+ cache.incr("does_not_exist", -1)
def test_decr(self):
"Dummy cache values can't be decremented"
- cache.set('answer', 42)
+ cache.set("answer", 42)
with self.assertRaises(ValueError):
- cache.decr('answer')
+ cache.decr("answer")
with self.assertRaises(ValueError):
- cache.decr('does_not_exist')
+ cache.decr("does_not_exist")
with self.assertRaises(ValueError):
- cache.decr('does_not_exist', -1)
+ cache.decr("does_not_exist", -1)
def test_touch(self):
"""Dummy cache can't do touch()."""
- self.assertIs(cache.touch('whatever'), False)
+ self.assertIs(cache.touch("whatever"), False)
def test_data_types(self):
"All data types are ignored equally by the dummy cache"
tests = {
- 'string': 'this is a string',
- 'int': 42,
- 'bool': True,
- 'list': [1, 2, 3, 4],
- 'tuple': (1, 2, 3, 4),
- 'dict': {'A': 1, 'B': 2},
- 'function': f,
- 'class': C,
+ "string": "this is a string",
+ "int": 42,
+ "bool": True,
+ "list": [1, 2, 3, 4],
+ "tuple": (1, 2, 3, 4),
+ "dict": {"A": 1, "B": 2},
+ "function": f,
+ "class": C,
}
for key, value in tests.items():
with self.subTest(key=key):
@@ -173,9 +190,9 @@ class DummyCacheTests(SimpleTestCase):
def test_expiration(self):
"Expiration has no effect on the dummy cache"
- cache.set('expire1', 'very quickly', 1)
- cache.set('expire2', 'very quickly', 1)
- cache.set('expire3', 'very quickly', 1)
+ cache.set("expire1", "very quickly", 1)
+ cache.set("expire2", "very quickly", 1)
+ cache.set("expire3", "very quickly", 1)
time.sleep(2)
self.assertIsNone(cache.get("expire1"))
@@ -187,10 +204,10 @@ class DummyCacheTests(SimpleTestCase):
def test_unicode(self):
"Unicode values are ignored by the dummy cache"
stuff = {
- 'ascii': 'ascii_value',
- 'unicode_ascii': 'Iñtërnâtiônàlizætiøn1',
- 'Iñtërnâtiônàlizætiøn': 'Iñtërnâtiônàlizætiøn2',
- 'ascii2': {'x': 1}
+ "ascii": "ascii_value",
+ "unicode_ascii": "Iñtërnâtiônàlizætiøn1",
+ "Iñtërnâtiônàlizætiøn": "Iñtërnâtiônàlizætiøn2",
+ "ascii2": {"x": 1},
}
for (key, value) in stuff.items():
with self.subTest(key=key):
@@ -199,22 +216,22 @@ class DummyCacheTests(SimpleTestCase):
def test_set_many(self):
"set_many does nothing for the dummy cache backend"
- self.assertEqual(cache.set_many({'a': 1, 'b': 2}), [])
- self.assertEqual(cache.set_many({'a': 1, 'b': 2}, timeout=2, version='1'), [])
+ self.assertEqual(cache.set_many({"a": 1, "b": 2}), [])
+ self.assertEqual(cache.set_many({"a": 1, "b": 2}, timeout=2, version="1"), [])
def test_set_many_invalid_key(self):
- msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ':1:key with spaces'
+ msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ":1:key with spaces"
with self.assertWarnsMessage(CacheKeyWarning, msg):
- cache.set_many({'key with spaces': 'foo'})
+ cache.set_many({"key with spaces": "foo"})
def test_delete_many(self):
"delete_many does nothing for the dummy cache backend"
- cache.delete_many(['a', 'b'])
+ cache.delete_many(["a", "b"])
def test_delete_many_invalid_key(self):
- msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ':1:key with spaces'
+ msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ":1:key with spaces"
with self.assertWarnsMessage(CacheKeyWarning, msg):
- cache.delete_many(['key with spaces'])
+ cache.delete_many(["key with spaces"])
def test_clear(self):
"clear does nothing for the dummy cache backend"
@@ -222,45 +239,45 @@ class DummyCacheTests(SimpleTestCase):
def test_incr_version(self):
"Dummy cache versions can't be incremented"
- cache.set('answer', 42)
+ cache.set("answer", 42)
with self.assertRaises(ValueError):
- cache.incr_version('answer')
+ cache.incr_version("answer")
with self.assertRaises(ValueError):
- cache.incr_version('does_not_exist')
+ cache.incr_version("does_not_exist")
def test_decr_version(self):
"Dummy cache versions can't be decremented"
- cache.set('answer', 42)
+ cache.set("answer", 42)
with self.assertRaises(ValueError):
- cache.decr_version('answer')
+ cache.decr_version("answer")
with self.assertRaises(ValueError):
- cache.decr_version('does_not_exist')
+ cache.decr_version("does_not_exist")
def test_get_or_set(self):
- self.assertEqual(cache.get_or_set('mykey', 'default'), 'default')
- self.assertIsNone(cache.get_or_set('mykey', None))
+ self.assertEqual(cache.get_or_set("mykey", "default"), "default")
+ self.assertIsNone(cache.get_or_set("mykey", None))
def test_get_or_set_callable(self):
def my_callable():
- return 'default'
+ return "default"
- self.assertEqual(cache.get_or_set('mykey', my_callable), 'default')
- self.assertEqual(cache.get_or_set('mykey', my_callable()), 'default')
+ self.assertEqual(cache.get_or_set("mykey", my_callable), "default")
+ self.assertEqual(cache.get_or_set("mykey", my_callable()), "default")
def custom_key_func(key, key_prefix, version):
"A customized cache key function"
- return 'CUSTOM-' + '-'.join([key_prefix, str(version), key])
+ return "CUSTOM-" + "-".join([key_prefix, str(version), key])
_caches_setting_base = {
- 'default': {},
- 'prefix': {'KEY_PREFIX': 'cacheprefix{}'.format(os.getpid())},
- 'v2': {'VERSION': 2},
- 'custom_key': {'KEY_FUNCTION': custom_key_func},
- 'custom_key2': {'KEY_FUNCTION': 'cache.tests.custom_key_func'},
- 'cull': {'OPTIONS': {'MAX_ENTRIES': 30}},
- 'zero_cull': {'OPTIONS': {'CULL_FREQUENCY': 0, 'MAX_ENTRIES': 30}},
+ "default": {},
+ "prefix": {"KEY_PREFIX": "cacheprefix{}".format(os.getpid())},
+ "v2": {"VERSION": 2},
+ "custom_key": {"KEY_FUNCTION": custom_key_func},
+ "custom_key2": {"KEY_FUNCTION": "cache.tests.custom_key_func"},
+ "cull": {"OPTIONS": {"MAX_ENTRIES": 30}},
+ "zero_cull": {"OPTIONS": {"CULL_FREQUENCY": 0, "MAX_ENTRIES": 30}},
}
@@ -299,8 +316,8 @@ class BaseCacheTests:
def test_default_used_when_none_is_set(self):
"""If None is cached, get() returns it instead of the default."""
- cache.set('key_default_none', None)
- self.assertIsNone(cache.get('key_default_none', default='default'))
+ cache.set("key_default_none", None)
+ self.assertIsNone(cache.get("key_default_none", default="default"))
def test_add(self):
# A key can be added to a cache
@@ -310,15 +327,15 @@ class BaseCacheTests:
def test_prefix(self):
# Test for same cache key conflicts between shared backend
- cache.set('somekey', 'value')
+ cache.set("somekey", "value")
# should not be set in the prefixed cache
- self.assertIs(caches['prefix'].has_key('somekey'), False)
+ self.assertIs(caches["prefix"].has_key("somekey"), False)
- caches['prefix'].set('somekey', 'value2')
+ caches["prefix"].set("somekey", "value2")
- self.assertEqual(cache.get('somekey'), 'value')
- self.assertEqual(caches['prefix'].get('somekey'), 'value2')
+ self.assertEqual(cache.get("somekey"), "value")
+ self.assertEqual(caches["prefix"].get("somekey"), "value2")
def test_non_existent(self):
"""Nonexistent cache keys return as None/default."""
@@ -327,23 +344,25 @@ class BaseCacheTests:
def test_get_many(self):
# Multiple cache keys can be returned using get_many
- cache.set_many({'a': 'a', 'b': 'b', 'c': 'c', 'd': 'd'})
- self.assertEqual(cache.get_many(['a', 'c', 'd']), {'a': 'a', 'c': 'c', 'd': 'd'})
- self.assertEqual(cache.get_many(['a', 'b', 'e']), {'a': 'a', 'b': 'b'})
- self.assertEqual(cache.get_many(iter(['a', 'b', 'e'])), {'a': 'a', 'b': 'b'})
- cache.set_many({'x': None, 'y': 1})
- self.assertEqual(cache.get_many(['x', 'y']), {'x': None, 'y': 1})
+ cache.set_many({"a": "a", "b": "b", "c": "c", "d": "d"})
+ self.assertEqual(
+ cache.get_many(["a", "c", "d"]), {"a": "a", "c": "c", "d": "d"}
+ )
+ self.assertEqual(cache.get_many(["a", "b", "e"]), {"a": "a", "b": "b"})
+ self.assertEqual(cache.get_many(iter(["a", "b", "e"])), {"a": "a", "b": "b"})
+ cache.set_many({"x": None, "y": 1})
+ self.assertEqual(cache.get_many(["x", "y"]), {"x": None, "y": 1})
def test_delete(self):
# Cache keys can be deleted
- cache.set_many({'key1': 'spam', 'key2': 'eggs'})
+ cache.set_many({"key1": "spam", "key2": "eggs"})
self.assertEqual(cache.get("key1"), "spam")
self.assertIs(cache.delete("key1"), True)
self.assertIsNone(cache.get("key1"))
self.assertEqual(cache.get("key2"), "eggs")
def test_delete_nonexistent(self):
- self.assertIs(cache.delete('nonexistent_key'), False)
+ self.assertIs(cache.delete("nonexistent_key"), False)
def test_has_key(self):
# The cache can be inspected for cache keys
@@ -352,64 +371,64 @@ class BaseCacheTests:
self.assertIs(cache.has_key("goodbye1"), False)
cache.set("no_expiry", "here", None)
self.assertIs(cache.has_key("no_expiry"), True)
- cache.set('null', None)
- self.assertIs(cache.has_key('null'), True)
+ cache.set("null", None)
+ self.assertIs(cache.has_key("null"), True)
def test_in(self):
# The in operator can be used to inspect cache contents
cache.set("hello2", "goodbye2")
self.assertIn("hello2", cache)
self.assertNotIn("goodbye2", cache)
- cache.set('null', None)
- self.assertIn('null', cache)
+ cache.set("null", None)
+ self.assertIn("null", cache)
def test_incr(self):
# Cache values can be incremented
- cache.set('answer', 41)
- self.assertEqual(cache.incr('answer'), 42)
- self.assertEqual(cache.get('answer'), 42)
- self.assertEqual(cache.incr('answer', 10), 52)
- self.assertEqual(cache.get('answer'), 52)
- self.assertEqual(cache.incr('answer', -10), 42)
+ cache.set("answer", 41)
+ self.assertEqual(cache.incr("answer"), 42)
+ self.assertEqual(cache.get("answer"), 42)
+ self.assertEqual(cache.incr("answer", 10), 52)
+ self.assertEqual(cache.get("answer"), 52)
+ self.assertEqual(cache.incr("answer", -10), 42)
with self.assertRaises(ValueError):
- cache.incr('does_not_exist')
+ cache.incr("does_not_exist")
with self.assertRaises(ValueError):
- cache.incr('does_not_exist', -1)
- cache.set('null', None)
+ cache.incr("does_not_exist", -1)
+ cache.set("null", None)
with self.assertRaises(self.incr_decr_type_error):
- cache.incr('null')
+ cache.incr("null")
def test_decr(self):
# Cache values can be decremented
- cache.set('answer', 43)
- self.assertEqual(cache.decr('answer'), 42)
- self.assertEqual(cache.get('answer'), 42)
- self.assertEqual(cache.decr('answer', 10), 32)
- self.assertEqual(cache.get('answer'), 32)
- self.assertEqual(cache.decr('answer', -10), 42)
+ cache.set("answer", 43)
+ self.assertEqual(cache.decr("answer"), 42)
+ self.assertEqual(cache.get("answer"), 42)
+ self.assertEqual(cache.decr("answer", 10), 32)
+ self.assertEqual(cache.get("answer"), 32)
+ self.assertEqual(cache.decr("answer", -10), 42)
with self.assertRaises(ValueError):
- cache.decr('does_not_exist')
+ cache.decr("does_not_exist")
with self.assertRaises(ValueError):
- cache.incr('does_not_exist', -1)
- cache.set('null', None)
+ cache.incr("does_not_exist", -1)
+ cache.set("null", None)
with self.assertRaises(self.incr_decr_type_error):
- cache.decr('null')
+ cache.decr("null")
def test_close(self):
- self.assertTrue(hasattr(cache, 'close'))
+ self.assertTrue(hasattr(cache, "close"))
cache.close()
def test_data_types(self):
# Many different data types can be cached
tests = {
- 'string': 'this is a string',
- 'int': 42,
- 'bool': True,
- 'list': [1, 2, 3, 4],
- 'tuple': (1, 2, 3, 4),
- 'dict': {'A': 1, 'B': 2},
- 'function': f,
- 'class': C,
+ "string": "this is a string",
+ "int": 42,
+ "bool": True,
+ "list": [1, 2, 3, 4],
+ "tuple": (1, 2, 3, 4),
+ "dict": {"A": 1, "B": 2},
+ "function": f,
+ "class": C,
}
for key, value in tests.items():
with self.subTest(key=key):
@@ -423,8 +442,8 @@ class BaseCacheTests:
my_poll = Poll.objects.create(question="Well?")
self.assertEqual(Poll.objects.count(), 1)
pub_date = my_poll.pub_date
- cache.set('question', my_poll)
- cached_poll = cache.get('question')
+ cache.set("question", my_poll)
+ cached_poll = cache.get("question")
self.assertEqual(cached_poll.pub_date, pub_date)
# We only want the default expensive calculation run once
self.assertEqual(expensive_calculation.num_runs, 1)
@@ -435,10 +454,10 @@ class BaseCacheTests:
Poll.objects.all().delete()
Poll.objects.create(question="What?")
self.assertEqual(expensive_calculation.num_runs, 1)
- defer_qs = Poll.objects.all().defer('question')
+ defer_qs = Poll.objects.all().defer("question")
self.assertEqual(defer_qs.count(), 1)
self.assertEqual(expensive_calculation.num_runs, 1)
- cache.set('deferred_queryset', defer_qs)
+ cache.set("deferred_queryset", defer_qs)
# cache set should not re-evaluate default functions
self.assertEqual(expensive_calculation.num_runs, 1)
@@ -448,20 +467,20 @@ class BaseCacheTests:
Poll.objects.all().delete()
Poll.objects.create(question="What?")
self.assertEqual(expensive_calculation.num_runs, 1)
- defer_qs = Poll.objects.all().defer('question')
+ defer_qs = Poll.objects.all().defer("question")
self.assertEqual(defer_qs.count(), 1)
- cache.set('deferred_queryset', defer_qs)
+ cache.set("deferred_queryset", defer_qs)
self.assertEqual(expensive_calculation.num_runs, 1)
runs_before_cache_read = expensive_calculation.num_runs
- cache.get('deferred_queryset')
+ cache.get("deferred_queryset")
# We only want the default expensive calculation run on creation and set
self.assertEqual(expensive_calculation.num_runs, runs_before_cache_read)
def test_expiration(self):
# Cache values can be set to expire
- cache.set('expire1', 'very quickly', 1)
- cache.set('expire2', 'very quickly', 1)
- cache.set('expire3', 'very quickly', 1)
+ cache.set("expire1", "very quickly", 1)
+ cache.set("expire2", "very quickly", 1)
+ cache.set("expire3", "very quickly", 1)
time.sleep(2)
self.assertIsNone(cache.get("expire1"))
@@ -472,27 +491,27 @@ class BaseCacheTests:
def test_touch(self):
# cache.touch() updates the timeout.
- cache.set('expire1', 'very quickly', timeout=1)
- self.assertIs(cache.touch('expire1', timeout=4), True)
+ cache.set("expire1", "very quickly", timeout=1)
+ self.assertIs(cache.touch("expire1", timeout=4), True)
time.sleep(2)
- self.assertIs(cache.has_key('expire1'), True)
+ self.assertIs(cache.has_key("expire1"), True)
time.sleep(3)
- self.assertIs(cache.has_key('expire1'), False)
+ self.assertIs(cache.has_key("expire1"), False)
# cache.touch() works without the timeout argument.
- cache.set('expire1', 'very quickly', timeout=1)
- self.assertIs(cache.touch('expire1'), True)
+ cache.set("expire1", "very quickly", timeout=1)
+ self.assertIs(cache.touch("expire1"), True)
time.sleep(2)
- self.assertIs(cache.has_key('expire1'), True)
+ self.assertIs(cache.has_key("expire1"), True)
- self.assertIs(cache.touch('nonexistent'), False)
+ self.assertIs(cache.touch("nonexistent"), False)
def test_unicode(self):
# Unicode values can be cached
stuff = {
- 'ascii': 'ascii_value',
- 'unicode_ascii': 'Iñtërnâtiônàlizætiøn1',
- 'Iñtërnâtiônàlizætiøn': 'Iñtërnâtiônàlizætiøn2',
- 'ascii2': {'x': 1}
+ "ascii": "ascii_value",
+ "unicode_ascii": "Iñtërnâtiônàlizætiøn1",
+ "Iñtërnâtiônàlizætiøn": "Iñtërnâtiônàlizætiøn2",
+ "ascii2": {"x": 1},
}
# Test `set`
for (key, value) in stuff.items():
@@ -518,24 +537,25 @@ class BaseCacheTests:
def test_binary_string(self):
# Binary strings should be cacheable
from zlib import compress, decompress
- value = 'value_to_be_compressed'
+
+ value = "value_to_be_compressed"
compressed_value = compress(value.encode())
# Test set
- cache.set('binary1', compressed_value)
- compressed_result = cache.get('binary1')
+ cache.set("binary1", compressed_value)
+ compressed_result = cache.get("binary1")
self.assertEqual(compressed_value, compressed_result)
self.assertEqual(value, decompress(compressed_result).decode())
# Test add
- self.assertIs(cache.add('binary1-add', compressed_value), True)
- compressed_result = cache.get('binary1-add')
+ self.assertIs(cache.add("binary1-add", compressed_value), True)
+ compressed_result = cache.get("binary1-add")
self.assertEqual(compressed_value, compressed_result)
self.assertEqual(value, decompress(compressed_result).decode())
# Test set_many
- cache.set_many({'binary1-set_many': compressed_value})
- compressed_result = cache.get('binary1-set_many')
+ cache.set_many({"binary1-set_many": compressed_value})
+ compressed_result = cache.get("binary1-set_many")
self.assertEqual(compressed_value, compressed_result)
self.assertEqual(value, decompress(compressed_result).decode())
@@ -547,7 +567,7 @@ class BaseCacheTests:
def test_set_many_returns_empty_list_on_success(self):
"""set_many() returns an empty list when all keys are inserted."""
- failing_keys = cache.set_many({'key1': 'spam', 'key2': 'eggs'})
+ failing_keys = cache.set_many({"key1": "spam", "key2": "eggs"})
self.assertEqual(failing_keys, [])
def test_set_many_expiration(self):
@@ -559,7 +579,7 @@ class BaseCacheTests:
def test_delete_many(self):
# Multiple keys can be deleted using delete_many
- cache.set_many({'key1': 'spam', 'key2': 'eggs', 'key3': 'ham'})
+ cache.set_many({"key1": "spam", "key2": "eggs", "key3": "ham"})
cache.delete_many(["key1", "key2"])
self.assertIsNone(cache.get("key1"))
self.assertIsNone(cache.get("key2"))
@@ -567,7 +587,7 @@ class BaseCacheTests:
def test_clear(self):
# The cache can be emptied using clear
- cache.set_many({'key1': 'spam', 'key2': 'eggs'})
+ cache.set_many({"key1": "spam", "key2": "eggs"})
cache.clear()
self.assertIsNone(cache.get("key1"))
self.assertIsNone(cache.get("key2"))
@@ -578,54 +598,56 @@ class BaseCacheTests:
treated as an absolute expiration timestamp instead of a relative
offset (#12399).
"""
- cache.set('key1', 'eggs', 60 * 60 * 24 * 30 + 1) # 30 days + 1 second
- self.assertEqual(cache.get('key1'), 'eggs')
+ cache.set("key1", "eggs", 60 * 60 * 24 * 30 + 1) # 30 days + 1 second
+ self.assertEqual(cache.get("key1"), "eggs")
- self.assertIs(cache.add('key2', 'ham', 60 * 60 * 24 * 30 + 1), True)
- self.assertEqual(cache.get('key2'), 'ham')
+ self.assertIs(cache.add("key2", "ham", 60 * 60 * 24 * 30 + 1), True)
+ self.assertEqual(cache.get("key2"), "ham")
- cache.set_many({'key3': 'sausage', 'key4': 'lobster bisque'}, 60 * 60 * 24 * 30 + 1)
- self.assertEqual(cache.get('key3'), 'sausage')
- self.assertEqual(cache.get('key4'), 'lobster bisque')
+ cache.set_many(
+ {"key3": "sausage", "key4": "lobster bisque"}, 60 * 60 * 24 * 30 + 1
+ )
+ self.assertEqual(cache.get("key3"), "sausage")
+ self.assertEqual(cache.get("key4"), "lobster bisque")
def test_forever_timeout(self):
"""
Passing in None into timeout results in a value that is cached forever
"""
- cache.set('key1', 'eggs', None)
- self.assertEqual(cache.get('key1'), 'eggs')
+ cache.set("key1", "eggs", None)
+ self.assertEqual(cache.get("key1"), "eggs")
- self.assertIs(cache.add('key2', 'ham', None), True)
- self.assertEqual(cache.get('key2'), 'ham')
- self.assertIs(cache.add('key1', 'new eggs', None), False)
- self.assertEqual(cache.get('key1'), 'eggs')
+ self.assertIs(cache.add("key2", "ham", None), True)
+ self.assertEqual(cache.get("key2"), "ham")
+ self.assertIs(cache.add("key1", "new eggs", None), False)
+ self.assertEqual(cache.get("key1"), "eggs")
- cache.set_many({'key3': 'sausage', 'key4': 'lobster bisque'}, None)
- self.assertEqual(cache.get('key3'), 'sausage')
- self.assertEqual(cache.get('key4'), 'lobster bisque')
+ cache.set_many({"key3": "sausage", "key4": "lobster bisque"}, None)
+ self.assertEqual(cache.get("key3"), "sausage")
+ self.assertEqual(cache.get("key4"), "lobster bisque")
- cache.set('key5', 'belgian fries', timeout=1)
- self.assertIs(cache.touch('key5', timeout=None), True)
+ cache.set("key5", "belgian fries", timeout=1)
+ self.assertIs(cache.touch("key5", timeout=None), True)
time.sleep(2)
- self.assertEqual(cache.get('key5'), 'belgian fries')
+ self.assertEqual(cache.get("key5"), "belgian fries")
def test_zero_timeout(self):
"""
Passing in zero into timeout results in a value that is not cached
"""
- cache.set('key1', 'eggs', 0)
- self.assertIsNone(cache.get('key1'))
+ cache.set("key1", "eggs", 0)
+ self.assertIsNone(cache.get("key1"))
- self.assertIs(cache.add('key2', 'ham', 0), True)
- self.assertIsNone(cache.get('key2'))
+ self.assertIs(cache.add("key2", "ham", 0), True)
+ self.assertIsNone(cache.get("key2"))
- cache.set_many({'key3': 'sausage', 'key4': 'lobster bisque'}, 0)
- self.assertIsNone(cache.get('key3'))
- self.assertIsNone(cache.get('key4'))
+ cache.set_many({"key3": "sausage", "key4": "lobster bisque"}, 0)
+ self.assertIsNone(cache.get("key3"))
+ self.assertIsNone(cache.get("key4"))
- cache.set('key5', 'belgian fries', timeout=5)
- self.assertIs(cache.touch('key5', timeout=0), True)
- self.assertIsNone(cache.get('key5'))
+ cache.set("key5", "belgian fries", timeout=5)
+ self.assertIs(cache.touch("key5", timeout=0), True)
+ self.assertIsNone(cache.get("key5"))
def test_float_timeout(self):
# Make sure a timeout given as a float doesn't crash anything.
@@ -641,31 +663,31 @@ class BaseCacheTests:
# Create initial cache key entries. This will overflow the cache,
# causing a cull.
for i in range(1, initial_count):
- cull_cache.set('cull%d' % i, 'value', 1000)
+ cull_cache.set("cull%d" % i, "value", 1000)
count = 0
# Count how many keys are left in the cache.
for i in range(1, initial_count):
- if cull_cache.has_key('cull%d' % i):
+ if cull_cache.has_key("cull%d" % i):
count += 1
self.assertEqual(count, final_count)
def test_cull(self):
- self._perform_cull_test('cull', 50, 29)
+ self._perform_cull_test("cull", 50, 29)
def test_zero_cull(self):
- self._perform_cull_test('zero_cull', 50, 19)
+ self._perform_cull_test("zero_cull", 50, 19)
def test_cull_delete_when_store_empty(self):
try:
- cull_cache = caches['cull']
+ cull_cache = caches["cull"]
except InvalidCacheBackendError:
self.skipTest("Culling isn't implemented.")
old_max_entries = cull_cache._max_entries
# Force _cull to delete on first cached record.
cull_cache._max_entries = -1
try:
- cull_cache.set('force_cull_delete', 'value', 1000)
- self.assertIs(cull_cache.has_key('force_cull_delete'), True)
+ cull_cache.set("force_cull_delete", "value", 1000)
+ self.assertIs(cull_cache.has_key("force_cull_delete"), True)
finally:
cull_cache._max_entries = old_max_entries
@@ -685,16 +707,16 @@ class BaseCacheTests:
cache.key_func = key_func or func
tests = [
- ('add', [key, 1]),
- ('get', [key]),
- ('set', [key, 1]),
- ('incr', [key]),
- ('decr', [key]),
- ('touch', [key]),
- ('delete', [key]),
- ('get_many', [[key, 'b']]),
- ('set_many', [{key: 1, 'b': 2}]),
- ('delete_many', [[key, 'b']]),
+ ("add", [key, 1]),
+ ("get", [key]),
+ ("set", [key, 1]),
+ ("incr", [key]),
+ ("decr", [key]),
+ ("touch", [key]),
+ ("delete", [key]),
+ ("get_many", [[key, "b"]]),
+ ("set_many", [{key: 1, "b": 2}]),
+ ("delete_many", [[key, "b"]]),
]
try:
for operation, args in tests:
@@ -707,15 +729,15 @@ class BaseCacheTests:
def test_invalid_key_characters(self):
# memcached doesn't allow whitespace or control characters in keys.
- key = 'key with spaces and 清'
+ key = "key with spaces and 清"
self._perform_invalid_key_test(key, KEY_ERRORS_WITH_MEMCACHED_MSG % key)
def test_invalid_key_length(self):
# memcached limits key length to 250.
- key = ('a' * 250) + '清'
+ key = ("a" * 250) + "清"
expected_warning = (
- 'Cache key will cause errors if used with memcached: '
- '%r (longer than %s)' % (key, 250)
+ "Cache key will cause errors if used with memcached: "
+ "%r (longer than %s)" % (key, 250)
)
self._perform_invalid_key_test(key, expected_warning)
@@ -723,296 +745,330 @@ class BaseCacheTests:
# Custom make_key() that adds a version to the key and exceeds the
# limit.
def key_func(key, *args):
- return key + ':1'
+ return key + ":1"
- key = 'a' * 249
+ key = "a" * 249
expected_warning = (
- 'Cache key will cause errors if used with memcached: '
- '%r (longer than %s)' % (key_func(key), 250)
+ "Cache key will cause errors if used with memcached: "
+ "%r (longer than %s)" % (key_func(key), 250)
)
self._perform_invalid_key_test(key, expected_warning, key_func=key_func)
def test_cache_versioning_get_set(self):
# set, using default version = 1
- cache.set('answer1', 42)
- self.assertEqual(cache.get('answer1'), 42)
- self.assertEqual(cache.get('answer1', version=1), 42)
- self.assertIsNone(cache.get('answer1', version=2))
+ cache.set("answer1", 42)
+ self.assertEqual(cache.get("answer1"), 42)
+ self.assertEqual(cache.get("answer1", version=1), 42)
+ self.assertIsNone(cache.get("answer1", version=2))
- self.assertIsNone(caches['v2'].get('answer1'))
- self.assertEqual(caches['v2'].get('answer1', version=1), 42)
- self.assertIsNone(caches['v2'].get('answer1', version=2))
+ self.assertIsNone(caches["v2"].get("answer1"))
+ self.assertEqual(caches["v2"].get("answer1", version=1), 42)
+ self.assertIsNone(caches["v2"].get("answer1", version=2))
# set, default version = 1, but manually override version = 2
- cache.set('answer2', 42, version=2)
- self.assertIsNone(cache.get('answer2'))
- self.assertIsNone(cache.get('answer2', version=1))
- self.assertEqual(cache.get('answer2', version=2), 42)
+ cache.set("answer2", 42, version=2)
+ self.assertIsNone(cache.get("answer2"))
+ self.assertIsNone(cache.get("answer2", version=1))
+ self.assertEqual(cache.get("answer2", version=2), 42)
- self.assertEqual(caches['v2'].get('answer2'), 42)
- self.assertIsNone(caches['v2'].get('answer2', version=1))
- self.assertEqual(caches['v2'].get('answer2', version=2), 42)
+ self.assertEqual(caches["v2"].get("answer2"), 42)
+ self.assertIsNone(caches["v2"].get("answer2", version=1))
+ self.assertEqual(caches["v2"].get("answer2", version=2), 42)
# v2 set, using default version = 2
- caches['v2'].set('answer3', 42)
- self.assertIsNone(cache.get('answer3'))
- self.assertIsNone(cache.get('answer3', version=1))
- self.assertEqual(cache.get('answer3', version=2), 42)
+ caches["v2"].set("answer3", 42)
+ self.assertIsNone(cache.get("answer3"))
+ self.assertIsNone(cache.get("answer3", version=1))
+ self.assertEqual(cache.get("answer3", version=2), 42)
- self.assertEqual(caches['v2'].get('answer3'), 42)
- self.assertIsNone(caches['v2'].get('answer3', version=1))
- self.assertEqual(caches['v2'].get('answer3', version=2), 42)
+ self.assertEqual(caches["v2"].get("answer3"), 42)
+ self.assertIsNone(caches["v2"].get("answer3", version=1))
+ self.assertEqual(caches["v2"].get("answer3", version=2), 42)
# v2 set, default version = 2, but manually override version = 1
- caches['v2'].set('answer4', 42, version=1)
- self.assertEqual(cache.get('answer4'), 42)
- self.assertEqual(cache.get('answer4', version=1), 42)
- self.assertIsNone(cache.get('answer4', version=2))
+ caches["v2"].set("answer4", 42, version=1)
+ self.assertEqual(cache.get("answer4"), 42)
+ self.assertEqual(cache.get("answer4", version=1), 42)
+ self.assertIsNone(cache.get("answer4", version=2))
- self.assertIsNone(caches['v2'].get('answer4'))
- self.assertEqual(caches['v2'].get('answer4', version=1), 42)
- self.assertIsNone(caches['v2'].get('answer4', version=2))
+ self.assertIsNone(caches["v2"].get("answer4"))
+ self.assertEqual(caches["v2"].get("answer4", version=1), 42)
+ self.assertIsNone(caches["v2"].get("answer4", version=2))
def test_cache_versioning_add(self):
# add, default version = 1, but manually override version = 2
- self.assertIs(cache.add('answer1', 42, version=2), True)
- self.assertIsNone(cache.get('answer1', version=1))
- self.assertEqual(cache.get('answer1', version=2), 42)
+ self.assertIs(cache.add("answer1", 42, version=2), True)
+ self.assertIsNone(cache.get("answer1", version=1))
+ self.assertEqual(cache.get("answer1", version=2), 42)
- self.assertIs(cache.add('answer1', 37, version=2), False)
- self.assertIsNone(cache.get('answer1', version=1))
- self.assertEqual(cache.get('answer1', version=2), 42)
+ self.assertIs(cache.add("answer1", 37, version=2), False)
+ self.assertIsNone(cache.get("answer1", version=1))
+ self.assertEqual(cache.get("answer1", version=2), 42)
- self.assertIs(cache.add('answer1', 37, version=1), True)
- self.assertEqual(cache.get('answer1', version=1), 37)
- self.assertEqual(cache.get('answer1', version=2), 42)
+ self.assertIs(cache.add("answer1", 37, version=1), True)
+ self.assertEqual(cache.get("answer1", version=1), 37)
+ self.assertEqual(cache.get("answer1", version=2), 42)
# v2 add, using default version = 2
- self.assertIs(caches['v2'].add('answer2', 42), True)
- self.assertIsNone(cache.get('answer2', version=1))
- self.assertEqual(cache.get('answer2', version=2), 42)
+ self.assertIs(caches["v2"].add("answer2", 42), True)
+ self.assertIsNone(cache.get("answer2", version=1))
+ self.assertEqual(cache.get("answer2", version=2), 42)
- self.assertIs(caches['v2'].add('answer2', 37), False)
- self.assertIsNone(cache.get('answer2', version=1))
- self.assertEqual(cache.get('answer2', version=2), 42)
+ self.assertIs(caches["v2"].add("answer2", 37), False)
+ self.assertIsNone(cache.get("answer2", version=1))
+ self.assertEqual(cache.get("answer2", version=2), 42)
- self.assertIs(caches['v2'].add('answer2', 37, version=1), True)
- self.assertEqual(cache.get('answer2', version=1), 37)
- self.assertEqual(cache.get('answer2', version=2), 42)
+ self.assertIs(caches["v2"].add("answer2", 37, version=1), True)
+ self.assertEqual(cache.get("answer2", version=1), 37)
+ self.assertEqual(cache.get("answer2", version=2), 42)
# v2 add, default version = 2, but manually override version = 1
- self.assertIs(caches['v2'].add('answer3', 42, version=1), True)
- self.assertEqual(cache.get('answer3', version=1), 42)
- self.assertIsNone(cache.get('answer3', version=2))
+ self.assertIs(caches["v2"].add("answer3", 42, version=1), True)
+ self.assertEqual(cache.get("answer3", version=1), 42)
+ self.assertIsNone(cache.get("answer3", version=2))
- self.assertIs(caches['v2'].add('answer3', 37, version=1), False)
- self.assertEqual(cache.get('answer3', version=1), 42)
- self.assertIsNone(cache.get('answer3', version=2))
+ self.assertIs(caches["v2"].add("answer3", 37, version=1), False)
+ self.assertEqual(cache.get("answer3", version=1), 42)
+ self.assertIsNone(cache.get("answer3", version=2))
- self.assertIs(caches['v2'].add('answer3', 37), True)
- self.assertEqual(cache.get('answer3', version=1), 42)
- self.assertEqual(cache.get('answer3', version=2), 37)
+ self.assertIs(caches["v2"].add("answer3", 37), True)
+ self.assertEqual(cache.get("answer3", version=1), 42)
+ self.assertEqual(cache.get("answer3", version=2), 37)
def test_cache_versioning_has_key(self):
- cache.set('answer1', 42)
+ cache.set("answer1", 42)
# has_key
- self.assertIs(cache.has_key('answer1'), True)
- self.assertIs(cache.has_key('answer1', version=1), True)
- self.assertIs(cache.has_key('answer1', version=2), False)
+ self.assertIs(cache.has_key("answer1"), True)
+ self.assertIs(cache.has_key("answer1", version=1), True)
+ self.assertIs(cache.has_key("answer1", version=2), False)
- self.assertIs(caches['v2'].has_key('answer1'), False)
- self.assertIs(caches['v2'].has_key('answer1', version=1), True)
- self.assertIs(caches['v2'].has_key('answer1', version=2), False)
+ self.assertIs(caches["v2"].has_key("answer1"), False)
+ self.assertIs(caches["v2"].has_key("answer1", version=1), True)
+ self.assertIs(caches["v2"].has_key("answer1", version=2), False)
def test_cache_versioning_delete(self):
- cache.set('answer1', 37, version=1)
- cache.set('answer1', 42, version=2)
- self.assertIs(cache.delete('answer1'), True)
- self.assertIsNone(cache.get('answer1', version=1))
- self.assertEqual(cache.get('answer1', version=2), 42)
-
- cache.set('answer2', 37, version=1)
- cache.set('answer2', 42, version=2)
- self.assertIs(cache.delete('answer2', version=2), True)
- self.assertEqual(cache.get('answer2', version=1), 37)
- self.assertIsNone(cache.get('answer2', version=2))
-
- cache.set('answer3', 37, version=1)
- cache.set('answer3', 42, version=2)
- self.assertIs(caches['v2'].delete('answer3'), True)
- self.assertEqual(cache.get('answer3', version=1), 37)
- self.assertIsNone(cache.get('answer3', version=2))
-
- cache.set('answer4', 37, version=1)
- cache.set('answer4', 42, version=2)
- self.assertIs(caches['v2'].delete('answer4', version=1), True)
- self.assertIsNone(cache.get('answer4', version=1))
- self.assertEqual(cache.get('answer4', version=2), 42)
+ cache.set("answer1", 37, version=1)
+ cache.set("answer1", 42, version=2)
+ self.assertIs(cache.delete("answer1"), True)
+ self.assertIsNone(cache.get("answer1", version=1))
+ self.assertEqual(cache.get("answer1", version=2), 42)
+
+ cache.set("answer2", 37, version=1)
+ cache.set("answer2", 42, version=2)
+ self.assertIs(cache.delete("answer2", version=2), True)
+ self.assertEqual(cache.get("answer2", version=1), 37)
+ self.assertIsNone(cache.get("answer2", version=2))
+
+ cache.set("answer3", 37, version=1)
+ cache.set("answer3", 42, version=2)
+ self.assertIs(caches["v2"].delete("answer3"), True)
+ self.assertEqual(cache.get("answer3", version=1), 37)
+ self.assertIsNone(cache.get("answer3", version=2))
+
+ cache.set("answer4", 37, version=1)
+ cache.set("answer4", 42, version=2)
+ self.assertIs(caches["v2"].delete("answer4", version=1), True)
+ self.assertIsNone(cache.get("answer4", version=1))
+ self.assertEqual(cache.get("answer4", version=2), 42)
def test_cache_versioning_incr_decr(self):
- cache.set('answer1', 37, version=1)
- cache.set('answer1', 42, version=2)
- self.assertEqual(cache.incr('answer1'), 38)
- self.assertEqual(cache.get('answer1', version=1), 38)
- self.assertEqual(cache.get('answer1', version=2), 42)
- self.assertEqual(cache.decr('answer1'), 37)
- self.assertEqual(cache.get('answer1', version=1), 37)
- self.assertEqual(cache.get('answer1', version=2), 42)
-
- cache.set('answer2', 37, version=1)
- cache.set('answer2', 42, version=2)
- self.assertEqual(cache.incr('answer2', version=2), 43)
- self.assertEqual(cache.get('answer2', version=1), 37)
- self.assertEqual(cache.get('answer2', version=2), 43)
- self.assertEqual(cache.decr('answer2', version=2), 42)
- self.assertEqual(cache.get('answer2', version=1), 37)
- self.assertEqual(cache.get('answer2', version=2), 42)
-
- cache.set('answer3', 37, version=1)
- cache.set('answer3', 42, version=2)
- self.assertEqual(caches['v2'].incr('answer3'), 43)
- self.assertEqual(cache.get('answer3', version=1), 37)
- self.assertEqual(cache.get('answer3', version=2), 43)
- self.assertEqual(caches['v2'].decr('answer3'), 42)
- self.assertEqual(cache.get('answer3', version=1), 37)
- self.assertEqual(cache.get('answer3', version=2), 42)
-
- cache.set('answer4', 37, version=1)
- cache.set('answer4', 42, version=2)
- self.assertEqual(caches['v2'].incr('answer4', version=1), 38)
- self.assertEqual(cache.get('answer4', version=1), 38)
- self.assertEqual(cache.get('answer4', version=2), 42)
- self.assertEqual(caches['v2'].decr('answer4', version=1), 37)
- self.assertEqual(cache.get('answer4', version=1), 37)
- self.assertEqual(cache.get('answer4', version=2), 42)
+ cache.set("answer1", 37, version=1)
+ cache.set("answer1", 42, version=2)
+ self.assertEqual(cache.incr("answer1"), 38)
+ self.assertEqual(cache.get("answer1", version=1), 38)
+ self.assertEqual(cache.get("answer1", version=2), 42)
+ self.assertEqual(cache.decr("answer1"), 37)
+ self.assertEqual(cache.get("answer1", version=1), 37)
+ self.assertEqual(cache.get("answer1", version=2), 42)
+
+ cache.set("answer2", 37, version=1)
+ cache.set("answer2", 42, version=2)
+ self.assertEqual(cache.incr("answer2", version=2), 43)
+ self.assertEqual(cache.get("answer2", version=1), 37)
+ self.assertEqual(cache.get("answer2", version=2), 43)
+ self.assertEqual(cache.decr("answer2", version=2), 42)
+ self.assertEqual(cache.get("answer2", version=1), 37)
+ self.assertEqual(cache.get("answer2", version=2), 42)
+
+ cache.set("answer3", 37, version=1)
+ cache.set("answer3", 42, version=2)
+ self.assertEqual(caches["v2"].incr("answer3"), 43)
+ self.assertEqual(cache.get("answer3", version=1), 37)
+ self.assertEqual(cache.get("answer3", version=2), 43)
+ self.assertEqual(caches["v2"].decr("answer3"), 42)
+ self.assertEqual(cache.get("answer3", version=1), 37)
+ self.assertEqual(cache.get("answer3", version=2), 42)
+
+ cache.set("answer4", 37, version=1)
+ cache.set("answer4", 42, version=2)
+ self.assertEqual(caches["v2"].incr("answer4", version=1), 38)
+ self.assertEqual(cache.get("answer4", version=1), 38)
+ self.assertEqual(cache.get("answer4", version=2), 42)
+ self.assertEqual(caches["v2"].decr("answer4", version=1), 37)
+ self.assertEqual(cache.get("answer4", version=1), 37)
+ self.assertEqual(cache.get("answer4", version=2), 42)
def test_cache_versioning_get_set_many(self):
# set, using default version = 1
- cache.set_many({'ford1': 37, 'arthur1': 42})
- self.assertEqual(cache.get_many(['ford1', 'arthur1']), {'ford1': 37, 'arthur1': 42})
- self.assertEqual(cache.get_many(['ford1', 'arthur1'], version=1), {'ford1': 37, 'arthur1': 42})
- self.assertEqual(cache.get_many(['ford1', 'arthur1'], version=2), {})
+ cache.set_many({"ford1": 37, "arthur1": 42})
+ self.assertEqual(
+ cache.get_many(["ford1", "arthur1"]), {"ford1": 37, "arthur1": 42}
+ )
+ self.assertEqual(
+ cache.get_many(["ford1", "arthur1"], version=1),
+ {"ford1": 37, "arthur1": 42},
+ )
+ self.assertEqual(cache.get_many(["ford1", "arthur1"], version=2), {})
- self.assertEqual(caches['v2'].get_many(['ford1', 'arthur1']), {})
- self.assertEqual(caches['v2'].get_many(['ford1', 'arthur1'], version=1), {'ford1': 37, 'arthur1': 42})
- self.assertEqual(caches['v2'].get_many(['ford1', 'arthur1'], version=2), {})
+ self.assertEqual(caches["v2"].get_many(["ford1", "arthur1"]), {})
+ self.assertEqual(
+ caches["v2"].get_many(["ford1", "arthur1"], version=1),
+ {"ford1": 37, "arthur1": 42},
+ )
+ self.assertEqual(caches["v2"].get_many(["ford1", "arthur1"], version=2), {})
# set, default version = 1, but manually override version = 2
- cache.set_many({'ford2': 37, 'arthur2': 42}, version=2)
- self.assertEqual(cache.get_many(['ford2', 'arthur2']), {})
- self.assertEqual(cache.get_many(['ford2', 'arthur2'], version=1), {})
- self.assertEqual(cache.get_many(['ford2', 'arthur2'], version=2), {'ford2': 37, 'arthur2': 42})
+ cache.set_many({"ford2": 37, "arthur2": 42}, version=2)
+ self.assertEqual(cache.get_many(["ford2", "arthur2"]), {})
+ self.assertEqual(cache.get_many(["ford2", "arthur2"], version=1), {})
+ self.assertEqual(
+ cache.get_many(["ford2", "arthur2"], version=2),
+ {"ford2": 37, "arthur2": 42},
+ )
- self.assertEqual(caches['v2'].get_many(['ford2', 'arthur2']), {'ford2': 37, 'arthur2': 42})
- self.assertEqual(caches['v2'].get_many(['ford2', 'arthur2'], version=1), {})
- self.assertEqual(caches['v2'].get_many(['ford2', 'arthur2'], version=2), {'ford2': 37, 'arthur2': 42})
+ self.assertEqual(
+ caches["v2"].get_many(["ford2", "arthur2"]), {"ford2": 37, "arthur2": 42}
+ )
+ self.assertEqual(caches["v2"].get_many(["ford2", "arthur2"], version=1), {})
+ self.assertEqual(
+ caches["v2"].get_many(["ford2", "arthur2"], version=2),
+ {"ford2": 37, "arthur2": 42},
+ )
# v2 set, using default version = 2
- caches['v2'].set_many({'ford3': 37, 'arthur3': 42})
- self.assertEqual(cache.get_many(['ford3', 'arthur3']), {})
- self.assertEqual(cache.get_many(['ford3', 'arthur3'], version=1), {})
- self.assertEqual(cache.get_many(['ford3', 'arthur3'], version=2), {'ford3': 37, 'arthur3': 42})
+ caches["v2"].set_many({"ford3": 37, "arthur3": 42})
+ self.assertEqual(cache.get_many(["ford3", "arthur3"]), {})
+ self.assertEqual(cache.get_many(["ford3", "arthur3"], version=1), {})
+ self.assertEqual(
+ cache.get_many(["ford3", "arthur3"], version=2),
+ {"ford3": 37, "arthur3": 42},
+ )
- self.assertEqual(caches['v2'].get_many(['ford3', 'arthur3']), {'ford3': 37, 'arthur3': 42})
- self.assertEqual(caches['v2'].get_many(['ford3', 'arthur3'], version=1), {})
- self.assertEqual(caches['v2'].get_many(['ford3', 'arthur3'], version=2), {'ford3': 37, 'arthur3': 42})
+ self.assertEqual(
+ caches["v2"].get_many(["ford3", "arthur3"]), {"ford3": 37, "arthur3": 42}
+ )
+ self.assertEqual(caches["v2"].get_many(["ford3", "arthur3"], version=1), {})
+ self.assertEqual(
+ caches["v2"].get_many(["ford3", "arthur3"], version=2),
+ {"ford3": 37, "arthur3": 42},
+ )
# v2 set, default version = 2, but manually override version = 1
- caches['v2'].set_many({'ford4': 37, 'arthur4': 42}, version=1)
- self.assertEqual(cache.get_many(['ford4', 'arthur4']), {'ford4': 37, 'arthur4': 42})
- self.assertEqual(cache.get_many(['ford4', 'arthur4'], version=1), {'ford4': 37, 'arthur4': 42})
- self.assertEqual(cache.get_many(['ford4', 'arthur4'], version=2), {})
+ caches["v2"].set_many({"ford4": 37, "arthur4": 42}, version=1)
+ self.assertEqual(
+ cache.get_many(["ford4", "arthur4"]), {"ford4": 37, "arthur4": 42}
+ )
+ self.assertEqual(
+ cache.get_many(["ford4", "arthur4"], version=1),
+ {"ford4": 37, "arthur4": 42},
+ )
+ self.assertEqual(cache.get_many(["ford4", "arthur4"], version=2), {})
- self.assertEqual(caches['v2'].get_many(['ford4', 'arthur4']), {})
- self.assertEqual(caches['v2'].get_many(['ford4', 'arthur4'], version=1), {'ford4': 37, 'arthur4': 42})
- self.assertEqual(caches['v2'].get_many(['ford4', 'arthur4'], version=2), {})
+ self.assertEqual(caches["v2"].get_many(["ford4", "arthur4"]), {})
+ self.assertEqual(
+ caches["v2"].get_many(["ford4", "arthur4"], version=1),
+ {"ford4": 37, "arthur4": 42},
+ )
+ self.assertEqual(caches["v2"].get_many(["ford4", "arthur4"], version=2), {})
def test_incr_version(self):
- cache.set('answer', 42, version=2)
- self.assertIsNone(cache.get('answer'))
- self.assertIsNone(cache.get('answer', version=1))
- self.assertEqual(cache.get('answer', version=2), 42)
- self.assertIsNone(cache.get('answer', version=3))
-
- self.assertEqual(cache.incr_version('answer', version=2), 3)
- self.assertIsNone(cache.get('answer'))
- self.assertIsNone(cache.get('answer', version=1))
- self.assertIsNone(cache.get('answer', version=2))
- self.assertEqual(cache.get('answer', version=3), 42)
-
- caches['v2'].set('answer2', 42)
- self.assertEqual(caches['v2'].get('answer2'), 42)
- self.assertIsNone(caches['v2'].get('answer2', version=1))
- self.assertEqual(caches['v2'].get('answer2', version=2), 42)
- self.assertIsNone(caches['v2'].get('answer2', version=3))
-
- self.assertEqual(caches['v2'].incr_version('answer2'), 3)
- self.assertIsNone(caches['v2'].get('answer2'))
- self.assertIsNone(caches['v2'].get('answer2', version=1))
- self.assertIsNone(caches['v2'].get('answer2', version=2))
- self.assertEqual(caches['v2'].get('answer2', version=3), 42)
+ cache.set("answer", 42, version=2)
+ self.assertIsNone(cache.get("answer"))
+ self.assertIsNone(cache.get("answer", version=1))
+ self.assertEqual(cache.get("answer", version=2), 42)
+ self.assertIsNone(cache.get("answer", version=3))
+
+ self.assertEqual(cache.incr_version("answer", version=2), 3)
+ self.assertIsNone(cache.get("answer"))
+ self.assertIsNone(cache.get("answer", version=1))
+ self.assertIsNone(cache.get("answer", version=2))
+ self.assertEqual(cache.get("answer", version=3), 42)
+
+ caches["v2"].set("answer2", 42)
+ self.assertEqual(caches["v2"].get("answer2"), 42)
+ self.assertIsNone(caches["v2"].get("answer2", version=1))
+ self.assertEqual(caches["v2"].get("answer2", version=2), 42)
+ self.assertIsNone(caches["v2"].get("answer2", version=3))
+
+ self.assertEqual(caches["v2"].incr_version("answer2"), 3)
+ self.assertIsNone(caches["v2"].get("answer2"))
+ self.assertIsNone(caches["v2"].get("answer2", version=1))
+ self.assertIsNone(caches["v2"].get("answer2", version=2))
+ self.assertEqual(caches["v2"].get("answer2", version=3), 42)
with self.assertRaises(ValueError):
- cache.incr_version('does_not_exist')
+ cache.incr_version("does_not_exist")
- cache.set('null', None)
- self.assertEqual(cache.incr_version('null'), 2)
+ cache.set("null", None)
+ self.assertEqual(cache.incr_version("null"), 2)
def test_decr_version(self):
- cache.set('answer', 42, version=2)
- self.assertIsNone(cache.get('answer'))
- self.assertIsNone(cache.get('answer', version=1))
- self.assertEqual(cache.get('answer', version=2), 42)
-
- self.assertEqual(cache.decr_version('answer', version=2), 1)
- self.assertEqual(cache.get('answer'), 42)
- self.assertEqual(cache.get('answer', version=1), 42)
- self.assertIsNone(cache.get('answer', version=2))
-
- caches['v2'].set('answer2', 42)
- self.assertEqual(caches['v2'].get('answer2'), 42)
- self.assertIsNone(caches['v2'].get('answer2', version=1))
- self.assertEqual(caches['v2'].get('answer2', version=2), 42)
-
- self.assertEqual(caches['v2'].decr_version('answer2'), 1)
- self.assertIsNone(caches['v2'].get('answer2'))
- self.assertEqual(caches['v2'].get('answer2', version=1), 42)
- self.assertIsNone(caches['v2'].get('answer2', version=2))
+ cache.set("answer", 42, version=2)
+ self.assertIsNone(cache.get("answer"))
+ self.assertIsNone(cache.get("answer", version=1))
+ self.assertEqual(cache.get("answer", version=2), 42)
+
+ self.assertEqual(cache.decr_version("answer", version=2), 1)
+ self.assertEqual(cache.get("answer"), 42)
+ self.assertEqual(cache.get("answer", version=1), 42)
+ self.assertIsNone(cache.get("answer", version=2))
+
+ caches["v2"].set("answer2", 42)
+ self.assertEqual(caches["v2"].get("answer2"), 42)
+ self.assertIsNone(caches["v2"].get("answer2", version=1))
+ self.assertEqual(caches["v2"].get("answer2", version=2), 42)
+
+ self.assertEqual(caches["v2"].decr_version("answer2"), 1)
+ self.assertIsNone(caches["v2"].get("answer2"))
+ self.assertEqual(caches["v2"].get("answer2", version=1), 42)
+ self.assertIsNone(caches["v2"].get("answer2", version=2))
with self.assertRaises(ValueError):
- cache.decr_version('does_not_exist', version=2)
+ cache.decr_version("does_not_exist", version=2)
- cache.set('null', None, version=2)
- self.assertEqual(cache.decr_version('null', version=2), 1)
+ cache.set("null", None, version=2)
+ self.assertEqual(cache.decr_version("null", version=2), 1)
def test_custom_key_func(self):
# Two caches with different key functions aren't visible to each other
- cache.set('answer1', 42)
- self.assertEqual(cache.get('answer1'), 42)
- self.assertIsNone(caches['custom_key'].get('answer1'))
- self.assertIsNone(caches['custom_key2'].get('answer1'))
+ cache.set("answer1", 42)
+ self.assertEqual(cache.get("answer1"), 42)
+ self.assertIsNone(caches["custom_key"].get("answer1"))
+ self.assertIsNone(caches["custom_key2"].get("answer1"))
- caches['custom_key'].set('answer2', 42)
- self.assertIsNone(cache.get('answer2'))
- self.assertEqual(caches['custom_key'].get('answer2'), 42)
- self.assertEqual(caches['custom_key2'].get('answer2'), 42)
+ caches["custom_key"].set("answer2", 42)
+ self.assertIsNone(cache.get("answer2"))
+ self.assertEqual(caches["custom_key"].get("answer2"), 42)
+ self.assertEqual(caches["custom_key2"].get("answer2"), 42)
@override_settings(CACHE_MIDDLEWARE_ALIAS=DEFAULT_CACHE_ALIAS)
def test_cache_write_unpicklable_object(self):
fetch_middleware = FetchFromCacheMiddleware(empty_response)
- request = self.factory.get('/cache/test')
+ request = self.factory.get("/cache/test")
request._cache_update_cache = True
- get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
+ get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
+ request
+ )
self.assertIsNone(get_cache_data)
- content = 'Testing cookie serialization.'
+ content = "Testing cookie serialization."
def get_response(req):
response = HttpResponse(content)
- response.set_cookie('foo', 'bar')
+ response.set_cookie("foo", "bar")
return response
update_middleware = UpdateCacheMiddleware(get_response)
@@ -1032,59 +1088,63 @@ class BaseCacheTests:
def test_add_fail_on_pickleerror(self):
# Shouldn't fail silently if trying to cache an unpicklable type.
with self.assertRaises(pickle.PickleError):
- cache.add('unpicklable', Unpicklable())
+ cache.add("unpicklable", Unpicklable())
def test_set_fail_on_pickleerror(self):
with self.assertRaises(pickle.PickleError):
- cache.set('unpicklable', Unpicklable())
+ cache.set("unpicklable", Unpicklable())
def test_get_or_set(self):
- self.assertIsNone(cache.get('projector'))
- self.assertEqual(cache.get_or_set('projector', 42), 42)
- self.assertEqual(cache.get('projector'), 42)
- self.assertIsNone(cache.get_or_set('null', None))
+ self.assertIsNone(cache.get("projector"))
+ self.assertEqual(cache.get_or_set("projector", 42), 42)
+ self.assertEqual(cache.get("projector"), 42)
+ self.assertIsNone(cache.get_or_set("null", None))
# Previous get_or_set() stores None in the cache.
- self.assertIsNone(cache.get('null', 'default'))
+ self.assertIsNone(cache.get("null", "default"))
def test_get_or_set_callable(self):
def my_callable():
- return 'value'
+ return "value"
- self.assertEqual(cache.get_or_set('mykey', my_callable), 'value')
- self.assertEqual(cache.get_or_set('mykey', my_callable()), 'value')
+ self.assertEqual(cache.get_or_set("mykey", my_callable), "value")
+ self.assertEqual(cache.get_or_set("mykey", my_callable()), "value")
- self.assertIsNone(cache.get_or_set('null', lambda: None))
+ self.assertIsNone(cache.get_or_set("null", lambda: None))
# Previous get_or_set() stores None in the cache.
- self.assertIsNone(cache.get('null', 'default'))
+ self.assertIsNone(cache.get("null", "default"))
def test_get_or_set_version(self):
msg = "get_or_set() missing 1 required positional argument: 'default'"
- self.assertEqual(cache.get_or_set('brian', 1979, version=2), 1979)
+ self.assertEqual(cache.get_or_set("brian", 1979, version=2), 1979)
with self.assertRaisesMessage(TypeError, msg):
- cache.get_or_set('brian')
+ cache.get_or_set("brian")
with self.assertRaisesMessage(TypeError, msg):
- cache.get_or_set('brian', version=1)
- self.assertIsNone(cache.get('brian', version=1))
- self.assertEqual(cache.get_or_set('brian', 42, version=1), 42)
- self.assertEqual(cache.get_or_set('brian', 1979, version=2), 1979)
- self.assertIsNone(cache.get('brian', version=3))
+ cache.get_or_set("brian", version=1)
+ self.assertIsNone(cache.get("brian", version=1))
+ self.assertEqual(cache.get_or_set("brian", 42, version=1), 42)
+ self.assertEqual(cache.get_or_set("brian", 1979, version=2), 1979)
+ self.assertIsNone(cache.get("brian", version=3))
def test_get_or_set_racing(self):
- with mock.patch('%s.%s' % (settings.CACHES['default']['BACKEND'], 'add')) as cache_add:
+ with mock.patch(
+ "%s.%s" % (settings.CACHES["default"]["BACKEND"], "add")
+ ) as cache_add:
# Simulate cache.add() failing to add a value. In that case, the
# default value should be returned.
cache_add.return_value = False
- self.assertEqual(cache.get_or_set('key', 'default'), 'default')
+ self.assertEqual(cache.get_or_set("key", "default"), "default")
-@override_settings(CACHES=caches_setting_for_tests(
- BACKEND='django.core.cache.backends.db.DatabaseCache',
- # Spaces are used in the table name to ensure quoting/escaping is working
- LOCATION='test cache table'
-))
+@override_settings(
+ CACHES=caches_setting_for_tests(
+ BACKEND="django.core.cache.backends.db.DatabaseCache",
+ # Spaces are used in the table name to ensure quoting/escaping is working
+ LOCATION="test cache table",
+ )
+)
class DBCacheTests(BaseCacheTests, TransactionTestCase):
- available_apps = ['cache']
+ available_apps = ["cache"]
def setUp(self):
# The super calls needs to happen first for the settings override.
@@ -1097,26 +1157,26 @@ class DBCacheTests(BaseCacheTests, TransactionTestCase):
self.drop_table()
def create_table(self):
- management.call_command('createcachetable', verbosity=0)
+ management.call_command("createcachetable", verbosity=0)
def drop_table(self):
with connection.cursor() as cursor:
- table_name = connection.ops.quote_name('test cache table')
- cursor.execute('DROP TABLE %s' % table_name)
+ table_name = connection.ops.quote_name("test cache table")
+ cursor.execute("DROP TABLE %s" % table_name)
def test_get_many_num_queries(self):
- cache.set_many({'a': 1, 'b': 2})
- cache.set('expired', 'expired', 0.01)
+ cache.set_many({"a": 1, "b": 2})
+ cache.set("expired", "expired", 0.01)
with self.assertNumQueries(1):
- self.assertEqual(cache.get_many(['a', 'b']), {'a': 1, 'b': 2})
+ self.assertEqual(cache.get_many(["a", "b"]), {"a": 1, "b": 2})
time.sleep(0.02)
with self.assertNumQueries(2):
- self.assertEqual(cache.get_many(['a', 'b', 'expired']), {'a': 1, 'b': 2})
+ self.assertEqual(cache.get_many(["a", "b", "expired"]), {"a": 1, "b": 2})
def test_delete_many_num_queries(self):
- cache.set_many({'a': 1, 'b': 2, 'c': 3})
+ cache.set_many({"a": 1, "b": 2, "c": 3})
with self.assertNumQueries(1):
- cache.delete_many(['a', 'b', 'c'])
+ cache.delete_many(["a", "b", "c"])
def test_cull_queries(self):
old_max_entries = cache._max_entries
@@ -1124,23 +1184,24 @@ class DBCacheTests(BaseCacheTests, TransactionTestCase):
cache._max_entries = -1
with CaptureQueriesContext(connection) as captured_queries:
try:
- cache.set('force_cull', 'value', 1000)
+ cache.set("force_cull", "value", 1000)
finally:
cache._max_entries = old_max_entries
- num_count_queries = sum('COUNT' in query['sql'] for query in captured_queries)
+ num_count_queries = sum("COUNT" in query["sql"] for query in captured_queries)
self.assertEqual(num_count_queries, 1)
# Column names are quoted.
for query in captured_queries:
- sql = query['sql']
- if 'expires' in sql:
- self.assertIn(connection.ops.quote_name('expires'), sql)
- if 'cache_key' in sql:
- self.assertIn(connection.ops.quote_name('cache_key'), sql)
+ sql = query["sql"]
+ if "expires" in sql:
+ self.assertIn(connection.ops.quote_name("expires"), sql)
+ if "cache_key" in sql:
+ self.assertIn(connection.ops.quote_name("cache_key"), sql)
def test_delete_cursor_rowcount(self):
"""
The rowcount attribute should not be checked on a closed cursor.
"""
+
class MockedCursorWrapper(CursorWrapper):
is_closed = False
@@ -1151,29 +1212,34 @@ class DBCacheTests(BaseCacheTests, TransactionTestCase):
@property
def rowcount(self):
if self.is_closed:
- raise Exception('Cursor is closed.')
+ raise Exception("Cursor is closed.")
return self.cursor.rowcount
- cache.set_many({'a': 1, 'b': 2})
- with mock.patch('django.db.backends.utils.CursorWrapper', MockedCursorWrapper):
- self.assertIs(cache.delete('a'), True)
+ cache.set_many({"a": 1, "b": 2})
+ with mock.patch("django.db.backends.utils.CursorWrapper", MockedCursorWrapper):
+ self.assertIs(cache.delete("a"), True)
def test_zero_cull(self):
- self._perform_cull_test('zero_cull', 50, 18)
+ self._perform_cull_test("zero_cull", 50, 18)
def test_second_call_doesnt_crash(self):
out = io.StringIO()
- management.call_command('createcachetable', stdout=out)
- self.assertEqual(out.getvalue(), "Cache table 'test cache table' already exists.\n" * len(settings.CACHES))
-
- @override_settings(CACHES=caches_setting_for_tests(
- BACKEND='django.core.cache.backends.db.DatabaseCache',
- # Use another table name to avoid the 'table already exists' message.
- LOCATION='createcachetable_dry_run_mode'
- ))
+ management.call_command("createcachetable", stdout=out)
+ self.assertEqual(
+ out.getvalue(),
+ "Cache table 'test cache table' already exists.\n" * len(settings.CACHES),
+ )
+
+ @override_settings(
+ CACHES=caches_setting_for_tests(
+ BACKEND="django.core.cache.backends.db.DatabaseCache",
+ # Use another table name to avoid the 'table already exists' message.
+ LOCATION="createcachetable_dry_run_mode",
+ )
+ )
def test_createcachetable_dry_run_mode(self):
out = io.StringIO()
- management.call_command('createcachetable', dry_run=True, stdout=out)
+ management.call_command("createcachetable", dry_run=True, stdout=out)
output = out.getvalue()
self.assertTrue(output.startswith("CREATE TABLE"))
@@ -1185,8 +1251,8 @@ class DBCacheTests(BaseCacheTests, TransactionTestCase):
self.drop_table()
out = io.StringIO()
management.call_command(
- 'createcachetable',
- 'test cache table',
+ "createcachetable",
+ "test cache table",
verbosity=2,
stdout=out,
)
@@ -1194,12 +1260,12 @@ class DBCacheTests(BaseCacheTests, TransactionTestCase):
def test_has_key_query_columns_quoted(self):
with CaptureQueriesContext(connection) as captured_queries:
- cache.has_key('key')
+ cache.has_key("key")
self.assertEqual(len(captured_queries), 1)
- sql = captured_queries[0]['sql']
+ sql = captured_queries[0]["sql"]
# Column names are quoted.
- self.assertIn(connection.ops.quote_name('expires'), sql)
- self.assertIn(connection.ops.quote_name('cache_key'), sql)
+ self.assertIn(connection.ops.quote_name("expires"), sql)
+ self.assertIn(connection.ops.quote_name("cache_key"), sql)
@override_settings(USE_TZ=True)
@@ -1211,37 +1277,37 @@ class DBCacheRouter:
"""A router that puts the cache table on the 'other' database."""
def db_for_read(self, model, **hints):
- if model._meta.app_label == 'django_cache':
- return 'other'
+ if model._meta.app_label == "django_cache":
+ return "other"
return None
def db_for_write(self, model, **hints):
- if model._meta.app_label == 'django_cache':
- return 'other'
+ if model._meta.app_label == "django_cache":
+ return "other"
return None
def allow_migrate(self, db, app_label, **hints):
- if app_label == 'django_cache':
- return db == 'other'
+ if app_label == "django_cache":
+ return db == "other"
return None
@override_settings(
CACHES={
- 'default': {
- 'BACKEND': 'django.core.cache.backends.db.DatabaseCache',
- 'LOCATION': 'my_cache_table',
+ "default": {
+ "BACKEND": "django.core.cache.backends.db.DatabaseCache",
+ "LOCATION": "my_cache_table",
},
},
)
class CreateCacheTableForDBCacheTests(TestCase):
- databases = {'default', 'other'}
+ databases = {"default", "other"}
@override_settings(DATABASE_ROUTERS=[DBCacheRouter()])
def test_createcachetable_observes_database_router(self):
# cache table should not be created on 'default'
- with self.assertNumQueries(0, using='default'):
- management.call_command('createcachetable', database='default', verbosity=0)
+ with self.assertNumQueries(0, using="default"):
+ management.call_command("createcachetable", database="default", verbosity=0)
# cache table should be created on 'other'
# Queries:
# 1: check table doesn't already exist
@@ -1249,13 +1315,12 @@ class CreateCacheTableForDBCacheTests(TestCase):
# 3: create the table
# 4: create the index
# 5: release savepoint (if transactional DDL is supported)
- num = 5 if connections['other'].features.can_rollback_ddl else 3
- with self.assertNumQueries(num, using='other'):
- management.call_command('createcachetable', database='other', verbosity=0)
+ num = 5 if connections["other"].features.can_rollback_ddl else 3
+ with self.assertNumQueries(num, using="other"):
+ management.call_command("createcachetable", database="other", verbosity=0)
class PicklingSideEffect:
-
def __init__(self, cache):
self.cache = cache
self.locked = False
@@ -1265,59 +1330,64 @@ class PicklingSideEffect:
return {}
-limit_locmem_entries = override_settings(CACHES=caches_setting_for_tests(
- BACKEND='django.core.cache.backends.locmem.LocMemCache',
- OPTIONS={'MAX_ENTRIES': 9},
-))
+limit_locmem_entries = override_settings(
+ CACHES=caches_setting_for_tests(
+ BACKEND="django.core.cache.backends.locmem.LocMemCache",
+ OPTIONS={"MAX_ENTRIES": 9},
+ )
+)
-@override_settings(CACHES=caches_setting_for_tests(
- BACKEND='django.core.cache.backends.locmem.LocMemCache',
-))
+@override_settings(
+ CACHES=caches_setting_for_tests(
+ BACKEND="django.core.cache.backends.locmem.LocMemCache",
+ )
+)
class LocMemCacheTests(BaseCacheTests, TestCase):
-
def setUp(self):
super().setUp()
# LocMem requires a hack to make the other caches
# share a data store with the 'normal' cache.
- caches['prefix']._cache = cache._cache
- caches['prefix']._expire_info = cache._expire_info
+ caches["prefix"]._cache = cache._cache
+ caches["prefix"]._expire_info = cache._expire_info
- caches['v2']._cache = cache._cache
- caches['v2']._expire_info = cache._expire_info
+ caches["v2"]._cache = cache._cache
+ caches["v2"]._expire_info = cache._expire_info
- caches['custom_key']._cache = cache._cache
- caches['custom_key']._expire_info = cache._expire_info
+ caches["custom_key"]._cache = cache._cache
+ caches["custom_key"]._expire_info = cache._expire_info
- caches['custom_key2']._cache = cache._cache
- caches['custom_key2']._expire_info = cache._expire_info
+ caches["custom_key2"]._cache = cache._cache
+ caches["custom_key2"]._expire_info = cache._expire_info
- @override_settings(CACHES={
- 'default': {'BACKEND': 'django.core.cache.backends.locmem.LocMemCache'},
- 'other': {
- 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
- 'LOCATION': 'other'
- },
- })
+ @override_settings(
+ CACHES={
+ "default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"},
+ "other": {
+ "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
+ "LOCATION": "other",
+ },
+ }
+ )
def test_multiple_caches(self):
"Multiple locmem caches are isolated"
- cache.set('value', 42)
- self.assertEqual(caches['default'].get('value'), 42)
- self.assertIsNone(caches['other'].get('value'))
+ cache.set("value", 42)
+ self.assertEqual(caches["default"].get("value"), 42)
+ self.assertIsNone(caches["other"].get("value"))
def test_locking_on_pickle(self):
"""#20613/#18541 -- Ensures pickling is done outside of the lock."""
bad_obj = PicklingSideEffect(cache)
- cache.set('set', bad_obj)
+ cache.set("set", bad_obj)
self.assertFalse(bad_obj.locked, "Cache was locked during pickling")
- self.assertIs(cache.add('add', bad_obj), True)
+ self.assertIs(cache.add("add", bad_obj), True)
self.assertFalse(bad_obj.locked, "Cache was locked during pickling")
def test_incr_decr_timeout(self):
"""incr/decr does not modify expiry time (matches memcached behavior)"""
- key = 'value'
+ key = "value"
_key = cache.make_key(key)
cache.set(key, 1, timeout=cache.default_timeout * 10)
expire = cache._expire_info[_key]
@@ -1373,18 +1443,22 @@ class LocMemCacheTests(BaseCacheTests, TestCase):
# one cache backend setting that points at your cache server.
configured_caches = {}
for _cache_params in settings.CACHES.values():
- configured_caches[_cache_params['BACKEND']] = _cache_params
+ configured_caches[_cache_params["BACKEND"]] = _cache_params
-PyLibMCCache_params = configured_caches.get('django.core.cache.backends.memcached.PyLibMCCache')
-PyMemcacheCache_params = configured_caches.get('django.core.cache.backends.memcached.PyMemcacheCache')
+PyLibMCCache_params = configured_caches.get(
+ "django.core.cache.backends.memcached.PyLibMCCache"
+)
+PyMemcacheCache_params = configured_caches.get(
+ "django.core.cache.backends.memcached.PyMemcacheCache"
+)
# The memcached backends don't support cull-related options like `MAX_ENTRIES`.
-memcached_excluded_caches = {'cull', 'zero_cull'}
+memcached_excluded_caches = {"cull", "zero_cull"}
-RedisCache_params = configured_caches.get('django.core.cache.backends.redis.RedisCache')
+RedisCache_params = configured_caches.get("django.core.cache.backends.redis.RedisCache")
# The redis backend does not support cull-related options like `MAX_ENTRIES`.
-redis_excluded_caches = {'cull', 'zero_cull'}
+redis_excluded_caches = {"cull", "zero_cull"}
class BaseMemcachedTests(BaseCacheTests):
@@ -1395,15 +1469,15 @@ class BaseMemcachedTests(BaseCacheTests):
def test_location_multiple_servers(self):
locations = [
- ['server1.tld', 'server2:11211'],
- 'server1.tld;server2:11211',
- 'server1.tld,server2:11211',
+ ["server1.tld", "server2:11211"],
+ "server1.tld;server2:11211",
+ "server1.tld,server2:11211",
]
for location in locations:
with self.subTest(location=location):
- params = {'BACKEND': self.base_params['BACKEND'], 'LOCATION': location}
- with self.settings(CACHES={'default': params}):
- self.assertEqual(cache._servers, ['server1.tld', 'server2:11211'])
+ params = {"BACKEND": self.base_params["BACKEND"], "LOCATION": location}
+ with self.settings(CACHES={"default": params}):
+ self.assertEqual(cache._servers, ["server1.tld", "server2:11211"])
def _perform_invalid_key_test(self, key, expected_warning):
"""
@@ -1412,16 +1486,16 @@ class BaseMemcachedTests(BaseCacheTests):
"""
msg = expected_warning.replace(key, cache.make_key(key))
tests = [
- ('add', [key, 1]),
- ('get', [key]),
- ('set', [key, 1]),
- ('incr', [key]),
- ('decr', [key]),
- ('touch', [key]),
- ('delete', [key]),
- ('get_many', [[key, 'b']]),
- ('set_many', [{key: 1, 'b': 2}]),
- ('delete_many', [[key, 'b']]),
+ ("add", [key, 1]),
+ ("get", [key]),
+ ("set", [key, 1]),
+ ("incr", [key]),
+ ("decr", [key]),
+ ("touch", [key]),
+ ("delete", [key]),
+ ("get_many", [[key, "b"]]),
+ ("set_many", [{key: 1, "b": 2}]),
+ ("delete_many", [[key, "b"]]),
]
for operation, args in tests:
with self.subTest(operation=operation):
@@ -1431,44 +1505,48 @@ class BaseMemcachedTests(BaseCacheTests):
def test_invalid_with_version_key_length(self):
# make_key() adds a version to the key and exceeds the limit.
- key = 'a' * 248
+ key = "a" * 248
expected_warning = (
- 'Cache key will cause errors if used with memcached: '
- '%r (longer than %s)' % (key, 250)
+ "Cache key will cause errors if used with memcached: "
+ "%r (longer than %s)" % (key, 250)
)
self._perform_invalid_key_test(key, expected_warning)
def test_default_never_expiring_timeout(self):
# Regression test for #22845
- with self.settings(CACHES=caches_setting_for_tests(
- base=self.base_params,
- exclude=memcached_excluded_caches,
- TIMEOUT=None)):
- cache.set('infinite_foo', 'bar')
- self.assertEqual(cache.get('infinite_foo'), 'bar')
+ with self.settings(
+ CACHES=caches_setting_for_tests(
+ base=self.base_params, exclude=memcached_excluded_caches, TIMEOUT=None
+ )
+ ):
+ cache.set("infinite_foo", "bar")
+ self.assertEqual(cache.get("infinite_foo"), "bar")
def test_default_far_future_timeout(self):
# Regression test for #22845
- with self.settings(CACHES=caches_setting_for_tests(
+ with self.settings(
+ CACHES=caches_setting_for_tests(
base=self.base_params,
exclude=memcached_excluded_caches,
# 60*60*24*365, 1 year
- TIMEOUT=31536000)):
- cache.set('future_foo', 'bar')
- self.assertEqual(cache.get('future_foo'), 'bar')
+ TIMEOUT=31536000,
+ )
+ ):
+ cache.set("future_foo", "bar")
+ self.assertEqual(cache.get("future_foo"), "bar")
def test_memcached_deletes_key_on_failed_set(self):
# By default memcached allows objects up to 1MB. For the cache_db session
# backend to always use the current session, memcached needs to delete
# the old key if it fails to set.
- max_value_length = 2 ** 20
+ max_value_length = 2**20
- cache.set('small_value', 'a')
- self.assertEqual(cache.get('small_value'), 'a')
+ cache.set("small_value", "a")
+ self.assertEqual(cache.get("small_value"), "a")
- large_value = 'a' * (max_value_length + 1)
+ large_value = "a" * (max_value_length + 1)
try:
- cache.set('small_value', large_value)
+ cache.set("small_value", large_value)
except Exception:
# Most clients (e.g. pymemcache or pylibmc) raise when the value is
# too large. This test is primarily checking that the key was
@@ -1476,7 +1554,7 @@ class BaseMemcachedTests(BaseCacheTests):
# not important.
pass
# small_value should be deleted, or set if configured to accept larger values
- value = cache.get('small_value')
+ value = cache.get("small_value")
self.assertTrue(value is None or value == large_value)
def test_close(self):
@@ -1484,7 +1562,9 @@ class BaseMemcachedTests(BaseCacheTests):
# connection is closed when the request is complete.
signals.request_finished.disconnect(close_old_connections)
try:
- with mock.patch.object(cache._class, 'disconnect_all', autospec=True) as mock_disconnect:
+ with mock.patch.object(
+ cache._class, "disconnect_all", autospec=True
+ ) as mock_disconnect:
signals.request_finished.send(self.__class__)
self.assertIs(mock_disconnect.called, self.should_disconnect_on_close)
finally:
@@ -1493,16 +1573,19 @@ class BaseMemcachedTests(BaseCacheTests):
def test_set_many_returns_failing_keys(self):
def fail_set_multi(mapping, *args, **kwargs):
return mapping.keys()
- with mock.patch.object(cache._class, 'set_multi', side_effect=fail_set_multi):
- failing_keys = cache.set_many({'key': 'value'})
- self.assertEqual(failing_keys, ['key'])
+
+ with mock.patch.object(cache._class, "set_multi", side_effect=fail_set_multi):
+ failing_keys = cache.set_many({"key": "value"})
+ self.assertEqual(failing_keys, ["key"])
@unittest.skipUnless(PyLibMCCache_params, "PyLibMCCache backend not configured")
-@override_settings(CACHES=caches_setting_for_tests(
- base=PyLibMCCache_params,
- exclude=memcached_excluded_caches,
-))
+@override_settings(
+ CACHES=caches_setting_for_tests(
+ base=PyLibMCCache_params,
+ exclude=memcached_excluded_caches,
+ )
+)
class PyLibMCCacheTests(BaseMemcachedTests, TestCase):
base_params = PyLibMCCache_params
# libmemcached manages its own connections.
@@ -1512,41 +1595,45 @@ class PyLibMCCacheTests(BaseMemcachedTests, TestCase):
def incr_decr_type_error(self):
return cache._lib.ClientError
- @override_settings(CACHES=caches_setting_for_tests(
- base=PyLibMCCache_params,
- exclude=memcached_excluded_caches,
- OPTIONS={
- 'binary': True,
- 'behaviors': {'tcp_nodelay': True},
- },
- ))
+ @override_settings(
+ CACHES=caches_setting_for_tests(
+ base=PyLibMCCache_params,
+ exclude=memcached_excluded_caches,
+ OPTIONS={
+ "binary": True,
+ "behaviors": {"tcp_nodelay": True},
+ },
+ )
+ )
def test_pylibmc_options(self):
self.assertTrue(cache._cache.binary)
- self.assertEqual(cache._cache.behaviors['tcp_nodelay'], int(True))
+ self.assertEqual(cache._cache.behaviors["tcp_nodelay"], int(True))
def test_pylibmc_client_servers(self):
- backend = self.base_params['BACKEND']
+ backend = self.base_params["BACKEND"]
tests = [
- ('unix:/run/memcached/socket', '/run/memcached/socket'),
- ('/run/memcached/socket', '/run/memcached/socket'),
- ('localhost', 'localhost'),
- ('localhost:11211', 'localhost:11211'),
- ('[::1]', '[::1]'),
- ('[::1]:11211', '[::1]:11211'),
- ('127.0.0.1', '127.0.0.1'),
- ('127.0.0.1:11211', '127.0.0.1:11211'),
+ ("unix:/run/memcached/socket", "/run/memcached/socket"),
+ ("/run/memcached/socket", "/run/memcached/socket"),
+ ("localhost", "localhost"),
+ ("localhost:11211", "localhost:11211"),
+ ("[::1]", "[::1]"),
+ ("[::1]:11211", "[::1]:11211"),
+ ("127.0.0.1", "127.0.0.1"),
+ ("127.0.0.1:11211", "127.0.0.1:11211"),
]
for location, expected in tests:
- settings = {'default': {'BACKEND': backend, 'LOCATION': location}}
+ settings = {"default": {"BACKEND": backend, "LOCATION": location}}
with self.subTest(location), self.settings(CACHES=settings):
self.assertEqual(cache.client_servers, [expected])
-@unittest.skipUnless(PyMemcacheCache_params, 'PyMemcacheCache backend not configured')
-@override_settings(CACHES=caches_setting_for_tests(
- base=PyMemcacheCache_params,
- exclude=memcached_excluded_caches,
-))
+@unittest.skipUnless(PyMemcacheCache_params, "PyMemcacheCache backend not configured")
+@override_settings(
+ CACHES=caches_setting_for_tests(
+ base=PyMemcacheCache_params,
+ exclude=memcached_excluded_caches,
+ )
+)
class PyMemcacheCacheTests(BaseMemcachedTests, TestCase):
base_params = PyMemcacheCache_params
@@ -1556,29 +1643,35 @@ class PyMemcacheCacheTests(BaseMemcachedTests, TestCase):
def test_pymemcache_highest_pickle_version(self):
self.assertEqual(
- cache._cache.default_kwargs['serde']._serialize_func.keywords['pickle_version'],
+ cache._cache.default_kwargs["serde"]._serialize_func.keywords[
+ "pickle_version"
+ ],
pickle.HIGHEST_PROTOCOL,
)
for cache_key in settings.CACHES:
for client_key, client in caches[cache_key]._cache.clients.items():
with self.subTest(cache_key=cache_key, server=client_key):
self.assertEqual(
- client.serde._serialize_func.keywords['pickle_version'],
+ client.serde._serialize_func.keywords["pickle_version"],
pickle.HIGHEST_PROTOCOL,
)
- @override_settings(CACHES=caches_setting_for_tests(
- base=PyMemcacheCache_params,
- exclude=memcached_excluded_caches,
- OPTIONS={'no_delay': True},
- ))
+ @override_settings(
+ CACHES=caches_setting_for_tests(
+ base=PyMemcacheCache_params,
+ exclude=memcached_excluded_caches,
+ OPTIONS={"no_delay": True},
+ )
+ )
def test_pymemcache_options(self):
- self.assertIs(cache._cache.default_kwargs['no_delay'], True)
+ self.assertIs(cache._cache.default_kwargs["no_delay"], True)
-@override_settings(CACHES=caches_setting_for_tests(
- BACKEND='django.core.cache.backends.filebased.FileBasedCache',
-))
+@override_settings(
+ CACHES=caches_setting_for_tests(
+ BACKEND="django.core.cache.backends.filebased.FileBasedCache",
+ )
+)
class FileBasedCacheTests(BaseCacheTests, TestCase):
"""
Specific test cases for the file-based cache.
@@ -1591,8 +1684,8 @@ class FileBasedCacheTests(BaseCacheTests, TestCase):
# hence settings are manipulated directly here and the setting_changed signal
# is triggered manually.
for cache_params in settings.CACHES.values():
- cache_params['LOCATION'] = self.dirname
- setting_changed.send(self.__class__, setting='CACHES', enter=False)
+ cache_params["LOCATION"] = self.dirname
+ setting_changed.send(self.__class__, setting="CACHES", enter=False)
def tearDown(self):
super().tearDown()
@@ -1603,41 +1696,43 @@ class FileBasedCacheTests(BaseCacheTests, TestCase):
return tempfile.mkdtemp()
def test_ignores_non_cache_files(self):
- fname = os.path.join(self.dirname, 'not-a-cache-file')
- with open(fname, 'w'):
+ fname = os.path.join(self.dirname, "not-a-cache-file")
+ with open(fname, "w"):
os.utime(fname, None)
cache.clear()
- self.assertTrue(os.path.exists(fname),
- 'Expected cache.clear to ignore non cache files')
+ self.assertTrue(
+ os.path.exists(fname), "Expected cache.clear to ignore non cache files"
+ )
os.remove(fname)
def test_clear_does_not_remove_cache_dir(self):
cache.clear()
- self.assertTrue(os.path.exists(self.dirname),
- 'Expected cache.clear to keep the cache dir')
+ self.assertTrue(
+ os.path.exists(self.dirname), "Expected cache.clear to keep the cache dir"
+ )
def test_creates_cache_dir_if_nonexistent(self):
os.rmdir(self.dirname)
- cache.set('foo', 'bar')
+ cache.set("foo", "bar")
self.assertTrue(os.path.exists(self.dirname))
def test_get_ignores_enoent(self):
- cache.set('foo', 'bar')
- os.unlink(cache._key_to_file('foo'))
+ cache.set("foo", "bar")
+ os.unlink(cache._key_to_file("foo"))
# Returns the default instead of erroring.
- self.assertEqual(cache.get('foo', 'baz'), 'baz')
+ self.assertEqual(cache.get("foo", "baz"), "baz")
@skipIf(
- sys.platform == 'win32',
- 'Windows only partially supports umasks and chmod.',
+ sys.platform == "win32",
+ "Windows only partially supports umasks and chmod.",
)
def test_cache_dir_permissions(self):
os.rmdir(self.dirname)
- dir_path = Path(self.dirname) / 'nested' / 'filebasedcache'
+ dir_path = Path(self.dirname) / "nested" / "filebasedcache"
for cache_params in settings.CACHES.values():
- cache_params['LOCATION'] = dir_path
- setting_changed.send(self.__class__, setting='CACHES', enter=False)
- cache.set('foo', 'bar')
+ cache_params["LOCATION"] = dir_path
+ setting_changed.send(self.__class__, setting="CACHES", enter=False)
+ cache.set("foo", "bar")
self.assertIs(dir_path.exists(), True)
tests = [
dir_path,
@@ -1650,27 +1745,29 @@ class FileBasedCacheTests(BaseCacheTests, TestCase):
self.assertEqual(dir_mode, 0o700)
def test_get_does_not_ignore_non_filenotfound_exceptions(self):
- with mock.patch('builtins.open', side_effect=OSError):
+ with mock.patch("builtins.open", side_effect=OSError):
with self.assertRaises(OSError):
- cache.get('foo')
+ cache.get("foo")
def test_empty_cache_file_considered_expired(self):
- cache_file = cache._key_to_file('foo')
- with open(cache_file, 'wb') as fh:
- fh.write(b'')
- with open(cache_file, 'rb') as fh:
+ cache_file = cache._key_to_file("foo")
+ with open(cache_file, "wb") as fh:
+ fh.write(b"")
+ with open(cache_file, "rb") as fh:
self.assertIs(cache._is_expired(fh), True)
@unittest.skipUnless(RedisCache_params, "Redis backend not configured")
-@override_settings(CACHES=caches_setting_for_tests(
- base=RedisCache_params,
- exclude=redis_excluded_caches,
-))
+@override_settings(
+ CACHES=caches_setting_for_tests(
+ base=RedisCache_params,
+ exclude=redis_excluded_caches,
+ )
+)
class RedisCacheTests(BaseCacheTests, TestCase):
-
def setUp(self):
import redis
+
super().setUp()
self.lib = redis
@@ -1718,7 +1815,7 @@ class RedisCacheTests(BaseCacheTests, TestCase):
def test_serializer_dumps(self):
self.assertEqual(cache._cache._serializer.dumps(123), 123)
self.assertIsInstance(cache._cache._serializer.dumps(True), bytes)
- self.assertIsInstance(cache._cache._serializer.dumps('abc'), bytes)
+ self.assertIsInstance(cache._cache._serializer.dumps("abc"), bytes)
class FileBasedCachePathLibTests(FileBasedCacheTests):
@@ -1727,61 +1824,65 @@ class FileBasedCachePathLibTests(FileBasedCacheTests):
return Path(tmp_dir)
-@override_settings(CACHES={
- 'default': {
- 'BACKEND': 'cache.liberal_backend.CacheClass',
- },
-})
+@override_settings(
+ CACHES={
+ "default": {
+ "BACKEND": "cache.liberal_backend.CacheClass",
+ },
+ }
+)
class CustomCacheKeyValidationTests(SimpleTestCase):
"""
Tests for the ability to mixin a custom ``validate_key`` method to
a custom cache backend that otherwise inherits from a builtin
backend, and override the default key validation. Refs #6447.
"""
+
def test_custom_key_validation(self):
# this key is both longer than 250 characters, and has spaces
- key = 'some key with spaces' * 15
- val = 'a value'
+ key = "some key with spaces" * 15
+ val = "a value"
cache.set(key, val)
self.assertEqual(cache.get(key), val)
@override_settings(
CACHES={
- 'default': {
- 'BACKEND': 'cache.closeable_cache.CacheClass',
+ "default": {
+ "BACKEND": "cache.closeable_cache.CacheClass",
}
}
)
class CacheClosingTests(SimpleTestCase):
-
def test_close(self):
self.assertFalse(cache.closed)
signals.request_finished.send(self.__class__)
self.assertTrue(cache.closed)
def test_close_only_initialized(self):
- with self.settings(CACHES={
- 'cache_1': {
- 'BACKEND': 'cache.closeable_cache.CacheClass',
- },
- 'cache_2': {
- 'BACKEND': 'cache.closeable_cache.CacheClass',
- },
- }):
+ with self.settings(
+ CACHES={
+ "cache_1": {
+ "BACKEND": "cache.closeable_cache.CacheClass",
+ },
+ "cache_2": {
+ "BACKEND": "cache.closeable_cache.CacheClass",
+ },
+ }
+ ):
self.assertEqual(caches.all(initialized_only=True), [])
signals.request_finished.send(self.__class__)
self.assertEqual(caches.all(initialized_only=True), [])
DEFAULT_MEMORY_CACHES_SETTINGS = {
- 'default': {
- 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
- 'LOCATION': 'unique-snowflake',
+ "default": {
+ "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
+ "LOCATION": "unique-snowflake",
}
}
NEVER_EXPIRING_CACHES_SETTINGS = copy.deepcopy(DEFAULT_MEMORY_CACHES_SETTINGS)
-NEVER_EXPIRING_CACHES_SETTINGS['default']['TIMEOUT'] = None
+NEVER_EXPIRING_CACHES_SETTINGS["default"]["TIMEOUT"] = None
class DefaultNonExpiringCacheKeyTests(SimpleTestCase):
@@ -1789,6 +1890,7 @@ class DefaultNonExpiringCacheKeyTests(SimpleTestCase):
Settings having Cache arguments with a TIMEOUT=None create Caches that will
set non-expiring keys.
"""
+
def setUp(self):
# The 5 minute (300 seconds) default expiration time for keys is
# defined in the implementation of the initializer method of the
@@ -1796,7 +1898,7 @@ class DefaultNonExpiringCacheKeyTests(SimpleTestCase):
self.DEFAULT_TIMEOUT = caches[DEFAULT_CACHE_ALIAS].default_timeout
def tearDown(self):
- del(self.DEFAULT_TIMEOUT)
+ del self.DEFAULT_TIMEOUT
def test_default_expiration_time_for_keys_is_5_minutes(self):
"""The default expiration time of a cache key is 5 minutes.
@@ -1850,53 +1952,71 @@ class DefaultNonExpiringCacheKeyTests(SimpleTestCase):
@override_settings(
- CACHE_MIDDLEWARE_KEY_PREFIX='settingsprefix',
+ CACHE_MIDDLEWARE_KEY_PREFIX="settingsprefix",
CACHE_MIDDLEWARE_SECONDS=1,
CACHES={
- 'default': {
- 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
+ "default": {
+ "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
},
},
USE_I18N=False,
- ALLOWED_HOSTS=['.example.com'],
+ ALLOWED_HOSTS=[".example.com"],
)
class CacheUtils(SimpleTestCase):
"""TestCase for django.utils.cache functions."""
- host = 'www.example.com'
- path = '/cache/test/'
+
+ host = "www.example.com"
+ path = "/cache/test/"
factory = RequestFactory(HTTP_HOST=host)
def tearDown(self):
cache.clear()
- def _get_request_cache(self, method='GET', query_string=None, update_cache=None):
- request = self._get_request(self.host, self.path,
- method, query_string=query_string)
+ def _get_request_cache(self, method="GET", query_string=None, update_cache=None):
+ request = self._get_request(
+ self.host, self.path, method, query_string=query_string
+ )
request._cache_update_cache = update_cache if update_cache else True
return request
def test_patch_vary_headers(self):
headers = (
# Initial vary, new headers, resulting vary.
- (None, ('Accept-Encoding',), 'Accept-Encoding'),
- ('Accept-Encoding', ('accept-encoding',), 'Accept-Encoding'),
- ('Accept-Encoding', ('ACCEPT-ENCODING',), 'Accept-Encoding'),
- ('Cookie', ('Accept-Encoding',), 'Cookie, Accept-Encoding'),
- ('Cookie, Accept-Encoding', ('Accept-Encoding',), 'Cookie, Accept-Encoding'),
- ('Cookie, Accept-Encoding', ('Accept-Encoding', 'cookie'), 'Cookie, Accept-Encoding'),
- (None, ('Accept-Encoding', 'COOKIE'), 'Accept-Encoding, COOKIE'),
- ('Cookie, Accept-Encoding', ('Accept-Encoding', 'cookie'), 'Cookie, Accept-Encoding'),
- ('Cookie , Accept-Encoding', ('Accept-Encoding', 'cookie'), 'Cookie, Accept-Encoding'),
- ('*', ('Accept-Language', 'Cookie'), '*'),
- ('Accept-Language, Cookie', ('*',), '*'),
+ (None, ("Accept-Encoding",), "Accept-Encoding"),
+ ("Accept-Encoding", ("accept-encoding",), "Accept-Encoding"),
+ ("Accept-Encoding", ("ACCEPT-ENCODING",), "Accept-Encoding"),
+ ("Cookie", ("Accept-Encoding",), "Cookie, Accept-Encoding"),
+ (
+ "Cookie, Accept-Encoding",
+ ("Accept-Encoding",),
+ "Cookie, Accept-Encoding",
+ ),
+ (
+ "Cookie, Accept-Encoding",
+ ("Accept-Encoding", "cookie"),
+ "Cookie, Accept-Encoding",
+ ),
+ (None, ("Accept-Encoding", "COOKIE"), "Accept-Encoding, COOKIE"),
+ (
+ "Cookie, Accept-Encoding",
+ ("Accept-Encoding", "cookie"),
+ "Cookie, Accept-Encoding",
+ ),
+ (
+ "Cookie , Accept-Encoding",
+ ("Accept-Encoding", "cookie"),
+ "Cookie, Accept-Encoding",
+ ),
+ ("*", ("Accept-Language", "Cookie"), "*"),
+ ("Accept-Language, Cookie", ("*",), "*"),
)
for initial_vary, newheaders, resulting_vary in headers:
with self.subTest(initial_vary=initial_vary, newheaders=newheaders):
response = HttpResponse()
if initial_vary is not None:
- response.headers['Vary'] = initial_vary
+ response.headers["Vary"] = initial_vary
patch_vary_headers(response, newheaders)
- self.assertEqual(response.headers['Vary'], resulting_vary)
+ self.assertEqual(response.headers["Vary"], resulting_vary)
def test_get_cache_key(self):
request = self.factory.get(self.path)
@@ -1908,20 +2028,20 @@ class CacheUtils(SimpleTestCase):
self.assertEqual(
get_cache_key(request),
- 'views.decorators.cache.cache_page.settingsprefix.GET.'
- '18a03f9c9649f7d684af5db3524f5c99.d41d8cd98f00b204e9800998ecf8427e'
+ "views.decorators.cache.cache_page.settingsprefix.GET."
+ "18a03f9c9649f7d684af5db3524f5c99.d41d8cd98f00b204e9800998ecf8427e",
)
# A specified key_prefix is taken into account.
- key_prefix = 'localprefix'
+ key_prefix = "localprefix"
learn_cache_key(request, response, key_prefix=key_prefix)
self.assertEqual(
get_cache_key(request, key_prefix=key_prefix),
- 'views.decorators.cache.cache_page.localprefix.GET.'
- '18a03f9c9649f7d684af5db3524f5c99.d41d8cd98f00b204e9800998ecf8427e'
+ "views.decorators.cache.cache_page.localprefix.GET."
+ "18a03f9c9649f7d684af5db3524f5c99.d41d8cd98f00b204e9800998ecf8427e",
)
def test_get_cache_key_with_query(self):
- request = self.factory.get(self.path, {'test': 1})
+ request = self.factory.get(self.path, {"test": 1})
response = HttpResponse()
# Expect None if no headers have been set yet.
self.assertIsNone(get_cache_key(request))
@@ -1930,76 +2050,92 @@ class CacheUtils(SimpleTestCase):
# The querystring is taken into account.
self.assertEqual(
get_cache_key(request),
- 'views.decorators.cache.cache_page.settingsprefix.GET.'
- 'beaf87a9a99ee81c673ea2d67ccbec2a.d41d8cd98f00b204e9800998ecf8427e'
+ "views.decorators.cache.cache_page.settingsprefix.GET."
+ "beaf87a9a99ee81c673ea2d67ccbec2a.d41d8cd98f00b204e9800998ecf8427e",
)
def test_cache_key_varies_by_url(self):
"""
get_cache_key keys differ by fully-qualified URL instead of path
"""
- request1 = self.factory.get(self.path, HTTP_HOST='sub-1.example.com')
+ request1 = self.factory.get(self.path, HTTP_HOST="sub-1.example.com")
learn_cache_key(request1, HttpResponse())
- request2 = self.factory.get(self.path, HTTP_HOST='sub-2.example.com')
+ request2 = self.factory.get(self.path, HTTP_HOST="sub-2.example.com")
learn_cache_key(request2, HttpResponse())
self.assertNotEqual(get_cache_key(request1), get_cache_key(request2))
def test_learn_cache_key(self):
request = self.factory.head(self.path)
response = HttpResponse()
- response.headers['Vary'] = 'Pony'
+ response.headers["Vary"] = "Pony"
# Make sure that the Vary header is added to the key hash
learn_cache_key(request, response)
self.assertEqual(
get_cache_key(request),
- 'views.decorators.cache.cache_page.settingsprefix.GET.'
- '18a03f9c9649f7d684af5db3524f5c99.d41d8cd98f00b204e9800998ecf8427e'
+ "views.decorators.cache.cache_page.settingsprefix.GET."
+ "18a03f9c9649f7d684af5db3524f5c99.d41d8cd98f00b204e9800998ecf8427e",
)
def test_patch_cache_control(self):
tests = (
# Initial Cache-Control, kwargs to patch_cache_control, expected Cache-Control parts
- (None, {'private': True}, {'private'}),
- ('', {'private': True}, {'private'}),
+ (None, {"private": True}, {"private"}),
+ ("", {"private": True}, {"private"}),
# no-cache.
- ('', {'no_cache': 'Set-Cookie'}, {'no-cache=Set-Cookie'}),
- ('', {'no-cache': 'Set-Cookie'}, {'no-cache=Set-Cookie'}),
- ('no-cache=Set-Cookie', {'no_cache': True}, {'no-cache'}),
- ('no-cache=Set-Cookie,no-cache=Link', {'no_cache': True}, {'no-cache'}),
- ('no-cache=Set-Cookie', {'no_cache': 'Link'}, {'no-cache=Set-Cookie', 'no-cache=Link'}),
+ ("", {"no_cache": "Set-Cookie"}, {"no-cache=Set-Cookie"}),
+ ("", {"no-cache": "Set-Cookie"}, {"no-cache=Set-Cookie"}),
+ ("no-cache=Set-Cookie", {"no_cache": True}, {"no-cache"}),
+ ("no-cache=Set-Cookie,no-cache=Link", {"no_cache": True}, {"no-cache"}),
(
- 'no-cache=Set-Cookie,no-cache=Link',
- {'no_cache': 'Custom'},
- {'no-cache=Set-Cookie', 'no-cache=Link', 'no-cache=Custom'},
+ "no-cache=Set-Cookie",
+ {"no_cache": "Link"},
+ {"no-cache=Set-Cookie", "no-cache=Link"},
+ ),
+ (
+ "no-cache=Set-Cookie,no-cache=Link",
+ {"no_cache": "Custom"},
+ {"no-cache=Set-Cookie", "no-cache=Link", "no-cache=Custom"},
),
# Test whether private/public attributes are mutually exclusive
- ('private', {'private': True}, {'private'}),
- ('private', {'public': True}, {'public'}),
- ('public', {'public': True}, {'public'}),
- ('public', {'private': True}, {'private'}),
- ('must-revalidate,max-age=60,private', {'public': True}, {'must-revalidate', 'max-age=60', 'public'}),
- ('must-revalidate,max-age=60,public', {'private': True}, {'must-revalidate', 'max-age=60', 'private'}),
- ('must-revalidate,max-age=60', {'public': True}, {'must-revalidate', 'max-age=60', 'public'}),
+ ("private", {"private": True}, {"private"}),
+ ("private", {"public": True}, {"public"}),
+ ("public", {"public": True}, {"public"}),
+ ("public", {"private": True}, {"private"}),
+ (
+ "must-revalidate,max-age=60,private",
+ {"public": True},
+ {"must-revalidate", "max-age=60", "public"},
+ ),
+ (
+ "must-revalidate,max-age=60,public",
+ {"private": True},
+ {"must-revalidate", "max-age=60", "private"},
+ ),
+ (
+ "must-revalidate,max-age=60",
+ {"public": True},
+ {"must-revalidate", "max-age=60", "public"},
+ ),
)
- cc_delim_re = re.compile(r'\s*,\s*')
+ cc_delim_re = re.compile(r"\s*,\s*")
for initial_cc, newheaders, expected_cc in tests:
with self.subTest(initial_cc=initial_cc, newheaders=newheaders):
response = HttpResponse()
if initial_cc is not None:
- response.headers['Cache-Control'] = initial_cc
+ response.headers["Cache-Control"] = initial_cc
patch_cache_control(response, **newheaders)
- parts = set(cc_delim_re.split(response.headers['Cache-Control']))
+ parts = set(cc_delim_re.split(response.headers["Cache-Control"]))
self.assertEqual(parts, expected_cc)
@override_settings(
CACHES={
- 'default': {
- 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
- 'KEY_PREFIX': 'cacheprefix',
+ "default": {
+ "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
+ "KEY_PREFIX": "cacheprefix",
},
},
)
@@ -2009,15 +2145,15 @@ class PrefixedCacheUtils(CacheUtils):
@override_settings(
CACHE_MIDDLEWARE_SECONDS=60,
- CACHE_MIDDLEWARE_KEY_PREFIX='test',
+ CACHE_MIDDLEWARE_KEY_PREFIX="test",
CACHES={
- 'default': {
- 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
+ "default": {
+ "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
},
},
)
class CacheHEADTest(SimpleTestCase):
- path = '/cache/test/'
+ path = "/cache/test/"
factory = RequestFactory()
def tearDown(self):
@@ -2027,7 +2163,7 @@ class CacheHEADTest(SimpleTestCase):
return UpdateCacheMiddleware(lambda req: HttpResponse(msg))(request)
def test_head_caches_correctly(self):
- test_content = 'test content'
+ test_content = "test content"
request = self.factory.head(self.path)
request._cache_update_cache = True
@@ -2035,37 +2171,41 @@ class CacheHEADTest(SimpleTestCase):
request = self.factory.head(self.path)
request._cache_update_cache = True
- get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
+ get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
+ request
+ )
self.assertIsNotNone(get_cache_data)
self.assertEqual(test_content.encode(), get_cache_data.content)
def test_head_with_cached_get(self):
- test_content = 'test content'
+ test_content = "test content"
request = self.factory.get(self.path)
request._cache_update_cache = True
self._set_cache(request, test_content)
request = self.factory.head(self.path)
- get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
+ get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
+ request
+ )
self.assertIsNotNone(get_cache_data)
self.assertEqual(test_content.encode(), get_cache_data.content)
@override_settings(
- CACHE_MIDDLEWARE_KEY_PREFIX='settingsprefix',
+ CACHE_MIDDLEWARE_KEY_PREFIX="settingsprefix",
CACHES={
- 'default': {
- 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
+ "default": {
+ "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
},
},
LANGUAGES=[
- ('en', 'English'),
- ('es', 'Spanish'),
+ ("en", "English"),
+ ("es", "Spanish"),
],
)
class CacheI18nTest(SimpleTestCase):
- path = '/cache/test/'
+ path = "/cache/test/"
factory = RequestFactory()
def tearDown(self):
@@ -2077,16 +2217,20 @@ class CacheI18nTest(SimpleTestCase):
lang = translation.get_language()
response = HttpResponse()
key = learn_cache_key(request, response)
- self.assertIn(lang, key, "Cache keys should include the language name when translation is active")
+ self.assertIn(
+ lang,
+ key,
+ "Cache keys should include the language name when translation is active",
+ )
key2 = get_cache_key(request)
self.assertEqual(key, key2)
def check_accept_language_vary(self, accept_language, vary, reference_key):
request = self.factory.get(self.path)
- request.META['HTTP_ACCEPT_LANGUAGE'] = accept_language
- request.META['HTTP_ACCEPT_ENCODING'] = 'gzip;q=1.0, identity; q=0.5, *;q=0'
+ request.META["HTTP_ACCEPT_LANGUAGE"] = accept_language
+ request.META["HTTP_ACCEPT_ENCODING"] = "gzip;q=1.0, identity; q=0.5, *;q=0"
response = HttpResponse()
- response.headers['Vary'] = vary
+ response.headers["Vary"] = vary
key = learn_cache_key(request, response)
key2 = get_cache_key(request)
self.assertEqual(key, reference_key)
@@ -2095,57 +2239,47 @@ class CacheI18nTest(SimpleTestCase):
@override_settings(USE_I18N=True, USE_TZ=False)
def test_cache_key_i18n_translation_accept_language(self):
lang = translation.get_language()
- self.assertEqual(lang, 'en')
+ self.assertEqual(lang, "en")
request = self.factory.get(self.path)
- request.META['HTTP_ACCEPT_ENCODING'] = 'gzip;q=1.0, identity; q=0.5, *;q=0'
+ request.META["HTTP_ACCEPT_ENCODING"] = "gzip;q=1.0, identity; q=0.5, *;q=0"
response = HttpResponse()
- response.headers['Vary'] = 'accept-encoding'
+ response.headers["Vary"] = "accept-encoding"
key = learn_cache_key(request, response)
- self.assertIn(lang, key, "Cache keys should include the language name when translation is active")
+ self.assertIn(
+ lang,
+ key,
+ "Cache keys should include the language name when translation is active",
+ )
self.check_accept_language_vary(
- 'en-us',
- 'cookie, accept-language, accept-encoding',
- key
+ "en-us", "cookie, accept-language, accept-encoding", key
)
self.check_accept_language_vary(
- 'en-US',
- 'cookie, accept-encoding, accept-language',
- key
+ "en-US", "cookie, accept-encoding, accept-language", key
)
self.check_accept_language_vary(
- 'en-US,en;q=0.8',
- 'accept-encoding, accept-language, cookie',
- key
+ "en-US,en;q=0.8", "accept-encoding, accept-language, cookie", key
)
self.check_accept_language_vary(
- 'en-US,en;q=0.8,ko;q=0.6',
- 'accept-language, cookie, accept-encoding',
- key
+ "en-US,en;q=0.8,ko;q=0.6", "accept-language, cookie, accept-encoding", key
)
self.check_accept_language_vary(
- 'ko-kr,ko;q=0.8,en-us;q=0.5,en;q=0.3 ',
- 'accept-encoding, cookie, accept-language',
- key
+ "ko-kr,ko;q=0.8,en-us;q=0.5,en;q=0.3 ",
+ "accept-encoding, cookie, accept-language",
+ key,
)
self.check_accept_language_vary(
- 'ko-KR,ko;q=0.8,en-US;q=0.6,en;q=0.4',
- 'accept-language, accept-encoding, cookie',
- key
+ "ko-KR,ko;q=0.8,en-US;q=0.6,en;q=0.4",
+ "accept-language, accept-encoding, cookie",
+ key,
)
self.check_accept_language_vary(
- 'ko;q=1.0,en;q=0.5',
- 'cookie, accept-language, accept-encoding',
- key
+ "ko;q=1.0,en;q=0.5", "cookie, accept-language, accept-encoding", key
)
self.check_accept_language_vary(
- 'ko, en',
- 'cookie, accept-encoding, accept-language',
- key
+ "ko, en", "cookie, accept-encoding, accept-language", key
)
self.check_accept_language_vary(
- 'ko-KR, en-US',
- 'accept-encoding, accept-language, cookie',
- key
+ "ko-KR, en-US", "accept-encoding, accept-language, cookie", key
)
@override_settings(USE_I18N=False, USE_TZ=True)
@@ -2154,7 +2288,11 @@ class CacheI18nTest(SimpleTestCase):
tz = timezone.get_current_timezone_name()
response = HttpResponse()
key = learn_cache_key(request, response)
- self.assertIn(tz, key, "Cache keys should include the time zone name when time zones are active")
+ self.assertIn(
+ tz,
+ key,
+ "Cache keys should include the time zone name when time zones are active",
+ )
key2 = get_cache_key(request)
self.assertEqual(key, key2)
@@ -2165,8 +2303,16 @@ class CacheI18nTest(SimpleTestCase):
tz = timezone.get_current_timezone_name()
response = HttpResponse()
key = learn_cache_key(request, response)
- self.assertNotIn(lang, key, "Cache keys shouldn't include the language name when i18n isn't active")
- self.assertNotIn(tz, key, "Cache keys shouldn't include the time zone name when i18n isn't active")
+ self.assertNotIn(
+ lang,
+ key,
+ "Cache keys shouldn't include the language name when i18n isn't active",
+ )
+ self.assertNotIn(
+ tz,
+ key,
+ "Cache keys shouldn't include the time zone name when i18n isn't active",
+ )
@override_settings(
CACHE_MIDDLEWARE_KEY_PREFIX="test",
@@ -2182,26 +2328,32 @@ class CacheI18nTest(SimpleTestCase):
return UpdateCacheMiddleware(get_response)(request)
# cache with non empty request.GET
- request = self.factory.get(self.path, {'foo': 'bar', 'other': 'true'})
+ request = self.factory.get(self.path, {"foo": "bar", "other": "true"})
request._cache_update_cache = True
- get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
+ get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
+ request
+ )
# first access, cache must return None
self.assertIsNone(get_cache_data)
- content = 'Check for cache with QUERY_STRING'
+ content = "Check for cache with QUERY_STRING"
def get_response(req):
return HttpResponse(content)
UpdateCacheMiddleware(get_response)(request)
- get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
+ get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
+ request
+ )
# cache must return content
self.assertIsNotNone(get_cache_data)
self.assertEqual(get_cache_data.content, content.encode())
# different QUERY_STRING, cache must be empty
- request = self.factory.get(self.path, {'foo': 'bar', 'somethingelse': 'true'})
+ request = self.factory.get(self.path, {"foo": "bar", "somethingelse": "true"})
request._cache_update_cache = True
- get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
+ get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
+ request
+ )
self.assertIsNone(get_cache_data)
# i18n tests
@@ -2210,23 +2362,29 @@ class CacheI18nTest(SimpleTestCase):
request = self.factory.get(self.path)
request._cache_update_cache = True
- set_cache(request, 'en', en_message)
- get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
+ set_cache(request, "en", en_message)
+ get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
+ request
+ )
# The cache can be recovered
self.assertIsNotNone(get_cache_data)
self.assertEqual(get_cache_data.content, en_message.encode())
# change the session language and set content
request = self.factory.get(self.path)
request._cache_update_cache = True
- set_cache(request, 'es', es_message)
+ set_cache(request, "es", es_message)
# change again the language
- translation.activate('en')
+ translation.activate("en")
# retrieve the content from cache
- get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
+ get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
+ request
+ )
self.assertEqual(get_cache_data.content, en_message.encode())
# change again the language
- translation.activate('es')
- get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
+ translation.activate("es")
+ get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
+ request
+ )
self.assertEqual(get_cache_data.content, es_message.encode())
# reset the language
translation.deactivate()
@@ -2237,23 +2395,27 @@ class CacheI18nTest(SimpleTestCase):
)
def test_middleware_doesnt_cache_streaming_response(self):
request = self.factory.get(self.path)
- get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
+ get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
+ request
+ )
self.assertIsNone(get_cache_data)
def get_stream_response(req):
- return StreamingHttpResponse(['Check for cache with streaming content.'])
+ return StreamingHttpResponse(["Check for cache with streaming content."])
UpdateCacheMiddleware(get_stream_response)(request)
- get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
+ get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(
+ request
+ )
self.assertIsNone(get_cache_data)
@override_settings(
CACHES={
- 'default': {
- 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
- 'KEY_PREFIX': 'cacheprefix'
+ "default": {
+ "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
+ "KEY_PREFIX": "cacheprefix",
},
},
)
@@ -2262,25 +2424,25 @@ class PrefixedCacheI18nTest(CacheI18nTest):
def hello_world_view(request, value):
- return HttpResponse('Hello World %s' % value)
+ return HttpResponse("Hello World %s" % value)
def csrf_view(request):
- return HttpResponse(csrf(request)['csrf_token'])
+ return HttpResponse(csrf(request)["csrf_token"])
@override_settings(
- CACHE_MIDDLEWARE_ALIAS='other',
- CACHE_MIDDLEWARE_KEY_PREFIX='middlewareprefix',
+ CACHE_MIDDLEWARE_ALIAS="other",
+ CACHE_MIDDLEWARE_KEY_PREFIX="middlewareprefix",
CACHE_MIDDLEWARE_SECONDS=30,
CACHES={
- 'default': {
- 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
+ "default": {
+ "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
},
- 'other': {
- 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
- 'LOCATION': 'other',
- 'TIMEOUT': '1',
+ "other": {
+ "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
+ "LOCATION": "other",
+ "TIMEOUT": "1",
},
},
)
@@ -2288,8 +2450,8 @@ class CacheMiddlewareTest(SimpleTestCase):
factory = RequestFactory()
def setUp(self):
- self.default_cache = caches['default']
- self.other_cache = caches['other']
+ self.default_cache = caches["default"]
+ self.other_cache = caches["other"]
def tearDown(self):
self.default_cache.clear()
@@ -2308,56 +2470,60 @@ class CacheMiddlewareTest(SimpleTestCase):
# Now test object attributes against values defined in setUp above
self.assertEqual(middleware.cache_timeout, 30)
- self.assertEqual(middleware.key_prefix, 'middlewareprefix')
- self.assertEqual(middleware.cache_alias, 'other')
+ self.assertEqual(middleware.key_prefix, "middlewareprefix")
+ self.assertEqual(middleware.cache_alias, "other")
self.assertEqual(middleware.cache, self.other_cache)
# If more arguments are being passed in construction, it's being used
# as a decorator. First, test with "defaults":
- as_view_decorator = CacheMiddleware(empty_response, cache_alias=None, key_prefix=None)
+ as_view_decorator = CacheMiddleware(
+ empty_response, cache_alias=None, key_prefix=None
+ )
- self.assertEqual(as_view_decorator.cache_timeout, 30) # Timeout value for 'default' cache, i.e. 30
- self.assertEqual(as_view_decorator.key_prefix, '')
+ self.assertEqual(
+ as_view_decorator.cache_timeout, 30
+ ) # Timeout value for 'default' cache, i.e. 30
+ self.assertEqual(as_view_decorator.key_prefix, "")
# Value of DEFAULT_CACHE_ALIAS from django.core.cache
- self.assertEqual(as_view_decorator.cache_alias, 'default')
+ self.assertEqual(as_view_decorator.cache_alias, "default")
self.assertEqual(as_view_decorator.cache, self.default_cache)
# Next, test with custom values:
as_view_decorator_with_custom = CacheMiddleware(
- hello_world_view, cache_timeout=60, cache_alias='other', key_prefix='foo'
+ hello_world_view, cache_timeout=60, cache_alias="other", key_prefix="foo"
)
self.assertEqual(as_view_decorator_with_custom.cache_timeout, 60)
- self.assertEqual(as_view_decorator_with_custom.key_prefix, 'foo')
- self.assertEqual(as_view_decorator_with_custom.cache_alias, 'other')
+ self.assertEqual(as_view_decorator_with_custom.key_prefix, "foo")
+ self.assertEqual(as_view_decorator_with_custom.cache_alias, "other")
self.assertEqual(as_view_decorator_with_custom.cache, self.other_cache)
def test_update_cache_middleware_constructor(self):
middleware = UpdateCacheMiddleware(empty_response)
self.assertEqual(middleware.cache_timeout, 30)
self.assertIsNone(middleware.page_timeout)
- self.assertEqual(middleware.key_prefix, 'middlewareprefix')
- self.assertEqual(middleware.cache_alias, 'other')
+ self.assertEqual(middleware.key_prefix, "middlewareprefix")
+ self.assertEqual(middleware.cache_alias, "other")
self.assertEqual(middleware.cache, self.other_cache)
def test_fetch_cache_middleware_constructor(self):
middleware = FetchFromCacheMiddleware(empty_response)
- self.assertEqual(middleware.key_prefix, 'middlewareprefix')
- self.assertEqual(middleware.cache_alias, 'other')
+ self.assertEqual(middleware.key_prefix, "middlewareprefix")
+ self.assertEqual(middleware.cache_alias, "other")
self.assertEqual(middleware.cache, self.other_cache)
def test_middleware(self):
middleware = CacheMiddleware(hello_world_view)
- prefix_middleware = CacheMiddleware(hello_world_view, key_prefix='prefix1')
+ prefix_middleware = CacheMiddleware(hello_world_view, key_prefix="prefix1")
timeout_middleware = CacheMiddleware(hello_world_view, cache_timeout=1)
- request = self.factory.get('/view/')
+ request = self.factory.get("/view/")
# Put the request through the request middleware
result = middleware.process_request(request)
self.assertIsNone(result)
- response = hello_world_view(request, '1')
+ response = hello_world_view(request, "1")
# Now put the response through the response middleware
response = middleware.process_response(request, response)
@@ -2365,7 +2531,7 @@ class CacheMiddlewareTest(SimpleTestCase):
# Repeating the request should result in a cache hit
result = middleware.process_request(request)
self.assertIsNotNone(result)
- self.assertEqual(result.content, b'Hello World 1')
+ self.assertEqual(result.content, b"Hello World 1")
# The same request through a different middleware won't hit
result = prefix_middleware.process_request(request)
@@ -2374,84 +2540,88 @@ class CacheMiddlewareTest(SimpleTestCase):
# The same request with a timeout _will_ hit
result = timeout_middleware.process_request(request)
self.assertIsNotNone(result)
- self.assertEqual(result.content, b'Hello World 1')
+ self.assertEqual(result.content, b"Hello World 1")
def test_view_decorator(self):
# decorate the same view with different cache decorators
default_view = cache_page(3)(hello_world_view)
- default_with_prefix_view = cache_page(3, key_prefix='prefix1')(hello_world_view)
+ default_with_prefix_view = cache_page(3, key_prefix="prefix1")(hello_world_view)
- explicit_default_view = cache_page(3, cache='default')(hello_world_view)
- explicit_default_with_prefix_view = cache_page(3, cache='default', key_prefix='prefix1')(hello_world_view)
+ explicit_default_view = cache_page(3, cache="default")(hello_world_view)
+ explicit_default_with_prefix_view = cache_page(
+ 3, cache="default", key_prefix="prefix1"
+ )(hello_world_view)
- other_view = cache_page(1, cache='other')(hello_world_view)
- other_with_prefix_view = cache_page(1, cache='other', key_prefix='prefix2')(hello_world_view)
+ other_view = cache_page(1, cache="other")(hello_world_view)
+ other_with_prefix_view = cache_page(1, cache="other", key_prefix="prefix2")(
+ hello_world_view
+ )
- request = self.factory.get('/view/')
+ request = self.factory.get("/view/")
# Request the view once
- response = default_view(request, '1')
- self.assertEqual(response.content, b'Hello World 1')
+ response = default_view(request, "1")
+ self.assertEqual(response.content, b"Hello World 1")
# Request again -- hit the cache
- response = default_view(request, '2')
- self.assertEqual(response.content, b'Hello World 1')
+ response = default_view(request, "2")
+ self.assertEqual(response.content, b"Hello World 1")
# Requesting the same view with the explicit cache should yield the same result
- response = explicit_default_view(request, '3')
- self.assertEqual(response.content, b'Hello World 1')
+ response = explicit_default_view(request, "3")
+ self.assertEqual(response.content, b"Hello World 1")
# Requesting with a prefix will hit a different cache key
- response = explicit_default_with_prefix_view(request, '4')
- self.assertEqual(response.content, b'Hello World 4')
+ response = explicit_default_with_prefix_view(request, "4")
+ self.assertEqual(response.content, b"Hello World 4")
# Hitting the same view again gives a cache hit
- response = explicit_default_with_prefix_view(request, '5')
- self.assertEqual(response.content, b'Hello World 4')
+ response = explicit_default_with_prefix_view(request, "5")
+ self.assertEqual(response.content, b"Hello World 4")
# And going back to the implicit cache will hit the same cache
- response = default_with_prefix_view(request, '6')
- self.assertEqual(response.content, b'Hello World 4')
+ response = default_with_prefix_view(request, "6")
+ self.assertEqual(response.content, b"Hello World 4")
# Requesting from an alternate cache won't hit cache
- response = other_view(request, '7')
- self.assertEqual(response.content, b'Hello World 7')
+ response = other_view(request, "7")
+ self.assertEqual(response.content, b"Hello World 7")
# But a repeated hit will hit cache
- response = other_view(request, '8')
- self.assertEqual(response.content, b'Hello World 7')
+ response = other_view(request, "8")
+ self.assertEqual(response.content, b"Hello World 7")
# And prefixing the alternate cache yields yet another cache entry
- response = other_with_prefix_view(request, '9')
- self.assertEqual(response.content, b'Hello World 9')
+ response = other_with_prefix_view(request, "9")
+ self.assertEqual(response.content, b"Hello World 9")
# But if we wait a couple of seconds...
time.sleep(2)
# ... the default cache will still hit
- caches['default']
- response = default_view(request, '11')
- self.assertEqual(response.content, b'Hello World 1')
+ caches["default"]
+ response = default_view(request, "11")
+ self.assertEqual(response.content, b"Hello World 1")
# ... the default cache with a prefix will still hit
- response = default_with_prefix_view(request, '12')
- self.assertEqual(response.content, b'Hello World 4')
+ response = default_with_prefix_view(request, "12")
+ self.assertEqual(response.content, b"Hello World 4")
# ... the explicit default cache will still hit
- response = explicit_default_view(request, '13')
- self.assertEqual(response.content, b'Hello World 1')
+ response = explicit_default_view(request, "13")
+ self.assertEqual(response.content, b"Hello World 1")
# ... the explicit default cache with a prefix will still hit
- response = explicit_default_with_prefix_view(request, '14')
- self.assertEqual(response.content, b'Hello World 4')
+ response = explicit_default_with_prefix_view(request, "14")
+ self.assertEqual(response.content, b"Hello World 4")
# .. but a rapidly expiring cache won't hit
- response = other_view(request, '15')
- self.assertEqual(response.content, b'Hello World 15')
+ response = other_view(request, "15")
+ self.assertEqual(response.content, b"Hello World 15")
# .. even if it has a prefix
- response = other_with_prefix_view(request, '16')
- self.assertEqual(response.content, b'Hello World 16')
+ response = other_with_prefix_view(request, "16")
+ self.assertEqual(response.content, b"Hello World 16")
def test_cache_page_timeout(self):
# Page timeout takes precedence over the "max-age" section of the
@@ -2465,32 +2635,34 @@ class CacheMiddlewareTest(SimpleTestCase):
view = cache_page(timeout=page_timeout)(
cache_control(max_age=max_age)(hello_world_view)
)
- request = self.factory.get('/view/')
- response = view(request, '1')
- self.assertEqual(response.content, b'Hello World 1')
+ request = self.factory.get("/view/")
+ response = view(request, "1")
+ self.assertEqual(response.content, b"Hello World 1")
time.sleep(1)
- response = view(request, '2')
+ response = view(request, "2")
self.assertEqual(
response.content,
- b'Hello World 1' if page_timeout > max_age else b'Hello World 2',
+ b"Hello World 1" if page_timeout > max_age else b"Hello World 2",
)
cache.clear()
def test_cached_control_private_not_cached(self):
"""Responses with 'Cache-Control: private' are not cached."""
- view_with_private_cache = cache_page(3)(cache_control(private=True)(hello_world_view))
- request = self.factory.get('/view/')
- response = view_with_private_cache(request, '1')
- self.assertEqual(response.content, b'Hello World 1')
- response = view_with_private_cache(request, '2')
- self.assertEqual(response.content, b'Hello World 2')
+ view_with_private_cache = cache_page(3)(
+ cache_control(private=True)(hello_world_view)
+ )
+ request = self.factory.get("/view/")
+ response = view_with_private_cache(request, "1")
+ self.assertEqual(response.content, b"Hello World 1")
+ response = view_with_private_cache(request, "2")
+ self.assertEqual(response.content, b"Hello World 2")
def test_sensitive_cookie_not_cached(self):
"""
Django must prevent caching of responses that set a user-specific (and
maybe security sensitive) cookie in response to a cookie-less request.
"""
- request = self.factory.get('/view/')
+ request = self.factory.get("/view/")
csrf_middleware = CsrfViewMiddleware(csrf_view)
csrf_middleware.process_view(request, csrf_view, (), {})
cache_middleware = CacheMiddleware(csrf_middleware)
@@ -2504,15 +2676,15 @@ class CacheMiddlewareTest(SimpleTestCase):
def test_304_response_has_http_caching_headers_but_not_cached(self):
original_view = mock.Mock(return_value=HttpResponseNotModified())
view = cache_page(2)(original_view)
- request = self.factory.get('/view/')
+ request = self.factory.get("/view/")
# The view shouldn't be cached on the second call.
view(request).close()
response = view(request)
response.close()
self.assertEqual(original_view.call_count, 2)
self.assertIsInstance(response, HttpResponseNotModified)
- self.assertIn('Cache-Control', response)
- self.assertIn('Expires', response)
+ self.assertIn("Cache-Control", response)
+ self.assertIn("Expires", response)
def test_per_thread(self):
"""The cache instance is different for each thread."""
@@ -2531,11 +2703,11 @@ class CacheMiddlewareTest(SimpleTestCase):
@override_settings(
- CACHE_MIDDLEWARE_KEY_PREFIX='settingsprefix',
+ CACHE_MIDDLEWARE_KEY_PREFIX="settingsprefix",
CACHE_MIDDLEWARE_SECONDS=1,
CACHES={
- 'default': {
- 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
+ "default": {
+ "BACKEND": "django.core.cache.backends.locmem.LocMemCache",
},
},
USE_I18N=False,
@@ -2549,7 +2721,8 @@ class TestWithTemplateResponse(SimpleTestCase):
content being complete (which is not necessarily always the case
with a TemplateResponse)
"""
- path = '/cache/test/'
+
+ path = "/cache/test/"
factory = RequestFactory()
def tearDown(self):
@@ -2558,30 +2731,46 @@ class TestWithTemplateResponse(SimpleTestCase):
def test_patch_vary_headers(self):
headers = (
# Initial vary, new headers, resulting vary.
- (None, ('Accept-Encoding',), 'Accept-Encoding'),
- ('Accept-Encoding', ('accept-encoding',), 'Accept-Encoding'),
- ('Accept-Encoding', ('ACCEPT-ENCODING',), 'Accept-Encoding'),
- ('Cookie', ('Accept-Encoding',), 'Cookie, Accept-Encoding'),
- ('Cookie, Accept-Encoding', ('Accept-Encoding',), 'Cookie, Accept-Encoding'),
- ('Cookie, Accept-Encoding', ('Accept-Encoding', 'cookie'), 'Cookie, Accept-Encoding'),
- (None, ('Accept-Encoding', 'COOKIE'), 'Accept-Encoding, COOKIE'),
- ('Cookie, Accept-Encoding', ('Accept-Encoding', 'cookie'), 'Cookie, Accept-Encoding'),
- ('Cookie , Accept-Encoding', ('Accept-Encoding', 'cookie'), 'Cookie, Accept-Encoding'),
+ (None, ("Accept-Encoding",), "Accept-Encoding"),
+ ("Accept-Encoding", ("accept-encoding",), "Accept-Encoding"),
+ ("Accept-Encoding", ("ACCEPT-ENCODING",), "Accept-Encoding"),
+ ("Cookie", ("Accept-Encoding",), "Cookie, Accept-Encoding"),
+ (
+ "Cookie, Accept-Encoding",
+ ("Accept-Encoding",),
+ "Cookie, Accept-Encoding",
+ ),
+ (
+ "Cookie, Accept-Encoding",
+ ("Accept-Encoding", "cookie"),
+ "Cookie, Accept-Encoding",
+ ),
+ (None, ("Accept-Encoding", "COOKIE"), "Accept-Encoding, COOKIE"),
+ (
+ "Cookie, Accept-Encoding",
+ ("Accept-Encoding", "cookie"),
+ "Cookie, Accept-Encoding",
+ ),
+ (
+ "Cookie , Accept-Encoding",
+ ("Accept-Encoding", "cookie"),
+ "Cookie, Accept-Encoding",
+ ),
)
for initial_vary, newheaders, resulting_vary in headers:
with self.subTest(initial_vary=initial_vary, newheaders=newheaders):
- template = engines['django'].from_string("This is a test")
+ template = engines["django"].from_string("This is a test")
response = TemplateResponse(HttpRequest(), template)
if initial_vary is not None:
- response.headers['Vary'] = initial_vary
+ response.headers["Vary"] = initial_vary
patch_vary_headers(response, newheaders)
- self.assertEqual(response.headers['Vary'], resulting_vary)
+ self.assertEqual(response.headers["Vary"], resulting_vary)
def test_get_cache_key(self):
request = self.factory.get(self.path)
- template = engines['django'].from_string("This is a test")
+ template = engines["django"].from_string("This is a test")
response = TemplateResponse(HttpRequest(), template)
- key_prefix = 'localprefix'
+ key_prefix = "localprefix"
# Expect None if no headers have been set yet.
self.assertIsNone(get_cache_key(request))
# Set headers to an empty list.
@@ -2589,20 +2778,20 @@ class TestWithTemplateResponse(SimpleTestCase):
self.assertEqual(
get_cache_key(request),
- 'views.decorators.cache.cache_page.settingsprefix.GET.'
- '58a0a05c8a5620f813686ff969c26853.d41d8cd98f00b204e9800998ecf8427e'
+ "views.decorators.cache.cache_page.settingsprefix.GET."
+ "58a0a05c8a5620f813686ff969c26853.d41d8cd98f00b204e9800998ecf8427e",
)
# A specified key_prefix is taken into account.
learn_cache_key(request, response, key_prefix=key_prefix)
self.assertEqual(
get_cache_key(request, key_prefix=key_prefix),
- 'views.decorators.cache.cache_page.localprefix.GET.'
- '58a0a05c8a5620f813686ff969c26853.d41d8cd98f00b204e9800998ecf8427e'
+ "views.decorators.cache.cache_page.localprefix.GET."
+ "58a0a05c8a5620f813686ff969c26853.d41d8cd98f00b204e9800998ecf8427e",
)
def test_get_cache_key_with_query(self):
- request = self.factory.get(self.path, {'test': 1})
- template = engines['django'].from_string("This is a test")
+ request = self.factory.get(self.path, {"test": 1})
+ template = engines["django"].from_string("This is a test")
response = TemplateResponse(HttpRequest(), template)
# Expect None if no headers have been set yet.
self.assertIsNone(get_cache_key(request))
@@ -2611,39 +2800,41 @@ class TestWithTemplateResponse(SimpleTestCase):
# The querystring is taken into account.
self.assertEqual(
get_cache_key(request),
- 'views.decorators.cache.cache_page.settingsprefix.GET.'
- '0f1c2d56633c943073c4569d9a9502fe.d41d8cd98f00b204e9800998ecf8427e'
+ "views.decorators.cache.cache_page.settingsprefix.GET."
+ "0f1c2d56633c943073c4569d9a9502fe.d41d8cd98f00b204e9800998ecf8427e",
)
class TestMakeTemplateFragmentKey(SimpleTestCase):
def test_without_vary_on(self):
- key = make_template_fragment_key('a.fragment')
- self.assertEqual(key, 'template.cache.a.fragment.d41d8cd98f00b204e9800998ecf8427e')
+ key = make_template_fragment_key("a.fragment")
+ self.assertEqual(
+ key, "template.cache.a.fragment.d41d8cd98f00b204e9800998ecf8427e"
+ )
def test_with_one_vary_on(self):
- key = make_template_fragment_key('foo', ['abc'])
- self.assertEqual(key, 'template.cache.foo.493e283d571a73056196f1a68efd0f66')
+ key = make_template_fragment_key("foo", ["abc"])
+ self.assertEqual(key, "template.cache.foo.493e283d571a73056196f1a68efd0f66")
def test_with_many_vary_on(self):
- key = make_template_fragment_key('bar', ['abc', 'def'])
- self.assertEqual(key, 'template.cache.bar.17c1a507a0cb58384f4c639067a93520')
+ key = make_template_fragment_key("bar", ["abc", "def"])
+ self.assertEqual(key, "template.cache.bar.17c1a507a0cb58384f4c639067a93520")
def test_proper_escaping(self):
- key = make_template_fragment_key('spam', ['abc:def%'])
- self.assertEqual(key, 'template.cache.spam.06c8ae8e8c430b69fb0a6443504153dc')
+ key = make_template_fragment_key("spam", ["abc:def%"])
+ self.assertEqual(key, "template.cache.spam.06c8ae8e8c430b69fb0a6443504153dc")
def test_with_ints_vary_on(self):
- key = make_template_fragment_key('foo', [1, 2, 3, 4, 5])
- self.assertEqual(key, 'template.cache.foo.7ae8fd2e0d25d651c683bdeebdb29461')
+ key = make_template_fragment_key("foo", [1, 2, 3, 4, 5])
+ self.assertEqual(key, "template.cache.foo.7ae8fd2e0d25d651c683bdeebdb29461")
def test_with_unicode_vary_on(self):
- key = make_template_fragment_key('foo', ['42º', '😀'])
- self.assertEqual(key, 'template.cache.foo.7ced1c94e543668590ba39b3c08b0237')
+ key = make_template_fragment_key("foo", ["42º", "😀"])
+ self.assertEqual(key, "template.cache.foo.7ced1c94e543668590ba39b3c08b0237")
def test_long_vary_on(self):
- key = make_template_fragment_key('foo', ['x' * 10000])
- self.assertEqual(key, 'template.cache.foo.3670b349b5124aa56bdb50678b02b23a')
+ key = make_template_fragment_key("foo", ["x" * 10000])
+ self.assertEqual(key, "template.cache.foo.3670b349b5124aa56bdb50678b02b23a")
class CacheHandlerTest(SimpleTestCase):
@@ -2651,8 +2842,8 @@ class CacheHandlerTest(SimpleTestCase):
"""
Attempting to retrieve the same alias should yield the same instance.
"""
- cache1 = caches['default']
- cache2 = caches['default']
+ cache1 = caches["default"]
+ cache2 = caches["default"]
self.assertIs(cache1, cache2)
@@ -2664,7 +2855,7 @@ class CacheHandlerTest(SimpleTestCase):
c = []
def runner():
- c.append(caches['default'])
+ c.append(caches["default"])
for x in range(2):
t = threading.Thread(target=runner)
@@ -2676,32 +2867,36 @@ class CacheHandlerTest(SimpleTestCase):
def test_nonexistent_alias(self):
msg = "The connection 'nonexistent' doesn't exist."
with self.assertRaisesMessage(InvalidCacheBackendError, msg):
- caches['nonexistent']
+ caches["nonexistent"]
def test_nonexistent_backend(self):
- test_caches = CacheHandler({
- 'invalid_backend': {
- 'BACKEND': 'django.nonexistent.NonexistentBackend',
- },
- })
+ test_caches = CacheHandler(
+ {
+ "invalid_backend": {
+ "BACKEND": "django.nonexistent.NonexistentBackend",
+ },
+ }
+ )
msg = (
"Could not find backend 'django.nonexistent.NonexistentBackend': "
"No module named 'django.nonexistent'"
)
with self.assertRaisesMessage(InvalidCacheBackendError, msg):
- test_caches['invalid_backend']
+ test_caches["invalid_backend"]
def test_all(self):
- test_caches = CacheHandler({
- 'cache_1': {
- 'BACKEND': 'django.core.cache.backends.dummy.DummyCache',
- },
- 'cache_2': {
- 'BACKEND': 'django.core.cache.backends.dummy.DummyCache',
- },
- })
+ test_caches = CacheHandler(
+ {
+ "cache_1": {
+ "BACKEND": "django.core.cache.backends.dummy.DummyCache",
+ },
+ "cache_2": {
+ "BACKEND": "django.core.cache.backends.dummy.DummyCache",
+ },
+ }
+ )
self.assertEqual(test_caches.all(initialized_only=True), [])
- cache_1 = test_caches['cache_1']
+ cache_1 = test_caches["cache_1"]
self.assertEqual(test_caches.all(initialized_only=True), [cache_1])
self.assertEqual(len(test_caches.all()), 2)
# .all() initializes all caches.
diff --git a/tests/cache/tests_async.py b/tests/cache/tests_async.py
index 4e84f35db2..3ac4e446e6 100644
--- a/tests/cache/tests_async.py
+++ b/tests/cache/tests_async.py
@@ -6,80 +6,83 @@ from django.test import SimpleTestCase, override_settings
from .tests import KEY_ERRORS_WITH_MEMCACHED_MSG
-@override_settings(CACHES={
- 'default': {
- 'BACKEND': 'django.core.cache.backends.dummy.DummyCache',
+@override_settings(
+ CACHES={
+ "default": {
+ "BACKEND": "django.core.cache.backends.dummy.DummyCache",
+ }
}
-})
+)
class AsyncDummyCacheTests(SimpleTestCase):
async def test_simple(self):
"""Dummy cache backend ignores cache set calls."""
- await cache.aset('key', 'value')
- self.assertIsNone(await cache.aget('key'))
+ await cache.aset("key", "value")
+ self.assertIsNone(await cache.aget("key"))
async def test_aadd(self):
"""Add doesn't do anything in dummy cache backend."""
- self.assertIs(await cache.aadd('key', 'value'), True)
- self.assertIs(await cache.aadd('key', 'new_value'), True)
- self.assertIsNone(await cache.aget('key'))
+ self.assertIs(await cache.aadd("key", "value"), True)
+ self.assertIs(await cache.aadd("key", "new_value"), True)
+ self.assertIsNone(await cache.aget("key"))
async def test_non_existent(self):
"""Nonexistent keys aren't found in the dummy cache backend."""
- self.assertIsNone(await cache.aget('does_not_exist'))
- self.assertEqual(await cache.aget('does_not_exist', 'default'), 'default')
+ self.assertIsNone(await cache.aget("does_not_exist"))
+ self.assertEqual(await cache.aget("does_not_exist", "default"), "default")
async def test_aget_many(self):
"""aget_many() returns nothing for the dummy cache backend."""
- await cache.aset_many({'a': 'a', 'b': 'b', 'c': 'c', 'd': 'd'})
- self.assertEqual(await cache.aget_many(['a', 'c', 'd']), {})
- self.assertEqual(await cache.aget_many(['a', 'b', 'e']), {})
+ await cache.aset_many({"a": "a", "b": "b", "c": "c", "d": "d"})
+ self.assertEqual(await cache.aget_many(["a", "c", "d"]), {})
+ self.assertEqual(await cache.aget_many(["a", "b", "e"]), {})
async def test_aget_many_invalid_key(self):
- msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ':1:key with spaces'
+ msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ":1:key with spaces"
with self.assertWarnsMessage(CacheKeyWarning, msg):
- await cache.aget_many(['key with spaces'])
+ await cache.aget_many(["key with spaces"])
async def test_adelete(self):
"""
Cache deletion is transparently ignored on the dummy cache backend.
"""
- await cache.aset_many({'key1': 'spam', 'key2': 'eggs'})
- self.assertIsNone(await cache.aget('key1'))
- self.assertIs(await cache.adelete('key1'), False)
- self.assertIsNone(await cache.aget('key1'))
- self.assertIsNone(await cache.aget('key2'))
+ await cache.aset_many({"key1": "spam", "key2": "eggs"})
+ self.assertIsNone(await cache.aget("key1"))
+ self.assertIs(await cache.adelete("key1"), False)
+ self.assertIsNone(await cache.aget("key1"))
+ self.assertIsNone(await cache.aget("key2"))
async def test_ahas_key(self):
"""ahas_key() doesn't ever return True for the dummy cache backend."""
- await cache.aset('hello1', 'goodbye1')
- self.assertIs(await cache.ahas_key('hello1'), False)
- self.assertIs(await cache.ahas_key('goodbye1'), False)
+ await cache.aset("hello1", "goodbye1")
+ self.assertIs(await cache.ahas_key("hello1"), False)
+ self.assertIs(await cache.ahas_key("goodbye1"), False)
async def test_aincr(self):
"""Dummy cache values can't be incremented."""
- await cache.aset('answer', 42)
+ await cache.aset("answer", 42)
with self.assertRaises(ValueError):
- await cache.aincr('answer')
+ await cache.aincr("answer")
with self.assertRaises(ValueError):
- await cache.aincr('does_not_exist')
+ await cache.aincr("does_not_exist")
with self.assertRaises(ValueError):
- await cache.aincr('does_not_exist', -1)
+ await cache.aincr("does_not_exist", -1)
async def test_adecr(self):
"""Dummy cache values can't be decremented."""
- await cache.aset('answer', 42)
+ await cache.aset("answer", 42)
with self.assertRaises(ValueError):
- await cache.adecr('answer')
+ await cache.adecr("answer")
with self.assertRaises(ValueError):
- await cache.adecr('does_not_exist')
+ await cache.adecr("does_not_exist")
with self.assertRaises(ValueError):
- await cache.adecr('does_not_exist', -1)
+ await cache.adecr("does_not_exist", -1)
async def test_atouch(self):
- self.assertIs(await cache.atouch('key'), False)
+ self.assertIs(await cache.atouch("key"), False)
async def test_data_types(self):
"""All data types are ignored equally by the dummy cache."""
+
def f():
return 42
@@ -88,37 +91,37 @@ class AsyncDummyCacheTests(SimpleTestCase):
return 24
data = {
- 'string': 'this is a string',
- 'int': 42,
- 'list': [1, 2, 3, 4],
- 'tuple': (1, 2, 3, 4),
- 'dict': {'A': 1, 'B': 2},
- 'function': f,
- 'class': C,
+ "string": "this is a string",
+ "int": 42,
+ "list": [1, 2, 3, 4],
+ "tuple": (1, 2, 3, 4),
+ "dict": {"A": 1, "B": 2},
+ "function": f,
+ "class": C,
}
- await cache.aset('data', data)
- self.assertIsNone(await cache.aget('data'))
+ await cache.aset("data", data)
+ self.assertIsNone(await cache.aget("data"))
async def test_expiration(self):
"""Expiration has no effect on the dummy cache."""
- await cache.aset('expire1', 'very quickly', 1)
- await cache.aset('expire2', 'very quickly', 1)
- await cache.aset('expire3', 'very quickly', 1)
+ await cache.aset("expire1", "very quickly", 1)
+ await cache.aset("expire2", "very quickly", 1)
+ await cache.aset("expire3", "very quickly", 1)
await asyncio.sleep(2)
- self.assertIsNone(await cache.aget('expire1'))
+ self.assertIsNone(await cache.aget("expire1"))
- self.assertIs(await cache.aadd('expire2', 'new_value'), True)
- self.assertIsNone(await cache.aget('expire2'))
- self.assertIs(await cache.ahas_key('expire3'), False)
+ self.assertIs(await cache.aadd("expire2", "new_value"), True)
+ self.assertIsNone(await cache.aget("expire2"))
+ self.assertIs(await cache.ahas_key("expire3"), False)
async def test_unicode(self):
"""Unicode values are ignored by the dummy cache."""
tests = {
- 'ascii': 'ascii_value',
- 'unicode_ascii': 'Iñtërnâtiônàlizætiøn1',
- 'Iñtërnâtiônàlizætiøn': 'Iñtërnâtiônàlizætiøn2',
- 'ascii2': {'x': 1},
+ "ascii": "ascii_value",
+ "unicode_ascii": "Iñtërnâtiônàlizætiøn1",
+ "Iñtërnâtiônàlizætiøn": "Iñtërnâtiônàlizætiøn2",
+ "ascii2": {"x": 1},
}
for key, value in tests.items():
with self.subTest(key=key):
@@ -127,25 +130,25 @@ class AsyncDummyCacheTests(SimpleTestCase):
async def test_aset_many(self):
"""aset_many() does nothing for the dummy cache backend."""
- self.assertEqual(await cache.aset_many({'a': 1, 'b': 2}), [])
+ self.assertEqual(await cache.aset_many({"a": 1, "b": 2}), [])
self.assertEqual(
- await cache.aset_many({'a': 1, 'b': 2}, timeout=2, version='1'),
+ await cache.aset_many({"a": 1, "b": 2}, timeout=2, version="1"),
[],
)
async def test_aset_many_invalid_key(self):
- msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ':1:key with spaces'
+ msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ":1:key with spaces"
with self.assertWarnsMessage(CacheKeyWarning, msg):
- await cache.aset_many({'key with spaces': 'foo'})
+ await cache.aset_many({"key with spaces": "foo"})
async def test_adelete_many(self):
"""adelete_many() does nothing for the dummy cache backend."""
- await cache.adelete_many(['a', 'b'])
+ await cache.adelete_many(["a", "b"])
async def test_adelete_many_invalid_key(self):
- msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ':1:key with spaces'
+ msg = KEY_ERRORS_WITH_MEMCACHED_MSG % ":1:key with spaces"
with self.assertWarnsMessage(CacheKeyWarning, msg):
- await cache.adelete_many({'key with spaces': 'foo'})
+ await cache.adelete_many({"key with spaces": "foo"})
async def test_aclear(self):
"""aclear() does nothing for the dummy cache backend."""
@@ -157,31 +160,31 @@ class AsyncDummyCacheTests(SimpleTestCase):
async def test_aincr_version(self):
"""Dummy cache versions can't be incremented."""
- await cache.aset('answer', 42)
+ await cache.aset("answer", 42)
with self.assertRaises(ValueError):
- await cache.aincr_version('answer')
+ await cache.aincr_version("answer")
with self.assertRaises(ValueError):
- await cache.aincr_version('answer', version=2)
+ await cache.aincr_version("answer", version=2)
with self.assertRaises(ValueError):
- await cache.aincr_version('does_not_exist')
+ await cache.aincr_version("does_not_exist")
async def test_adecr_version(self):
"""Dummy cache versions can't be decremented."""
- await cache.aset('answer', 42)
+ await cache.aset("answer", 42)
with self.assertRaises(ValueError):
- await cache.adecr_version('answer')
+ await cache.adecr_version("answer")
with self.assertRaises(ValueError):
- await cache.adecr_version('answer', version=2)
+ await cache.adecr_version("answer", version=2)
with self.assertRaises(ValueError):
- await cache.adecr_version('does_not_exist')
+ await cache.adecr_version("does_not_exist")
async def test_aget_or_set(self):
- self.assertEqual(await cache.aget_or_set('key', 'default'), 'default')
- self.assertIsNone(await cache.aget_or_set('key', None))
+ self.assertEqual(await cache.aget_or_set("key", "default"), "default")
+ self.assertIsNone(await cache.aget_or_set("key", None))
async def test_aget_or_set_callable(self):
def my_callable():
- return 'default'
+ return "default"
- self.assertEqual(await cache.aget_or_set('key', my_callable), 'default')
- self.assertEqual(await cache.aget_or_set('key', my_callable()), 'default')
+ self.assertEqual(await cache.aget_or_set("key", my_callable), "default")
+ self.assertEqual(await cache.aget_or_set("key", my_callable()), "default")