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
|
from __future__ import with_statement
import pytest
import time
import redis
from redis.exceptions import ConnectionError
def wait_for_message(pubsub, timeout=0.1, ignore_subscribe_messages=False):
now = time.time()
timeout = now + timeout
while now < timeout:
message = pubsub.get_message(
ignore_subscribe_messages=ignore_subscribe_messages)
if message is not None:
return message
time.sleep(0.01)
now = time.time()
return None
def make_message(type, channel, data, pattern=None):
return {
'type': type,
'pattern': pattern,
'channel': channel,
'data': data
}
class TestPubSubSubscribeUnsubscribe(object):
def test_subscribe_unsubscribe(self, r):
p = r.pubsub()
assert p.subscribe('foo', 'bar') is None
# should be 2 messages indicating that we've subscribed
assert wait_for_message(p) == make_message('subscribe', 'foo', 1)
assert wait_for_message(p) == make_message('subscribe', 'bar', 2)
assert p.unsubscribe('foo', 'bar') is None
# should be 2 messages indicating that we've unsubscribed
assert wait_for_message(p) == make_message('unsubscribe', 'foo', 1)
assert wait_for_message(p) == make_message('unsubscribe', 'bar', 0)
def test_pattern_subscribe_unsubscribe(self, r):
p = r.pubsub()
assert p.psubscribe('f*', 'b*') is None
# should be 2 messages indicating that we've subscribed
assert wait_for_message(p) == make_message('psubscribe', 'f*', 1)
assert wait_for_message(p) == make_message('psubscribe', 'b*', 2)
assert p.punsubscribe('f*', 'b*') is None
# should be 2 messages indicating that we've unsubscribed
assert wait_for_message(p) == make_message('punsubscribe', 'f*', 1)
assert wait_for_message(p) == make_message('punsubscribe', 'b*', 0)
def test_resubscribe_to_channels_on_reconnection(self, r):
channels = ['foo', 'bar']
p = r.pubsub()
assert p.subscribe(*channels) is None
for i, channel in enumerate(channels):
i += 1 # enumerate is 0 index, but we want 1 based indexing
assert wait_for_message(p) == make_message('subscribe', channel, i)
# manually disconnect
p.connection.disconnect()
# calling get_message again reconnects and resubscribes
for i, channel in enumerate(channels):
i += 1 # enumerate is 0 index, but we want 1 based indexing
assert wait_for_message(p) == make_message('subscribe', channel, i)
def test_resubscribe_to_patterns_on_reconnection(self, r):
patterns = ['f*', 'b*']
p = r.pubsub()
assert p.psubscribe(*patterns) is None
for i, pattern in enumerate(patterns):
i += 1 # enumerate is 0 index, but we want 1 based indexing
assert wait_for_message(p) == make_message(
'psubscribe', pattern, i)
# manually disconnect
p.connection.disconnect()
# calling get_message again reconnects and resubscribes
for i, pattern in enumerate(patterns):
i += 1 # enumerate is 0 index, but we want 1 based indexing
assert wait_for_message(p) == make_message(
'psubscribe', pattern, i)
def test_ignore_all_subscribe_messages(self, r):
p = r.pubsub(ignore_subscribe_messages=True)
checks = (
(p.subscribe, 'foo'),
(p.unsubscribe, 'foo'),
(p.psubscribe, 'f*'),
(p.unsubscribe, 'f*'),
)
for func, channel in checks:
assert func(channel) is None
assert wait_for_message(p) is None
def test_ignore_individual_subscribe_messages(self, r):
p = r.pubsub()
checks = (
(p.subscribe, 'foo'),
(p.unsubscribe, 'foo'),
(p.psubscribe, 'f*'),
(p.unsubscribe, 'f*'),
)
for func, channel in checks:
assert func(channel) is None
message = wait_for_message(p, ignore_subscribe_messages=True)
assert message is None
class TestPubSubRedisDown(object):
def test_channel_subscribe(self, r):
r = redis.Redis(host='localhost', port=6390)
p = r.pubsub()
with pytest.raises(ConnectionError):
p.subscribe('foo')
|
