diff options
Diffstat (limited to 'cpp/src/tests')
| -rw-r--r-- | cpp/src/tests/Makefile.am | 2 | ||||
| -rw-r--r-- | cpp/src/tests/brokertest.py | 4 | ||||
| -rwxr-xr-x | cpp/src/tests/ha_store_tests.py | 130 | ||||
| -rwxr-xr-x | cpp/src/tests/ha_test.py | 258 | ||||
| -rwxr-xr-x | cpp/src/tests/ha_tests.py | 224 |
5 files changed, 395 insertions, 223 deletions
diff --git a/cpp/src/tests/Makefile.am b/cpp/src/tests/Makefile.am index 2a66aefc03..fb47aea875 100644 --- a/cpp/src/tests/Makefile.am +++ b/cpp/src/tests/Makefile.am @@ -348,7 +348,9 @@ EXTRA_DIST += \ run_msg_group_tests \ ipv6_test \ run_ha_tests \ + ha_test.py \ ha_tests.py \ + ha_store_tests.py \ test_env.ps1.in check_LTLIBRARIES += libdlclose_noop.la diff --git a/cpp/src/tests/brokertest.py b/cpp/src/tests/brokertest.py index dd09e8aa27..0ab0d13424 100644 --- a/cpp/src/tests/brokertest.py +++ b/cpp/src/tests/brokertest.py @@ -425,8 +425,8 @@ class Cluster: self._brokers.append(self.test.broker(self.args+args, name, expect, wait, port=port, show_cmd=show_cmd)) return self._brokers[-1] - def ready(self): - for b in self: b.ready() + def ready(self, timeout=30, **kwargs): + for b in self: b.ready(**kwargs) def start_n(self, count, expect=EXPECT_RUNNING, wait=True, args=[], show_cmd=False): for i in range(count): self.start(expect=expect, wait=wait, args=args, show_cmd=show_cmd) diff --git a/cpp/src/tests/ha_store_tests.py b/cpp/src/tests/ha_store_tests.py new file mode 100755 index 0000000000..d1eaca1b87 --- /dev/null +++ b/cpp/src/tests/ha_store_tests.py @@ -0,0 +1,130 @@ +#!/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. +# + +""" +This module contains tests for HA functionality that requires a store. +It is not included as part of "make check" since it will not function +without a store. Currently it can be run from a build of the message +store. +""" + +import os, signal, sys, time, imp, re, subprocess, glob, random, logging, shutil, math, unittest, random +import traceback +from qpid.messaging import Message, NotFound, ConnectionError, ReceiverError, Connection, Timeout, Disposition, REJECTED, Empty +from qpid.datatypes import uuid4 +from brokertest import * +from ha_test import * +from threading import Thread, Lock, Condition +from logging import getLogger, WARN, ERROR, DEBUG, INFO +from qpidtoollibs import BrokerAgent +from uuid import UUID + + +class StoreTests(BrokerTest): + """Test for HA with persistence.""" + + def test_store_recovery(self): + """Verify basic store and recover functionality""" + cluster = HaCluster(self, 2) + sn = cluster[0].connect().session() + s = sn.sender("qq;{create:always,node:{durable:true}}") + sk = sn.sender("xx/k;{create:always,node:{type:topic, durable:true, x-declare:{type:'direct'}, x-bindings:[{exchange:xx,key:k,queue:qq}]}}") + s.send(Message("foo", durable=True)) + s.send(Message("bar", durable=True)) + sk.send(Message("baz", durable=True)) + r = cluster[0].connect().session().receiver("qq") + self.assertEqual(r.fetch().content, "foo") + r.session.acknowledge() + # FIXME aconway 2012-09-21: sending this message is an ugly hack to flush + # the dequeue operation on qq. + s.send(Message("flush", durable=True)) + + def verify(broker, x_count): + sn = broker.connect().session() + assert_browse(sn, "qq", [ "bar", "baz", "flush" ]+ (x_count)*["x"]) + sn.sender("xx/k").send(Message("x", durable=True)) + assert_browse(sn, "qq", [ "bar", "baz", "flush" ]+ (x_count+1)*["x"]) + + verify(cluster[0], 0) + cluster.bounce(0, promote_next=False) + cluster[0].promote() + cluster[0].wait_status("active") + verify(cluster[0], 1) + cluster.kill(0, promote_next=False) + cluster[1].promote() + cluster[1].wait_status("active") + verify(cluster[1], 2) + cluster.bounce(1, promote_next=False) + cluster[1].promote() + cluster[1].wait_status("active") + verify(cluster[1], 3) + + def test_catchup_store(self): + """Verify that a backup erases queue data from store recovery before + doing catch-up from the primary.""" + cluster = HaCluster(self, 2) + sn = cluster[0].connect().session() + s1 = sn.sender("q1;{create:always,node:{durable:true}}") + for m in ["foo","bar"]: s1.send(Message(m, durable=True)) + s2 = sn.sender("q2;{create:always,node:{durable:true}}") + sk2 = sn.sender("ex/k2;{create:always,node:{type:topic, durable:true, x-declare:{type:'direct'}, x-bindings:[{exchange:ex,key:k2,queue:q2}]}}") + sk2.send(Message("hello", durable=True)) + # Wait for backup to catch up. + cluster[1].assert_browse_backup("q1", ["foo","bar"]) + cluster[1].assert_browse_backup("q2", ["hello"]) + + # Make changes that the backup doesn't see + cluster.kill(1, promote_next=False) + time.sleep(1) # FIXME aconway 2012-09-25: + r1 = cluster[0].connect().session().receiver("q1") + for m in ["foo", "bar"]: self.assertEqual(r1.fetch().content, m) + r1.session.acknowledge() + for m in ["x","y","z"]: s1.send(Message(m, durable=True)) + # Use old connection to unbind + us = cluster[0].connect_old().session(str(uuid4())) + us.exchange_unbind(exchange="ex", binding_key="k2", queue="q2") + us.exchange_bind(exchange="ex", binding_key="k1", queue="q1") + # Restart both brokers from store to get inconsistent sequence numbering. + cluster.bounce(0, promote_next=False) + cluster[0].promote() + cluster[0].wait_status("active") + cluster.restart(1) + cluster[1].wait_status("ready") + + # Verify state + cluster[0].assert_browse("q1", ["x","y","z"]) + cluster[1].assert_browse_backup("q1", ["x","y","z"]) + sn = cluster[0].connect().session() # FIXME aconway 2012-09-25: should fail over! + sn.sender("ex/k1").send("boo") + cluster[0].assert_browse_backup("q1", ["x","y","z", "boo"]) + cluster[1].assert_browse_backup("q1", ["x","y","z", "boo"]) + sn.sender("ex/k2").send("hoo") # q2 was unbound so this should be dropped. + sn.sender("q2").send("end") # mark the end of the queue for assert_browse + cluster[0].assert_browse("q2", ["hello", "end"]) + cluster[1].assert_browse_backup("q2", ["hello", "end"]) + +if __name__ == "__main__": + shutil.rmtree("brokertest.tmp", True) + qpid_ha = os.getenv("QPID_HA_EXEC") + if qpid_ha and os.path.exists(qpid_ha): + os.execvp("qpid-python-test", + ["qpid-python-test", "-m", "ha_store_tests"] + sys.argv[1:]) + else: + print "Skipping ha_store_tests, %s not available"%(qpid_ha) diff --git a/cpp/src/tests/ha_test.py b/cpp/src/tests/ha_test.py new file mode 100755 index 0000000000..18a969a07b --- /dev/null +++ b/cpp/src/tests/ha_test.py @@ -0,0 +1,258 @@ +#!/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, shutil, math, unittest, random +import traceback +from qpid.messaging import Message, NotFound, ConnectionError, ReceiverError, Connection, Timeout, Disposition, REJECTED, Empty +from qpid.datatypes import uuid4 +from brokertest import * +from threading import Thread, Lock, Condition +from logging import getLogger, WARN, ERROR, DEBUG, INFO +from qpidtoollibs import BrokerAgent +from uuid import UUID + +log = getLogger(__name__) + +class QmfAgent(object): + """Access to a QMF broker agent.""" + def __init__(self, address, **kwargs): + self._connection = Connection.establish( + address, client_properties={"qpid.ha-admin":1}, **kwargs) + self._agent = BrokerAgent(self._connection) + + def __getattr__(self, name): + a = getattr(self._agent, name) + return a + +class Credentials(object): + """SASL credentials: username, password, and mechanism""" + def __init__(self, username, password, mechanism): + (self.username, self.password, self.mechanism) = (username, password, mechanism) + + def __str__(self): return "Credentials%s"%(self.tuple(),) + + def tuple(self): return (self.username, self.password, self.mechanism) + + def add_user(self, url): return "%s/%s@%s"%(self.username, self.password, url) + +class HaBroker(Broker): + """Start a broker with HA enabled + @param client_cred: (user, password, mechanism) for admin clients started by the HaBroker. + """ + def __init__(self, test, args=[], brokers_url=None, ha_cluster=True, ha_replicate="all", + client_credentials=None, **kwargs): + assert BrokerTest.ha_lib, "Cannot locate HA plug-in" + args = copy(args) + args += ["--load-module", BrokerTest.ha_lib, + "--log-enable=debug+:ha::", + # FIXME aconway 2012-02-13: workaround slow link failover. + "--link-maintenace-interval=0.1", + "--ha-cluster=%s"%ha_cluster] + if ha_replicate is not None: + args += [ "--ha-replicate=%s"%ha_replicate ] + if brokers_url: args += [ "--ha-brokers-url", brokers_url ] + Broker.__init__(self, test, args, **kwargs) + self.qpid_ha_path=os.path.join(os.getenv("PYTHON_COMMANDS"), "qpid-ha") + assert os.path.exists(self.qpid_ha_path) + self.qpid_config_path=os.path.join(os.getenv("PYTHON_COMMANDS"), "qpid-config") + assert os.path.exists(self.qpid_config_path) + getLogger().setLevel(ERROR) # Hide expected WARNING log messages from failover. + self.qpid_ha_script=import_script(self.qpid_ha_path) + self._agent = None + self.client_credentials = client_credentials + + def __str__(self): return Broker.__str__(self) + + def qpid_ha(self, args): + cred = self.client_credentials + url = self.host_port() + if cred: + url =cred.add_user(url) + args = args + ["--sasl-mechanism", cred.mechanism] + self.qpid_ha_script.main_except(["", "-b", url]+args) + + def promote(self): self.ready(); self.qpid_ha(["promote"]) + def set_client_url(self, url): self.qpid_ha(["set", "--public-url", url]) + def set_brokers_url(self, url): self.qpid_ha(["set", "--brokers-url", url]) + def replicate(self, from_broker, queue): self.qpid_ha(["replicate", from_broker, queue]) + + def agent(self): + if not self._agent: + cred = self.client_credentials + if cred: + self._agent = QmfAgent(cred.add_user(self.host_port()), sasl_mechanisms=cred.mechanism) + else: + self._agent = QmfAgent(self.host_port()) + return self._agent + + def ha_status(self): + hb = self.agent().getHaBroker() + hb.update() + return hb.status + + def wait_status(self, status): + def try_get_status(): + self._status = "<unknown>" + # Ignore ConnectionError, the broker may not be up yet. + try: + self._status = self.ha_status() + return self._status == status; + except ConnectionError: return False + assert retry(try_get_status, timeout=20), "%s expected=%r, actual=%r"%( + self, status, self._status) + + # FIXME aconway 2012-05-01: do direct python call to qpid-config code. + def qpid_config(self, args): + assert subprocess.call( + [self.qpid_config_path, "--broker", self.host_port()]+args) == 0 + + def config_replicate(self, from_broker, queue): + self.qpid_config(["add", "queue", "--start-replica", from_broker, queue]) + + def config_declare(self, queue, replication): + self.qpid_config(["add", "queue", queue, "--replicate", replication]) + + def connect_admin(self, **kwargs): + cred = self.client_credentials + if cred: + return Broker.connect( + self, client_properties={"qpid.ha-admin":1}, + username=cred.username, password=cred.password, sasl_mechanisms=cred.mechanism, + **kwargs) + else: + return Broker.connect(self, client_properties={"qpid.ha-admin":1}, **kwargs) + + def wait_backup(self, address): + """Wait for address to become valid on a backup broker.""" + bs = self.connect_admin().session() + try: wait_address(bs, address) + finally: bs.connection.close() + + def assert_browse(self, queue, expected, **kwargs): + """Verify queue contents by browsing.""" + bs = self.connect().session() + try: + wait_address(bs, queue) + assert_browse_retry(bs, queue, expected, **kwargs) + finally: bs.connection.close() + + def assert_browse_backup(self, queue, expected, **kwargs): + """Combines wait_backup and assert_browse_retry.""" + bs = self.connect_admin().session() + try: + wait_address(bs, queue) + assert_browse_retry(bs, queue, expected, **kwargs) + finally: bs.connection.close() + + def assert_connect_fail(self): + try: + self.connect() + self.test.fail("Expected ConnectionError") + except ConnectionError: pass + + def try_connect(self): + try: return self.connect() + except ConnectionError: return None + + def ready(self): + return Broker.ready(self, client_properties={"qpid.ha-admin":1}) + + +class HaCluster(object): + _cluster_count = 0 + + def __init__(self, test, n, promote=True, **kwargs): + """Start a cluster of n brokers""" + self.test = test + self.kwargs = kwargs + self._brokers = [] + self.id = HaCluster._cluster_count + self.broker_id = 0 + HaCluster._cluster_count += 1 + for i in xrange(n): self.start(False) + self.update_urls() + self[0].promote() + + def next_name(self): + name="cluster%s-%s"%(self.id, self.broker_id) + self.broker_id += 1 + return name + + def start(self, update_urls=True, args=[]): + """Start a new broker in the cluster""" + b = HaBroker(self.test, name=self.next_name(), **self.kwargs) + self._brokers.append(b) + if update_urls: self.update_urls() + return b + + def update_urls(self): + self.url = ",".join([b.host_port() for b in self]) + if len(self) > 1: # No failover addresses on a 1 cluster. + for b in self: b.set_brokers_url(self.url) + + def connect(self, i): + """Connect with reconnect_urls""" + return self[i].connect(reconnect=True, reconnect_urls=self.url.split(",")) + + def kill(self, i, promote_next=True): + """Kill broker i, promote broker i+1""" + self[i].expect = EXPECT_EXIT_FAIL + self[i].kill() + if promote_next: self[(i+1) % len(self)].promote() + + def restart(self, i): + """Start a broker with the same port, name and data directory. It will get + a separate log file: foo.n.log""" + b = self._brokers[i] + self._brokers[i] = HaBroker( + self.test, name=b.name, port=b.port(), brokers_url=self.url, + **self.kwargs) + + def bounce(self, i, promote_next=True): + """Stop and restart a broker in a cluster.""" + if (len(self) == 1): + self.kill(i, promote_next=False) + self.restart(i) + self[i].ready() + if promote_next: self[i].promote() + else: + self.kill(i, promote_next) + self.restart(i) + + # Behave like a list of brokers. + def __len__(self): return len(self._brokers) + def __getitem__(self,index): return self._brokers[index] + def __iter__(self): return self._brokers.__iter__() + +def wait_address(session, address): + """Wait for an address to become valid.""" + def check(): + try: + session.sender(address) + return True + except NotFound: return False + assert retry(check), "Timed out waiting for address %s"%(address) + +def valid_address(session, address): + """Test if an address is valid""" + try: + session.receiver(address) + return True + except NotFound: return False diff --git a/cpp/src/tests/ha_tests.py b/cpp/src/tests/ha_tests.py index 31142de293..de87c49d21 100755 --- a/cpp/src/tests/ha_tests.py +++ b/cpp/src/tests/ha_tests.py @@ -23,230 +23,12 @@ import traceback from qpid.messaging import Message, NotFound, ConnectionError, ReceiverError, Connection, Timeout, Disposition, REJECTED, Empty from qpid.datatypes import uuid4 from brokertest import * +from ha_test import * from threading import Thread, Lock, Condition from logging import getLogger, WARN, ERROR, DEBUG, INFO from qpidtoollibs import BrokerAgent from uuid import UUID -log = getLogger(__name__) - -class QmfAgent(object): - """Access to a QMF broker agent.""" - def __init__(self, address, **kwargs): - self._connection = Connection.establish( - address, client_properties={"qpid.ha-admin":1}, **kwargs) - self._agent = BrokerAgent(self._connection) - - def __getattr__(self, name): - a = getattr(self._agent, name) - return a - -class Credentials(object): - """SASL credentials: username, password, and mechanism""" - def __init__(self, username, password, mechanism): - (self.username, self.password, self.mechanism) = (username, password, mechanism) - - def __str__(self): return "Credentials%s"%(self.tuple(),) - - def tuple(self): return (self.username, self.password, self.mechanism) - - def add_user(self, url): return "%s/%s@%s"%(self.username, self.password, url) - -class HaBroker(Broker): - """Start a broker with HA enabled - @param client_cred: (user, password, mechanism) for admin clients started by the HaBroker. - """ - def __init__(self, test, args=[], brokers_url=None, ha_cluster=True, ha_replicate="all", - client_credentials=None, **kwargs): - assert BrokerTest.ha_lib, "Cannot locate HA plug-in" - args = copy(args) - args += ["--load-module", BrokerTest.ha_lib, - "--log-enable=debug+:ha::", - # FIXME aconway 2012-02-13: workaround slow link failover. - "--link-maintenace-interval=0.1", - "--ha-cluster=%s"%ha_cluster] - if ha_replicate is not None: - args += [ "--ha-replicate=%s"%ha_replicate ] - if brokers_url: args += [ "--ha-brokers-url", brokers_url ] - Broker.__init__(self, test, args, **kwargs) - self.qpid_ha_path=os.path.join(os.getenv("PYTHON_COMMANDS"), "qpid-ha") - assert os.path.exists(self.qpid_ha_path) - self.qpid_config_path=os.path.join(os.getenv("PYTHON_COMMANDS"), "qpid-config") - assert os.path.exists(self.qpid_config_path) - getLogger().setLevel(ERROR) # Hide expected WARNING log messages from failover. - self.qpid_ha_script=import_script(self.qpid_ha_path) - self._agent = None - self.client_credentials = client_credentials - - def __str__(self): return Broker.__str__(self) - - def qpid_ha(self, args): - cred = self.client_credentials - url = self.host_port() - if cred: - url =cred.add_user(url) - args = args + ["--sasl-mechanism", cred.mechanism] - self.qpid_ha_script.main_except(["", "-b", url]+args) - - def promote(self): self.qpid_ha(["promote"]) - def set_client_url(self, url): self.qpid_ha(["set", "--public-url", url]) - def set_brokers_url(self, url): self.qpid_ha(["set", "--brokers-url", url]) - def replicate(self, from_broker, queue): self.qpid_ha(["replicate", from_broker, queue]) - - def agent(self): - if not self._agent: - cred = self.client_credentials - if cred: - self._agent = QmfAgent(cred.add_user(self.host_port()), sasl_mechanisms=cred.mechanism) - else: - self._agent = QmfAgent(self.host_port()) - return self._agent - - def ha_status(self): - hb = self.agent().getHaBroker() - hb.update() - return hb.status - - def wait_status(self, status): - def try_get_status(): - self._status = self.ha_status() - # Ignore ConnectionError, the broker may not be up yet. - try: - self._status = self.ha_status() - return self._status == status; - except ConnectionError: return False - assert retry(try_get_status, timeout=20), "%s expected=%r, actual=%r"%( - self, status, self._status) - - # FIXME aconway 2012-05-01: do direct python call to qpid-config code. - def qpid_config(self, args): - assert subprocess.call( - [self.qpid_config_path, "--broker", self.host_port()]+args) == 0 - - def config_replicate(self, from_broker, queue): - self.qpid_config(["add", "queue", "--start-replica", from_broker, queue]) - - def config_declare(self, queue, replication): - self.qpid_config(["add", "queue", queue, "--replicate", replication]) - - def connect_admin(self, **kwargs): - cred = self.client_credentials - if cred: - return Broker.connect( - self, client_properties={"qpid.ha-admin":1}, - username=cred.username, password=cred.password, sasl_mechanisms=cred.mechanism, - **kwargs) - else: - return Broker.connect(self, client_properties={"qpid.ha-admin":1}, **kwargs) - - def wait_backup(self, address): - """Wait for address to become valid on a backup broker.""" - bs = self.connect_admin().session() - try: wait_address(bs, address) - finally: bs.connection.close() - - def assert_browse(self, queue, expected, **kwargs): - """Verify queue contents by browsing.""" - bs = self.connect().session() - try: - wait_address(bs, queue) - assert_browse_retry(bs, queue, expected, **kwargs) - finally: bs.connection.close() - - def assert_browse_backup(self, queue, expected, **kwargs): - """Combines wait_backup and assert_browse_retry.""" - bs = self.connect_admin().session() - try: - wait_address(bs, queue) - assert_browse_retry(bs, queue, expected, **kwargs) - finally: bs.connection.close() - - def assert_connect_fail(self): - try: - self.connect() - self.test.fail("Expected ConnectionError") - except ConnectionError: pass - - def try_connect(self): - try: return self.connect() - except ConnectionError: return None - -class HaCluster(object): - _cluster_count = 0 - - def __init__(self, test, n, promote=True, **kwargs): - """Start a cluster of n brokers""" - self.test = test - self.kwargs = kwargs - self._brokers = [] - self.id = HaCluster._cluster_count - self.broker_id = 0 - HaCluster._cluster_count += 1 - for i in xrange(n): self.start(False) - self.update_urls() - self[0].promote() - - def next_name(self): - name="cluster%s-%s"%(self.id, self.broker_id) - self.broker_id += 1 - return name - - def start(self, update_urls=True, args=[]): - """Start a new broker in the cluster""" - b = HaBroker(self.test, name=self.next_name(), **self.kwargs) - self._brokers.append(b) - if update_urls: self.update_urls() - return b - - def update_urls(self): - self.url = ",".join([b.host_port() for b in self]) - if len(self) > 1: # No failover addresses on a 1 cluster. - for b in self: b.set_brokers_url(self.url) - - def connect(self, i): - """Connect with reconnect_urls""" - return self[i].connect(reconnect=True, reconnect_urls=self.url.split(",")) - - def kill(self, i, promote_next=True): - """Kill broker i, promote broker i+1""" - self[i].expect = EXPECT_EXIT_FAIL - self[i].kill() - if promote_next: self[(i+1) % len(self)].promote() - - def restart(self, i): - """Start a broker with the same port, name and data directory. It will get - a separate log file: foo.n.log""" - b = self._brokers[i] - self._brokers[i] = HaBroker( - self.test, name=b.name, port=b.port(), brokers_url=self.url, - **self.kwargs) - - def bounce(self, i, promote_next=True): - """Stop and restart a broker in a cluster.""" - self.kill(i, promote_next) - self.restart(i) - - # Behave like a list of brokers. - def __len__(self): return len(self._brokers) - def __getitem__(self,index): return self._brokers[index] - def __iter__(self): return self._brokers.__iter__() - -def wait_address(session, address): - """Wait for an address to become valid.""" - def check(): - try: - session.sender(address) - return True - except NotFound: return False - assert retry(check), "Timed out waiting for address %s"%(address) - -def valid_address(session, address): - """Test if an address is valid""" - try: - session.receiver(address) - return True - except NotFound: return False - class ReplicationTests(BrokerTest): """Correctness tests for HA replication.""" @@ -927,7 +709,7 @@ class LongTests(BrokerTest): if dead is not None: brokers.restart(dead) # Restart backup - brokers[dead].ready(client_properties={"qpid.ha-admin":1}) + brokers[dead].ready() dead = None i += 1 except: @@ -1031,5 +813,5 @@ if __name__ == "__main__": os.execvp("qpid-python-test", ["qpid-python-test", "-m", "ha_tests"] + sys.argv[1:]) else: - print "Skipping ha_tests, qpid_ha not available" + print "Skipping ha_tests, %s not available"%(qpid_ha) |