path: root/qpid/cpp/src/tests/cluster_tests.py

#!/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 os, signal, sys, time, imp, re, subprocess, glob, random, logging
import cluster_test_logs
from qpid import datatypes, messaging
from brokertest import *
from qpid.harness import Skipped
from qpid.messaging import Message, Empty, Disposition, REJECTED, util
from threading import Thread, Lock, Condition
from logging import getLogger
from itertools import chain
from tempfile import NamedTemporaryFile

log = getLogger("qpid.cluster_tests")

# Note: brokers that shut themselves down due to critical error during
# normal operation will still have an exit code of 0. Brokers that
# shut down because of an error found during initialize will exit with
# a non-0 code. Hence the apparently inconsistent use of EXPECT_EXIT_OK
# and EXPECT_EXIT_FAIL in some of the tests below.

# TODO aconway 2010-03-11: resolve this - ideally any exit due to an error
# should give non-0 exit status.

# Import scripts as modules
qpid_cluster=import_script(checkenv("QPID_CLUSTER_EXEC"))

def readfile(filename):
    """Returns te content of file named filename as a string"""
    f = file(filename)
    try: return f.read()
    finally: f.close()

class ShortTests(BrokerTest):
    """Short cluster functionality tests."""

    def test_message_replication(self):
        """Test basic cluster message replication."""
        # Start a cluster, send some messages to member 0.
        cluster = self.cluster(2)
        s0 = cluster[0].connect().session()
        s0.sender("q; {create:always}").send(Message("x"))
        s0.sender("q; {create:always}").send(Message("y"))
        s0.connection.close()

        # Verify messages available on member 1.
        s1 = cluster[1].connect().session()
        m = s1.receiver("q", capacity=1).fetch(timeout=1)
        s1.acknowledge()
        self.assertEqual("x", m.content)
        s1.connection.close()

        # Start member 2 and verify messages available.
        s2 = cluster.start().connect().session()
        m = s2.receiver("q", capacity=1).fetch(timeout=1)
        s2.acknowledge()
        self.assertEqual("y", m.content)
        s2.connection.close()

    def test_store_direct_update_match(self):
        """Verify that brokers stores an identical message whether they receive it
        direct from clients or during an update, no header or other differences"""
        cluster = self.cluster(0, args=["--load-module", self.test_store_lib])
        cluster.start(args=["--test-store-dump", "direct.dump"])
        # Try messages with various headers
        cluster[0].send_message("q", Message(durable=True, content="foobar",
                                             subject="subject",
                                             reply_to="reply_to",
                                             properties={"n":10}))
        # Try messages of different sizes
        for size in range(0,10000,100):
            cluster[0].send_message("q", Message(content="x"*size, durable=True))
        # Try sending via named exchange
        c = cluster[0].connect_old()
        s = c.session(str(qpid.datatypes.uuid4()))
        s.exchange_bind(exchange="amq.direct", binding_key="foo", queue="q")
        props = s.delivery_properties(routing_key="foo", delivery_mode=2)
        s.message_transfer(
            destination="amq.direct",
            message=qpid.datatypes.Message(props, "content"))

        # Try message with TTL and differnet headers/properties
        cluster[0].send_message("q", Message(durable=True, ttl=100000))
        cluster[0].send_message("q", Message(durable=True, properties={}, ttl=100000))
        cluster[0].send_message("q", Message(durable=True, properties={"x":10}, ttl=100000))

        # Now update a new member and compare their dumps.
        cluster.start(args=["--test-store-dump", "updatee.dump"])
        assert readfile("direct.dump") == readfile("updatee.dump")

        os.remove("direct.dump")
        os.remove("updatee.dump")

    def test_sasl(self):
        """Test SASL authentication and encryption in a cluster"""
        sasl_config=os.path.join(self.rootdir, "sasl_config")
        acl=os.path.join(os.getcwd(), "policy.acl")
        aclf=file(acl,"w")
        # Must allow cluster-user (zag) access to credentials exchange.
        aclf.write("""
acl allow zag@QPID publish exchange name=qpid.cluster-credentials
acl allow zig@QPID all all
acl deny all all
""")
        aclf.close()
        cluster = self.cluster(1, args=["--auth", "yes",
                                        "--sasl-config", sasl_config,
                                        "--load-module", os.getenv("ACL_LIB"),
                                        "--acl-file", acl,
                                        "--cluster-username=zag",
                                        "--cluster-password=zag",
                                        "--cluster-mechanism=PLAIN"
                                        ])

        # Valid user/password, ensure queue is created.
        c = cluster[0].connect(username="zig", password="zig")
        c.session().sender("ziggy;{create:always,node:{x-declare:{exclusive:true}}}")
        c.close()
        cluster.start()                 # Start second node.

        # Check queue is created on second node.
        c = cluster[1].connect(username="zig", password="zig")
        c.session().receiver("ziggy;{assert:always}")
        c.close()
        for b in cluster: b.ready()     # Make sure all brokers still running.

        # Valid user, bad password
        try:
            cluster[0].connect(username="zig", password="foo").close()
            self.fail("Expected exception")
        except messaging.exceptions.ConnectionError: pass
        for b in cluster: b.ready()     # Make sure all brokers still running.

        # Bad user ID
        try:
            cluster[0].connect(username="foo", password="bar").close()
            self.fail("Expected exception")
        except messaging.exceptions.ConnectionError: pass
        for b in cluster: b.ready()     # Make sure all brokers still running.

        # Action disallowed by ACL
        c = cluster[0].connect(username="zag", password="zag")
        try:
            s = c.session()
            s.sender("zaggy;{create:always}")
            s.close()
            self.fail("Expected exception")
        except messaging.exceptions.UnauthorizedAccess: pass
        # make sure the queue was not created at the other node.
        c = cluster[1].connect(username="zig", password="zig")
        try:
            s = c.session()
            s.sender("zaggy;{assert:always}")
            s.close()
            self.fail("Expected exception")
        except messaging.exceptions.NotFound: pass

    def test_sasl_join_good(self):
        """Verify SASL authentication between brokers when joining a cluster."""
        sasl_config=os.path.join(self.rootdir, "sasl_config")
        # Test with a valid username/password
        cluster = self.cluster(1, args=["--auth", "yes",
                                        "--sasl-config", sasl_config,
                                        "--cluster-username=zig",
                                        "--cluster-password=zig",
                                        "--cluster-mechanism=PLAIN"
                                        ])
        cluster.start()
        c = cluster[1].connect(username="zag", password="zag", mechanism="PLAIN")

    def test_sasl_join_bad_password(self):
        # Test with an invalid password
        cluster = self.cluster(1, args=["--auth", "yes",
                                        "--sasl-config", os.path.join(self.rootdir, "sasl_config"),
                                        "--cluster-username=zig",
                                        "--cluster-password=bad",
                                        "--cluster-mechanism=PLAIN"
                                        ])
        cluster.start(wait=False, expect=EXPECT_EXIT_FAIL)
        assert cluster[1].log_contains("critical Unexpected error: connection-forced: Authentication failed")

    def test_sasl_join_wrong_user(self):
        # Test with a valid user that is not the cluster user.
        cluster = self.cluster(0, args=["--auth", "yes",
                                        "--sasl-config", os.path.join(self.rootdir, "sasl_config")])
        cluster.start(args=["--cluster-username=zig",
                            "--cluster-password=zig",
                            "--cluster-mechanism=PLAIN"
                            ])

        cluster.start(wait=False, expect=EXPECT_EXIT_FAIL,
                      args=["--cluster-username=zag",
                            "--cluster-password=zag",
                            "--cluster-mechanism=PLAIN"
                            ])
        assert cluster[1].log_contains("critical Unexpected error: unauthorized-access: unauthorized-access: Unauthorized user zag@QPID for qpid.cluster-credentials, should be zig")

    def test_user_id_update(self):
        """Ensure that user-id of an open session is updated to new cluster members"""
        sasl_config=os.path.join(self.rootdir, "sasl_config")
        cluster = self.cluster(1, args=["--auth", "yes", "--sasl-config", sasl_config,
                                        "--cluster-mechanism=ANONYMOUS"])
        c = cluster[0].connect(username="zig", password="zig")
        s = c.session().sender("q;{create:always}")
        s.send(Message("x", user_id="zig")) # Message sent before start new broker
        cluster.start()
        s.send(Message("y", user_id="zig")) # Messsage sent after start of new broker
        # Verify brokers are healthy and messages are on the queue.
        self.assertEqual("x", cluster[0].get_message("q").content)
        self.assertEqual("y", cluster[1].get_message("q").content)

    def test_link_events(self):
        """Regression test for https://bugzilla.redhat.com/show_bug.cgi?id=611543"""
        args = ["--mgmt-pub-interval", 1] # Publish management information every second.
        broker1 = self.cluster(1, args)[0]
        broker2 = self.cluster(1, args)[0]
        qp = self.popen(["qpid-printevents", broker1.host_port()], EXPECT_RUNNING)
        qr = self.popen(["qpid-route", "route", "add",
                         broker1.host_port(), broker2.host_port(),
                         "amq.fanout", "key"
                         ], EXPECT_EXIT_OK)
        # Look for link event in printevents output.
        retry(lambda: find_in_file("brokerLinkUp", qp.outfile("out")))
        broker1.ready()
        broker2.ready()
        qr.wait()

    def test_queue_cleaner(self):
        """ Regression test to ensure that cleanup of expired messages works correctly """
        cluster = self.cluster(2, args=["--queue-purge-interval", 3])

        s0 = cluster[0].connect().session()
        sender = s0.sender("my-lvq; {create: always, node:{x-declare:{arguments:{'qpid.last_value_queue':1}}}}")
        #send 10 messages that will all expire and be cleaned up
        for i in range(1, 10):
            msg = Message("message-%s" % i)
            msg.properties["qpid.LVQ_key"] = "a"
            msg.ttl = 0.1
            sender.send(msg)
        #wait for queue cleaner to run
        time.sleep(3)

        #test all is ok by sending and receiving a message
        msg = Message("non-expiring")
        msg.properties["qpid.LVQ_key"] = "b"
        sender.send(msg)
        s0.connection.close()
        s1 = cluster[1].connect().session()
        m = s1.receiver("my-lvq", capacity=1).fetch(timeout=1)
        s1.acknowledge()
        self.assertEqual("non-expiring", m.content)
        s1.connection.close()

        for b in cluster: b.ready()     # Make sure all brokers still running.


    def test_amqfailover_visible(self):
        """Verify that the amq.failover exchange can be seen by
        QMF-based tools - regression test for BZ615300."""
        broker1 = self.cluster(1)[0]
        broker2 = self.cluster(1)[0]
        qs = subprocess.Popen(["qpid-stat", "-e", "-b", broker1.host_port()],  stdout=subprocess.PIPE)
        out = qs.communicate()[0]
        assert out.find("amq.failover") > 0

    def evaluate_address(self, session, address):
        """Create a receiver just to evaluate an address for its side effects"""
        r = session.receiver(address)
        r.close()

    def test_expire_fanout(self):
        """Regression test for QPID-2874: Clustered broker crashes in assertion in
        cluster/ExpiryPolicy.cpp.
        Caused by a fan-out message being updated as separate messages"""
        cluster = self.cluster(1)
        session0 = cluster[0].connect().session()
        # Create 2 queues bound to fanout exchange.
        self.evaluate_address(session0, "q1;{create:always,node:{x-bindings:[{exchange:'amq.fanout',queue:q1}]}}")
        self.evaluate_address(session0, "q2;{create:always,node:{x-bindings:[{exchange:'amq.fanout',queue:q2}]}}")
        queues = ["q1", "q2"]
        # Send a fanout message with a long timeout
        s = session0.sender("amq.fanout")
        s.send(Message("foo", ttl=100), sync=False)
        # Start a new member, check the messages
        cluster.start()
        session1 = cluster[1].connect().session()
        for q in queues: self.assert_browse(session1, "q1", ["foo"])

    def test_route_update(self):
        """Regression test for https://issues.apache.org/jira/browse/QPID-2982
        Links and bridges associated with routes were not replicated on update.
        This meant extra management objects and caused an exit if a management
        client was attached.
        """
        args=["--mgmt-pub-interval=1","--log-enable=trace+:management"]

        # First broker will be killed.
        cluster0 = self.cluster(1, args=args)
        cluster1 = self.cluster(1, args=args)
        assert 0 == subprocess.call(
            ["qpid-route", "route", "add", cluster0[0].host_port(),
             cluster1[0].host_port(), "dummy-exchange", "dummy-key", "-d"])
        cluster0.start()

        # Wait for qpid-tool:list on cluster0[0] to generate expected output.
        pattern = re.compile("org.apache.qpid.broker.*link")
        qpid_tool = subprocess.Popen(["qpid-tool", cluster0[0].host_port()],
                                     stdin=subprocess.PIPE, stdout=subprocess.PIPE)
        class Scanner(Thread):
            def __init__(self): self.found = False; Thread.__init__(self)
            def run(self):
                for l in qpid_tool.stdout:
                    if pattern.search(l): self.found = True; return
        scanner = Scanner()
        scanner.start()
        start = time.time()
        try:
            # Wait up to 5 second timeout for scanner to find expected output
            while not scanner.found and time.time() < start + 5:
                qpid_tool.stdin.write("list\n") # Ask qpid-tool to list
                for b in cluster0: b.ready() # Raise if any brokers are down
        finally:
            qpid_tool.stdin.write("quit\n")
            qpid_tool.wait()
            scanner.join()
        assert scanner.found
        # Regression test for https://issues.apache.org/jira/browse/QPID-3235
        # Inconsistent stats when changing elder.

        # Force a change of elder
        cluster0.start()
        for b in cluster0: b.ready()
        cluster0[0].expect=EXPECT_EXIT_FAIL # About to die.
        cluster0[0].kill()
        time.sleep(2) # Allow a management interval to pass.
        for b in cluster0[1:]: b.ready()
        # Verify logs are consistent
        cluster_test_logs.verify_logs()

    def test_redelivered(self):
        """Verify that redelivered flag is set correctly on replayed messages"""
        cluster = self.cluster(2, expect=EXPECT_EXIT_FAIL)
        url = "amqp:tcp:%s,tcp:%s" % (cluster[0].host_port(), cluster[1].host_port())
        queue = "my-queue"
        cluster[0].declare_queue(queue)
        self.sender = self.popen(
            ["qpid-send",
             "--broker", url,
             "--address", queue,
             "--sequence=true",
             "--send-eos=1",
             "--messages=100000",
             "--connection-options={%s}"%(Cluster.CONNECTION_OPTIONS)
             ])
        self.receiver = self.popen(
            ["qpid-receive",
             "--broker", url,
             "--address", queue,
             "--ignore-duplicates",
             "--check-redelivered",
             "--connection-options={%s}"%(Cluster.CONNECTION_OPTIONS),
             "--forever"
             ])
        time.sleep(1)#give sender enough time to have some messages to replay
        cluster[0].kill()
        self.sender.wait()
        self.receiver.wait()
        cluster[1].kill()

    class BlockedSend(Thread):
        """Send a message, send is expected to block.
        Verify that it does block (for a given timeout), then allow
        waiting till it unblocks when it is expected to do so."""
        def __init__(self, sender, msg):
            self.sender, self.msg = sender, msg
            self.blocked = True
            self.condition = Condition()
            self.timeout = 0.1    # Time to wait for expected results.
            Thread.__init__(self)
        def run(self):
            try:
                self.sender.send(self.msg, sync=True)
                self.condition.acquire()
                try:
                    self.blocked = False
                    self.condition.notify()
                finally: self.condition.release()
            except Exception,e: print "BlockedSend exception: %s"%e
        def start(self):
            Thread.start(self)
            time.sleep(self.timeout)
            assert self.blocked         # Expected to block
        def assert_blocked(self): assert self.blocked
        def wait(self):                 # Now expecting to unblock
            self.condition.acquire()
            try:
                while self.blocked:
                    self.condition.wait(self.timeout)
                    if self.blocked: raise Exception("Timed out waiting for send to unblock")
            finally: self.condition.release()
            self.join()

    def queue_flowlimit_test(self, brokers):
        """Verify that the queue's flowlimit configuration and state are
        correctly replicated.
        The brokers argument allows this test to run on single broker,
        cluster of 2 pre-startd brokers or cluster where second broker
        starts after queue is in flow control.
        """
        # configure a queue with a specific flow limit on first broker
        ssn0 = brokers.first().connect().session()
        s0 = ssn0.sender("flq; {create:always, node:{type:queue, x-declare:{arguments:{'qpid.flow_stop_count':5, 'qpid.flow_resume_count':3}}}}")
        brokers.first().startQmf()
        q1 = [q for q in brokers.first().qmf_session.getObjects(_class="queue") if q.name == "flq"][0]
        oid = q1.getObjectId()
        self.assertEqual(q1.name, "flq")
        self.assertEqual(q1.arguments, {u'qpid.flow_stop_count': 5L, u'qpid.flow_resume_count': 3L})
        assert not q1.flowStopped
        self.assertEqual(q1.flowStoppedCount, 0)

        # fill the queue on one broker until flow control is active
        for x in range(5): s0.send(Message(str(x)))
        sender = ShortTests.BlockedSend(s0, Message(str(6)))
        sender.start()                  # Tests that sender does block
        # Verify the broker queue goes into a flowStopped state
        deadline = time.time() + 1
        while not q1.flowStopped and time.time() < deadline: q1.update()
        assert q1.flowStopped
        self.assertEqual(q1.flowStoppedCount, 1)
        sender.assert_blocked()         # Still blocked

        # Now verify the  both brokers in cluster have same configuration
        brokers.second().startQmf()
        qs = brokers.second().qmf_session.getObjects(_objectId=oid)
        self.assertEqual(len(qs), 1)
        q2 = qs[0]
        self.assertEqual(q2.name, "flq")
        self.assertEqual(q2.arguments, {u'qpid.flow_stop_count': 5L, u'qpid.flow_resume_count': 3L})
        assert q2.flowStopped
        self.assertEqual(q2.flowStoppedCount, 1)

        # now drain the queue using a session to the other broker
        ssn1 = brokers.second().connect().session()
        r1 = ssn1.receiver("flq", capacity=6)
        for x in range(4):
            r1.fetch(timeout=0)
            ssn1.acknowledge()
        sender.wait()                   # Verify no longer blocked.

        # and re-verify state of queue on both brokers
        q1.update()
        assert not q1.flowStopped
        q2.update()
        assert not q2.flowStopped

        ssn0.connection.close()
        ssn1.connection.close()
        cluster_test_logs.verify_logs()

    def test_queue_flowlimit(self):
        """Test flow limits on a standalone broker"""
        broker = self.broker()
        class Brokers:
            def first(self): return broker
            def second(self): return broker
        self.queue_flowlimit_test(Brokers())

    def test_queue_flowlimit_cluster(self):
        cluster = self.cluster(2)
        class Brokers:
            def first(self): return cluster[0]
            def second(self): return cluster[1]
        self.queue_flowlimit_test(Brokers())

    def test_queue_flowlimit_cluster_join(self):
        cluster = self.cluster(1)
        class Brokers:
            def first(self): return cluster[0]
            def second(self):
                if len(cluster) == 1: cluster.start()
                return cluster[1]
        self.queue_flowlimit_test(Brokers())

    def test_queue_flowlimit_replicate(self):
        """ Verify that a queue which is in flow control BUT has drained BELOW
        the flow control 'stop' threshold, is correctly replicated when a new
        broker is added to the cluster.
        """

        class AsyncSender(Thread):
            """Send a fixed number of msgs from a sender in a separate thread
            so it may block without blocking the test.
            """
            def __init__(self, broker, address, count=1, size=4):
                Thread.__init__(self)
                self.daemon = True
                self.broker = broker
                self.queue = address
                self.count = count
                self.size = size
                self.done = False

            def run(self):
                self.sender = subprocess.Popen(["qpid-send",
                                                "--capacity=1",
                                                "--content-size=%s" % self.size,
                                                "--messages=%s" % self.count,
                                                "--failover-updates",
                                                "--connection-options={%s}"%(Cluster.CONNECTION_OPTIONS),
                                                "--address=%s" % self.queue,
                                                "--broker=%s" % self.broker.host_port()])
                self.sender.wait()
                self.done = True

        cluster = self.cluster(2)
        # create a queue with rather draconian flow control settings
        ssn0 = cluster[0].connect().session()
        s0 = ssn0.sender("flq; {create:always, node:{type:queue, x-declare:{arguments:{'qpid.flow_stop_count':100, 'qpid.flow_resume_count':20}}}}")

        # fire off the sending thread to broker[0], and wait until the queue
        # hits flow control on broker[1]
        sender = AsyncSender(cluster[0], "flq", count=110);
        sender.start();

        cluster[1].startQmf()
        q_obj = [q for q in cluster[1].qmf_session.getObjects(_class="queue") if q.name == "flq"][0]
        deadline = time.time() + 10
        while not q_obj.flowStopped and time.time() < deadline:
            q_obj.update()
        assert q_obj.flowStopped
        assert not sender.done
        assert q_obj.msgDepth < 110

        # Now drain enough messages on broker[1] to drop below the flow stop
        # threshold, but not relieve flow control...
        receiver = subprocess.Popen(["qpid-receive",
                                     "--messages=15",
                                     "--timeout=1",
                                     "--print-content=no",
                                     "--failover-updates",
                                     "--connection-options={%s}"%(Cluster.CONNECTION_OPTIONS),
                                     "--ack-frequency=1",
                                     "--address=flq",
                                     "--broker=%s" % cluster[1].host_port()])
        receiver.wait()
        q_obj.update()
        assert q_obj.flowStopped
        assert not sender.done
        current_depth = q_obj.msgDepth

        # add a new broker to the cluster, and verify that the queue is in flow
        # control on that broker
        cluster.start()
        cluster[2].startQmf()
        q_obj = [q for q in cluster[2].qmf_session.getObjects(_class="queue") if q.name == "flq"][0]
        assert q_obj.flowStopped
        assert q_obj.msgDepth == current_depth

        # now drain the queue on broker[2], and verify that the sender becomes
        # unblocked
        receiver = subprocess.Popen(["qpid-receive",
                                     "--messages=95",
                                     "--timeout=1",
                                     "--print-content=no",
                                     "--failover-updates",
                                     "--connection-options={%s}"%(Cluster.CONNECTION_OPTIONS),
                                     "--ack-frequency=1",
                                     "--address=flq",
                                     "--broker=%s" % cluster[2].host_port()])
        receiver.wait()
        q_obj.update()
        assert not q_obj.flowStopped
        self.assertEqual(q_obj.msgDepth, 0)

        # verify that the sender has become unblocked
        sender.join(timeout=5)
        assert not sender.isAlive()
        assert sender.done

    def test_blocked_queue_delete(self):
        """Verify that producers which are blocked on a queue due to flow
        control are unblocked when that queue is deleted.
        """

        cluster = self.cluster(2)
        cluster[0].startQmf()
        cluster[1].startQmf()

        # configure a queue with a specific flow limit on first broker
        ssn0 = cluster[0].connect().session()
        s0 = ssn0.sender("flq; {create:always, node:{type:queue, x-declare:{arguments:{'qpid.flow_stop_count':5, 'qpid.flow_resume_count':3}}}}")
        q1 = [q for q in cluster[0].qmf_session.getObjects(_class="queue") if q.name == "flq"][0]
        oid = q1.getObjectId()
        self.assertEqual(q1.name, "flq")
        self.assertEqual(q1.arguments, {u'qpid.flow_stop_count': 5L, u'qpid.flow_resume_count': 3L})
        assert not q1.flowStopped
        self.assertEqual(q1.flowStoppedCount, 0)

        # fill the queue on one broker until flow control is active
        for x in range(5): s0.send(Message(str(x)))
        sender = ShortTests.BlockedSend(s0, Message(str(6)))
        sender.start()                  # Tests that sender does block
        # Verify the broker queue goes into a flowStopped state
        deadline = time.time() + 1
        while not q1.flowStopped and time.time() < deadline: q1.update()
        assert q1.flowStopped
        self.assertEqual(q1.flowStoppedCount, 1)
        sender.assert_blocked()         # Still blocked

        # Now verify the  both brokers in cluster have same configuration
        qs = cluster[1].qmf_session.getObjects(_objectId=oid)
        self.assertEqual(len(qs), 1)
        q2 = qs[0]
        self.assertEqual(q2.name, "flq")
        self.assertEqual(q2.arguments, {u'qpid.flow_stop_count': 5L, u'qpid.flow_resume_count': 3L})
        assert q2.flowStopped
        self.assertEqual(q2.flowStoppedCount, 1)

        # now delete the blocked queue from other broker
        ssn1 = cluster[1].connect().session()
        self.evaluate_address(ssn1, "flq;{delete:always}")
        sender.wait()                   # Verify no longer blocked.

        ssn0.connection.close()
        ssn1.connection.close()
        cluster_test_logs.verify_logs()


    def test_alternate_exchange_update(self):
        """Verify that alternate-exchange on exchanges and queues is propagated to new members of a cluster. """
        cluster = self.cluster(1)
        s0 = cluster[0].connect().session()
        # create alt queue bound to amq.fanout exchange, will be destination for alternate exchanges
        self.evaluate_address(s0, "alt;{create:always,node:{x-bindings:[{exchange:'amq.fanout',queue:alt}]}}")
        # create direct exchange ex with alternate-exchange amq.fanout and no queues bound
        self.evaluate_address(s0, "ex;{create:always,node:{type:topic, x-declare:{type:'direct', alternate-exchange:'amq.fanout'}}}")
        # create queue q with alternate-exchange amq.fanout
        self.evaluate_address(s0, "q;{create:always,node:{type:queue, x-declare:{alternate-exchange:'amq.fanout'}}}")

        def verify(broker):
            s = broker.connect().session()
            # Verify unmatched message goes to ex's alternate.
            s.sender("ex").send("foo")
            self.assertEqual("foo", s.receiver("alt").fetch(timeout=0).content)
            # Verify rejected message goes to q's alternate.
            s.sender("q").send("bar")
            msg = s.receiver("q").fetch(timeout=0)
            self.assertEqual("bar", msg.content)
            s.acknowledge(msg, Disposition(REJECTED)) # Reject the message
            self.assertEqual("bar", s.receiver("alt").fetch(timeout=0).content)

        verify(cluster[0])
        cluster.start()
        verify(cluster[1])

    def test_binding_order(self):
        """Regression test for binding order inconsistency in cluster"""
        cluster = self.cluster(1)
        c0 = cluster[0].connect()
        s0 = c0.session()
        # Declare multiple queues bound to same key on amq.topic
        def declare(q,max=0):
            if max: declare = 'x-declare:{arguments:{"qpid.max_count":%d, "qpid.flow_stop_count":0}}'%max
            else: declare = 'x-declare:{}'
            bind='x-bindings:[{queue:%s,key:key,exchange:"amq.topic"}]'%(q)
            s0.sender("%s;{create:always,node:{%s,%s}}" % (q,declare,bind))
        declare('d',max=4)              # Only one with a limit
        for q in ['c', 'b','a']: declare(q)
        # Add a cluster member, send enough messages to exceed the max count
        cluster.start()
        try:
            s = s0.sender('amq.topic/key')
            for m in xrange(1,6): s.send(Message(str(m)))
            self.fail("Expected capacity exceeded exception")
        except messaging.exceptions.TargetCapacityExceeded: pass
        c1 = cluster[1].connect()
        s1 = c1.session()
        s0 = c0.session()        # Old session s0 is broken by exception.
        # Verify queue contents are consistent.
        for q in ['a','b','c','d']:
            self.assertEqual(self.browse(s0, q), self.browse(s1, q))
        # Verify queue contents are "best effort"
        for q in ['a','b','c']: self.assert_browse(s1,q,[str(n) for n in xrange(1,6)])
        self.assert_browse(s1,'d',[str(n) for n in xrange(1,5)])

    def test_deleted_exchange(self):
        """QPID-3215: cached exchange reference can cause cluster inconsistencies
        if exchange is deleted/recreated
        Verify stand-alone case
        """
        cluster = self.cluster()
        # Verify we do not route message via an exchange that has been destroyed.
        cluster.start()
        s0 = cluster[0].connect().session()
        self.evaluate_address(s0, "ex;{create:always,node:{type:topic}}")
        self.evaluate_address(s0, "q;{create:always,node:{x-bindings:[{exchange:'ex',queue:q,key:foo}]}}")
        send0 = s0.sender("ex/foo")
        send0.send("foo")
        self.assert_browse(s0, "q", ["foo"])
        self.evaluate_address(s0, "ex;{delete:always}")
        try:
            send0.send("bar")     # Should fail, exchange is deleted.
            self.fail("Expected not-found exception")
        except qpid.messaging.NotFound: pass
        self.assert_browse(cluster[0].connect().session(), "q", ["foo"])

    def test_deleted_exchange_inconsistent(self):
        """QPID-3215: cached exchange reference can cause cluster inconsistencies
        if exchange is deleted/recreated

        Verify cluster inconsistency.
        """
        cluster = self.cluster()
        cluster.start()
        s0 = cluster[0].connect().session()
        self.evaluate_address(s0, "ex;{create:always,node:{type:topic}}")
        self.evaluate_address(s0, "q;{create:always,node:{x-bindings:[{exchange:'ex',queue:q,key:foo}]}}")
        send0 = s0.sender("ex/foo")
        send0.send("foo")
        self.assert_browse(s0, "q", ["foo"])

        cluster.start()
        s1 = cluster[1].connect().session()
        self.evaluate_address(s0, "ex;{delete:always}")
        try:
            send0.send("bar")
            self.fail("Expected not-found exception")
        except qpid.messaging.NotFound: pass

        self.assert_browse(s1, "q", ["foo"])


    def test_ttl_consistent(self):
        """Ensure we don't get inconsistent errors with message that have TTL very close together"""
        messages = [ Message(str(i), ttl=i/1000.0) for i in xrange(0,1000)]
        messages.append(Message("x"))
        cluster = self.cluster(2)
        sender = cluster[0].connect().session().sender("q;{create:always}")

        def fetch(b):
            receiver = b.connect().session().receiver("q;{create:always}")
            while receiver.fetch().content != "x": pass

        for m in messages: sender.send(m, sync=False)
        for m in messages: sender.send(m, sync=False)
        fetch(cluster[0])
        fetch(cluster[1])
        for m in messages: sender.send(m, sync=False)
        cluster.start()
        fetch(cluster[2])


    def _verify_federation(self, src_broker, src, dst_broker, dst, timeout=30):
        """ Prove that traffic can pass between two federated brokers.
        """
        tot_time = 0
        active = False
        send_session = src_broker.connect().session()
        sender = send_session.sender(src)
        receive_session = dst_broker.connect().session()
        receiver = receive_session.receiver(dst)
        while not active and tot_time < timeout:
            sender.send(Message("Hello from Source!"))
            try:
                receiver.fetch(timeout = 1)
                receive_session.acknowledge()
                # Get this far without Empty exception, and the link is good!
                active = True
                while True:
                    # Keep receiving msgs, as several may have accumulated
                    receiver.fetch(timeout = 1)
                    receive_session.acknowledge()
            except Empty:
                if not active:
                    tot_time += 1
        receiver.close()
        receive_session.close()
        sender.close()
        send_session.close()
        return active

    def test_federation_failover(self):
        """
        Verify that federation operates across failures occuring in a cluster.
        Specifically:
        1) Destination cluster learns of membership changes in the source
        cluster
        2) Destination cluster replicates the current state of the source
        cluster to newly-added members
        """

        # 2 node cluster source, 2 node cluster destination
        src_cluster = self.cluster(2, expect=EXPECT_EXIT_FAIL)
        src_cluster.ready();
        dst_cluster = self.cluster(2, expect=EXPECT_EXIT_FAIL)
        dst_cluster.ready();

        cmd = self.popen(["qpid-config",
                          "--broker", src_cluster[0].host_port(),
                          "add", "queue", "srcQ"], EXPECT_EXIT_OK)
        cmd.wait()

        cmd = self.popen(["qpid-config",
                          "--broker", dst_cluster[0].host_port(),
                          "add", "exchange", "fanout", "destX"], EXPECT_EXIT_OK)
        cmd.wait()

        cmd = self.popen(["qpid-config",
                          "--broker", dst_cluster[0].host_port(),
                          "add", "queue", "destQ"], EXPECT_EXIT_OK)
        cmd.wait()

        cmd = self.popen(["qpid-config",
                          "--broker", dst_cluster[0].host_port(),
                          "bind", "destX", "destQ"], EXPECT_EXIT_OK)
        cmd.wait()

        # federate the srcQ to the destination exchange
        dst_cluster[0].startQmf()
        dst_broker = dst_cluster[0].qmf_session.getObjects(_class="broker")[0]
        result = dst_broker.connect(src_cluster[0].host(), src_cluster[0].port(), False, "PLAIN",
                                    "guest", "guest", "tcp")
        self.assertEqual(result.status, 0, result);

        link = dst_cluster[0].qmf_session.getObjects(_class="link")[0]
        result = link.bridge(False, "srcQ", "destX", "", "", "", True, False, False, 10)
        self.assertEqual(result.status, 0, result)

        # check that traffic passes
        assert self._verify_federation(src_cluster[0], "srcQ", dst_cluster[0], "destQ")

        # add src[2] broker to source cluster
        src_cluster.start(expect=EXPECT_EXIT_FAIL);
        src_cluster.ready();
        assert self._verify_federation(src_cluster[2], "srcQ", dst_cluster[0], "destQ")

        # Kill src[0]. dst[0] should fail over to src[1]
        src_cluster[0].kill()
        for b in src_cluster[1:]: b.ready()
        assert self._verify_federation(src_cluster[1], "srcQ", dst_cluster[0], "destQ")

        # Kill src[1], dst[0] should fail over to src[2]
        src_cluster[1].kill()
        for b in src_cluster[2:]: b.ready()
        assert self._verify_federation(src_cluster[2], "srcQ", dst_cluster[0], "destQ")

        # Kill dest[0], force failover to dest[1]
        dst_cluster[0].kill()
        for b in dst_cluster[1:]: b.ready()
        assert self._verify_federation(src_cluster[2], "srcQ", dst_cluster[1], "destQ")

        # Add dest[2]
        # dest[1] syncs dest[2] to current remote state
        dst_cluster.start(expect=EXPECT_EXIT_FAIL);
        for b in dst_cluster[1:]: b.ready()
        assert self._verify_federation(src_cluster[2], "srcQ", dst_cluster[1], "destQ")

        # Kill dest[1], force failover to dest[2]
        dst_cluster[1].kill()
        for b in dst_cluster[2:]: b.ready()
        assert self._verify_federation(src_cluster[2], "srcQ", dst_cluster[2], "destQ")

        for i in range(2, len(src_cluster)): src_cluster[i].kill()
        for i in range(2, len(dst_cluster)): dst_cluster[i].kill()


    def test_federation_multilink_failover(self):
        """
        Verify that multi-link federation operates across failures occuring in
        a cluster.
        """

        # 1 node cluster source, 1 node cluster destination
        src_cluster = self.cluster(1, expect=EXPECT_EXIT_FAIL)
        src_cluster.ready();
        dst_cluster = self.cluster(1, expect=EXPECT_EXIT_FAIL)
        dst_cluster.ready();

        # federate a direct binding across two separate links

        # first, create a direct exchange bound to two queues using different
        # bindings
        cmd = self.popen(["qpid-config",
                          "--broker", src_cluster[0].host_port(),
                          "add", "exchange", "direct", "FedX"],
                         EXPECT_EXIT_OK)
        cmd.wait()

        cmd = self.popen(["qpid-config",
                          "--broker", dst_cluster[0].host_port(),
                          "add", "exchange", "direct", "FedX"],
                         EXPECT_EXIT_OK)
        cmd.wait()

        cmd = self.popen(["qpid-config",
                          "--broker", dst_cluster[0].host_port(),
                          "add", "queue", "destQ1"],
                         EXPECT_EXIT_OK)
        cmd.wait()

        cmd = self.popen(["qpid-config",
                          "--broker", dst_cluster[0].host_port(),
                          "bind", "FedX", "destQ1", "one"],
                         EXPECT_EXIT_OK)
        cmd.wait()

        cmd = self.popen(["qpid-config",
                          "--broker", dst_cluster[0].host_port(),
                          "add", "queue", "destQ2"],
                         EXPECT_EXIT_OK)
        cmd.wait()

        cmd = self.popen(["qpid-config",
                          "--broker", dst_cluster[0].host_port(),
                          "bind", "FedX", "destQ2", "two"],
                         EXPECT_EXIT_OK)
        cmd.wait()

        # Create two separate links between the dst and source brokers, bind
        # each to different keys
        dst_cluster[0].startQmf()
        dst_broker = dst_cluster[0].qmf_session.getObjects(_class="broker")[0]

        for _l in [("link1", "bridge1", "one"),
                   ("link2", "bridge2", "two")]:
            result = dst_broker.create("link", _l[0],
                                       {"host":src_cluster[0].host(),
                                        "port":src_cluster[0].port()},
                                       False)
            self.assertEqual(result.status, 0, result);
            result = dst_broker.create("bridge", _l[1],
                                       {"link":_l[0],
                                        "src":"FedX",
                                        "dest":"FedX",
                                        "key":_l[2]}, False)
            self.assertEqual(result.status, 0);

        # check that traffic passes
        assert self._verify_federation(src_cluster[0], "FedX/one", dst_cluster[0], "destQ1")
        assert self._verify_federation(src_cluster[0], "FedX/two", dst_cluster[0], "destQ2")

        # add new member, verify traffic
        src_cluster.start(expect=EXPECT_EXIT_FAIL);
        src_cluster.ready();

        dst_cluster.start(expect=EXPECT_EXIT_FAIL);
        dst_cluster.ready();

        assert self._verify_federation(src_cluster[0], "FedX/one", dst_cluster[0], "destQ1")
        assert self._verify_federation(src_cluster[0], "FedX/two", dst_cluster[0], "destQ2")

        src_cluster[0].kill()
        for b in src_cluster[1:]: b.ready()

        assert self._verify_federation(src_cluster[1], "FedX/one", dst_cluster[0], "destQ1")
        assert self._verify_federation(src_cluster[1], "FedX/two", dst_cluster[0], "destQ2")

        dst_cluster[0].kill()
        for b in dst_cluster[1:]: b.ready()

        assert self._verify_federation(src_cluster[1], "FedX/one", dst_cluster[1], "destQ1")
        assert self._verify_federation(src_cluster[1], "FedX/two", dst_cluster[1], "destQ2")

        for i in range(1, len(src_cluster)): src_cluster[i].kill()
        for i in range(1, len(dst_cluster)): dst_cluster[i].kill()



