From 12048dcd9d13f1c2baeffb4b65c4ce65c0155234 Mon Sep 17 00:00:00 2001 From: Kim van der Riet Date: Wed, 3 Jun 2009 15:25:56 +0000 Subject: Python cluster test improvements and some additional tests git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@781431 13f79535-47bb-0310-9956-ffa450edef68 --- cpp/src/tests/cluster.py | 433 ++++++++++++++-------------------------- cpp/src/tests/run_cluster_tests | 30 +-- cpp/src/tests/testlib.py | 382 +++++++++++++++++++++++++++++------ 3 files changed, 491 insertions(+), 354 deletions(-) (limited to 'cpp/src') diff --git a/cpp/src/tests/cluster.py b/cpp/src/tests/cluster.py index 87c9b363a8..ae4cc7aa85 100755 --- a/cpp/src/tests/cluster.py +++ b/cpp/src/tests/cluster.py @@ -29,10 +29,9 @@ class ClusterTests(TestBaseCluster): try: clusterName = "cluster-01" self.createCheckCluster(clusterName, 5) - self.checkNumBrokers(5) self.stopCheckCluster(clusterName) except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_02_MultipleClusterInitialization(self): @@ -40,13 +39,13 @@ class ClusterTests(TestBaseCluster): try: for i in range(0, 5): clusterName = "cluster-02.%d" % i - self.createCluster(clusterName, 5) + self.createCheckCluster(clusterName, 5) self.checkNumBrokers(25) self.killCluster("cluster-02.2") self.checkNumBrokers(20) self.stopCheckAll() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_03_AddRemoveNodes(self): @@ -66,7 +65,7 @@ class ClusterTests(TestBaseCluster): self.checkNumClusterBrokers(clusterName, 7) self.stopCheckAll() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_04_RemoveRestoreNodes(self): @@ -93,7 +92,7 @@ class ClusterTests(TestBaseCluster): self.checkNumClusterBrokers(clusterName, 6) self.stopCheckAll() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_05_KillAllNodesThenRecover(self): @@ -105,170 +104,126 @@ class ClusterTests(TestBaseCluster): self.createCheckCluster(clusterName, 6) self.stopCheckAll() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_06_PublishConsume(self): """Publish then consume 100 messages from a single cluster""" try: - clusterName = "cluster-06" - self.createCheckCluster(clusterName, 3) - self.sendReceiveMsgs(0, clusterName, "test-exchange-06", "test-queue-06", 100) - self.stopCheckAll() + dh = TestBaseCluster.DirectExchangeTestHelper(self, "cluster-06", 3, "test-exchange-06", ["test-queue-06"]) + dh.sendMsgs(100) + dh.finalizeTest() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_07_MultiplePublishConsume(self): """Staggered publish and consume on a single cluster""" try: - clusterName = "cluster-07" - exchangeName = "test-exchange-07" - queueName = "test-queue-07" - self.createCheckCluster(clusterName, 3) - self.createBindDirectExchangeQueue(0, clusterName, exchangeName, queueName) - txMsgs = self.sendMsgs(0, clusterName, exchangeName, queueName, 20) # 20, 0 - rxMsgs = self.receiveMsgs(1, clusterName, queueName, 10) # 10, 10 - txMsgs += self.sendMsgs(2, clusterName, exchangeName, queueName, 20) # 30, 10 - rxMsgs += self.receiveMsgs(0, clusterName, queueName, 20) # 10, 30 - txMsgs += self.sendMsgs(1, clusterName, exchangeName, queueName, 20) # 30, 30 - rxMsgs += self.receiveMsgs(2, clusterName, queueName, 20) # 10, 50 - txMsgs += self.sendMsgs(0, clusterName, exchangeName, queueName, 20) # 30, 50 - rxMsgs += self.receiveMsgs(1, clusterName, queueName, 30) # 0, 80 - self.stopCheckAll() - if txMsgs != rxMsgs: - print "txMsgs=%s" % txMsgs - print "rxMsgs=%s" % rxMsgs - self.fail("Send - receive message mismatch") + dh = TestBaseCluster.DirectExchangeTestHelper(self, "cluster-07", 3, "test-exchange-07", ["test-queue-07"]) + # tx rx nodes + # 0 0 0 1 2 + dh.sendMsgs(20) # 20 0 * + dh.receiveMsgs(10, 1) # 20 10 * + dh.sendMsgs(20, 2) # 40 10 * + dh.receiveMsgs(20, 0) # 40 30 * + dh.sendMsgs(20, 1) # 60 30 * + dh.receiveMsgs(20, 2) # 60 50 * + dh.sendMsgs(20, 0) # 80 50 * + dh.finalizeTest() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_08_MsgPublishConsumeAddRemoveNodes(self): """Staggered publish and consume interleaved with adding and removing nodes on a single cluster""" try: - clusterName = "cluster-08" - exchangeName = "test-exchange-08" - queueName = "test-queue-08" - self.createCheckCluster(clusterName, 3) - self.createBindDirectExchangeQueue(0, clusterName, exchangeName, queueName) - txMsgs = self.sendMsgs(0, clusterName, exchangeName, queueName, 20) # 20, 0 - for i in range(3,6): - self.createClusterNode(i, clusterName) - self.checkNumClusterBrokers(clusterName, 6) - txMsgs += self.sendMsgs(1, clusterName, exchangeName, queueName, 20) # 40, 0 - self.killNode(0, clusterName) - self.checkNumClusterBrokers(clusterName, 5) - rxMsgs = self.receiveMsgs(2, clusterName, queueName, 10) # 30, 10 - self.killNode(2, clusterName) - self.checkNumClusterBrokers(clusterName, 4) - rxMsgs += self.receiveMsgs(3, clusterName, queueName, 20) # 10, 30 - self.createClusterNode(6, clusterName) - self.checkNumClusterBrokers(clusterName, 5) - txMsgs += self.sendMsgs(4, clusterName, exchangeName, queueName, 20) # 30, 30 - rxMsgs += self.receiveMsgs(5, clusterName, queueName, 20) # 10, 50 - self.createClusterNode(7, clusterName) - self.checkNumClusterBrokers(clusterName, 6) - txMsgs += self.sendMsgs(6, clusterName, exchangeName, queueName, 20) # 30, 50 - rxMsgs += self.receiveMsgs(1, clusterName, queueName, 30) # 0, 80 - self.stopCheckAll() - if txMsgs != rxMsgs: - print "txMsgs=%s" % txMsgs - print "rxMsgs=%s" % rxMsgs - self.fail("Send - receive message mismatch") + dh = TestBaseCluster.DirectExchangeTestHelper(self, "cluster-08", 3, "test-exchange-08", ["test-queue-08"]) + # tx rx nodes + # 0 0 0 1 2 + dh.sendMsgs(20) # 20 0 * + dh.addNodes(2) # 0 1 2 3 4 + dh.sendMsgs(20, 1) # 40 0 * + dh.killNode(0) # . 1 2 3 4 + dh.receiveMsgs(10, 2) # 40 10 * + dh.killNode(2) # . 1 . 3 4 + dh.receiveMsgs(20, 3) # 40 30 * + dh.addNodes() # . 1 . 3 4 5 + dh.sendMsgs(20, 4) # 60 30 * + dh.receiveMsgs(20, 5) # 60 50 * + dh.addNodes() # . 1 . 3 4 5 6 + dh.sendMsgs(20, 6) # 80 50 * + dh.killNode(5) # . 1 . 3 4 . 6 + dh.finalizeTest() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_09_MsgPublishConsumeRemoveRestoreNodes(self): """Publish and consume messages interleaved with adding and restoring previous nodes on a single cluster""" try: - clusterName = "cluster-09" - exchangeName = "test-exchange-09" - queueName = "test-queue-09" - self.createCheckCluster(clusterName, 6) - self.createBindDirectExchangeQueue(0, clusterName, exchangeName, queueName) - txMsgs = self.sendMsgs(0, clusterName, exchangeName, queueName, 20) # 20, 0 - self.killNode(2, clusterName) - self.checkNumClusterBrokers(clusterName, 5) - txMsgs += self.sendMsgs(1, clusterName, exchangeName, queueName, 20) # 40, 0 - self.killNode(0, clusterName) - self.checkNumClusterBrokers(clusterName, 4) - rxMsgs = self.receiveMsgs(3, clusterName, queueName, 10) # 30, 10 - self.killNode(4, clusterName) - self.checkNumClusterBrokers(clusterName, 3) - rxMsgs += self.receiveMsgs(5, clusterName, queueName, 20) # 10, 30 - self.createClusterNode(2, clusterName) - self.checkNumClusterBrokers(clusterName, 4) - txMsgs += self.sendMsgs(1, clusterName, exchangeName, queueName, 20) # 30, 30 - self.createClusterNode(0, clusterName) - self.checkNumClusterBrokers(clusterName, 5) - rxMsgs += self.receiveMsgs(2, clusterName, queueName, 20) # 10, 50 - self.createClusterNode(4, clusterName) - self.checkNumClusterBrokers(clusterName, 6) - txMsgs += self.sendMsgs(0, clusterName, exchangeName, queueName, 20) # 30, 50 - rxMsgs += self.receiveMsgs(4, clusterName, queueName, 30) # 0, 80 - self.stopCheckAll() - if txMsgs != rxMsgs: - print "txMsgs=%s" % txMsgs - print "rxMsgs=%s" % rxMsgs - self.fail("Send - receive message mismatch") + dh = TestBaseCluster.DirectExchangeTestHelper(self, "cluster-09", 6, "test-exchange-09", ["test-queue-09"]) + # tx rx nodes + # 0 0 0 1 2 3 4 5 + dh.sendMsgs(20) # 20 0 * + dh.killNode(2) # 0 1 . 3 4 5 + dh.sendMsgs(20, 1) # 40 0 * + dh.killNode(0) # . 1 . 3 4 5 + dh.receiveMsgs(10, 3) # 40 10 * + dh.killNode(4) # . 1 . 3 . 5 + dh.receiveMsgs(20, 5) # 40 30 * + dh.restoreNode(2) # . 1 2 3 . 5 + dh.sendMsgs(20, 1) # 60 30 * + dh.restoreNode(0) # 0 1 2 3 . 5 + dh.receiveMsgs(20, 0) # 60 50 * + dh.killNode(2) # 0 1 . 3 . 5 + dh.restoreNode(2) # 0 1 2 3 . 5 + dh.sendMsgs(20, 2) # 80 50 * + dh.finalizeTest() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_10_LinearNodeKillCreateProgression(self): """Publish and consume messages while linearly killing all original nodes and replacing them with new ones""" try: - clusterName = "cluster-10" - exchangeName = "test-exchange-10" - queueName = "test-queue-10" - self.createCheckCluster(clusterName, 4) - self.createBindDirectExchangeQueue(2, clusterName, exchangeName, queueName) - txMsgs = self.sendMsgs(2, clusterName, exchangeName, queueName, 20) - rxMsgs = self.receiveMsgs(3, clusterName, queueName, 10) - for i in range(0, 16): - self.killNode(i, clusterName) - self.createClusterNode(i+4, clusterName) - self.checkNumClusterBrokers(clusterName, 4) - txMsgs += self.sendMsgs(i+1, clusterName, exchangeName, queueName, 20) - rxMsgs += self.receiveMsgs(i+2, clusterName, queueName, 20) - rxMsgs += self.receiveMsgs(16, clusterName, queueName, 10) - self.stopCheckAll() - if txMsgs != rxMsgs: - print "txMsgs=%s" % txMsgs - print "rxMsgs=%s" % rxMsgs - self.fail("Send - receive message mismatch") + dh = TestBaseCluster.DirectExchangeTestHelper(self, "cluster-10", 4, "test-exchange-10", ["test-queue-10"]) + # tx rx nodes + # 0 0 0 1 2 3 + dh.sendMsgs(20) # 20 0 * + dh.receiveMsgs(10, 1) # 20 10 * + for i in range(0, 16): # First loop: + dh.killNode(i) # . 1 2 3 + dh.addNodes() # . 1 2 3 4 + dh.sendMsgs(20, i+1) # 40 10 * + dh.receiveMsgs(20, i+2) # 40 30 * + # After loop: + # 340 330 . . . . . . . . . . . . . . . . 16 17 18 19 + dh.finalizeTest() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_11_CircularNodeKillRestoreProgression(self): """Publish and consume messages while circularly killing all original nodes and restoring them again""" try: - clusterName = "cluster-11" - exchangeName = "test-exchange-11" - queueName = "test-queue-11" - self.createCheckCluster(clusterName, 4) - self.createBindDirectExchangeQueue(2, clusterName, exchangeName, queueName) - txMsgs = self.sendMsgs(3, clusterName, exchangeName, queueName, 20) - rxMsgs = self.receiveMsgs(0, clusterName, queueName, 10) - self.killNode(0, clusterName) - self.killNode(1, clusterName) - for i in range(0, 16): - self.killNode((i + 2) % 4, clusterName) - self.createClusterNode(i % 4, clusterName) - self.checkNumClusterBrokers(clusterName, 2) - txMsgs += self.sendMsgs((i + 3) % 4, clusterName, exchangeName, queueName, 20) - rxMsgs += self.receiveMsgs((i + 4) % 4, clusterName, queueName, 20) - rxMsgs += self.receiveMsgs(3, clusterName, queueName, 10) - self.stopCheckAll() - if txMsgs != rxMsgs: - print "txMsgs=%s" % txMsgs - print "rxMsgs=%s" % rxMsgs - self.fail("Send - receive message mismatch") + dh = TestBaseCluster.DirectExchangeTestHelper(self, "cluster-11", 4, "test-exchange-11", ["test-queue-11"]) + # tx rx nodes + # 0 0 0 1 2 3 + dh.sendMsgs(20, 3) # 20 0 * + dh.receiveMsgs(10) # 20 10 * + dh.killNode(0) # . 1 2 3 + dh.killNode(1) # . . 2 3 + for i in range(0, 16): # First loop: + dh.killNode((i + 2) % 4) # . . . 3 + dh.restoreNode(i % 4) # 0 . . 3 + dh.sendMsgs(20, (i + 3) % 4) # 40 10 * + dh.receiveMsgs(20, (i + 4) % 4) # 40 30 * + # After loop: + # 340 330 . . 2 3 + dh.finalizeTest() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_12_KillAllNodesRecoverMessages(self): @@ -277,180 +232,94 @@ class ClusterTests(TestBaseCluster): print " No store loaded, skipped" return try: - clusterName = "cluster-12" - exchangeName = "test-exchange-12" - queueName = "test-queue-12" - self.createCheckCluster(clusterName, 4) - self.createBindDirectExchangeQueue(2, clusterName, exchangeName, queueName) - txMsgs = self.sendMsgs(0, clusterName, exchangeName, queueName, 20) - rxMsgs = self.receiveMsgs(1, clusterName, queueName, 10) - txMsgs += self.sendMsgs(2, clusterName, exchangeName, queueName, 20) - rxMsgs += self.receiveMsgs(3, clusterName, queueName, 20) - self.killNode(0, clusterName) - self.createClusterNode(4, clusterName) - self.checkNumClusterBrokers(clusterName, 4) - txMsgs += self.sendMsgs(4, clusterName, exchangeName, queueName, 20) - rxMsgs += self.receiveMsgs(1, clusterName, queueName, 20) - self.killNode(2, clusterName) - self.createClusterNode(0, clusterName) - self.createClusterNode(5, clusterName) - self.checkNumClusterBrokers(clusterName, 5) - txMsgs += self.sendMsgs(0, clusterName, exchangeName, queueName, 20) - rxMsgs += self.receiveMsgs(3, clusterName, queueName, 20) - self.killAllClusters() - self.checkNumClusterBrokers(clusterName, 0) - self.createCluster(clusterName) - self.createClusterNode(3, clusterName) # last node to be used - self.createClusterNode(0, clusterName) - self.createClusterNode(1, clusterName) - self.createClusterNode(2, clusterName) - self.createClusterNode(4, clusterName) - self.createClusterNode(5, clusterName) - rxMsgs += self.receiveMsgs(0, clusterName, queueName, 10) - if txMsgs != rxMsgs: - print "txMsgs=%s" % txMsgs - print "rxMsgs=%s" % rxMsgs - self.fail("Send - receive message mismatch") + dh = TestBaseCluster.DirectExchangeTestHelper(self, "cluster-12", 4, "test-exchange-12", ["test-queue-12"]) + # tx rx nodes + # 0 0 0 1 2 3 + dh.sendMsgs(20, 2) # 20 0 * + dh.receiveMsgs(10, 1) # 20 10 * + dh.killNode(1) # 0 . 2 3 + dh.sendMsgs(20, 0) # 40 10 * + dh.receiveMsgs(20, 3) # 40 30 * + dh.killNode(2) # 0 . . 3 + dh.addNodes(2) # 0 . . 3 4 5 + dh.sendMsgs(20, 5) # 60 30 * + dh.receiveMsgs(20, 4) # 60 50 * + dh.killCluster() # cluster does not exist + dh.restoreCluster() # 60 50 . . . . . . + dh.restoreNodes() # 0 1 2 3 4 5 + dh.finalizeTest() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_13_TopicExchange(self): - """Create topic exchange in a cluster and make sure it replicates correctly""" + """Create topic exchange in a cluster and make sure it behaves correctly""" try: - clusterName = "cluster-13" - self.createCheckCluster(clusterName, 4) - topicExchangeName = "test-exchange-13" - topicQueueNameKeyList = {"test-queue-13-A" : "#.A", "test-queue-13-B" : "#.B", "test-queue-13-C" : "C.#", "test-queue-13-D" : "D.#"} - self.createBindTopicExchangeQueues(2, clusterName, topicExchangeName, topicQueueNameKeyList) - - # Place initial messages - txMsgsA = txMsgsC = self.sendMsgs(3, clusterName, topicExchangeName, "C.hello.A", 10) # (10, 0, 10, 0) - self.sendMsgs(2, clusterName, topicExchangeName, "hello", 10) # Should not go to any queue - txMsgsD = self.sendMsgs(1, clusterName, topicExchangeName, "D.hello.A", 10) # (20, 0, 10, 10) - txMsgsA += txMsgsD - txMsgsB = self.sendMsgs(0, clusterName, topicExchangeName, "hello.B", 20) # (20, 20, 10, 10) + topicQueueNameKeyList = {"test-queue-13-A" : "#.A", "test-queue-13-B" : "*.B", "test-queue-13-C" : "C.#", "test-queue-13-D" : "D.*"} + th = TestBaseCluster.TopicExchangeTestHelper(self, "cluster-13", 4, "test-exchange-13", topicQueueNameKeyList) + # Place initial messages + th.sendMsgs("C.hello.A", 10) + th.sendMsgs("hello.world", 10) # matches none of the queues + th.sendMsgs("D.hello.A", 10) + th.sendMsgs("hello.B", 20) + th.sendMsgs("D.hello", 20) # Kill and add some nodes - self.killNode(0, clusterName) - self.killNode(2, clusterName) - self.createClusterNode(4, clusterName) - self.createClusterNode(5, clusterName) - self.checkNumClusterBrokers(clusterName, 4) + th.killNode(0) + th.killNode(2) + th.addNodes(2) # Pull 10 messages from each queue - rxMsgsA = self.receiveMsgs(1, clusterName, "test-queue-13-A", 10) # (10, 20, 10, 10) - rxMsgsB = self.receiveMsgs(3, clusterName, "test-queue-13-B", 10) # (10, 10, 10, 10) - rxMsgsC = self.receiveMsgs(4, clusterName, "test-queue-13-C", 10) # (10, 10, 0, 10) - rxMsgsD = self.receiveMsgs(5, clusterName, "test-queue-13-D", 10) # (10, 10, 0, 0) + th.receiveMsgs(10) # Kill and add another node - self.killNode(4, clusterName) - self.createClusterNode(6, clusterName) - self.checkNumClusterBrokers(clusterName, 4) + th.killNode(4) + th.addNodes() # Add two more queues - self.createBindTopicExchangeQueues(6, clusterName, topicExchangeName, {"test-queue-13-E" : "#.bye.A", "test-queue-13-F" : "#.bye.B"}) + th.addQueues({"test-queue-13-E" : "#.bye.A", "test-queue-13-F" : "#.bye.B"}) # Place more messages - txMsgs = self.sendMsgs(3, clusterName, topicExchangeName, "C.bye.A", 10) # (20, 10, 10, 0, 10, 0) - txMsgsA += txMsgs - txMsgsC += txMsgs - txMsgsE = txMsgs - self.sendMsgs(1, clusterName, topicExchangeName, "bye", 20) # Should not go to any queue - txMsgs = self.sendMsgs(5, clusterName, topicExchangeName, "D.bye.B", 20) # (20, 30, 10, 20, 10, 20) - txMsgsB += txMsgs - txMsgsD += txMsgs - txMsgsF = txMsgs + th.sendMsgs("C.bye.A", 10) + th.sendMsgs("hello.bye", 20) # matches none of the queues + th.sendMsgs("hello.bye.B", 20) + th.sendMsgs("D.bye", 20) # Kill all nodes but one - self.killNode(1, clusterName) - self.killNode(3, clusterName) - self.killNode(6, clusterName) - self.checkNumClusterBrokers(clusterName, 1) - # Pull all remaining messages from each queue - rxMsgsA += self.receiveMsgs(5, clusterName, "test-queue-13-A", 20) - rxMsgsB += self.receiveMsgs(5, clusterName, "test-queue-13-B", 30) - rxMsgsC += self.receiveMsgs(5, clusterName, "test-queue-13-C", 10) - rxMsgsD += self.receiveMsgs(5, clusterName, "test-queue-13-D", 20) - rxMsgsE = self.receiveMsgs(5, clusterName, "test-queue-13-E", 10) - rxMsgsF = self.receiveMsgs(5, clusterName, "test-queue-13-F", 20) - # Check messages - self.stopCheckAll() - if txMsgsA != rxMsgsA: - self.fail("Send - receive message mismatch for queue A") - if txMsgsB != rxMsgsB: - self.fail("Send - receive message mismatch for queue B") - if txMsgsC != rxMsgsC: - self.fail("Send - receive message mismatch for queue C") - if txMsgsD != rxMsgsD: - self.fail("Send - receive message mismatch for queue D") - if txMsgsE != rxMsgsE: - self.fail("Send - receive message mismatch for queue E") - if txMsgsF != rxMsgsF: - self.fail("Send - receive message mismatch for queue F") + th.killNode(1) + th.killNode(3) + th.killNode(6) + # Pull all remaining messages from each queue and check messages + th.finalizeTest() except: - self.killAllClusters() + self.killAllClusters(True) raise def test_Cluster_14_FanoutExchange(self): - """Create fanout exchange in a cluster and make sure it replicates correctly""" + """Create fanout exchange in a cluster and make sure it behaves correctly""" try: - clusterName = "cluster-14" - self.createCheckCluster(clusterName, 4) - fanoutExchangeName = "test-exchange-14" fanoutQueueNameList = ["test-queue-14-A", "test-queue-14-B", "test-queue-14-C"] - self.createBindFanoutExchangeQueues(2, clusterName, fanoutExchangeName, fanoutQueueNameList) - + fh = TestBaseCluster.FanoutExchangeTestHelper(self, "cluster-14", 4, "test-exchange-14", fanoutQueueNameList) # Place initial 20 messages, retrieve 10 - txMsg = self.sendMsgs(3, clusterName, fanoutExchangeName, None, 20) - rxMsgA = self.receiveMsgs(1, clusterName, "test-queue-14-A", 10) - rxMsgB = self.receiveMsgs(3, clusterName, "test-queue-14-B", 10) - rxMsgC = self.receiveMsgs(0, clusterName, "test-queue-14-C", 10) + fh.sendMsgs(20) + fh.receiveMsgs(10) # Kill and add some nodes - self.killNode(0, clusterName) - self.killNode(2, clusterName) - self.createClusterNode(4, clusterName) - self.createClusterNode(5, clusterName) - self.checkNumClusterBrokers(clusterName, 4) + fh.killNode(0) + fh.killNode(2) + fh.addNodes(2) # Place another 20 messages, retrieve 20 - txMsg += self.sendMsgs(3, clusterName, fanoutExchangeName, None, 20) - rxMsgA += self.receiveMsgs(1, clusterName, "test-queue-14-A", 20) - rxMsgB += self.receiveMsgs(3, clusterName, "test-queue-14-B", 20) - rxMsgC += self.receiveMsgs(4, clusterName, "test-queue-14-C", 20) + fh.sendMsgs(20) + fh.receiveMsgs(20) # Kill and add another node - self.killNode(4, clusterName) - self.createClusterNode(6, clusterName) - self.checkNumClusterBrokers(clusterName, 4) + fh.killNode(4) + fh.addNodes() # Add another 2 queues - self.createBindFanoutExchangeQueues(6, clusterName, fanoutExchangeName, ["test-queue-14-D", "test-queue-14-E"]) + fh.addQueues(["test-queue-14-D", "test-queue-14-E"]) # Place another 20 messages, retrieve 20 - tmp = self.sendMsgs(3, clusterName, fanoutExchangeName, None, 20) - txMsg += tmp - rxMsgA += self.receiveMsgs(1, clusterName, "test-queue-14-A", 20) - rxMsgB += self.receiveMsgs(3, clusterName, "test-queue-14-B", 20) - rxMsgC += self.receiveMsgs(6, clusterName, "test-queue-14-C", 20) - rxMsgD = self.receiveMsgs(6, clusterName, "test-queue-14-D", 10) - rxMsgE = self.receiveMsgs(6, clusterName, "test-queue-14-E", 10) + fh.sendMsgs(20) + fh.receiveMsgs(20) # Kill all nodes but one - self.killNode(1, clusterName) - self.killNode(3, clusterName) - self.killNode(6, clusterName) - self.checkNumClusterBrokers(clusterName, 1) - # Pull all remaining messages from each queue - rxMsgA += self.receiveMsgs(5, clusterName, "test-queue-14-A", 10) - rxMsgB += self.receiveMsgs(5, clusterName, "test-queue-14-B", 10) - rxMsgC += self.receiveMsgs(5, clusterName, "test-queue-14-C", 10) - rxMsgD += self.receiveMsgs(5, clusterName, "test-queue-14-D", 10) - rxMsgE += self.receiveMsgs(5, clusterName, "test-queue-14-E", 10) + fh.killNode(1) + fh.killNode(3) + fh.killNode(6) # Check messages - self.stopCheckAll() - if txMsg != rxMsgA: - self.fail("Send - receive message mismatch for queue A") - if txMsg != rxMsgB: - self.fail("Send - receive message mismatch for queue B") - if txMsg != rxMsgC: - self.fail("Send - receive message mismatch for queue C") - if tmp != rxMsgD: - self.fail("Send - receive message mismatch for queue D") - if tmp != rxMsgE: - self.fail("Send - receive message mismatch for queue E") + fh.finalizeTest() except: - self.killAllClusters() + self.killAllClusters(True) raise # Start the test here diff --git a/cpp/src/tests/run_cluster_tests b/cpp/src/tests/run_cluster_tests index d2f7a77865..103896cd3d 100755 --- a/cpp/src/tests/run_cluster_tests +++ b/cpp/src/tests/run_cluster_tests @@ -1,4 +1,4 @@ -#!/bin/sh +#!/bin/bash # # Licensed to the Apache Software Foundation (ASF) under one @@ -23,13 +23,13 @@ TEST_DIR=${top_builddir}/src/tests # Check AIS requirements -id -nG | grep '\' >/dev/null || NOGROUP="You are not a member of the ais group." -ps -u root | grep 'aisexec\|corosync' >/dev/null || NOAISEXEC="The aisexec or corosync daemon is not running as root" +id -nG | grep '\' > /dev/null || NOGROUP="You are not a member of the ais group." +ps -u root | grep 'aisexec\|corosync' > /dev/null || NOAISEXEC="The aisexec or corosync daemon is not running as root" if test -n "${NOGROUP}" -o -n "${NOAISEXEC}"; then cat < highestDeadNode: + return highestNode + return highestDeadNode + + def killCluster(self): + """Kill all nodes in the cluster""" + self._testBaseCluster.killCluster(self._clusterName) + self._testBaseCluster.checkNumClusterBrokers(self._clusterName, 0) + self._deadNodes.extend(self._nodes) + self._deadNodes.sort() + del self._nodes[:] + + def restoreCluster(self, lastNode = None, restoreNodes = True): + """Restore a previously killed cluster""" + self._testBaseCluster.createCluster(self._clusterName) + if restoreNodes: + numNodes = len(self._deadNodes) + self.restoreNodes(lastNode) + self._testBaseCluster.checkNumClusterBrokers(self._clusterName, numNodes) + + def addNodes(self, numberOfNodes = 1): + """Add a fixed number of nodes to the cluster.""" + nodeStart = self._highestNodeNumber() + 1 + for i in range(0, numberOfNodes): + nodeNumber = nodeStart + i + self._testBaseCluster.createClusterNode(nodeNumber, self._clusterName) + self._nodes.append(nodeNumber) + self._testBaseCluster.checkNumClusterBrokers(self._clusterName, len(self._nodes)) + + def restoreNode(self, nodeNumber): + """Restore a cluster node that has been previously killed""" + if nodeNumber not in self._deadNodes: + raise Exception("restoreNode(): Node number %d not in dead node list %s" % (nodeNumber, self._deadNodes)) + self._testBaseCluster.createClusterNode(nodeNumber, self._clusterName) + self._deadNodes.remove(nodeNumber) + self._nodes.append(nodeNumber) + self._nodes.sort() + + def restoreNodes(self, lastNode = None): + """Restore all known cluster nodes that have been previously killed starting with a known last-used node""" + if len(self._nodes) == 0: # restore last-used node first + if lastNode == None: + lastNode = self._lastNode + self.restoreNode(lastNode) + while len(self._deadNodes) > 0: + self.restoreNode(self._deadNodes[0]) + + def killNode(self, nodeNumber): + """Kill a cluster node (if it is in the _nodes list).""" + if nodeNumber not in self._nodes: + raise Exception("killNode(): Node number %d not in node list %s" % (nodeNumber, self._nodes)) + self._testBaseCluster.killNode(nodeNumber, self._clusterName) + self._nodes.remove(nodeNumber) + self._deadNodes.append(nodeNumber) + self._deadNodes.sort() + + def sendMsgs(self, routingKey, numMsgs, nodeNumber = None, msgSize = None, wait = True): + """Send a fixed number of messages using the given routing key.""" + if nodeNumber == None: + nodeNumber = self._nodes[0] # Use first available node + msgs = self._testBaseCluster._makeMessageList(numMsgs, msgSize) + sender = self._testBaseCluster.createSender(nodeNumber, self._clusterName, self._exchangeName, routingKey) + sender.stdin.write(msgs) + sender.stdin.close() + if wait: + sender.wait() + self._lastNode = nodeNumber + return msgs.split() + + # TODO - this i/f is messy: one mumMsgs can be given, but a list of queues + # so assuming numMsgs for each queue + # A mechanism is needed to specify a different numMsgs per queue + def receiveMsgs(self, numMsgs, nodeNumber = None, queueNameList = None, wait = True): + """Receive a fixed number of messages from a named queue. If numMsgs == None, get all remaining messages.""" + if nodeNumber == None: + nodeNumber = self._nodes[0] # Use first available node + if queueNameList == None: + queueNameList = self._txMsgs.iterkeys() + for qn in queueNameList: + nm = numMsgs + if nm == None: + nm = len(self._txMsgs[qn]) - len(self._rxMsgs[qn]) # get all remaining messages + if nm > 0: + receiver = self._testBaseCluster.createReciever(nodeNumber, self._clusterName, qn, nm) + cnt = 0 + while cnt < nm: + rx = receiver.stdout.readline().strip() + if rx == "" and receiver.poll() != None: break + self._rxMsgs[qn].append(rx) + cnt = cnt + 1 + if wait: + receiver.wait() + self._lastNode = nodeNumber + + def receiveRemainingMsgs(self, nodeNumber = None, queueNameList = None, wait = True): + """Receive all remaining messages on named queue.""" + self.receiveMsgs(None, nodeNumber, queueNameList, wait) + + def checkMsgs(self): + """Return True if all expected messages have been received (ie the transmit and receive list are identical).""" + txMsgTot = 0 + rxMsgTot = 0 + for qn, txMsgList in self._txMsgs.iteritems(): + rxMsgList = self._rxMsgs[qn] + txMsgTot = txMsgTot + len(txMsgList) + rxMsgTot = rxMsgTot + len(rxMsgList) + if len(txMsgList) != len(rxMsgList): + return False + for i, m in enumerate(txMsgList): + if m != rxMsgList[i]: + return False + if txMsgTot == 0 and rxMsgTot == 0: + print "WARNING: No messages were either sent or received" + return True + + def finalizeTest(self): + """Recover all the remaining messages on all queues, then check that all expected messages were received.""" + self.receiveRemainingMsgs() + self._testBaseCluster.stopCheckAll() + if not self.checkMsgs(): + self._testBaseCluster.fail("Send - receive message mismatch") + self.printMsgs() + + def printMsgs(self, txMsgs = True, rxMsgs = True): + """Print all messages transmitted and received.""" + for qn, txMsgList in self._txMsgs.iteritems(): + print "Queue: %s" % qn + if txMsgs: + print " txMsgList = %s" % txMsgList + if rxMsgs: + rxMsgList = self._rxMsgs[qn] + print " rxMsgList = %s" % rxMsgList + + + class DirectExchangeTestHelper(TestHelper): + + def __init__(self, testBaseCluster, clusterName, numNodes, exchangeName, queueNameList): + TestBaseCluster.TestHelper.__init__(self, testBaseCluster, clusterName, numNodes, exchangeName, queueNameList) + self._testBaseCluster.addExchange(0, clusterName, "direct", exchangeName) + for qn in queueNameList: + self._bindQueue(qn, qn) + + def addQueues(self, queueNameList): + self._addQueues(queueNameList) + for qn in queueNameList: + self._bindQueue(qn, qn) + + def sendMsgs(self, numMsgs, nodeNumber = None, queueNameList = None, msgSize = None, wait = True): + if queueNameList == None: + queueNameList = self._txMsgs.iterkeys() + for qn in queueNameList: + self._txMsgs[qn].extend(TestBaseCluster.TestHelper.sendMsgs(self, qn, numMsgs, nodeNumber, msgSize, wait)) + + + class TopicExchangeTestHelper(TestHelper): + + def __init__(self, testBaseCluster, clusterName, numNodes, exchangeName, queueNameKeyList): + self._queueNameKeyList = queueNameKeyList + TestBaseCluster.TestHelper.__init__(self, testBaseCluster, clusterName, numNodes, exchangeName, queueNameKeyList.iterkeys()) + self._testBaseCluster.addExchange(0, clusterName, "topic", exchangeName) + for qn, bk in queueNameKeyList.iteritems(): + self._bindQueue(qn, bk) + + def addQueues(self, queueNameKeyList): + self._addQueues(queueNameKeyList.iterkeys()) + for qn, bk in queueNameKeyList.iteritems(): + self._bindQueue(qn, bk) + + def _prepareRegex(self, bk): + # This regex conversion is not very complete - there are other chars that should be escaped too + return "^%s$" % bk.replace(".", r"\.").replace("*", r"[^.]*").replace("#", ".*") + + def sendMsgs(self, routingKey, numMsgs, nodeNumber = None, msgSize = None, wait = True): + msgList = TestBaseCluster.TestHelper.sendMsgs(self, routingKey, numMsgs, nodeNumber, msgSize, wait) + for qn, bk in self._queueNameKeyList.iteritems(): + if re.match(self._prepareRegex(bk), routingKey): + self._txMsgs[qn].extend(msgList) + + + class FanoutExchangeTestHelper(TestHelper): + + def __init__(self, testBaseCluster, clusterName, numNodes, exchangeName, queueNameList): + TestBaseCluster.TestHelper.__init__(self, testBaseCluster, clusterName, numNodes, exchangeName, queueNameList) + self._testBaseCluster.addExchange(0, clusterName, "fanout", exchangeName) + for qn in queueNameList: + self._bindQueue(qn, "") + + def addQueues(self, queueNameList): + self._addQueues(queueNameList) + for qn in queueNameList: + self._bindQueue(qn, "") + + def sendMsgs(self, numMsgs, nodeNumber = None, msgSize = None, wait = True): + msgList = TestBaseCluster.TestHelper.sendMsgs(self, "", numMsgs, nodeNumber, msgSize, wait) + for ml in self._txMsgs.itervalues(): + ml.extend(msgList) \ No newline at end of file -- cgit v1.2.1