| # |
| # 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 __future__ import absolute_import |
| from qpid.datatypes import Message, RangedSet |
| from qpid.testlib import TestBase010 |
| from qpid.management import managementChannel, managementClient |
| from threading import Condition |
| from time import sleep |
| import qmf.console |
| import qpid.messaging |
| from qpid.messaging.exceptions import Empty |
| from qpidtoollibs import BrokerAgent |
| |
| class ManagementTest (TestBase010): |
| |
| def setup_access(self): |
| if 'broker_agent' not in self.__dict__: |
| self.conn2 = qpid.messaging.Connection(self.broker) |
| self.conn2.open() |
| self.broker_agent = BrokerAgent(self.conn2) |
| return self.broker_agent |
| |
| """ |
| Tests for the management hooks |
| """ |
| |
| def test_broker_connectivity_oldAPI (self): |
| """ |
| Call the "echo" method on the broker to verify it is alive and talking. |
| """ |
| session = self.session |
| |
| mc = managementClient () |
| mch = mc.addChannel (session) |
| |
| mc.syncWaitForStable (mch) |
| brokers = mc.syncGetObjects (mch, "broker") |
| self.assertEqual (len (brokers), 1) |
| broker = brokers[0] |
| args = {} |
| body = "Echo Message Body" |
| args["body"] = body |
| |
| for seq in range (1, 5): |
| args["sequence"] = seq |
| res = mc.syncCallMethod (mch, broker.id, broker.classKey, "echo", args) |
| self.assertEqual (res.status, 0) |
| self.assertEqual (res.statusText, "OK") |
| self.assertEqual (res.sequence, seq) |
| self.assertEqual (res.body, body) |
| mc.removeChannel (mch) |
| |
| def test_methods_sync (self): |
| """ |
| Call the "echo" method on the broker to verify it is alive and talking. |
| """ |
| session = self.session |
| self.startQmf() |
| |
| brokers = self.qmf.getObjects(_class="broker") |
| self.assertEqual(len(brokers), 1) |
| broker = brokers[0] |
| |
| body = "Echo Message Body" |
| for seq in range(1, 20): |
| res = broker.echo(seq, body) |
| self.assertEqual(res.status, 0) |
| self.assertEqual(res.text, "OK") |
| self.assertEqual(res.sequence, seq) |
| self.assertEqual(res.body, body) |
| |
| def test_get_objects(self): |
| self.startQmf() |
| |
| # get the package list, verify that the qpid broker package is there |
| packages = self.qmf.getPackages() |
| assert 'org.apache.qpid.broker' in packages |
| |
| # get the schema class keys for the broker, verify the broker table and link-down event |
| keys = self.qmf.getClasses('org.apache.qpid.broker') |
| broker = None |
| linkDown = None |
| for key in keys: |
| if key.getClassName() == "broker": broker = key |
| if key.getClassName() == "brokerLinkDown" : linkDown = key |
| assert broker |
| assert linkDown |
| |
| brokerObjs = self.qmf.getObjects(_class="broker") |
| assert len(brokerObjs) == 1 |
| brokerObjs = self.qmf.getObjects(_key=broker) |
| assert len(brokerObjs) == 1 |
| |
| def test_self_session_id (self): |
| self.startQmf() |
| sessionId = self.qmf_broker.getSessionId() |
| brokerSessions = self.qmf.getObjects(_class="session") |
| |
| found = False |
| for bs in brokerSessions: |
| if bs.name.endswith(sessionId): |
| found = True |
| self.assertEqual (found, True) |
| |
| def test_standard_exchanges (self): |
| self.startQmf() |
| |
| exchanges = self.qmf.getObjects(_class="exchange") |
| exchange = self.findExchange (exchanges, "") |
| self.assertEqual (exchange.type, "direct") |
| exchange = self.findExchange (exchanges, "amq.direct") |
| self.assertEqual (exchange.type, "direct") |
| exchange = self.findExchange (exchanges, "amq.topic") |
| self.assertEqual (exchange.type, "topic") |
| exchange = self.findExchange (exchanges, "amq.fanout") |
| self.assertEqual (exchange.type, "fanout") |
| exchange = self.findExchange (exchanges, "amq.match") |
| self.assertEqual (exchange.type, "headers") |
| exchange = self.findExchange (exchanges, "qpid.management") |
| self.assertEqual (exchange.type, "topic") |
| |
| def findExchange (self, exchanges, name): |
| for exchange in exchanges: |
| if exchange.name == name: |
| return exchange |
| return None |
| |
| def test_move_queued_messages_empty(self): |
| """ |
| Test that moving messages from an empty queue does not cause an error. |
| """ |
| self.startQmf() |
| session = self.session |
| "Set up source queue" |
| session.queue_declare(queue="src-queue-empty", exclusive=True, auto_delete=True) |
| |
| "Set up destination queue" |
| session.queue_declare(queue="dest-queue-empty", exclusive=True, auto_delete=True) |
| |
| queues = self.qmf.getObjects(_class="queue") |
| |
| "Move all messages from src-queue-empty to dest-queue-empty" |
| result = self.qmf.getObjects(_class="broker")[0].queueMoveMessages("src-queue-empty", "dest-queue-empty", 0, {}) |
| self.assertEqual (result.status, 0) |
| |
| sq = self.qmf.getObjects(_class="queue", name="src-queue-empty")[0] |
| dq = self.qmf.getObjects(_class="queue", name="dest-queue-empty")[0] |
| |
| self.assertEqual (sq.msgDepth,0) |
| self.assertEqual (dq.msgDepth,0) |
| |
| def test_move_queued_messages(self): |
| """ |
| Test ability to move messages from the head of one queue to another. |
| Need to test moveing all and N messages. |
| """ |
| self.startQmf() |
| session = self.session |
| "Set up source queue" |
| session.queue_declare(queue="src-queue", exclusive=True, auto_delete=True) |
| session.exchange_bind(queue="src-queue", exchange="amq.direct", binding_key="routing_key") |
| |
| twenty = range(1,21) |
| props = session.delivery_properties(routing_key="routing_key") |
| for count in twenty: |
| body = "Move Message %d" % count |
| src_msg = Message(props, body) |
| session.message_transfer(destination="amq.direct", message=src_msg) |
| |
| "Set up destination queue" |
| session.queue_declare(queue="dest-queue", exclusive=True, auto_delete=True) |
| session.exchange_bind(queue="dest-queue", exchange="amq.direct") |
| |
| queues = self.qmf.getObjects(_class="queue") |
| |
| "Move 10 messages from src-queue to dest-queue" |
| result = self.qmf.getObjects(_class="broker")[0].queueMoveMessages("src-queue", "dest-queue", 10, {}) |
| self.assertEqual (result.status, 0) |
| |
| sq = self.qmf.getObjects(_class="queue", name="src-queue")[0] |
| dq = self.qmf.getObjects(_class="queue", name="dest-queue")[0] |
| |
| self.assertEqual (sq.msgDepth,10) |
| self.assertEqual (dq.msgDepth,10) |
| |
| "Move all remaining messages to destination" |
| result = self.qmf.getObjects(_class="broker")[0].queueMoveMessages("src-queue", "dest-queue", 0, {}) |
| self.assertEqual (result.status,0) |
| |
| sq = self.qmf.getObjects(_class="queue", name="src-queue")[0] |
| dq = self.qmf.getObjects(_class="queue", name="dest-queue")[0] |
| |
| self.assertEqual (sq.msgDepth,0) |
| self.assertEqual (dq.msgDepth,20) |
| |
| "Use a bad source queue name" |
| result = self.qmf.getObjects(_class="broker")[0].queueMoveMessages("bad-src-queue", "dest-queue", 0, {}) |
| self.assertEqual (result.status,4) |
| |
| "Use a bad destination queue name" |
| result = self.qmf.getObjects(_class="broker")[0].queueMoveMessages("src-queue", "bad-dest-queue", 0, {}) |
| self.assertEqual (result.status,4) |
| |
| " Use a large qty (40) to move from dest-queue back to " |
| " src-queue- should move all " |
| result = self.qmf.getObjects(_class="broker")[0].queueMoveMessages("dest-queue", "src-queue", 40, {}) |
| self.assertEqual (result.status,0) |
| |
| sq = self.qmf.getObjects(_class="queue", name="src-queue")[0] |
| dq = self.qmf.getObjects(_class="queue", name="dest-queue")[0] |
| |
| self.assertEqual (sq.msgDepth,20) |
| self.assertEqual (dq.msgDepth,0) |
| |
| "Consume the messages of the queue and check they are all there in order" |
| session.message_subscribe(queue="src-queue", destination="tag") |
| session.message_flow(destination="tag", unit=session.credit_unit.message, value=0xFFFFFFFF) |
| session.message_flow(destination="tag", unit=session.credit_unit.byte, value=0xFFFFFFFF) |
| queue = session.incoming("tag") |
| for count in twenty: |
| consumed_msg = queue.get(timeout=1) |
| body = "Move Message %d" % count |
| self.assertEqual(body, consumed_msg.body) |
| |
| def test_purge_queue(self): |
| """ |
| Test ability to purge messages from the head of a queue. |
| Need to test moveing all, 1 (top message) and N messages. |
| """ |
| self.startQmf() |
| session = self.session |
| "Set up purge queue" |
| session.queue_declare(queue="purge-queue", exclusive=True, auto_delete=True) |
| session.exchange_bind(queue="purge-queue", exchange="amq.direct", binding_key="routing_key") |
| |
| twenty = range(1,21) |
| props = session.delivery_properties(routing_key="routing_key") |
| for count in twenty: |
| body = "Purge Message %d" % count |
| msg = Message(props, body) |
| session.message_transfer(destination="amq.direct", message=msg) |
| |
| pq = self.qmf.getObjects(_class="queue", name="purge-queue")[0] |
| |
| "Purge top message from purge-queue" |
| result = pq.purge(1, {}) |
| self.assertEqual (result.status, 0) |
| pq = self.qmf.getObjects(_class="queue", name="purge-queue")[0] |
| self.assertEqual (pq.msgDepth,19) |
| |
| "Purge top 9 messages from purge-queue" |
| result = pq.purge(9, {}) |
| self.assertEqual (result.status, 0) |
| pq = self.qmf.getObjects(_class="queue", name="purge-queue")[0] |
| self.assertEqual (pq.msgDepth,10) |
| |
| "Purge all messages from purge-queue" |
| result = pq.purge(0, {}) |
| self.assertEqual (result.status, 0) |
| pq = self.qmf.getObjects(_class="queue", name="purge-queue")[0] |
| self.assertEqual (pq.msgDepth,0) |
| |
| def test_reroute_priority_queue(self): |
| self.startQmf() |
| session = self.session |
| |
| #setup test queue supporting multiple priority levels |
| session.queue_declare(queue="test-queue", exclusive=True, auto_delete=True, arguments={'x-qpid-priorities':10}) |
| |
| #send some messages of varying priority to that queue: |
| for i in range(0, 5): |
| deliveryProps = session.delivery_properties(routing_key="test-queue", priority=i+5) |
| session.message_transfer(message=Message(deliveryProps, "Message %d" % (i+1))) |
| |
| |
| #declare and bind a queue to amq.fanout through which rerouted |
| #messages can be verified: |
| session.queue_declare(queue="rerouted", exclusive=True, auto_delete=True, arguments={'x-qpid-priorities':10}) |
| session.exchange_bind(queue="rerouted", exchange="amq.fanout") |
| |
| #reroute messages from test queue to amq.fanout (and hence to |
| #rerouted queue): |
| pq = self.qmf.getObjects(_class="queue", name="test-queue")[0] |
| result = pq.reroute(0, False, "amq.fanout", {}) |
| self.assertEqual(result.status, 0) |
| |
| #verify messages are all rerouted: |
| self.subscribe(destination="incoming", queue="rerouted") |
| incoming = session.incoming("incoming") |
| for i in range(0, 5): |
| msg = incoming.get(timeout=1) |
| self.assertEqual("Message %d" % (5-i), msg.body) |
| |
| |
| def test_reroute_queue(self): |
| """ |
| Test ability to reroute messages from the head of a queue. |
| Need to test moving all, 1 (top message) and N messages. |
| """ |
| self.startQmf() |
| session = self.session |
| "Set up test queue" |
| session.exchange_declare(exchange="alt.direct1", type="direct") |
| session.queue_declare(queue="alt-queue1", exclusive=True, auto_delete=True) |
| session.exchange_bind(queue="alt-queue1", exchange="alt.direct1", binding_key="routing_key") |
| session.exchange_declare(exchange="alt.direct2", type="direct") |
| session.queue_declare(queue="alt-queue2", exclusive=True, auto_delete=True) |
| session.exchange_bind(queue="alt-queue2", exchange="alt.direct2", binding_key="routing_key") |
| session.queue_declare(queue="reroute-queue", exclusive=True, auto_delete=True, alternate_exchange="alt.direct1") |
| session.exchange_bind(queue="reroute-queue", exchange="amq.direct", binding_key="routing_key") |
| |
| twenty = range(1,21) |
| props = session.delivery_properties(routing_key="routing_key") |
| mp = session.message_properties(application_headers={'x-qpid.trace' : 'A,B,C'}) |
| for count in twenty: |
| body = "Reroute Message %d" % count |
| msg = Message(props, mp, body) |
| session.message_transfer(destination="amq.direct", message=msg) |
| |
| pq = self.qmf.getObjects(_class="queue", name="reroute-queue")[0] |
| |
| "Reroute top message from reroute-queue to alternate exchange" |
| result = pq.reroute(1, True, "", {}) |
| self.assertEqual(result.status, 0) |
| pq.update() |
| aq = self.qmf.getObjects(_class="queue", name="alt-queue1")[0] |
| self.assertEqual(pq.msgDepth,19) |
| self.assertEqual(aq.msgDepth,1) |
| |
| "Verify that the trace was cleared on the rerouted message" |
| url = "%s://%s:%d" % (self.broker.scheme or "amqp", self.broker.host, self.broker.port or 5672) |
| conn = qpid.messaging.Connection(url) |
| conn.open() |
| sess = conn.session() |
| rx = sess.receiver("alt-queue1;{mode:browse}") |
| rm = rx.fetch(1) |
| self.assertEqual(rm.properties['x-qpid.trace'], '') |
| conn.close() |
| |
| "Reroute top 9 messages from reroute-queue to alt.direct2" |
| result = pq.reroute(9, False, "alt.direct2", {}) |
| self.assertEqual(result.status, 0) |
| pq.update() |
| aq = self.qmf.getObjects(_class="queue", name="alt-queue2")[0] |
| self.assertEqual(pq.msgDepth,10) |
| self.assertEqual(aq.msgDepth,9) |
| |
| "Reroute using a non-existent exchange" |
| result = pq.reroute(0, False, "amq.nosuchexchange", {}) |
| self.assertEqual(result.status, 4) |
| |
| "Reroute all messages from reroute-queue" |
| result = pq.reroute(0, False, "alt.direct2", {}) |
| self.assertEqual(result.status, 0) |
| pq.update() |
| aq = self.qmf.getObjects(_class="queue", name="alt-queue2")[0] |
| self.assertEqual(pq.msgDepth,0) |
| self.assertEqual(aq.msgDepth,19) |
| |
| "Make more messages" |
| twenty = range(1,21) |
| props = session.delivery_properties(routing_key="routing_key") |
| for count in twenty: |
| body = "Reroute Message %d" % count |
| msg = Message(props, body) |
| session.message_transfer(destination="amq.direct", message=msg) |
| |
| "Reroute onto the same queue" |
| result = pq.reroute(0, False, "amq.direct", {}) |
| self.assertEqual(result.status, 0) |
| pq.update() |
| self.assertEqual(pq.msgDepth,20) |
| |
| def test_reroute_alternate_exchange(self): |
| """ |
| Test that when rerouting, the alternate-exchange is considered if relevant |
| """ |
| self.startQmf() |
| session = self.session |
| # 1. Create 2 exchanges A and B (fanout) where B is the |
| # alternate exchange for A |
| session.exchange_declare(exchange="B", type="fanout") |
| session.exchange_declare(exchange="A", type="fanout", alternate_exchange="B") |
| |
| # 2. Bind queue X to B |
| session.queue_declare(queue="X", exclusive=True, auto_delete=True) |
| session.exchange_bind(queue="X", exchange="B") |
| |
| # 3. Send 1 message to queue Y |
| session.queue_declare(queue="Y", exclusive=True, auto_delete=True) |
| props = session.delivery_properties(routing_key="Y") |
| session.message_transfer(message=Message(props, "reroute me!")) |
| |
| # 4. Call reroute on queue Y and specify that messages should |
| # be sent to exchange A |
| y = self.qmf.getObjects(_class="queue", name="Y")[0] |
| result = y.reroute(1, False, "A", {}) |
| self.assertEqual(result.status, 0) |
| |
| # 5. verify that the message is rerouted through B (as A has |
| # no matching bindings) to X |
| self.subscribe(destination="x", queue="X") |
| self.assertEqual("reroute me!", session.incoming("x").get(timeout=1).body) |
| |
| # Cleanup |
| for e in ["A", "B"]: session.exchange_delete(exchange=e) |
| |
| def test_reroute_invalid_alt_exchange(self): |
| """ |
| Test that an error is returned for an attempt to reroute to |
| alternate exchange on a queue for which no such exchange has |
| been defined. |
| """ |
| self.startQmf() |
| session = self.session |
| # create queue with no alt-exchange, and send a message to it |
| session.queue_declare(queue="q", exclusive=True, auto_delete=True) |
| props = session.delivery_properties(routing_key="q") |
| session.message_transfer(message=Message(props, "don't reroute me!")) |
| |
| # attempt to reroute the message to alt-exchange |
| q = self.qmf.getObjects(_class="queue", name="q")[0] |
| result = q.reroute(1, True, "", {}) |
| # verify the attempt fails... |
| self.assertEqual(result.status, 4) #invalid parameter |
| |
| # ...and message is still on the queue |
| self.subscribe(destination="d", queue="q") |
| self.assertEqual("don't reroute me!", session.incoming("d").get(timeout=1).body) |
| |
| |
| def test_methods_async (self): |
| """ |
| """ |
| class Handler (qmf.console.Console): |
| def __init__(self): |
| self.cv = Condition() |
| self.xmtList = {} |
| self.rcvList = {} |
| |
| def methodResponse(self, broker, seq, response): |
| self.cv.acquire() |
| try: |
| self.rcvList[seq] = response |
| finally: |
| self.cv.release() |
| |
| def request(self, broker, count): |
| self.count = count |
| for idx in range(count): |
| self.cv.acquire() |
| try: |
| seq = broker.echo(idx, "Echo Message", _async = True) |
| self.xmtList[seq] = idx |
| finally: |
| self.cv.release() |
| |
| def check(self): |
| if self.count != len(self.xmtList): |
| return "fail (attempted send=%d, actual sent=%d)" % (self.count, len(self.xmtList)) |
| lost = 0 |
| mismatched = 0 |
| for seq in self.xmtList: |
| value = self.xmtList[seq] |
| if seq in self.rcvList: |
| result = self.rcvList.pop(seq) |
| if result.sequence != value: |
| mismatched += 1 |
| else: |
| lost += 1 |
| spurious = len(self.rcvList) |
| if lost == 0 and mismatched == 0 and spurious == 0: |
| return "pass" |
| else: |
| return "fail (lost=%d, mismatch=%d, spurious=%d)" % (lost, mismatched, spurious) |
| |
| handler = Handler() |
| self.startQmf(handler) |
| brokers = self.qmf.getObjects(_class="broker") |
| self.assertEqual(len(brokers), 1) |
| broker = brokers[0] |
| handler.request(broker, 20) |
| sleep(1) |
| self.assertEqual(handler.check(), "pass") |
| |
| def test_connection_close(self): |
| """ |
| Test management method for closing connection |
| """ |
| self.startQmf() |
| conn = self.connect() |
| session = conn.session("my-named-session") |
| |
| #using qmf find named session and close the corresponding connection: |
| qmf_ssn_object = [s for s in self.qmf.getObjects(_class="session") if s.name.endswith("my-named-session")][0] |
| qmf_ssn_object._connectionRef_.close() |
| |
| #check that connection is closed |
| try: |
| conn.session("another-session") |
| self.fail("Expected failure from closed connection") |
| except: None |
| |
| #make sure that the named session has been closed and the name can be re-used |
| conn = self.connect() |
| session = conn.session("my-named-session") |
| session.queue_declare(queue="whatever", exclusive=True, auto_delete=True) |
| |
| def test_immediate_method(self): |
| url = "%s://%s:%d" % (self.broker.scheme or "amqp", self.broker.host or "localhost", self.broker.port or 5672) |
| conn = qpid.messaging.Connection(url) |
| conn.open() |
| sess = conn.session() |
| replyTo = "qmf.default.direct/reply_immediate_method_test;{node:{type:topic}}" |
| agent_sender = sess.sender("qmf.default.direct/broker") |
| agent_receiver = sess.receiver(replyTo) |
| queue_create = sess.sender("test-queue-imm-method;{create:always,delete:always,node:{type:queue,durable:False,x-declare:{auto-delete:True}}}") |
| |
| method_request = {'_method_name':'reroute','_object_id':{'_object_name':'org.apache.qpid.broker:queue:test-queue-imm-method'}} |
| method_request['_arguments'] = {'request':0, 'useAltExchange':False, 'exchange':'amq.fanout'} |
| |
| reroute_call = qpid.messaging.Message(method_request) |
| reroute_call.properties['qmf.opcode'] = '_method_request' |
| reroute_call.properties['x-amqp-0-10.app-id'] = 'qmf2' |
| reroute_call.reply_to = replyTo |
| |
| agent_sender.send(reroute_call) |
| result = agent_receiver.fetch(3) |
| self.assertEqual(result.properties['qmf.opcode'], '_method_response') |
| |
| conn.close() |
| |
| def test_binding_count_on_queue(self): |
| self.startQmf() |
| conn = self.connect() |
| session = self.session |
| |
| QUEUE = "binding_test_queue" |
| EX_DIR = "binding_test_exchange_direct" |
| EX_FAN = "binding_test_exchange_fanout" |
| EX_TOPIC = "binding_test_exchange_topic" |
| EX_HDR = "binding_test_exchange_headers" |
| |
| # |
| # Create a test queue |
| # |
| session.queue_declare(queue=QUEUE, exclusive=True, auto_delete=True) |
| queue = self.qmf.getObjects(_class="queue", name=QUEUE)[0] |
| if not queue: |
| self.fail("Queue not found") |
| self.assertEqual(queue.bindingCount, 1, "wrong initial binding count") |
| |
| # |
| # Create an exchange of each supported type |
| # |
| session.exchange_declare(exchange=EX_DIR, type="direct") |
| session.exchange_declare(exchange=EX_FAN, type="fanout") |
| session.exchange_declare(exchange=EX_TOPIC, type="topic") |
| session.exchange_declare(exchange=EX_HDR, type="headers") |
| |
| # |
| # Bind each exchange to the test queue |
| # |
| match = {} |
| match['x-match'] = "all" |
| match['key'] = "value" |
| session.exchange_bind(exchange=EX_DIR, queue=QUEUE, binding_key="key1") |
| session.exchange_bind(exchange=EX_DIR, queue=QUEUE, binding_key="key2") |
| session.exchange_bind(exchange=EX_FAN, queue=QUEUE) |
| session.exchange_bind(exchange=EX_TOPIC, queue=QUEUE, binding_key="key1.#") |
| session.exchange_bind(exchange=EX_TOPIC, queue=QUEUE, binding_key="key2.#") |
| session.exchange_bind(exchange=EX_HDR, queue=QUEUE, binding_key="key1", arguments=match) |
| match['key2'] = "value2" |
| session.exchange_bind(exchange=EX_HDR, queue=QUEUE, binding_key="key2", arguments=match) |
| |
| # |
| # Verify that the queue's binding count accounts for the new bindings |
| # |
| queue.update() |
| self.assertEqual(queue.bindingCount, 8, |
| "added bindings not accounted for (expected 8, got %d)" % queue.bindingCount) |
| |
| # |
| # Remove some of the bindings |
| # |
| session.exchange_unbind(exchange=EX_DIR, queue=QUEUE, binding_key="key2") |
| session.exchange_unbind(exchange=EX_TOPIC, queue=QUEUE, binding_key="key2.#") |
| session.exchange_unbind(exchange=EX_HDR, queue=QUEUE, binding_key="key2") |
| |
| # |
| # Verify that the queue's binding count accounts for the deleted bindings |
| # |
| queue.update() |
| self.assertEqual(queue.bindingCount, 5, |
| "deleted bindings not accounted for (expected 5, got %d)" % queue.bindingCount) |
| # |
| # Delete the exchanges |
| # |
| session.exchange_delete(exchange=EX_DIR) |
| session.exchange_delete(exchange=EX_FAN) |
| session.exchange_delete(exchange=EX_TOPIC) |
| session.exchange_delete(exchange=EX_HDR) |
| |
| # |
| # Verify that the queue's binding count accounts for the lost bindings |
| # |
| queue.update() |
| self.assertEqual(queue.bindingCount, 1, |
| "deleted bindings not accounted for (expected 1, got %d)" % queue.bindingCount) |
| |
| def test_connection_stats(self): |
| """ |
| Test message in/out stats for connection |
| """ |
| agent = self.setup_access() |
| conn = self.connect() |
| session = conn.session("stats-session") |
| |
| #using qmf find named session and the corresponding connection: |
| conn_qmf = None |
| sessions = agent.getAllSessions() |
| for s in sessions: |
| if s.name.endswith("stats-session"): |
| conn_qmf = agent.getConnection(s.connectionRef) |
| |
| assert(conn_qmf) |
| |
| #send a message to a queue |
| session.queue_declare(queue="stats-q", exclusive=True, auto_delete=True) |
| session.message_transfer(message=Message(session.delivery_properties(routing_key="stats-q"), "abc")) |
| |
| #check the 'msgs sent from' stat for this connection |
| conn_qmf.update() |
| self.assertEqual(conn_qmf.msgsFromClient, 1) |
| |
| #receive message from queue |
| session.message_subscribe(destination="d", queue="stats-q") |
| incoming = session.incoming("d") |
| incoming.start() |
| self.assertEqual("abc", incoming.get(timeout=1).body) |
| |
| #check the 'msgs sent to' stat for this connection |
| conn_qmf.update() |
| self.assertEqual(conn_qmf.msgsToClient, 1) |
| |
| def test_timestamp_config(self): |
| """ |
| Test message timestamping control. |
| """ |
| self.startQmf() |
| conn = self.connect() |
| session = conn.session("timestamp-session") |
| |
| #verify that receive message timestamping is OFF by default |
| broker = self.qmf.getObjects(_class="broker")[0] |
| rc = broker.getTimestampConfig() |
| self.assertEqual(rc.status, 0) |
| self.assertEqual(rc.text, "OK") |
| |
| #try to enable it |
| rc = broker.setTimestampConfig(True) |
| self.assertEqual(rc.status, 0) |
| self.assertEqual(rc.text, "OK") |
| |
| rc = broker.getTimestampConfig() |
| self.assertEqual(rc.status, 0) |
| self.assertEqual(rc.text, "OK") |
| self.assertEqual(rc.receive, True) |
| |
| # setup a connection & session to the broker |
| url = "%s://%s:%d" % (self.broker.scheme or "amqp", self.broker.host or "localhost", self.broker.port or 5672) |
| conn = qpid.messaging.Connection(url) |
| conn.open() |
| sess = conn.session() |
| |
| #send a message to a queue |
| sender = sess.sender("ts-q; {create:sender, delete:receiver}") |
| sender.send( qpid.messaging.Message(content="abc") ) |
| |
| #receive message from queue, and verify timestamp is present |
| receiver = sess.receiver("ts-q") |
| try: |
| msg = receiver.fetch(timeout=1) |
| except Empty: |
| assert(False) |
| self.assertEqual("abc", msg.content) |
| self.assertEqual(True, "x-amqp-0-10.timestamp" in msg.properties) |
| assert(msg.properties["x-amqp-0-10.timestamp"]) |
| |
| #try to disable it |
| rc = broker.setTimestampConfig(False) |
| self.assertEqual(rc.status, 0) |
| self.assertEqual(rc.text, "OK") |
| |
| rc = broker.getTimestampConfig() |
| self.assertEqual(rc.status, 0) |
| self.assertEqual(rc.text, "OK") |
| self.assertEqual(rc.receive, False) |
| |
| #send another message to the queue |
| sender.send( qpid.messaging.Message(content="def") ) |
| |
| #receive message from queue, and verify timestamp is NOT PRESENT |
| receiver = sess.receiver("ts-q") |
| try: |
| msg = receiver.fetch(timeout=1) |
| except Empty: |
| assert(False) |
| self.assertEqual("def", msg.content) |
| self.assertEqual(False, "x-amqp-0-10.timestamp" in msg.properties) |
| |
