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
|
import pytest
import redis
from redis.connection import Connection
from redis.utils import HIREDIS_PACK_AVAILABLE
from .conftest import _get_client
class TestEncoding:
@pytest.fixture()
def r(self, request):
return _get_client(redis.Redis, request=request, decode_responses=True)
@pytest.fixture()
def r_no_decode(self, request):
return _get_client(redis.Redis, request=request, decode_responses=False)
def test_simple_encoding(self, r_no_decode):
unicode_string = chr(3456) + "abcd" + chr(3421)
r_no_decode["unicode-string"] = unicode_string.encode("utf-8")
cached_val = r_no_decode["unicode-string"]
assert isinstance(cached_val, bytes)
assert unicode_string == cached_val.decode("utf-8")
def test_simple_encoding_and_decoding(self, r):
unicode_string = chr(3456) + "abcd" + chr(3421)
r["unicode-string"] = unicode_string
cached_val = r["unicode-string"]
assert isinstance(cached_val, str)
assert unicode_string == cached_val
def test_memoryview_encoding(self, r_no_decode):
unicode_string = chr(3456) + "abcd" + chr(3421)
unicode_string_view = memoryview(unicode_string.encode("utf-8"))
r_no_decode["unicode-string-memoryview"] = unicode_string_view
cached_val = r_no_decode["unicode-string-memoryview"]
# The cached value won't be a memoryview because it's a copy from Redis
assert isinstance(cached_val, bytes)
assert unicode_string == cached_val.decode("utf-8")
def test_memoryview_encoding_and_decoding(self, r):
unicode_string = chr(3456) + "abcd" + chr(3421)
unicode_string_view = memoryview(unicode_string.encode("utf-8"))
r["unicode-string-memoryview"] = unicode_string_view
cached_val = r["unicode-string-memoryview"]
assert isinstance(cached_val, str)
assert unicode_string == cached_val
def test_list_encoding(self, r):
unicode_string = chr(3456) + "abcd" + chr(3421)
result = [unicode_string, unicode_string, unicode_string]
r.rpush("a", *result)
assert r.lrange("a", 0, -1) == result
class TestEncodingErrors:
def test_ignore(self, request):
r = _get_client(
redis.Redis,
request=request,
decode_responses=True,
encoding_errors="ignore",
)
r.set("a", b"foo\xff")
assert r.get("a") == "foo"
def test_replace(self, request):
r = _get_client(
redis.Redis,
request=request,
decode_responses=True,
encoding_errors="replace",
)
r.set("a", b"foo\xff")
assert r.get("a") == "foo\ufffd"
@pytest.mark.skipif(
HIREDIS_PACK_AVAILABLE,
reason="Packing via hiredis does not preserve memoryviews",
)
class TestMemoryviewsAreNotPacked:
def test_memoryviews_are_not_packed(self):
c = Connection()
arg = memoryview(b"some_arg")
arg_list = ["SOME_COMMAND", arg]
cmd = c.pack_command(*arg_list)
assert cmd[1] is arg
cmds = c.pack_commands([arg_list, arg_list])
assert cmds[1] is arg
assert cmds[3] is arg
class TestCommandsAreNotEncoded:
@pytest.fixture()
def r(self, request):
return _get_client(redis.Redis, request=request, encoding="utf-8")
def test_basic_command(self, r):
r.set("hello", "world")
class TestInvalidUserInput:
def test_boolean_fails(self, r):
with pytest.raises(redis.DataError):
r.set("a", True)
def test_none_fails(self, r):
with pytest.raises(redis.DataError):
r.set("a", None)
def test_user_type_fails(self, r):
class Foo:
def __str__(self):
return "Foo"
with pytest.raises(redis.DataError):
r.set("a", Foo())
|
