blob: d94a81515917b6d71ae4718aef5e428e6e0b1980 (
plain)
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
|
import types
from unittest import mock
import pytest
from redis.connection import Connection
from redis.exceptions import InvalidResponse
from redis.utils import HIREDIS_AVAILABLE
from .conftest import skip_if_server_version_lt
@pytest.mark.skipif(HIREDIS_AVAILABLE, reason="PythonParser only")
@pytest.mark.onlynoncluster
def test_invalid_response(r):
raw = b"x"
parser = r.connection._parser
with mock.patch.object(parser._buffer, "readline", return_value=raw):
with pytest.raises(InvalidResponse) as cm:
parser.read_response()
assert str(cm.value) == f"Protocol Error: {raw!r}"
@skip_if_server_version_lt("4.0.0")
@pytest.mark.redismod
def test_loading_external_modules(modclient):
def inner():
pass
modclient.load_external_module("myfuncname", inner)
assert getattr(modclient, "myfuncname") == inner
assert isinstance(getattr(modclient, "myfuncname"), types.FunctionType)
# and call it
from redis.commands import RedisModuleCommands
j = RedisModuleCommands.json
modclient.load_external_module("sometestfuncname", j)
# d = {'hello': 'world!'}
# mod = j(modclient)
# mod.set("fookey", ".", d)
# assert mod.get('fookey') == d
class TestConnection:
def test_disconnect(self):
conn = Connection()
mock_sock = mock.Mock()
conn._sock = mock_sock
conn.disconnect()
mock_sock.shutdown.assert_called_once()
mock_sock.close.assert_called_once()
assert conn._sock is None
def test_disconnect__shutdown_OSError(self):
"""An OSError on socket shutdown will still close the socket."""
conn = Connection()
mock_sock = mock.Mock()
conn._sock = mock_sock
conn._sock.shutdown.side_effect = OSError
conn.disconnect()
mock_sock.shutdown.assert_called_once()
mock_sock.close.assert_called_once()
assert conn._sock is None
def test_disconnect__close_OSError(self):
"""An OSError on socket close will still clear out the socket."""
conn = Connection()
mock_sock = mock.Mock()
conn._sock = mock_sock
conn._sock.close.side_effect = OSError
conn.disconnect()
mock_sock.shutdown.assert_called_once()
mock_sock.close.assert_called_once()
assert conn._sock is None
|
