| #!/usr/bin/env python |
| # |
| # 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. |
| # |
| |
| import sys |
| import qpid |
| from qpid.util import connect |
| from qpid.connection import Connection |
| from qpid.datatypes import uuid4 |
| from qpid.testlib import TestBase010 |
| from qmf.console import Session |
| from qpid.datatypes import Message |
| import qpid.messaging |
| from time import sleep |
| from os import environ, popen |
| |
| class ACLFile: |
| def __init__(self, policy='data_dir/policy.acl'): |
| self.f = open(policy,'w') |
| |
| def write(self,line): |
| self.f.write(line) |
| |
| def close(self): |
| self.f.close() |
| |
| class QueueredirectTests(TestBase010): |
| |
| def get_session(self, user, passwd): |
| socket = connect(self.broker.host, self.broker.port) |
| connection = Connection (sock=socket, username=user, password=passwd, |
| mechanism="PLAIN") |
| connection.start() |
| return connection.session(str(uuid4())) |
| |
| def reload_acl(self): |
| result = None |
| try: |
| self.broker_access.reloadAclFile() |
| except Exception, e: |
| result = str(e) |
| return result |
| |
| def get_acl_file(self): |
| return ACLFile(self.config.defines.get("policy-file", "data_dir/policy.acl")) |
| |
| def setUp(self): |
| aclf = self.get_acl_file() |
| aclf.write('acl allow all all\n') |
| aclf.close() |
| TestBase010.setUp(self) |
| self.startBrokerAccess() |
| self.reload_acl() |
| |
| def tearDown(self): |
| aclf = self.get_acl_file() |
| aclf.write('acl allow all all\n') |
| aclf.close() |
| self.reload_acl() |
| TestBase010.tearDown(self) |
| |
| |
| def redirect(self, srcQueue, tgtQueue, expectPass, failMessage): |
| try: |
| result = {} |
| result = self.broker_access.Redirect(srcQueue, tgtQueue) |
| if not expectPass: |
| self.fail("src:" + srcQueue + ", tgt:" + tgtQueue + " - " + failMessage) |
| except Exception, e: |
| if expectPass: |
| self.fail("src:" + srcQueue + ", tgt:" + tgtQueue + " - " + failMessage) |
| |
| def create_queue(self, session, name, autoDelete): |
| try: |
| session.queue_declare(queue=name, auto_delete=autoDelete) |
| except Exception, e: |
| self.fail("Should allow create queue " + name) |
| |
| def _start_qpid_send(self, queue, count, content="X", capacity=100): |
| """ Use the qpid-send client to generate traffic to a queue. |
| """ |
| command = "qpid-send" + \ |
| " -b" + " %s:%s" % (self.broker.host, self.broker.port) \ |
| + " -a " + str(queue) \ |
| + " --messages " + str(count) \ |
| + " --content-string " + str(content) \ |
| + " --capacity " + str(capacity) |
| return popen(command) |
| |
| def _start_qpid_receive(self, queue, count, timeout=5): |
| """ Use the qpid-receive client to consume from a queue. |
| Note well: prints one line of text to stdout for each consumed msg. |
| """ |
| command = "qpid-receive" + \ |
| " -b " + "%s:%s" % (self.broker.host, self.broker.port) \ |
| + " -a " + str(queue) \ |
| + " --messages " + str(count) \ |
| + " --timeout " + str(timeout) \ |
| + " --print-content yes" |
| return popen(command) |
| |
| |
| |
| #===================================== |
| # QT queue tests |
| #===================================== |
| |
| def test_010_deny_backing_up_a_nonexistant_queue(self): |
| session = self.get_session('bob','bob') |
| self.redirect("A010", "A010", False, "Should not allow redirect to non-existent queue A010") |
| session.close() |
| |
| def test_020_deny_destroy_redirect(self): |
| session = self.get_session('bob','bob') |
| self.create_queue(session, "A020", False) |
| self.redirect("A020", "", False, "Should not allow destroying redirect") |
| session.close() |
| |
| def test_030_deny_redirecting_to_nonexistent_queue(self): |
| session = self.get_session('bob','bob') |
| self.create_queue(session, "A030", False) |
| self.redirect("A030", "Axxx", False, "Should not allow redirect with non-existent queue Axxx") |
| session.close() |
| |
| def test_040_deny_queue_redirecting_to_itself(self): |
| session = self.get_session('bob','bob') |
| self.create_queue(session, "A040", False) |
| self.redirect("A040", "A040", False, "Should not allow redirect with itself") |
| session.close() |
| |
| def test_050_deny_redirecting_autodelete_queue(self): |
| session = self.get_session('bob','bob') |
| self.create_queue(session, "A050", True) |
| self.create_queue(session, "B050", False) |
| self.redirect("A050", "B050", False, "Should not allow redirect with autodelete source queue") |
| self.redirect("B050", "A050", False, "Should not allow redirect with autodelete target queue") |
| session.close() |
| |
| def test_100_create_redirect_queue_pair(self): |
| session = self.get_session('bob','bob') |
| self.create_queue(session, "A100", False) |
| self.create_queue(session, "B100", False) |
| self.redirect("A100", "B100", True, "Should allow redirect") |
| session.close() |
| |
| def test_110_deny_adding_second_redirect_to_queue(self): |
| session = self.get_session('bob','bob') |
| self.create_queue(session, "A110", False) |
| self.create_queue(session, "B110", False) |
| self.create_queue(session, "C110", False) |
| self.redirect("A110", "B110", True, "Should allow redirect") |
| self.redirect("A110", "C110", False, "Should deny second redirect") |
| self.redirect("C110", "B110", False, "Should deny second redirect") |
| session.close() |
| |
| def test_120_verify_redirect_to_target(self): |
| session = self.get_session('bob','bob') |
| self.create_queue(session, "A120", False) |
| self.create_queue(session, "B120", False) |
| |
| # Send messages to original queue |
| sndr1 = self._start_qpid_send("A120", count=5, content="A120-before-rebind"); |
| sndr1.close() |
| |
| # redirect |
| self.redirect("A120", "B120", True, "Should allow redirect") |
| |
| # Send messages to original queue |
| sndr2 = self._start_qpid_send("A120", count=3, content="A120-after-rebind"); |
| sndr2.close() |
| |
| # drain the queue |
| rcvr = self._start_qpid_receive("A120", |
| count=5) |
| count = 0; |
| x = rcvr.readline() # prints a line for each received msg |
| while x: |
| # print "Read from A120 " + x |
| count += 1; |
| x = rcvr.readline() |
| |
| self.assertEqual(count, 5) |
| |
| # drain the queue |
| rcvrB = self._start_qpid_receive("B120", |
| count=3) |
| count = 0; |
| x = rcvrB.readline() # prints a line for each received msg |
| while x: |
| # print "Read from B120 " + x |
| count += 1; |
| x = rcvrB.readline() |
| |
| self.assertEqual(count, 3) |
| |
| ###session.close() |
| |
| def test_140_verify_redirect_to_source(self): |
| session = self.get_session('bob','bob') |
| self.create_queue(session, "A140", False) |
| self.create_queue(session, "B140", False) |
| |
| # Send messages to target queue - these go onto B |
| sndr1 = self._start_qpid_send("B140", count=5, content="B140-before-rebind"); |
| sndr1.close() |
| |
| # redirect |
| self.redirect("A140", "B140", True, "Should allow redirect") |
| |
| # Send messages to target queue - these go onto A |
| sndr2 = self._start_qpid_send("B140", count=3, content="B140-after-rebind"); |
| sndr2.close() |
| |
| # drain the queue |
| rcvr = self._start_qpid_receive("B140", count=5) |
| count = 0; |
| x = rcvr.readline() # prints a line for each received msg |
| while x: |
| # print "Read from B140 " + x |
| count += 1; |
| x = rcvr.readline() |
| |
| self.assertEqual(count, 5) |
| |
| # drain the queue |
| rcvrB = self._start_qpid_receive("A140", count=3) |
| count = 0; |
| x = rcvrB.readline() # prints a line for each received msg |
| while x: |
| # print "Read from A140 " + x |
| count += 1; |
| x = rcvrB.readline() |
| |
| self.assertEqual(count, 3) |
| |
| ###session.close() |
| |
| def test_150_queue_deletion_destroys_redirect(self): |
| session = self.get_session('bob','bob') |
| self.create_queue(session, "A150", False) |
| self.create_queue(session, "B150", False) |
| self.create_queue(session, "C150", False) |
| |
| # redirect |
| self.redirect("A150", "B150", True, "Should allow redirect") |
| |
| self.redirect("A150", "C150", False, "A is already redirected") |
| |
| alice = BrokerAdmin(self.config.broker, "bob", "bob") |
| alice.delete_queue("B150") #should pass |
| |
| self.redirect("A150", "C150", True, "Should allow redirect") |
| session.close() |
| |
| ############################################################################################## |
| class BrokerAdmin: |
| def __init__(self, broker, username=None, password=None): |
| self.connection = qpid.messaging.Connection(broker) |
| if username: |
| self.connection.username = username |
| self.connection.password = password |
| self.connection.sasl_mechanisms = "PLAIN" |
| self.connection.open() |
| self.session = self.connection.session() |
| self.sender = self.session.sender("qmf.default.direct/broker") |
| self.reply_to = "responses-#; {create:always}" |
| self.receiver = self.session.receiver(self.reply_to) |
| |
| def invoke(self, method, arguments): |
| content = { |
| "_object_id": {"_object_name": "org.apache.qpid.broker:broker:amqp-broker"}, |
| "_method_name": method, |
| "_arguments": arguments |
| } |
| request = qpid.messaging.Message(reply_to=self.reply_to, content=content) |
| request.properties["x-amqp-0-10.app-id"] = "qmf2" |
| request.properties["qmf.opcode"] = "_method_request" |
| self.sender.send(request) |
| response = self.receiver.fetch() |
| self.session.acknowledge() |
| if response.properties['x-amqp-0-10.app-id'] == 'qmf2': |
| if response.properties['qmf.opcode'] == '_method_response': |
| return response.content['_arguments'] |
| elif response.properties['qmf.opcode'] == '_exception': |
| raise Exception(response.content['_values']) |
| else: raise Exception("Invalid response received, unexpected opcode: %s" % response.properties['qmf.opcode']) |
| else: raise Exception("Invalid response received, not a qmfv2 method: %s" % response.properties['x-amqp-0-10.app-id']) |
| |
| def create_exchange(self, name, exchange_type=None, options={}): |
| properties = options |
| if exchange_type: properties["exchange_type"] = exchange_type |
| self.invoke("create", {"type": "exchange", "name":name, "properties":properties}) |
| |
| def create_queue(self, name, properties={}): |
| self.invoke("create", {"type": "queue", "name":name, "properties":properties}) |
| |
| def delete_exchange(self, name): |
| self.invoke("delete", {"type": "exchange", "name":name}) |
| |
| def delete_queue(self, name): |
| self.invoke("delete", {"type": "queue", "name":name}) |
