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
|
import os
from time import sleep
import pytest
import redis
import kombu
from .common import BaseExchangeTypes, BasePriority, BasicFunctionality
def get_connection(
hostname, port, vhost, user_name=None, password=None,
transport_options=None):
credentials = f'{user_name}:{password}@' if user_name else ''
return kombu.Connection(
f'redis://{credentials}{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')
def test_failed_credentials():
"""Tests denied connection when wrong credentials were provided"""
with pytest.raises(redis.exceptions.ResponseError):
get_connection(
hostname=os.environ.get('REDIS_HOST', 'localhost'),
port=os.environ.get('REDIS_6379_TCP', '6379'),
vhost=None,
user_name='wrong_redis_user',
password='wrong_redis_password'
).connect()
@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'}
|
