This repository has been archived by the owner on Jan 9, 2024. It is now read-only.
/
test_encoding_cluster.py
78 lines (64 loc) · 3 KB
/
test_encoding_cluster.py
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
from __future__ import unicode_literals
import pytest
from rediscluster import RedisCluster
from redis._compat import unichr, unicode
from .conftest import _get_client
class TestEncodingCluster(object):
"""
We must import the entire class due to the seperate fixture that uses RedisCluster as client
class instead of the normal Redis instance.
FIXME: If possible, monkeypatching TestEncoding class would be preferred but kinda impossible in reality
"""
@pytest.fixture()
def r(self, request):
return _get_client(RedisCluster, request=request, decode_responses=True)
@pytest.fixture()
def r_no_decode(self, request):
return _get_client(
RedisCluster,
request=request,
decode_responses=False,
)
def test_simple_encoding(self, r_no_decode):
unicode_string = unichr(3456) + 'abcd' + unichr(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 = unichr(3456) + 'abcd' + unichr(3421)
r['unicode-string'] = unicode_string
cached_val = r['unicode-string']
assert isinstance(cached_val, unicode)
assert unicode_string == cached_val
def test_memoryview_encoding(self, r_no_decode):
unicode_string = unichr(3456) + 'abcd' + unichr(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 = unichr(3456) + 'abcd' + unichr(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, unicode)
assert unicode_string == cached_val
def test_list_encoding(self, r):
unicode_string = unichr(3456) + 'abcd' + unichr(3421)
result = [unicode_string, unicode_string, unicode_string]
r.rpush('a', *result)
assert r.lrange('a', 0, -1) == result
class TestEncodingErrors(object):
def test_ignore(self, request):
r = _get_client(RedisCluster, 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(RedisCluster, request=request, decode_responses=True,
encoding_errors='replace')
r.set('a', b'foo\xff')
assert r.get('a') == 'foo\ufffd'