NO-JIRA: HA Fix ha_tests.py failures with SWIG 0.10 client.

- Fix un-necessary re-sends in amqp0_10::SenderImpl::replay.
- Throw NotFound and UnauthorizedAccess correctly from amqp0_10::SessionImpl and ConnectionImpl
- Fix ha_test wait_address and valid_address re-using a session after it is closed by NotFound.

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

2 files changed, 29 insertions, 39 deletions

diff --git a/qpid/cpp/src/tests/ha_test.py b/qpid/cpp/src/tests/ha_test.py
index 1c131e7872..40ea3854c9 100755
--- a/qpid/cpp/src/tests/ha_test.py
+++ b/qpid/cpp/src/tests/ha_test.py
@@ -241,11 +241,11 @@ acl allow all all
 
     def wait_address(self, address):
         """Wait for address to become valid on the broker."""
-        bs = self.connect_admin().session()
-        try: wait_address(bs, address)
-        finally: bs.connection.close()
+        c = self.connect_admin()
+        try: wait_address(c, address)
+        finally: c.close()
 
-    def wait_backup(self, address): self.wait_address(address)
+    wait_backup = wait_address
 
     def browse(self, queue, timeout=0, transform=lambda m: m.content):
         c = self.connect_admin()
@@ -253,21 +253,15 @@ acl allow all all
             return browse(c.session(), queue, timeout, transform)
         finally: c.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()
+        c = self.connect_admin()
         try:
-            wait_address(bs, queue)
-            assert_browse_retry(bs, queue, expected, **kwargs)
-        finally: bs.connection.close()
+            wait_address(c, queue)
+            assert_browse_retry(c.session(), queue, expected, **kwargs)
+        finally: c.close()
+
+    assert_browse = assert_browse_backup
 
     def assert_connect_fail(self):
         try:
@@ -384,18 +378,17 @@ class HaCluster(object):
     def __iter__(self): return self._brokers.__iter__()
 
 
-def wait_address(session, address):
+def wait_address(connection, address):
     """Wait for an address to become valid."""
-    def check():
-        try: session.sender(address); return True
-        except  qm.NotFound: return False
-    assert retry(check), "Timed out waiting for address %s"%(address)
+    assert retry(lambda: valid_address(connection, address)), "Timed out waiting for address %s"%(address)
 
-def valid_address(session, address):
+def valid_address(connection, address):
     """Test if an address is valid"""
     try:
-        session.receiver(address)
+        s = connection.session().receiver(address)
+        s.session.close()
         return True
-    except qm.NotFound: return False
+    except qm.NotFound:
+        return False
 
 
diff --git a/qpid/cpp/src/tests/ha_tests.py b/qpid/cpp/src/tests/ha_tests.py
index 482296e47f..a43b939ee3 100755
--- a/qpid/cpp/src/tests/ha_tests.py
+++ b/qpid/cpp/src/tests/ha_tests.py
@@ -82,7 +82,7 @@ class ReplicationTests(HaBrokerTest):
         def verify(b, prefix, p):
             """Verify setup was replicated to backup b"""
             # Wait for configuration to replicate.
-            wait_address(b, prefix+"x");
+            wait_address(b.connection, prefix+"x");
             self.assert_browse_retry(b, prefix+"q1", ["b", "1", "4"])
 
             self.assertEqual(p.receiver(prefix+"q1").fetch(timeout=0).content, "b")
@@ -90,7 +90,7 @@ class ReplicationTests(HaBrokerTest):
             self.assert_browse_retry(b, prefix+"q1", ["1", "4"])
 
             self.assert_browse_retry(b, prefix+"q2", []) # configuration only
-            assert not valid_address(b, prefix+"q3")
+            assert not valid_address(b.connection, prefix+"q3")
 
             # Verify exchange with replicate=all
             b.sender(prefix+"e1/key1").send(qm.Message(prefix+"e1"))
@@ -104,8 +104,8 @@ class ReplicationTests(HaBrokerTest):
             self.assert_browse_retry(b, prefix+"q4", ["6","7"])
 
             # Verify deletes
-            assert not valid_address(b, prefix+"dq")
-            assert not valid_address(b, prefix+"de")
+            assert not valid_address(b.connection, prefix+"dq")
+            assert not valid_address(b.connection, prefix+"de")
 
         l = LogLevel(ERROR) # Hide expected WARNING log messages from failover.
         try:
@@ -130,7 +130,7 @@ class ReplicationTests(HaBrokerTest):
             # Test a series of messages, enqueue all then dequeue all.
             primary.agent.addQueue("foo")
             s = p.sender("foo")
-            wait_address(b, "foo")
+            wait_address(b.connection, "foo")
             msgs = [str(i) for i in range(10)]
             for m in msgs: s.send(qm.Message(m))
             self.assert_browse_retry(p, "foo", msgs)
@@ -168,11 +168,8 @@ class ReplicationTests(HaBrokerTest):
 
         msgs = [str(i) for i in range(30)]
         b1 = backup1.connect_admin().session()
-        wait_address(b1, "q");
-        self.assert_browse_retry(b1, "q", msgs)
-        b2 = backup2.connect_admin().session()
-        wait_address(b2, "q");
-        self.assert_browse_retry(b2, "q", msgs)
+        backup1.assert_browse_backup("q", msgs)
+        backup2.assert_browse_backup("q", msgs)
 
     def test_send_receive(self):
         """Verify sequence numbers of messages sent by qpid-send"""
@@ -556,16 +553,16 @@ class ReplicationTests(HaBrokerTest):
 
         s1 = cluster[1].connect_admin().session()
         cluster[1].wait_backup("q")
-        assert not valid_address(s1, "exad")
-        assert valid_address(s1, "ex")
-        assert valid_address(s1, "ad")
-        assert valid_address(s1, "time")
+        assert not valid_address(s1.connection, "exad")
+        assert valid_address(s1.connection, "ex")
+        assert valid_address(s1.connection, "ad")
+        assert valid_address(s1.connection, "time")
 
         # Verify that auto-delete queues are not kept alive by
         # replicating subscriptions
         ad.close()
         s0.sync()
-        assert not valid_address(s0, "ad")
+        assert not valid_address(s0.connection, "ad")
 
     def test_broker_info(self):
         """Check that broker information is correctly published via management"""