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
|
#
# Copyright (c) 2006 The Apache Software Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from qpid.client import Client, Closed
from qpid.queue import Empty
from qpid.content import Content
from qpid.testlib import testrunner, TestBase
class BasicTests(TestBase):
"""Tests for 'methods' on the amqp basic 'class'"""
def test_consume_no_local(self):
"""
Test that the no_local flag is honoured in the consume method
"""
channel = self.channel
#setup, declare two queues:
channel.queue_declare(queue="test-queue-1a", exclusive=True)
channel.queue_declare(queue="test-queue-1b", exclusive=True)
#establish two consumers one of which excludes delivery of locally sent messages
channel.basic_consume(consumer_tag="local_included", queue="test-queue-1a")
channel.basic_consume(consumer_tag="local_excluded", queue="test-queue-1b", no_local=True)
#send a message
channel.basic_publish(routing_key="test-queue-1a", content=Content("consume_no_local"))
channel.basic_publish(routing_key="test-queue-1b", content=Content("consume_no_local"))
#check the queues of the two consumers
excluded = self.client.queue("local_excluded")
included = self.client.queue("local_included")
msg = included.get(timeout=1)
self.assertEqual("consume_no_local", msg.content.body)
try:
excluded.get(timeout=1)
self.fail("Received locally published message though no_local=true")
except Empty: None
def test_consume_exclusive(self):
"""
Test that the exclusive flag is honoured in the consume method
"""
channel = self.channel
#setup, declare a queue:
channel.queue_declare(queue="test-queue-2", exclusive=True)
#check that an exclusive consumer prevents other consumer being created:
channel.basic_consume(consumer_tag="first", queue="test-queue-2", exclusive=True)
try:
channel.basic_consume(consumer_tag="second", queue="test-queue-2")
self.fail("Expected consume request to fail due to previous exclusive consumer")
except Closed, e:
self.assertChannelException(403, e.args[0])
#open new channel and cleanup last consumer:
channel = self.client.channel(2)
channel.channel_open()
#check that an exclusive consumer cannot be created if a consumer already exists:
channel.basic_consume(consumer_tag="first", queue="test-queue-2")
try:
channel.basic_consume(consumer_tag="second", queue="test-queue-2", exclusive=True)
self.fail("Expected exclusive consume request to fail due to previous consumer")
except Closed, e:
self.assertChannelException(403, e.args[0])
def test_consume_queue_errors(self):
"""
Test error conditions associated with the queue field of the consume method:
"""
channel = self.channel
try:
#queue specified but doesn't exist:
channel.basic_consume(queue="invalid-queue")
self.fail("Expected failure when consuming from non-existent queue")
except Closed, e:
self.assertChannelException(404, e.args[0])
channel = self.client.channel(2)
channel.channel_open()
try:
#queue not specified and none previously declared for channel:
channel.basic_consume(queue="")
self.fail("Expected failure when consuming from unspecified queue")
except Closed, e:
self.assertConnectionException(530, e.args[0])
def test_consume_unique_consumers(self):
"""
Ensure unique consumer tags are enforced
"""
channel = self.channel
#setup, declare a queue:
channel.queue_declare(queue="test-queue-3", exclusive=True)
#check that attempts to use duplicate tags are detected and prevented:
channel.basic_consume(consumer_tag="first", queue="test-queue-3")
try:
channel.basic_consume(consumer_tag="first", queue="test-queue-3")
self.fail("Expected consume request to fail due to non-unique tag")
except Closed, e:
self.assertConnectionException(530, e.args[0])
def test_basic_cancel(self):
"""
Test compliance of the basic.cancel method
"""
channel = self.channel
#setup, declare a queue:
channel.queue_declare(queue="test-queue-4", exclusive=True)
channel.basic_consume(consumer_tag="my-consumer", queue="test-queue-4")
channel.basic_publish(routing_key="test-queue-4", content=Content("One"))
#cancel should stop messages being delivered
channel.basic_cancel(consumer_tag="my-consumer")
channel.basic_publish(routing_key="test-queue-4", content=Content("Two"))
myqueue = self.client.queue("my-consumer")
msg = myqueue.get(timeout=1)
self.assertEqual("One", msg.content.body)
try:
msg = myqueue.get(timeout=1)
self.fail("Got message after cancellation: " + msg)
except Empty: None
#cancellation of non-existant consumers should be handled without error
channel.basic_cancel(consumer_tag="my-consumer")
channel.basic_cancel(consumer_tag="this-never-existed")
|
