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
120
121
122
123
|
from dogpile.cache.region import _backend_loader
from ._fixtures import _GenericBackendTest, _GenericMutexTest
from unittest import TestCase
from mock import patch, Mock
import pytest
import os
REDIS_HOST = '127.0.0.1'
REDIS_PORT = int(os.getenv('DOGPILE_REDIS_PORT', '6379'))
class _TestRedisConn(object):
@classmethod
def _check_backend_available(cls, backend):
try:
client = backend._create_client()
client.set("x", "y")
# on py3k it appears to return b"y"
assert client.get("x").decode("ascii") == "y"
client.delete("x")
except:
pytest.skip(
"redis is not running or "
"otherwise not functioning correctly")
class RedisTest(_TestRedisConn, _GenericBackendTest):
backend = 'dogpile.cache.redis'
config_args = {
"arguments": {
'host': REDIS_HOST,
'port': REDIS_PORT,
'db': 0,
"foo": "barf"
}
}
class RedisDistributedMutexTest(_TestRedisConn, _GenericMutexTest):
backend = 'dogpile.cache.redis'
config_args = {
"arguments": {
'host': REDIS_HOST,
'port': REDIS_PORT,
'db': 0,
'distributed_lock': True,
}
}
@patch('redis.StrictRedis', autospec=True)
class RedisConnectionTest(TestCase):
backend = 'dogpile.cache.redis'
@classmethod
def setup_class(cls):
try:
cls.backend_cls = _backend_loader.load(cls.backend)
cls.backend_cls({})
except ImportError:
pytest.skip("Backend %s not installed" % cls.backend)
def _test_helper(self, mock_obj, expected_args, connection_args=None):
if connection_args is None:
connection_args = expected_args
self.backend_cls(connection_args)
mock_obj.assert_called_once_with(**expected_args)
def test_connect_with_defaults(self, MockStrictRedis):
# The defaults, used if keys are missing from the arguments dict.
arguments = {
'host': 'localhost',
'password': None,
'port': 6379,
'db': 0,
}
self._test_helper(MockStrictRedis, arguments, {})
def test_connect_with_basics(self, MockStrictRedis):
arguments = {
'host': '127.0.0.1',
'password': None,
'port': 6379,
'db': 0,
}
self._test_helper(MockStrictRedis, arguments)
def test_connect_with_password(self, MockStrictRedis):
arguments = {
'host': '127.0.0.1',
'password': 'some password',
'port': 6379,
'db': 0,
}
self._test_helper(MockStrictRedis, arguments)
def test_connect_with_socket_timeout(self, MockStrictRedis):
arguments = {
'host': '127.0.0.1',
'port': 6379,
'socket_timeout': 0.5,
'password': None,
'db': 0,
}
self._test_helper(MockStrictRedis, arguments)
def test_connect_with_connection_pool(self, MockStrictRedis):
pool = Mock()
arguments = {
'connection_pool': pool,
'socket_timeout': 0.5
}
expected_args = {'connection_pool': pool}
self._test_helper(MockStrictRedis, expected_args,
connection_args=arguments)
def test_connect_with_url(self, MockStrictRedis):
arguments = {
'url': 'redis://redis:password@127.0.0.1:6379/0'
}
self._test_helper(MockStrictRedis.from_url, arguments)
|
