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
|
import socket
from redis.retry import Retry
from redis.sentinel import SentinelManagedConnection
from redis.backoff import NoBackoff
from unittest import mock
def test_connect_retry_on_timeout_error(master_host):
"""Test that the _connect function is retried in case of a timeout"""
connection_pool = mock.Mock()
connection_pool.get_master_address = mock.Mock(
return_value=(master_host[0], master_host[1])
)
conn = SentinelManagedConnection(
retry_on_timeout=True,
retry=Retry(NoBackoff(), 3),
connection_pool=connection_pool,
)
origin_connect = conn._connect
conn._connect = mock.Mock()
def mock_connect():
# connect only on the last retry
if conn._connect.call_count <= 2:
raise socket.timeout
else:
return origin_connect()
conn._connect.side_effect = mock_connect
conn.connect()
assert conn._connect.call_count == 3
assert connection_pool.get_master_address.call_count == 3
conn.disconnect()
|