# Some utility code for transaction tests
XA_RBROLLBACK = 1
XA_RBTIMEOUT = 2
XA_OK = 0
dtx_branch_counter = 0

class DtxStatusException(Exception):
    def __init__(self, expect, actual):
        self.expect = expect
        self.actual = actual

    def str(self):
        return "DtxStatusException(expect=%s, actual=%s)"%(self.expect, self.actual)

class DtxTestFixture:
    """Bundle together some common requirements for dtx tests."""
    def __init__(self, test, broker, name, exclusive=False):
        self.test = test
        self.broker = broker
        self.name = name
        # Use old API. DTX is not supported in messaging API.
        self.connection = broker.connect_old()
        self.session = self.connection.session(name, 1) # 1 second timeout
        self.queue = self.session.queue_declare(name, exclusive=exclusive)
        self.session.dtx_select()
        self.consumer = None

    def xid(self, id=None):
        if id is None: id = self.name
        return self.session.xid(format=0, global_id=id)

    def check_status(self, expect, actual):
        if expect != actual: raise DtxStatusException(expect, actual)

    def start(self, id=None, resume=False):
        self.check_status(XA_OK, self.session.dtx_start(xid=self.xid(id), resume=resume).status)

    def end(self, id=None, suspend=False):
        self.check_status(XA_OK, self.session.dtx_end(xid=self.xid(id), suspend=suspend).status)

    def prepare(self, id=None):
        self.check_status(XA_OK, self.session.dtx_prepare(xid=self.xid(id)).status)

    def commit(self, id=None, one_phase=True):
        self.check_status(
            XA_OK, self.session.dtx_commit(xid=self.xid(id), one_phase=one_phase).status)

    def rollback(self, id=None):
        self.check_status(XA_OK, self.session.dtx_rollback(xid=self.xid(id)).status)

    def set_timeout(self, timeout, id=None):
        self.session.dtx_set_timeout(xid=self.xid(id),timeout=timeout)

    def send(self, messages):
       for m in messages:
           dp=self.session.delivery_properties(routing_key=self.name)
           mp=self.session.message_properties()
           self.session.message_transfer(message=qpid.datatypes.Message(dp, mp, m))

    def accept(self):
        """Accept 1 message from queue"""
        consumer_tag="%s-consumer"%(self.name)
        self.session.message_subscribe(queue=self.name, destination=consumer_tag)
        self.session.message_flow(unit = self.session.credit_unit.message, value = 1, destination = consumer_tag)
        self.session.message_flow(unit = self.session.credit_unit.byte, value = 0xFFFFFFFFL, destination = consumer_tag)
        msg = self.session.incoming(consumer_tag).get(timeout=1)
        self.session.message_cancel(destination=consumer_tag)
        self.session.message_accept(qpid.datatypes.RangedSet(msg.id))
        return msg


    def verify(self, sessions, messages):
        for s in sessions:
            self.test.assert_browse(s, self.name, messages)

