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
|
import os
from time import sleep
import pytest
import kombu
from .common import BaseExchangeTypes, BasePriority, BasicFunctionality
def get_connection(
hostname, port, vhost, transport_options=None):
return kombu.Connection(
f'redis://{hostname}:{port}', transport_options=transport_options)
@pytest.fixture(params=[None, {'global_keyprefix': '_prefixed_'}])
def connection(request):
# this fixture yields plain connections to broker and TLS encrypted
return get_connection(
hostname=os.environ.get('REDIS_HOST', 'localhost'),
port=os.environ.get('REDIS_6379_TCP', '6379'),
vhost=getattr(
request.config, "slaveinput", {}
).get("slaveid", None),
transport_options=request.param
)
@pytest.fixture()
def invalid_connection():
return kombu.Connection('redis://localhost:12345')
@pytest.mark.env('redis')
@pytest.mark.flaky(reruns=5, reruns_delay=2)
class test_RedisBasicFunctionality(BasicFunctionality):
pass
@pytest.mark.env('redis')
@pytest.mark.flaky(reruns=5, reruns_delay=2)
class test_RedisBaseExchangeTypes(BaseExchangeTypes):
pass
@pytest.mark.env('redis')
@pytest.mark.flaky(reruns=5, reruns_delay=2)
class test_RedisPriority(BasePriority):
# Comparing to py-amqp transport has Redis transport several
# differences:
# 1. Order of priorities is reversed
# 2. drain_events() consumes only single value
# redis transport has lower numbers higher priority
PRIORITY_ORDER = 'desc'
def test_publish_consume(self, connection):
test_queue = kombu.Queue(
'priority_test', routing_key='priority_test', max_priority=10
)
received_messages = []
def callback(body, message):
received_messages.append(body)
message.ack()
with connection as conn:
with conn.channel() as channel:
producer = kombu.Producer(channel)
for msg, prio in [
[{'msg': 'first'}, 6],
[{'msg': 'second'}, 3],
[{'msg': 'third'}, 6],
]:
producer.publish(
msg,
retry=True,
exchange=test_queue.exchange,
routing_key=test_queue.routing_key,
declare=[test_queue],
serializer='pickle',
priority=prio
)
# Sleep to make sure that queue sorted based on priority
sleep(0.5)
consumer = kombu.Consumer(
conn, [test_queue], accept=['pickle']
)
consumer.register_callback(callback)
with consumer:
# drain_events() returns just on number in
# Virtual transports
conn.drain_events(timeout=1)
conn.drain_events(timeout=1)
conn.drain_events(timeout=1)
# Second message must be received first
assert received_messages[0] == {'msg': 'second'}
assert received_messages[1] == {'msg': 'first'}
assert received_messages[2] == {'msg': 'third'}
|
