diff options
Diffstat (limited to 'qpid/tests/src/py/qpid_tests/broker_0_9/query.py')
| -rw-r--r-- | qpid/tests/src/py/qpid_tests/broker_0_9/query.py | 224 |
1 files changed, 224 insertions, 0 deletions
diff --git a/qpid/tests/src/py/qpid_tests/broker_0_9/query.py b/qpid/tests/src/py/qpid_tests/broker_0_9/query.py new file mode 100644 index 0000000000..cb66d079e5 --- /dev/null +++ b/qpid/tests/src/py/qpid_tests/broker_0_9/query.py @@ -0,0 +1,224 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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 TestBase + +class QueryTests(TestBase): + """Tests for various query methods introduced in 0-10 and available in 0-9 for preview""" + + def test_exchange_query(self): + """ + Test that the exchange_query method works as expected + """ + channel = self.channel + #check returned type for the standard exchanges + self.assertEqual("direct", channel.exchange_query(name="amq.direct").type) + self.assertEqual("topic", channel.exchange_query(name="amq.topic").type) + self.assertEqual("fanout", channel.exchange_query(name="amq.fanout").type) + self.assertEqual("headers", channel.exchange_query(name="amq.match").type) + self.assertEqual("direct", channel.exchange_query(name="").type) + #declare an exchange + channel.exchange_declare(exchange="my-test-exchange", type= "direct", durable=False) + #check that the result of a query is as expected + response = channel.exchange_query(name="my-test-exchange") + self.assertEqual("direct", response.type) + self.assertEqual(False, response.durable) + self.assertEqual(False, response.not_found) + #delete the exchange + channel.exchange_delete(exchange="my-test-exchange") + #check that the query now reports not-found + self.assertEqual(True, channel.exchange_query(name="my-test-exchange").not_found) + + def test_binding_query_direct(self): + """ + Test that the binding_query method works as expected with the direct exchange + """ + self.binding_query_with_key("amq.direct") + + def test_binding_query_topic(self): + """ + Test that the binding_query method works as expected with the direct exchange + """ + self.binding_query_with_key("amq.topic") + + def binding_query_with_key(self, exchange_name): + channel = self.channel + #setup: create two queues + channel.queue_declare(queue="used-queue", exclusive=True) + channel.queue_declare(queue="unused-queue", exclusive=True) + + channel.queue_bind(exchange=exchange_name, queue="used-queue", routing_key="used-key") + + # test detection of any binding to specific queue + response = channel.binding_query(exchange=exchange_name, queue="used-queue") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(False, response.queue_not_matched) + + # test detection of specific binding to any queue + response = channel.binding_query(exchange=exchange_name, routing_key="used-key") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(False, response.key_not_matched) + + # test detection of specific binding to specific queue + response = channel.binding_query(exchange=exchange_name, queue="used-queue", routing_key="used-key") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(False, response.queue_not_matched) + self.assertEqual(False, response.key_not_matched) + + # test unmatched queue, unspecified binding + response = channel.binding_query(exchange=exchange_name, queue="unused-queue") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(True, response.queue_not_matched) + + # test unspecified queue, unmatched binding + response = channel.binding_query(exchange=exchange_name, routing_key="unused-key") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(True, response.key_not_matched) + + # test matched queue, unmatched binding + response = channel.binding_query(exchange=exchange_name, queue="used-queue", routing_key="unused-key") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(False, response.queue_not_matched) + self.assertEqual(True, response.key_not_matched) + + # test unmatched queue, matched binding + response = channel.binding_query(exchange=exchange_name, queue="unused-queue", routing_key="used-key") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(True, response.queue_not_matched) + self.assertEqual(False, response.key_not_matched) + + # test unmatched queue, unmatched binding + response = channel.binding_query(exchange=exchange_name, queue="unused-queue", routing_key="unused-key") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(True, response.queue_not_matched) + self.assertEqual(True, response.key_not_matched) + + #test exchange not found + self.assertEqual(True, channel.binding_query(exchange="unknown-exchange").exchange_not_found) + + #test queue not found + self.assertEqual(True, channel.binding_query(exchange=exchange_name, queue="unknown-queue").queue_not_found) + + + def test_binding_query_fanout(self): + """ + Test that the binding_query method works as expected with fanout exchange + """ + channel = self.channel + #setup + channel.queue_declare(queue="used-queue", exclusive=True) + channel.queue_declare(queue="unused-queue", exclusive=True) + channel.queue_bind(exchange="amq.fanout", queue="used-queue") + + # test detection of any binding to specific queue + response = channel.binding_query(exchange="amq.fanout", queue="used-queue") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(False, response.queue_not_matched) + + # test unmatched queue, unspecified binding + response = channel.binding_query(exchange="amq.fanout", queue="unused-queue") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(True, response.queue_not_matched) + + #test exchange not found + self.assertEqual(True, channel.binding_query(exchange="unknown-exchange").exchange_not_found) + + #test queue not found + self.assertEqual(True, channel.binding_query(exchange="amq.fanout", queue="unknown-queue").queue_not_found) + + def test_binding_query_header(self): + """ + Test that the binding_query method works as expected with headers exchanges + """ + channel = self.channel + #setup + channel.queue_declare(queue="used-queue", exclusive=True) + channel.queue_declare(queue="unused-queue", exclusive=True) + channel.queue_bind(exchange="amq.match", queue="used-queue", arguments={"x-match":"all", "a":"A"} ) + + # test detection of any binding to specific queue + response = channel.binding_query(exchange="amq.match", queue="used-queue") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(False, response.queue_not_matched) + + # test detection of specific binding to any queue + response = channel.binding_query(exchange="amq.match", arguments={"x-match":"all", "a":"A"}) + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(False, response.args_not_matched) + + # test detection of specific binding to specific queue + response = channel.binding_query(exchange="amq.match", queue="used-queue", arguments={"x-match":"all", "a":"A"}) + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(False, response.queue_not_matched) + self.assertEqual(False, response.args_not_matched) + + # test unmatched queue, unspecified binding + response = channel.binding_query(exchange="amq.match", queue="unused-queue") + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(True, response.queue_not_matched) + + # test unspecified queue, unmatched binding + response = channel.binding_query(exchange="amq.match", arguments={"x-match":"all", "b":"B"}) + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(True, response.args_not_matched) + + # test matched queue, unmatched binding + response = channel.binding_query(exchange="amq.match", queue="used-queue", arguments={"x-match":"all", "b":"B"}) + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(False, response.queue_not_matched) + self.assertEqual(True, response.args_not_matched) + + # test unmatched queue, matched binding + response = channel.binding_query(exchange="amq.match", queue="unused-queue", arguments={"x-match":"all", "a":"A"}) + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(True, response.queue_not_matched) + self.assertEqual(False, response.args_not_matched) + + # test unmatched queue, unmatched binding + response = channel.binding_query(exchange="amq.match", queue="unused-queue", arguments={"x-match":"all", "b":"B"}) + self.assertEqual(False, response.exchange_not_found) + self.assertEqual(False, response.queue_not_found) + self.assertEqual(True, response.queue_not_matched) + self.assertEqual(True, response.args_not_matched) + + #test exchange not found + self.assertEqual(True, channel.binding_query(exchange="unknown-exchange").exchange_not_found) + + #test queue not found + self.assertEqual(True, channel.binding_query(exchange="amq.match", queue="unknown-queue").queue_not_found) + |