1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
import copy
from typing import Iterable, cast
import pytest
from django.core.cache import caches
from pytest import LogCaptureFixture
from pytest_django.fixtures import SettingsWrapper
from redis.exceptions import ConnectionError
from django_redis.cache import RedisCache
from django_redis.client import ShardClient
def make_key(key: str, prefix: str, version: str) -> str:
return f"{prefix}#{version}#{key}"
def reverse_key(key: str) -> str:
return key.split("#", 2)[2]
@pytest.fixture
def ignore_exceptions_cache(settings: SettingsWrapper) -> RedisCache:
caches_setting = copy.deepcopy(settings.CACHES)
caches_setting["doesnotexist"]["OPTIONS"]["IGNORE_EXCEPTIONS"] = True
caches_setting["doesnotexist"]["OPTIONS"]["LOG_IGNORED_EXCEPTIONS"] = True
settings.CACHES = caches_setting
settings.DJANGO_REDIS_IGNORE_EXCEPTIONS = True
settings.DJANGO_REDIS_LOG_IGNORED_EXCEPTIONS = True
return cast(RedisCache, caches["doesnotexist"])
def test_get_django_omit_exceptions_many_returns_default_arg(
ignore_exceptions_cache: RedisCache,
):
assert ignore_exceptions_cache._ignore_exceptions is True
assert ignore_exceptions_cache.get_many(["key1", "key2", "key3"]) == {}
def test_get_django_omit_exceptions(
caplog: LogCaptureFixture, ignore_exceptions_cache: RedisCache
):
assert ignore_exceptions_cache._ignore_exceptions is True
assert ignore_exceptions_cache._log_ignored_exceptions is True
assert ignore_exceptions_cache.get("key") is None
assert ignore_exceptions_cache.get("key", "default") == "default"
assert ignore_exceptions_cache.get("key", default="default") == "default"
assert len(caplog.records) == 3
assert all(
record.levelname == "ERROR" and record.msg == "Exception ignored"
for record in caplog.records
)
def test_get_django_omit_exceptions_priority_1(settings: SettingsWrapper):
caches_setting = copy.deepcopy(settings.CACHES)
caches_setting["doesnotexist"]["OPTIONS"]["IGNORE_EXCEPTIONS"] = True
settings.CACHES = caches_setting
settings.DJANGO_REDIS_IGNORE_EXCEPTIONS = False
cache = cast(RedisCache, caches["doesnotexist"])
assert cache._ignore_exceptions is True
assert cache.get("key") is None
def test_get_django_omit_exceptions_priority_2(settings: SettingsWrapper):
caches_setting = copy.deepcopy(settings.CACHES)
caches_setting["doesnotexist"]["OPTIONS"]["IGNORE_EXCEPTIONS"] = False
settings.CACHES = caches_setting
settings.DJANGO_REDIS_IGNORE_EXCEPTIONS = True
cache = cast(RedisCache, caches["doesnotexist"])
assert cache._ignore_exceptions is False
with pytest.raises(ConnectionError):
cache.get("key")
@pytest.fixture
def key_prefix_cache(
cache: RedisCache, settings: SettingsWrapper
) -> Iterable[RedisCache]:
caches_setting = copy.deepcopy(settings.CACHES)
caches_setting["default"]["KEY_PREFIX"] = "*"
settings.CACHES = caches_setting
yield cache
@pytest.fixture
def with_prefix_cache() -> Iterable[RedisCache]:
with_prefix = cast(RedisCache, caches["with_prefix"])
yield with_prefix
with_prefix.clear()
class TestDjangoRedisCacheEscapePrefix:
def test_delete_pattern(
self, key_prefix_cache: RedisCache, with_prefix_cache: RedisCache
):
key_prefix_cache.set("a", "1")
with_prefix_cache.set("b", "2")
key_prefix_cache.delete_pattern("*")
assert key_prefix_cache.has_key("a") is False
assert with_prefix_cache.get("b") == "2"
def test_iter_keys(
self, key_prefix_cache: RedisCache, with_prefix_cache: RedisCache
):
if isinstance(key_prefix_cache.client, ShardClient):
pytest.skip("ShardClient doesn't support iter_keys")
key_prefix_cache.set("a", "1")
with_prefix_cache.set("b", "2")
assert list(key_prefix_cache.iter_keys("*")) == ["a"]
def test_keys(self, key_prefix_cache: RedisCache, with_prefix_cache: RedisCache):
key_prefix_cache.set("a", "1")
with_prefix_cache.set("b", "2")
keys = key_prefix_cache.keys("*")
assert "a" in keys
assert "b" not in keys
def test_custom_key_function(cache: RedisCache, settings: SettingsWrapper):
caches_setting = copy.deepcopy(settings.CACHES)
caches_setting["default"]["KEY_FUNCTION"] = "test_cache_options.make_key"
caches_setting["default"]["REVERSE_KEY_FUNCTION"] = "test_cache_options.reverse_key"
settings.CACHES = caches_setting
if isinstance(cache.client, ShardClient):
pytest.skip("ShardClient doesn't support get_client")
for key in ["foo-aa", "foo-ab", "foo-bb", "foo-bc"]:
cache.set(key, "foo")
res = cache.delete_pattern("*foo-a*")
assert bool(res) is True
keys = cache.keys("foo*")
assert set(keys) == {"foo-bb", "foo-bc"}
# ensure our custom function was actually called
assert {k.decode() for k in cache.client.get_client(write=False).keys("*")} == (
{"#1#foo-bc", "#1#foo-bb"}
)
|