class DtxTests(BrokerTest):

    def test_dtx_update(self):
        """Verify that DTX transaction state is updated to a new broker.
        Start a collection of transactions, then add a new cluster member,
        then verify they commit/rollback correctly on the new broker."""

        # Note: multiple test have been bundled into one to avoid the need to start/stop
        # multiple brokers per test.

        cluster=self.cluster(1)
        sessions = [cluster[0].connect().session()] # For verify

        # Transaction that will be open when new member joins, then committed.
        t1 = DtxTestFixture(self, cluster[0], "t1")
        t1.start()
        t1.send(["1", "2"])
        t1.verify(sessions, [])          # Not visible outside of transaction

        # Transaction that will be open when  new member joins, then rolled back.
        t2 = DtxTestFixture(self, cluster[0], "t2")
        t2.start()
        t2.send(["1", "2"])

        # Transaction that will be prepared when new member joins, then committed.
        t3 = DtxTestFixture(self, cluster[0], "t3")
        t3.start()
        t3.send(["1", "2"])
        t3.end()
        t3.prepare()
        t1.verify(sessions, [])          # Not visible outside of transaction

        # Transaction that will be prepared when  new member joins, then rolled back.
        t4 = DtxTestFixture(self, cluster[0], "t4")
        t4.start()
        t4.send(["1", "2"])
        t4.end()
        t4.prepare()

        # Transaction using an exclusive queue
        t5 = DtxTestFixture(self, cluster[0], "t5", exclusive=True)
        t5.start()
        t5.send(["1", "2"])

        # Accept messages in a transaction before/after join then commit
        # Note: Message sent outside transaction, we're testing transactional acceptance.
        t6 = DtxTestFixture(self, cluster[0], "t6")
        t6.send(["a","b","c"])
        t6.start()
        self.assertEqual(t6.accept().body, "a");
        t6.verify(sessions, ["b", "c"])

        # Accept messages in a transaction before/after join then roll back
        # Note: Message sent outside transaction, we're testing transactional acceptance.
        t7 = DtxTestFixture(self, cluster[0], "t7")
        t7.send(["a","b","c"])
        t7.start()
        self.assertEqual(t7.accept().body, "a");
        t7.verify(sessions, ["b", "c"])

        # Ended, suspended transactions across join.
        t8 = DtxTestFixture(self, cluster[0], "t8")
        t8.start(id="1")
        t8.send(["x"])
        t8.end(id="1", suspend=True)
        t8.start(id="2")
        t8.send(["y"])
        t8.end(id="2")
        t8.start()
        t8.send("z")


        # Start new cluster member
        cluster.start()
        sessions.append(cluster[1].connect().session())

        # Commit t1
        t1.send(["3","4"])
        t1.verify(sessions, [])
        t1.end()
        t1.commit(one_phase=True)
        t1.verify(sessions, ["1","2","3","4"])

        # Rollback t2
        t2.send(["3","4"])
        t2.end()
        t2.rollback()
        t2.verify(sessions, [])

        # Commit t3
        t3.commit(one_phase=False)
        t3.verify(sessions, ["1","2"])

        # Rollback t4
        t4.rollback()
        t4.verify(sessions, [])

        # Commit t5
        t5.send(["3","4"])
        t5.verify(sessions, [])
        t5.end()
        t5.commit(one_phase=True)
        t5.verify(sessions, ["1","2","3","4"])

        # Commit t6
        self.assertEqual(t6.accept().body, "b");
        t6.verify(sessions, ["c"])
        t6.end()
        t6.commit(one_phase=True)
        t6.session.close()              # Make sure they're not requeued by the session.
        t6.verify(sessions, ["c"])

        # Rollback t7
        self.assertEqual(t7.accept().body, "b");
        t7.verify(sessions, ["c"])
        t7.end()
        t7.rollback()
        t7.verify(sessions, ["a", "b", "c"])

        # Resume t8
        t8.end()
        t8.commit(one_phase=True)
        t8.start("1", resume=True)
        t8.end("1")
        t8.commit("1", one_phase=True)
        t8.commit("2", one_phase=True)
        t8.verify(sessions, ["z", "x","y"])


    def test_dtx_failover_rollback(self):
       """Kill a broker during a transaction, verify we roll back correctly"""
       cluster=self.cluster(1, expect=EXPECT_EXIT_FAIL)
       cluster.start(expect=EXPECT_RUNNING)

       # Test unprepared at crash
       t1 = DtxTestFixture(self, cluster[0], "t1")
       t1.send(["a"])                   # Not in transaction
       t1.start()
       t1.send(["b"])                   # In transaction

       # Test prepared at crash
       t2 = DtxTestFixture(self, cluster[0], "t2")
       t2.send(["a"])                   # Not in transaction
       t2.start()
       t2.send(["b"])                   # In transaction
       t2.end()
       t2.prepare()

       # Crash the broker
       cluster[0].kill()

       # Transactional changes should not appear
       s = cluster[1].connect().session();
       self.assert_browse(s, "t1", ["a"])
       self.assert_browse(s, "t2", ["a"])

    def test_dtx_timeout(self):
        """Verify that dtx timeout works"""
        cluster = self.cluster(1)
        t1 = DtxTestFixture(self, cluster[0], "t1")
        t1.start()
        t1.set_timeout(1)
        time.sleep(1.1)
        try:
            t1.end()
            self.fail("Expected rollback timeout.")
        except DtxStatusException, e:
            self.assertEqual(e.actual, XA_RBTIMEOUT)

