path: root/python/tests_0-10/management.py

#
# 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.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

class ManagementTest (TestBase010):
    """
    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 == 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(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=0xFFFFFFFFL)
        session.message_flow(destination="tag", unit=session.credit_unit.byte, value=0xFFFFFFFFL)
        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_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 = self.qmf.getObjects(_class="session", name="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_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)