QPID-4944: HA re-enable test_failover_send_receive

Appears to have been fixed at this point on trunk, not clear which checkins are
responsible. Test ran for 48 hours with no failures.

Other minor changes:
- Enable test_failover_send_receive
- Increase heartbeat interval.
- Reduce capacity of senders in failover test to be more aggressive.
- Use HaBrokerTest as test base

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1522711 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 35 insertions, 26 deletions

diff --git a/qpid/cpp/src/tests/brokertest.py b/qpid/cpp/src/tests/brokertest.py
index ba6c259a26..c8f8d3ff38 100644
--- a/qpid/cpp/src/tests/brokertest.py
+++ b/qpid/cpp/src/tests/brokertest.py
@@ -141,7 +141,7 @@ class Popen(subprocess.Popen):
         finally: f.close()
         log.debug("Started process %s: %s" % (self.pname, " ".join(self.cmd)))
 
-    def __str__(self): return "Popen<%s>"%(self.pname)
+    def __repr__(self): return "Popen<%s>"%(self.pname)
 
     def outfile(self, ext): return "%s.%s" % (self.pname, ext)
 
@@ -242,7 +242,7 @@ class Broker(Popen):
     _broker_count = 0
     _log_count = 0
 
-    def __str__(self): return "Broker<%s %s :%d>"%(self.log, self.pname, self.port())
+    def __repr__(self): return "<Broker:%s:%d>"%(self.log, self.port())
 
     def get_log(self):
         return os.path.abspath(self.log)
diff --git a/qpid/cpp/src/tests/ha_test.py b/qpid/cpp/src/tests/ha_test.py
index 3280540db7..3b5874875a 100755
--- a/qpid/cpp/src/tests/ha_test.py
+++ b/qpid/cpp/src/tests/ha_test.py
@@ -119,6 +119,9 @@ class HaBroker(Broker):
     """Start a broker with HA enabled
     @param client_cred: (user, password, mechanism) for admin clients started by the HaBroker.
     """
+
+    heartbeat=2
+
     def __init__(self, test, ha_port=None, args=[], brokers_url=None, ha_cluster=True,
                  ha_replicate="all", client_credentials=None, **kwargs):
         assert BrokerTest.ha_lib, "Cannot locate HA plug-in"
@@ -130,7 +133,7 @@ class HaBroker(Broker):
                  "--link-maintenance-interval=0.1",
                  # Heartbeat and negotiate time are needed so that a broker wont
                  # stall on an address that doesn't currently have a broker running.
-                 "--link-heartbeat-interval=1",
+                 "--link-heartbeat-interval=%s"%(HaBroker.heartbeat),
                  "--max-negotiate-time=1000",
                  "--ha-cluster=%s"%ha_cluster]
         if ha_replicate is not None:
@@ -157,15 +160,18 @@ acl allow all all
         self.client_credentials = client_credentials
         self.ha_port = ha_port
 
-    def __str__(self): return Broker.__str__(self)
+    def __repr__(self): return "<HaBroker:%s:%d>"%(self.log, self.port())
 
     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)
+        try:
+            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)
+        except Exception, e:
+            raise Exception("Error in qpid_ha -b %s %s: %s"%(url, args,e))
 
     def promote(self): self.ready(); self.qpid_ha(["promote"])
     def set_public_url(self, url): self.qpid_ha(["set", "--public-url", url])
diff --git a/qpid/cpp/src/tests/ha_tests.py b/qpid/cpp/src/tests/ha_tests.py
index 5fd0e2fa40..3c126072af 100755
--- a/qpid/cpp/src/tests/ha_tests.py
+++ b/qpid/cpp/src/tests/ha_tests.py
@@ -220,7 +220,8 @@ class ReplicationTests(HaBrokerTest):
             backup.connect_admin().close()
 
             # Test discovery: should connect to primary after reject by backup