class TxTests(BrokerTest):

    def test_tx_update(self):
        """Verify that transaction state is updated to a new broker"""

        def make_message(session, body=None, key=None, id=None):
            dp=session.delivery_properties(routing_key=key)
            mp=session.message_properties(correlation_id=id)
            return qpid.datatypes.Message(dp, mp, body)

        cluster=self.cluster(1)
        # Use old API. TX is not supported in messaging API.
        c = cluster[0].connect_old()
        s = c.session("tx-session", 1)
        s.queue_declare(queue="q")
        # Start transaction
        s.tx_select()
        s.message_transfer(message=make_message(s, "1", "q"))
        # Start new member mid-transaction
        cluster.start()
        # Do more work
        s.message_transfer(message=make_message(s, "2", "q"))
        # Commit the transaction and verify the results.
        s.tx_commit()
        for b in cluster: self.assert_browse(b.connect().session(), "q", ["1","2"])


class LongTests(BrokerTest):
    """Tests that can run for a long time if -DDURATION=<minutes> is set"""
    def duration(self):
        d = self.config.defines.get("DURATION")
        if d: return float(d)*60
        else: return 3                  # Default is to be quick

    def test_failover(self):
        """Test fail-over during continuous send-receive with errors"""

        # Original cluster will all be killed so expect exit with failure
        cluster = self.cluster(3, expect=EXPECT_EXIT_FAIL)
        for b in cluster: b.ready()     # Wait for brokers to be ready
        for b in cluster: ErrorGenerator(b)

        # Start sender and receiver threads
        cluster[0].declare_queue("test-queue")
        sender = NumberedSender(cluster[0], max_depth=1000)
        receiver = NumberedReceiver(cluster[0], sender=sender)
        receiver.start()
        sender.start()
        # Wait for sender & receiver to get up and running
        retry(lambda: receiver.received > 0)

        # Kill original brokers, start new ones for the duration.
        endtime = time.time() + self.duration()
        i = 0
        while time.time() < endtime:
            sender.sender.assert_running()
            receiver.receiver.assert_running()
            cluster[i].kill()
            i += 1
            b = cluster.start(expect=EXPECT_EXIT_FAIL)
            for b in cluster[i:]: b.ready()
            ErrorGenerator(b)
            time.sleep(5)
        sender.stop()
        receiver.stop()
        for i in range(i, len(cluster)): cluster[i].kill()

    def test_management(self, args=[]):
        """
        Stress test: Run management clients and other clients concurrently
        while killing and restarting brokers.
        """

        class ClientLoop(StoppableThread):
            """Run a client executable in a loop."""
            def __init__(self, broker, cmd):
                StoppableThread.__init__(self)
                self.broker=broker
                self.cmd = cmd          # Client command.
                self.lock = Lock()
                self.process = None     # Client process.
                self.start()

            def run(self):
                try:
                    while True:
                        self.lock.acquire()
                        try:
                            if self.stopped: break
                            self.process = self.broker.test.popen(
                                self.cmd, expect=EXPECT_UNKNOWN)
                        finally:
                            self.lock.release()
                        try:
                            exit = self.process.wait()
                        except OSError, e:
                            # Process may already have been killed by self.stop()
                            break
                        except Exception, e:
                            self.process.unexpected(
                                "client of %s: %s"%(self.broker.name, e))
                        self.lock.acquire()
                        try:
                            if self.stopped: break
                            if exit != 0:
                                self.process.unexpected(
                                    "client of %s exit code %s"%(self.broker.name, exit))
                        finally:
                            self.lock.release()
                except Exception, e:
                    self.error = RethrownException("Error in ClientLoop.run")

            def stop(self):
                """Stop the running client and wait for it to exit"""
                self.lock.acquire()
                try:
                    if self.stopped: return
                    self.stopped = True
                    if self.process:
                        try: self.process.kill() # Kill the client.
                        except OSError: pass # The client might not be running.
                finally: self.lock.release()
                StoppableThread.stop(self)

        # body of test_management()

        args += ["--mgmt-pub-interval", 1]
        args += ["--log-enable=trace+:management"]
        # Use store if present.
        if BrokerTest.store_lib: args +=["--load-module", BrokerTest.store_lib]
        cluster = self.cluster(3, args, expect=EXPECT_EXIT_FAIL) # brokers will be killed

        clients = [] # Per-broker list of clients that only connect to one broker.
        mclients = [] # Management clients that connect to every broker in the cluster.

        def start_clients(broker):
            """Start ordinary clients for a broker."""
            cmds=[
                ["qpid-tool", "localhost:%s"%(broker.port())],
                ["qpid-perftest", "--count=5000", "--durable=yes",
                 "--base-name", str(qpid.datatypes.uuid4()), "--port", broker.port()],
                ["qpid-txtest", "--queue-base-name", "tx-%s"%str(qpid.datatypes.uuid4()),
                 "--port", broker.port()],
                ["qpid-queue-stats", "-a", "localhost:%s" %(broker.port())]
                 ]
            clients.append([ClientLoop(broker, cmd) for cmd in cmds])

        def start_mclients(broker):
            """Start management clients that make multiple connections."""
            cmd = ["qpid-cluster", "-C", "localhost:%s" %(broker.port())]
            mclients.append(ClientLoop(broker, cmd))

        endtime = time.time() + self.duration()
        # For long duration, first run is a quarter of the duration.
        runtime = min(5.0, self.duration() / 3.0)
        alive = 0                       # First live cluster member
        for i in range(len(cluster)): start_clients(cluster[i])
        start_mclients(cluster[alive])

        while time.time() < endtime:
            time.sleep(runtime)
            runtime = 5                 # Remaining runs 5 seconds, frequent broker kills
            for b in cluster[alive:]: b.ready() # Check if a broker crashed.
            # Kill the first broker, expect the clients to fail.
            b = cluster[alive]
            b.ready()
            b.kill()
            # Stop the brokers clients and all the mclients.
            for c in clients[alive] + mclients:
                try: c.stop()
                except: pass            # Ignore expected errors due to broker shutdown.
            clients[alive] = []
            mclients = []
            # Start another broker and clients
            alive += 1
            cluster.start(expect=EXPECT_EXIT_FAIL)
            cluster[-1].ready()         # Wait till its ready
            start_clients(cluster[-1])
            start_mclients(cluster[alive])
        for c in chain(mclients, *clients):
            c.stop()
        for b in cluster[alive:]:
            b.ready() # Verify still alive
            b.kill()
        # Verify that logs are consistent
        cluster_test_logs.verify_logs()

    def test_management_qmf2(self):
        self.test_management(args=["--mgmt-qmf2=yes"])

    def test_connect_consistent(self):
        args=["--mgmt-pub-interval=1","--log-enable=trace+:management"]
        cluster = self.cluster(2, args=args)
        end = time.time() + self.duration()
        while (time.time() < end):  # Get a management interval
            for i in xrange(1000): cluster[0].connect().close()
        cluster_test_logs.verify_logs()

    def test_flowlimit_failover(self):
        """Test fail-over during continuous send-receive with flow control
        active.
        """

        # Original cluster will all be killed so expect exit with failure
        cluster = self.cluster(3, expect=EXPECT_EXIT_FAIL)
        for b in cluster: b.ready()     # Wait for brokers to be ready

        # create a queue with rather draconian flow control settings
        ssn0 = cluster[0].connect().session()
        s0 = ssn0.sender("test-queue; {create:always, node:{type:queue, x-declare:{arguments:{'qpid.flow_stop_count':2000, 'qpid.flow_resume_count':100}}}}")

        receiver = NumberedReceiver(cluster[0])
        receiver.start()
        sender = NumberedSender(cluster[0])
        sender.start()
        # Wait for senders & receiver to get up and running
        retry(lambda: receiver.received > 10)

        # Kill original brokers, start new ones for the duration.
        endtime = time.time() + self.duration();
        i = 0
        while time.time() < endtime:
            sender.sender.assert_running()
            receiver.receiver.assert_running()
            for b in cluster[i:]: b.ready() # Check if any broker crashed.
            cluster[i].kill()
            i += 1
            b = cluster.start(expect=EXPECT_EXIT_FAIL)
            time.sleep(5)
        sender.stop()
        receiver.stop()
        for i in range(i, len(cluster)): cluster[i].kill()

    def test_ttl_failover(self):
        """Test that messages with TTL don't cause problems in a cluster with failover"""

        class Client(StoppableThread):

            def __init__(self, broker):
                StoppableThread.__init__(self)
                self.connection = broker.connect(reconnect=True)
                self.auto_fetch_reconnect_urls(self.connection)
                self.session = self.connection.session()

            def auto_fetch_reconnect_urls(self, conn):
                """Replacment for qpid.messaging.util version which is noisy"""
                ssn = conn.session("auto-fetch-reconnect-urls")
                rcv = ssn.receiver("amq.failover")
                rcv.capacity = 10

                def main():
                    while True:
                        try:
                            msg = rcv.fetch()
                            qpid.messaging.util.set_reconnect_urls(conn, msg)
                            ssn.acknowledge(msg, sync=False)
                        except messaging.exceptions.LinkClosed: return
                        except messaging.exceptions.ConnectionError: return

                thread = Thread(name="auto-fetch-reconnect-urls", target=main)
                thread.setDaemon(True)
                thread.start()

            def stop(self):
                StoppableThread.stop(self)
                self.connection.detach()

        class Sender(Client):
            def __init__(self, broker, address):
                Client.__init__(self, broker)
                self.sent = 0    # Number of messages _reliably_ sent.
                self.sender = self.session.sender(address, capacity=1000)

            def send_counted(self, ttl):
                self.sender.send(Message(str(self.sent), ttl=ttl))
                self.sent += 1

            def run(self):
                while not self.stopped:
                    choice = random.randint(0,4)
                    if choice == 0: self.send_counted(None) # No ttl
                    elif choice == 1: self.send_counted(100000) # Large ttl
                    else: # Small ttl, might expire
                        self.sender.send(Message("", ttl=random.random()/10))
                self.sender.send(Message("z"), sync=True) # Chaser.

        class Receiver(Client):

            def __init__(self, broker, address):
                Client.__init__(self, broker)
                self.received = 0 # Number of  non-empty (reliable) messages received.
                self.receiver = self.session.receiver(address, capacity=1000)
            def run(self):
                try:
                    while True:
                        m = self.receiver.fetch(1)
                        if m.content == "z": break
                        if m.content:   # Ignore unreliable messages
                            # Ignore duplicates
                            if int(m.content) == self.received: self.received += 1
                except Exception,e: self.error = e

        # def test_ttl_failover

        # Original cluster will all be killed so expect exit with failure
        # Set small purge interval.
        cluster = self.cluster(3, expect=EXPECT_EXIT_FAIL, args=["--queue-purge-interval=1"])
        for b in cluster: b.ready()     # Wait for brokers to be ready

        # Python client failover produces noisy WARN logs, disable temporarily
        logger = logging.getLogger()
        log_level = logger.getEffectiveLevel()
        logger.setLevel(logging.ERROR)
        sender = None
        receiver = None
        try:
            # Start sender and receiver threads
            receiver = Receiver(cluster[0], "q;{create:always}")
            receiver.start()
            sender = Sender(cluster[0], "q;{create:always}")
            sender.start()
            # Wait for sender & receiver to get up and running
            retry(lambda: receiver.received > 0)

            # Kill brokers in a cycle.
            endtime = time.time() + self.duration()
            runtime = min(5.0, self.duration() / 4.0)
            i = 0
            while time.time() < endtime:
                for b in cluster[i:]: b.ready() # Check if any broker crashed.
                cluster[i].kill()
                i += 1
                b = cluster.start(expect=EXPECT_EXIT_FAIL)
                b.ready()
                time.sleep(runtime)
            sender.stop()
            receiver.stop()
            for b in cluster[i:]:
                b.ready()               # Check it didn't crash
                b.kill()
            self.assertEqual(sender.sent, receiver.received)
            cluster_test_logs.verify_logs()

        finally:
            # Detach to avoid slow reconnect attempts during shut-down if test fails.
            if sender: sender.connection.detach()
            if receiver: receiver.connection.detach()
            logger.setLevel(log_level)

    def test_msg_group_failover(self):
        """Test fail-over during continuous send-receive of grouped messages.
        """

        class GroupedTrafficGenerator(Thread):
            def __init__(self, url, queue, group_key):
                Thread.__init__(self)
                self.url = url
                self.queue = queue
                self.group_key = group_key
                self.status = -1

            def run(self):
                # generate traffic for approx 10 seconds (2011msgs / 200 per-sec)
                cmd = ["msg_group_test",
                       "--broker=%s" % self.url,
                       "--address=%s" % self.queue,
                       "--connection-options={%s}" % (Cluster.CONNECTION_OPTIONS),
                       "--group-key=%s" % self.group_key,
                       "--receivers=2",
                       "--senders=3",
                       "--messages=2011",
                       "--send-rate=200",
                       "--capacity=11",
                       "--ack-frequency=23",
                       "--allow-duplicates",
                       "--group-size=37",
                       "--randomize-group-size",
                       "--interleave=13"]
                #      "--trace"]
                self.generator = Popen( cmd );
                self.status = self.generator.wait()
                return self.status

            def results(self):
                self.join(timeout=30)  # 3x assumed duration
                if self.isAlive(): return -1
                return self.status

        # Original cluster will all be killed so expect exit with failure
        cluster = self.cluster(3, expect=EXPECT_EXIT_FAIL, args=["-t"])
        for b in cluster: b.ready()     # Wait for brokers to be ready

        # create a queue with rather draconian flow control settings
        ssn0 = cluster[0].connect().session()
        q_args = "{'qpid.group_header_key':'group-id', 'qpid.shared_msg_group':1}"
        s0 = ssn0.sender("test-group-q; {create:always, node:{type:queue, x-declare:{arguments:%s}}}" % q_args)

        # Kill original brokers, start new ones for the duration.
        endtime = time.time() + self.duration();
        i = 0
        while time.time() < endtime:
            traffic = GroupedTrafficGenerator( cluster[i].host_port(),
                                               "test-group-q", "group-id" )
            traffic.start()
            time.sleep(1)

            for x in range(2):
                for b in cluster[i:]: b.ready() # Check if any broker crashed.
                cluster[i].kill()
                i += 1
                b = cluster.start(expect=EXPECT_EXIT_FAIL)
                time.sleep(1)

            # wait for traffic to finish, verify success
            self.assertEqual(0, traffic.results())

        for i in range(i, len(cluster)): cluster[i].kill()


