svn commit: r1383093 - /qpid/trunk/qpid/cpp/src/tests/ha_tests.py

[ac...@apache.org]

Author: aconway
Date: Mon Sep 10 20:01:12 2012
New Revision: 1383093

URL: http://svn.apache.org/viewvc?rev=1383093&view=rev
Log:
QPID-4259, QPID-4262: HA broker can crash due to priority queue corruption

Already fixed on trunk. Comitting just the regression test.

Modified:
    qpid/trunk/qpid/cpp/src/tests/ha_tests.py

Modified: qpid/trunk/qpid/cpp/src/tests/ha_tests.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ha_tests.py?rev=1383093&r1=1383092&r2=1383093&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ha_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/ha_tests.py Mon Sep 10 20:01:12 2012
@@ -20,7 +20,7 @@
 
 import os, signal, sys, time, imp, re, subprocess, glob, random, logging, shutil, math, unittest
 import traceback
-from qpid.messaging import Message, NotFound, ConnectionError, ReceiverError, Connection, Timeout, Disposition, REJECTED
+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
@@ -36,7 +36,6 @@ class QmfAgent(object):
         self._connection = Connection.establish(
             address, client_properties={"qpid.ha-admin":1}, **kwargs)
         self._agent = BrokerAgent(self._connection)
-        assert self._agent.getHaBroker(), "HA module not loaded in broker at: %s"%(address)
 
     def __getattr__(self, name):
         a = getattr(self._agent, name)
@@ -805,6 +804,21 @@ acl deny all all
         cluster.bounce(1)
         verify(cluster[2])
 
+    def test_priority_reroute(self):
+        """Regression test for QPID-4262, rerouting messages from a priority queue
+        to itself causes a crash"""
+        cluster = HaCluster(self, 2)
+        primary = cluster[0]
+        session = primary.connect().session()
+        s = session.sender("pq; {create:always, node:{x-declare:{arguments:{'qpid.priorities':10}},x-bindings:[{exchange:'amq.fanout',queue:pq}]}}")
+        for m in xrange(100): s.send(Message(str(m), priority=m%10))
+        pq =  QmfAgent(primary.host_port()).getQueue("pq")
+        pq.reroute(request=0, useAltExchange=False, exchange="amq.fanout")
+        # Verify that consuming is in priority order
+        expect = [str(10*i+p) for p in xrange(9,-1,-1) for i in xrange(0,10) ]
+        actual = [m.content for m in primary.get_messages("pq", 100)]
+        self.assertEqual(expect, actual)
+
 def fairshare(msgs, limit, levels):
     """
     Generator to return prioritised messages in expected order for a given fairshare limit



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org