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
|
# Ubuntu CI Engine
# Copyright 2014 Canonical Ltd.
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License version 3, as
# published by the Free Software Foundation.
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import unittest
import subprocess
import deployers
from ucitests import fixtures
from ci_utils import amqp_utils
from ci_utils.testing import features
RABBIT_SERVICE = 'ci-airline-rabbit'
@features.requires(features.bootstrapped_juju)
class TestRabbit(deployers.DeployerTest):
def setUp(self):
super(TestRabbit, self).setUp()
self.create_user('tester', 's3cr3t')
self.addCleanup(self.delete_user, 'tester')
deployers.juju_expose(RABBIT_SERVICE)
self.addCleanup(deployers.juju_expose, RABBIT_SERVICE, False)
class AmqpConfig(object):
def __init__(inner):
ip, _ = self.get_ip_and_port(RABBIT_SERVICE)
inner.AMQP_HOST = ip
inner.AMQP_VHOST = '/'
inner.AMQP_USER = 'tester'
inner.AMQP_PASSWORD = 's3cr3t'
fixtures.patch(self, amqp_utils, 'get_config', AmqpConfig)
def run_ctl_cmd(self, args):
deployers.juju_run(RABBIT_SERVICE,
'sudo rabbitmqctl {}'.format(' '.join(args)))
def create_user(self, user_name, password=None):
if password is None:
password = user_name
self.run_ctl_cmd(['add_user', user_name, password])
self.run_ctl_cmd(['set_permissions', '-p', '/',
user_name, '".*" ".*" ".*"'])
def delete_user(self, user_name):
self.run_ctl_cmd(['delete_user', user_name])
def assertConsumed(self, channel, queue_name, callback, check):
tag = channel.basic_consume(callback=callback, queue=queue_name)
# Calling consume only setup the queue to trigger the callback when a
# message is received, we need some monitoring to ensure the messages
# are processed. For the present tests, a single call is expected.
while not check():
channel.wait()
# We're done with the queue, cancel the subscription
channel.basic_cancel(tag)
def test_acked_message(self):
amqp_utils.send('myqueue', 'hello')
conn, channel = amqp_utils.declare_queue('myqueue')
self.addCleanup(conn.close)
self.addCleanup(channel.close)
self.addCleanup(channel.queue_delete, 'myqueue')
channel.basic_qos(prefetch_size=0, prefetch_count=1, a_global=False)
self.ack_called = False
def ack(msg):
self.assertEqual('hello', msg.body)
msg.channel.basic_ack(msg.delivery_tag)
self.ack_called = True
self.assertConsumed(channel, 'myqueue', ack, lambda: self.ack_called)
def test_nacked_message(self):
amqp_utils.send('myqueue', 'hello')
conn, channel = amqp_utils.declare_queue('myqueue')
self.addCleanup(conn.close)
channel.basic_qos(prefetch_size=0, prefetch_count=1, a_global=False)
self.nack_called = False
def nack(msg):
self.assertEqual('hello', msg.body)
msg.channel.basic_reject(msg.delivery_tag, requeue=True)
self.nack_called = True
self.assertConsumed(channel, 'myqueue', nack, lambda: self.nack_called)
# Bounce the message by closing the channel otherwise it stays there
# waiting for another callback to consume it.
channel.close()
# Use a different channel to get the message again
conn, channel = amqp_utils.declare_queue('myqueue')
self.addCleanup(conn.close)
self.addCleanup(channel.close)
self.addCleanup(channel.queue_delete, 'myqueue')
channel.basic_qos(prefetch_size=0, prefetch_count=1, a_global=False)
self.ack_called = False
def ack(msg):
self.assertEqual('hello', msg.body)
msg.channel.basic_ack(msg.delivery_tag)
self.ack_called = True
self.assertConsumed(channel, 'myqueue', ack, lambda: self.ack_called)
if __name__ == '__main__':
unittest.main()
|