class StoreTests(BrokerTest):
    """
    Cluster tests that can only be run if there is a store available.
    """
    def args(self):
        assert BrokerTest.store_lib
        return ["--load-module", BrokerTest.store_lib]

    def test_store_loaded(self):
        """Ensure we are indeed loading a working store"""
        broker = self.broker(self.args(), name="recoverme", expect=EXPECT_EXIT_FAIL)
        m = Message("x", durable=True)
        broker.send_message("q", m)
        broker.kill()
        broker = self.broker(self.args(), name="recoverme")
        self.assertEqual("x", broker.get_message("q").content)

    def test_kill_restart(self):
        """Verify we can kill/resetart a broker with store in a cluster"""
        cluster = self.cluster(1, self.args())
        cluster.start("restartme", expect=EXPECT_EXIT_FAIL).kill()

        # Send a message, retrieve from the restarted broker
        cluster[0].send_message("q", "x")
        m = cluster.start("restartme").get_message("q")
        self.assertEqual("x", m.content)

    def stop_cluster(self,broker):
        """Clean shut-down of a cluster"""
        self.assertEqual(0, qpid_cluster.main(
            ["-kf", broker.host_port()]))

    def test_persistent_restart(self):
        """Verify persistent cluster shutdown/restart scenarios"""
        cluster = self.cluster(0, args=self.args() + ["--cluster-size=3"])
        a = cluster.start("a", expect=EXPECT_EXIT_OK, wait=False)
        b = cluster.start("b", expect=EXPECT_EXIT_OK, wait=False)
        c = cluster.start("c", expect=EXPECT_EXIT_FAIL, wait=True)
        a.send_message("q", Message("1", durable=True))
        # Kill & restart one member.
        c.kill()
        self.assertEqual(a.get_message("q").content, "1")
        a.send_message("q", Message("2", durable=True))
        c = cluster.start("c", expect=EXPECT_EXIT_OK)
        self.assertEqual(c.get_message("q").content, "2")
        # Shut down the entire cluster cleanly and bring it back up
        a.send_message("q", Message("3", durable=True))
        self.stop_cluster(a)
        a = cluster.start("a", wait=False)
        b = cluster.start("b", wait=False)
        c = cluster.start("c", wait=True)
        self.assertEqual(a.get_message("q").content, "3")

    def test_persistent_partial_failure(self):
        # Kill 2 members, shut down the last cleanly then restart
        # Ensure we use the clean database
        cluster = self.cluster(0, args=self.args() + ["--cluster-size=3"])
        a = cluster.start("a", expect=EXPECT_EXIT_FAIL, wait=False)
        b = cluster.start("b", expect=EXPECT_EXIT_FAIL, wait=False)
        c = cluster.start("c", expect=EXPECT_EXIT_OK, wait=True)
        a.send_message("q", Message("4", durable=True))
        a.kill()
        b.kill()
        self.assertEqual(c.get_message("q").content, "4")
        c.send_message("q", Message("clean", durable=True))
        self.stop_cluster(c)
        a = cluster.start("a", wait=False)
        b = cluster.start("b", wait=False)
        c = cluster.start("c", wait=True)
        self.assertEqual(a.get_message("q").content, "clean")

    def test_wrong_cluster_id(self):
        # Start a cluster1 broker, then try to restart in cluster2
        cluster1 = self.cluster(0, args=self.args())
        a = cluster1.start("a", expect=EXPECT_EXIT_OK)
        a.terminate()
        cluster2 = self.cluster(1, args=self.args())
        try:
            a = cluster2.start("a", expect=EXPECT_EXIT_FAIL)
            a.ready()
            self.fail("Expected exception")
        except: pass

    def test_wrong_shutdown_id(self):
        # Start 2 members and shut down.
        cluster = self.cluster(0, args=self.args()+["--cluster-size=2"])
        a = cluster.start("a", expect=EXPECT_EXIT_OK, wait=False)
        b = cluster.start("b", expect=EXPECT_EXIT_OK, wait=False)
        self.stop_cluster(a)
        self.assertEqual(a.wait(), 0)
        self.assertEqual(b.wait(), 0)

        # Restart with a different member and shut down.
        a = cluster.start("a", expect=EXPECT_EXIT_OK, wait=False)
        c = cluster.start("c", expect=EXPECT_EXIT_OK, wait=False)
        self.stop_cluster(a)
        self.assertEqual(a.wait(), 0)
        self.assertEqual(c.wait(), 0)
        # Mix members from both shutdown events, they should fail
        # TODO aconway 2010-03-11: can't predict the exit status of these
        # as it depends on the order of delivery of initial-status messages.
        # See comment at top of this file.
        a = cluster.start("a", expect=EXPECT_UNKNOWN, wait=False)
        b = cluster.start("b", expect=EXPECT_UNKNOWN, wait=False)
        self.assertRaises(Exception, lambda: a.ready())
        self.assertRaises(Exception, lambda: b.ready())

    def test_solo_store_clean(self):
        # A single node cluster should always leave a clean store.
        cluster = self.cluster(0, self.args())
        a = cluster.start("a", expect=EXPECT_EXIT_FAIL)
        a.send_message("q", Message("x", durable=True))
        a.kill()
        a = cluster.start("a")
        self.assertEqual(a.get_message("q").content, "x")

    def test_last_store_clean(self):
        # Verify that only the last node in a cluster to shut down has
        # a clean store. Start with cluster of 3, reduce to 1 then
        # increase again to ensure that a node that was once alone but
        # finally did not finish as the last node does not get a clean
        # store.
        cluster = self.cluster(0, self.args())
        a = cluster.start("a", expect=EXPECT_EXIT_FAIL)
        self.assertEqual(a.store_state(), "clean")
        b = cluster.start("b", expect=EXPECT_EXIT_FAIL)
        c = cluster.start("c", expect=EXPECT_EXIT_FAIL)
        self.assertEqual(b.store_state(), "dirty")
        self.assertEqual(c.store_state(), "dirty")
        retry(lambda: a.store_state() == "dirty")

        a.send_message("q", Message("x", durable=True))
        a.kill()
        b.kill()                # c is last man, will mark store clean
        retry(lambda: c.store_state() == "clean")
        a = cluster.start("a", expect=EXPECT_EXIT_FAIL) # c no longer last man
        retry(lambda: c.store_state() == "dirty")
        c.kill()                        # a is now last man
        retry(lambda: a.store_state() == "clean")
        a.kill()
        self.assertEqual(a.store_state(), "clean")
        self.assertEqual(b.store_state(), "dirty")
        self.assertEqual(c.store_state(), "dirty")

    def test_restart_clean(self):
        """Verify that we can re-start brokers one by one in a
        persistent cluster after a clean oshutdown"""
        cluster = self.cluster(0, self.args())
        a = cluster.start("a", expect=EXPECT_EXIT_OK)
        b = cluster.start("b", expect=EXPECT_EXIT_OK)
        c = cluster.start("c", expect=EXPECT_EXIT_OK)
        a.send_message("q", Message("x", durable=True))
        self.stop_cluster(a)
        a = cluster.start("a")
        b = cluster.start("b")
        c = cluster.start("c")
        self.assertEqual(c.get_message("q").content, "x")

    def test_join_sub_size(self):
        """Verify that after starting a cluster with cluster-size=N,
        we can join new members even if size < N-1"""
        cluster = self.cluster(0, self.args()+["--cluster-size=3"])
        a = cluster.start("a", wait=False, expect=EXPECT_EXIT_FAIL)
        b = cluster.start("b", wait=False, expect=EXPECT_EXIT_FAIL)
        c = cluster.start("c")
        a.send_message("q", Message("x", durable=True))
        a.send_message("q", Message("y", durable=True))
        a.kill()
        b.kill()
        a = cluster.start("a")
        self.assertEqual(c.get_message("q").content, "x")
        b = cluster.start("b")
        self.assertEqual(c.get_message("q").content, "y")