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
|
import pytest
import redis
from redis.connection import Connection
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'
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-16')
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())
|
