diff options
author | django-bot <ops@djangoproject.com> | 2022-02-03 20:24:19 +0100 |
---|---|---|
committer | Mariusz Felisiak <felisiak.mariusz@gmail.com> | 2022-02-07 20:37:05 +0100 |
commit | 9c19aff7c7561e3a82978a272ecdaad40dda5c00 (patch) | |
tree | f0506b668a013d0063e5fba3dbf4863b466713ba /tests/cache | |
parent | f68fa8b45dfac545cfc4111d4e52804c86db68d3 (diff) | |
download | django-9c19aff7c7561e3a82978a272ecdaad40dda5c00.tar.gz |
Refs #33476 -- Reformatted code with Black.
Diffstat (limited to 'tests/cache')
-rw-r--r-- | tests/cache/models.py | 2 | ||||
-rw-r--r-- | tests/cache/tests.py | 2041 | ||||
-rw-r--r-- | tests/cache/tests_async.py | 149 |
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") |