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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
import pytest
import redis
import valkey
from socketio import redis_manager
from socketio.redis_manager import RedisManager, parse_redis_sentinel_url
class TestPubSubManager:
def test_redis_not_installed(self):
saved_redis = redis_manager.redis
redis_manager.redis = None
with pytest.raises(RuntimeError):
RedisManager('redis://')._redis_connect()
assert RedisManager('unix:///var/sock/redis.sock') is not None
redis_manager.redis = saved_redis
def test_valkey_not_installed(self):
saved_valkey = redis_manager.valkey
redis_manager.valkey = None
with pytest.raises(RuntimeError):
RedisManager('valkey://')._redis_connect()
assert RedisManager('unix:///var/sock/redis.sock') is not None
redis_manager.valkey = saved_valkey
def test_redis_valkey_not_installed(self):
saved_redis = redis_manager.redis
redis_manager.redis = None
saved_valkey = redis_manager.valkey
redis_manager.valkey = None
with pytest.raises(RuntimeError):
RedisManager('redis://')._redis_connect()
with pytest.raises(RuntimeError):
RedisManager('valkey://')._redis_connect()
with pytest.raises(RuntimeError):
RedisManager('unix:///var/sock/redis.sock')._redis_connect()
redis_manager.redis = saved_redis
redis_manager.valkey = saved_valkey
def test_bad_url(self):
with pytest.raises(ValueError):
RedisManager('http://localhost:6379')._redis_connect()
def test_redis_connect(self):
urls = [
'redis://localhost:6379',
'redis://localhost:6379/0',
'redis://:password@localhost:6379',
'redis://:password@localhost:6379/0',
'redis://user:password@localhost:6379',
'redis://user:password@localhost:6379/0',
'rediss://localhost:6379',
'rediss://localhost:6379/0',
'rediss://:password@localhost:6379',
'rediss://:password@localhost:6379/0',
'rediss://user:password@localhost:6379',
'rediss://user:password@localhost:6379/0',
'unix:///var/sock/redis.sock',
'unix:///var/sock/redis.sock?db=0',
'unix://user@/var/sock/redis.sock',
'unix://user@/var/sock/redis.sock?db=0',
'redis+sentinel://192.168.0.1:6379,192.168.0.2:6379/'
]
for url in urls:
c = RedisManager(url)
assert c.redis is None
c._redis_connect()
assert isinstance(c.redis, redis.Redis)
def test_valkey_connect(self):
saved_redis = redis_manager.redis
redis_manager.redis = None
urls = [
'valkey://localhost:6379',
'valkey://localhost:6379/0',
'valkey://:password@localhost:6379',
'valkey://:password@localhost:6379/0',
'valkey://user:password@localhost:6379',
'valkey://user:password@localhost:6379/0',
'valkeys://localhost:6379',
'valkeys://localhost:6379/0',
'valkeys://:password@localhost:6379',
'valkeys://:password@localhost:6379/0',
'valkeys://user:password@localhost:6379',
'valkeys://user:password@localhost:6379/0',
'unix:///var/sock/redis.sock',
'unix:///var/sock/redis.sock?db=0',
'unix://user@/var/sock/redis.sock',
'unix://user@/var/sock/redis.sock?db=0',
'valkey+sentinel://192.168.0.1:6379,192.168.0.2:6379/'
]
for url in urls:
c = RedisManager(url)
assert c.redis is None
c._redis_connect()
assert isinstance(c.redis, valkey.Valkey)
redis_manager.redis = saved_redis
@pytest.mark.parametrize('rtype', ['redis', 'valkey'])
def test_sentinel_url_parser(self, rtype):
with pytest.raises(ValueError):
parse_redis_sentinel_url(f'{rtype}://localhost:6379/0')
assert parse_redis_sentinel_url(
f'{rtype}+sentinel://localhost:6379'
) == (
[('localhost', 6379)],
None,
{}
)
assert parse_redis_sentinel_url(
f'{rtype}+sentinel://192.168.0.1:6379,192.168.0.2:6379/'
) == (
[('192.168.0.1', 6379), ('192.168.0.2', 6379)],
None,
{}
)
assert parse_redis_sentinel_url(
f'{rtype}+sentinel://h1:6379,h2:6379/0'
) == (
[('h1', 6379), ('h2', 6379)],
None,
{'db': 0}
)
assert parse_redis_sentinel_url(
f'{rtype}+sentinel://'
'user:password@h1:6379,h2:6379,h1:6380/0/myredis'
) == (
[('h1', 6379), ('h2', 6379), ('h1', 6380)],
'myredis',
{'username': 'user', 'password': 'password', 'db': 0}
)
|