-            c = backup.connect(reconnect_urls=[primary.host_port(), backup.host_port()], reconnect=True)
+            c = backup.connect(reconnect_urls=[primary.host_port(), backup.host_port()],
+                               reconnect=True)
             s = c.session()
             sender = s.sender("q;{create:always}")
             backup.wait_backup("q")
@@ -933,20 +934,22 @@ class LongTests(HaBrokerTest):
         if d: return float(d)*60
         else: return 3                  # Default is to be quick
 
-    # FIXME aconway 2013-06-27: skip this test pending a fix for
-    # https://issues.apache.org/jira/browse/QPID-4944
-    def skip_test_failover_send_receive(self):
+    def test_failover_send_receive(self):
         """Test failover with continuous send-receive"""
         brokers = HaCluster(self, 3)
 
         # Start sender and receiver threads
         n = 10
-        senders = [NumberedSender(brokers[0], url=brokers.url,
-                                  max_depth=1024, failover_updates=False,
-                                  queue="test%s"%(i)) for i in xrange(n)]
-        receivers = [NumberedReceiver(brokers[0], url=brokers.url, sender=senders[i],
-                                      failover_updates=False,
-                                      queue="test%s"%(i)) for i in xrange(n)]
+        senders = [
+            NumberedSender(
+                brokers[0], url=brokers.url,max_depth=1024, failover_updates=False,
+                queue="test%s"%(i), args=["--capacity=10"]) for i in xrange(n)]
+
+        receivers = [
+            NumberedReceiver(
+                brokers[0], url=brokers.url, sender=senders[i],failover_updates=False,
+                queue="test%s"%(i), args=["--capacity=10"]) for i in xrange(n)]
+
         for r in receivers: r.start()
         for s in senders: s.start()
 
@@ -997,7 +1000,7 @@ class LongTests(HaBrokerTest):
         finally:
             for s in senders: s.stop()
             for r in receivers: r.stop()
-            dead = filter(lambda i: not brokers[i].is_running(), xrange(3))
+            dead = filter(lambda b: not b.is_running(), brokers)
             if dead: raise Exception("Brokers not running: %s"%dead)
 
     def test_qmf_order(self):
@@ -1206,7 +1209,7 @@ class ConfigurationTests(HaBrokerTest):
         cluster[0].set_brokers_url(cluster.url+",xxx:1234")
         self.assertRaises(Empty, r.fetch, 0) # Not updated for brokers URL
 
-class StoreTests(BrokerTest):
+class StoreTests(HaBrokerTest):
     """Test for HA with persistence."""
 
     def check_skip(self):
@@ -1254,7 +1257,7 @@ class StoreTests(BrokerTest):
         doing catch-up from the primary."""
         if self.check_skip(): return
         cluster = HaCluster(self, 2)
-        sn = cluster[0].connect(heartbeat=1).session()
+        sn = cluster[0].connect(heartbeat=HaBroker.heartbeat).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}}")
@@ -1265,7 +1268,7 @@ class StoreTests(BrokerTest):
         cluster[1].assert_browse_backup("q2", ["hello"])
         # Make changes that the backup doesn't see
         cluster.kill(1, promote_next=False, final=False)
-        r1 = cluster[0].connect(heartbeat=1).session().receiver("q1")
+        r1 = cluster[0].connect(heartbeat=HaBroker.heartbeat).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))
@@ -1284,7 +1287,7 @@ class StoreTests(BrokerTest):
         cluster[0].assert_browse("q1",  ["x","y","z"])
         cluster[1].assert_browse_backup("q1",  ["x","y","z"])
 
-        sn = cluster[0].connect(heartbeat=1).session()
+        sn = cluster[0].connect(heartbeat=HaBroker.heartbeat).session()
         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"])
@@ -1299,7 +1302,7 @@ def open_read(name):
         return f.read()
     finally: f.close()
 
-class TransactionTests(BrokerTest):
+class TransactionTests(HaBrokerTest):
 
     load_store=["--load-module", BrokerTest.test_store_lib]