# # 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. # # # Support library for qpid python tests. # import os, re, signal, subprocess, time, unittest class TestBase(unittest.TestCase): """ Base class for qpid tests. Provides broker start/stop/kill methods """ """ The following environment vars control if and how the test is run, and determine where many of the helper executables/libs are to be found. """ _storeLib = os.getenv("STORE_LIB") _storeEnable = _storeLib != None # Must be True for durability to be enabled during the test _qpiddExec = os.getenv("QPIDD_EXEC", "/usr/sbin/qpidd") _tempStoreDir = os.path.abspath(os.getenv("TMP_DATA_DIR", "/tmp/qpid")) """Global message counter ensures unique messages""" _msgCnt = 0 # --- Helper functions for parameter handling --- def _paramBool(self, key, val, keyOnly = False): if val == None: return "" if keyOnly: if val: return " --%s" % key else: return "" else: if val: return " --%s yes" % key else: return " --%s no" % key # --- Helper functions for message creation --- def _makeMessage(self, msgSize): msg = "Message-%04d" % self._msgCnt self._msgCnt = self._msgCnt + 1 msgLen = len(msg) if msgSize > msgLen: for i in range(msgLen, msgSize): if i == msgLen: msg += "-" else: msg += chr(ord('a') + (i % 26)) return msg def _makeMessageList(self, numMsgs, msgSize): if msgSize == None: msgSize = 12 msgs = "" for m in range(0, numMsgs): msgs += "%s\n" % self._makeMessage(msgSize) return msgs # --- Starting and stopping a broker --- def startBroker(self, qpiddArgs, logFile = None): """Start a single broker daemon, returns tuple (pid, port)""" if self._qpiddExec == None: raise Exception("Environment variable QPIDD is not set") cmd = "%s --daemon --port=0 %s" % (self._qpiddExec, qpiddArgs) portStr = os.popen(cmd).read() if len(portStr) == 0: err = "Broker daemon startup failed." if logFile != None: err += " See log file %s" % logFile raise Exception(err) port = int(portStr) pidStr = os.popen("%s -p %d -c" % (self._qpiddExec, port)).read() try: pid = int(pidStr) except: raise Exception("Unable to get pid: \"%s -p %d -c\" returned %s" % (self._qpiddExec, port, pidStr)) #print "started broker: pid=%d, port=%d args: %s" % (pid, port, qpiddArgs) return (pid, port) def killBroker(self, nodeTuple, ignoreFailures = False): """Kill a broker using kill -9""" try: os.kill(nodeTuple[self.PID], signal.SIGKILL) try: os.waitpid(nodeTuple[self.PID], 0) except: pass #print "killed broker: port=%d pid=%d" % (nodeTuple[self.PORT], nodeTuple[self.PID]) except: if ignoreFailures: print "WARNING: killBroker (port=%d pid=%d) failed - ignoring." % (nodeTuple[self.PORT], nodeTuple[self.PID]) else: raise def stopBroker(self, nodeTuple, ignoreFailures = False): """Stop a broker using qpidd -q""" try: ret = os.spawnl(os.P_WAIT, self._qpiddExec, self._qpiddExec, "--port=%d" % nodeTuple[self.PORT], "--quit", "--no-module-dir") if ret != 0: raise Exception("stopBroker(): port=%d: qpidd -q returned %d" % (nodeTuple[self.PORT], ret)) try: os.waitpid(nodeTuple[self.PID], 0) except: pass #print "stopped broker: port=%d pid=%d" % (nodeTuple[self.PORT], nodeTuple[self.PID]) except: if ignoreFailures: print "WARNING: stopBroker (port=%d pid=%d) failed - ignoring." % (nodeTuple[self.PORT], nodeTuple[self.PID]) else: raise class TestBaseCluster(TestBase): """ Base class for cluster tests. Provides methods for starting and stopping clusters and cluster nodes. """ """ The following environment vars control if and how the test is run, and determine where many of the helper executables/libs are to be found. """ _clusterLib = os.getenv("CLUSTER_LIB") _clusterTestEnable = _clusterLib != None # Must be True for these cluster tests to run _xmlLib = os.getenv("XML_LIB") _xmlEnable = _xmlLib != None _qpidConfigExec = os.getenv("QPID_CONFIG_EXEC", "/usr/bin/qpid-config") _qpidRouteExec = os.getenv("QPID_ROUTE_EXEC", "/usr/bin/qpid-route") _receiverExec = os.getenv("RECEIVER_EXEC", "/usr/libexec/qpid/test/receiver") _senderExec = os.getenv("SENDER_EXEC", "/usr/libexec/qpid/test/sender") """ _clusterDict is a dictionary of clusters: key = cluster name (string) val = dictionary of node numbers: key = integer node number val = tuple containing (pid, port) For example, two clusters "TestCluster0" and "TestCluster1" containing several nodes would look as follows: {"TestCluster0": {0: (pid0-0, port0-0), 1: (pid0-1, port0-1), ...}, "TestCluster1": {0: (pid1-0, port1-0), 1: (pid1-1, port1-1), ...}} where pidm-n and portm-n are the int pid and port for TestCluster m node n respectively. """ _clusterDict = {} """Index for (pid, port) tuple""" PID = 0 PORT = 1 def run(self, res): """ Skip cluster testing if env var RUN_CLUSTER_TESTS is not defined.""" if not self._clusterTestEnable: return unittest.TestCase.run(self, res) # --- Private helper / convenience functions --- def _checkPids(self, clusterName = None): for pid, port in self.getTupleList(): try: os.kill(pid, 0) except: raise Exception("_checkPids(): Broker with pid %d expected but does not exist! (crashed?)" % pid) # --- Starting cluster node(s) --- def createClusterNode(self, nodeNumber, clusterName): """Create a node and add it to the named cluster""" if self._tempStoreDir == None: raise Exception("Environment variable TMP_DATA_DIR is not set") if self._clusterLib == None: raise Exception("Environment variable LIBCLUSTER is not set") name = "%s-%d" % (clusterName, nodeNumber) dataDir = os.path.join(self._tempStoreDir, "cluster", name) logFile = "%s.log" % dataDir args = "--no-module-dir --load-module=%s --data-dir=%s --cluster-name=%s --auth=no --log-enable=notice+ --log-to-file=%s" % \ (self._clusterLib, dataDir, clusterName, logFile) if self._storeEnable: if self._storeLib == None: raise Exception("Environment variable LIBSTORE is not set") args += " --load-module %s" % self._storeLib self._clusterDict[clusterName][nodeNumber] = self.startBroker(args, logFile) def createCluster(self, clusterName, numberNodes = 0): """Create a cluster containing an initial number of nodes""" self._clusterDict[clusterName] = {} for n in range(0, numberNodes): self.createClusterNode(n, clusterName) def waitForNodes(self, clusterName): """Wait for all nodes to become active (ie finish cluster sync)""" # TODO - connect to each known node in cluster # Until this is done, wait a bit (hack) time.sleep(1) # --- Cluster and node status --- def getTupleList(self, clusterName = None): """Get list of (pid, port) tuples of all known cluster brokers""" tList = [] for c, l in self._clusterDict.iteritems(): if clusterName == None or c == clusterName: for t in l.itervalues(): tList.append(t) return tList def getNumBrokers(self): """Get total number of brokers in all known clusters""" return len(self.getTupleList()) def checkNumBrokers(self, expected = None, checkPids = True): """Check that the total number of brokers in all known clusters is the expected value""" if expected != None and self.getNumBrokers() != expected: raise Exception("Unexpected number of brokers: expected %d, found %d" % (expected, self.getNumBrokers())) if checkPids: self._checkPids() def getClusterTupleList(self, clusterName): """Get list of (pid, port) tuples of all nodes in named cluster""" if clusterName in self._clusterDict: return self._clusterDict[clusterName].values() return [] def getNumClusterBrokers(self, clusterName): """Get total number of brokers in named cluster""" return len(self.getClusterTupleList(clusterName)) def getNodeTuple(self, nodeNumber, clusterName): """Get the (pid, port) tuple for the given cluster node""" return self._clusterDict[clusterName][nodeNumber] def checkNumClusterBrokers(self, clusterName, expected = None, checkPids = True, waitForNodes = True): """Check that the total number of brokers in the named cluster is the expected value""" if expected != None and self.getNumClusterBrokers(clusterName) != expected: raise Exception("Unexpected number of brokers in cluster %s: expected %d, found %d" % \ (clusterName, expected, self.getNumClusterBrokers(clusterName))) if checkPids: self._checkPids(clusterName) if waitForNodes: self.waitForNodes(clusterName) def clusterExists(self, clusterName): """ Return True if clusterName exists, False otherwise""" return clusterName in self._clusterDict.keys() def clusterNodeExists(self, clusterName, nodeNumber): """ Return True if nodeNumber in clusterName exists, False otherwise""" if clusterName in self._clusterDict.keys(): return nodeNumber in self._clusterDict[nodeName] return False def createCheckCluster(self, clusterName, size): """Create a cluster using the given name and size, then check the number of brokers""" self.createCluster(clusterName, size) self.checkNumClusterBrokers(clusterName, size) # --- Kill cluster nodes using signal 9 --- def killNode(self, nodeNumber, clusterName, updateDict = True, ignoreFailures = False): """Kill the given node in the named cluster using kill -9""" self.killBroker(self.getNodeTuple(nodeNumber, clusterName), ignoreFailures) if updateDict: del(self._clusterDict[clusterName][nodeNumber]) def killCluster(self, clusterName, updateDict = True, ignoreFailures = False): """Kill all nodes in the named cluster""" for n in self._clusterDict[clusterName].iterkeys(): self.killNode(n, clusterName, False, ignoreFailures) if updateDict: del(self._clusterDict[clusterName]) def killClusterCheck(self, clusterName): """Kill the named cluster and check that the name is removed from the cluster dictionary""" self.killCluster(clusterName) if self.clusterExists(clusterName): raise Exception("Unable to kill cluster %s; %d nodes still exist" % \ (clusterName, self.getNumClusterBrokers(clusterName))) def killAllClusters(self, ignoreFailures = False): """Kill all known clusters""" for n in self._clusterDict.iterkeys(): self.killCluster(n, False, ignoreFailures) self._clusterDict.clear() def killAllClustersCheck(self, ignoreFailures = False): """Kill all known clusters and check that the cluster dictionary is empty""" self.killAllClusters(ignoreFailures) self.checkNumBrokers(0) # --- Stop cluster nodes using qpidd -q --- def stopNode(self, nodeNumber, clusterName, updateDict = True, ignoreFailures = False): """Stop the given node in the named cluster using qpidd -q""" self.stopBroker(self.getNodeTuple(nodeNumber, clusterName), ignoreFailures) if updateDict: del(self._clusterDict[clusterName][nodeNumber]) def stopAllClusters(self, ignoreFailures = False): """Stop all known clusters""" for n in self._clusterDict.iterkeys(): self.stopCluster(n, False, ignoreFailures) self._clusterDict.clear() def stopCluster(self, clusterName, updateDict = True, ignoreFailures = False): """Stop all nodes in the named cluster""" for n in self._clusterDict[clusterName].iterkeys(): self.stopNode(n, clusterName, False, ignoreFailures) if updateDict: del(self._clusterDict[clusterName]) def stopCheckCluster(self, clusterName, ignoreFailures = False): """Stop the named cluster and check that the name is removed from the cluster dictionary""" self.stopCluster(clusterName, True, ignoreFailures) if self.clusterExists(clusterName): raise Exception("Unable to kill cluster %s; %d nodes still exist" % (clusterName, self.getNumClusterBrokers(clusterName))) def stopAllCheck(self, ignoreFailures = False): """Kill all known clusters and check that the cluster dictionary is empty""" self.stopAllClusters() self.checkNumBrokers(0) # --- qpid-config functions --- def _qpidConfig(self, nodeNumber, clusterName, action): """Configure some aspect of a qpid broker using the qpid_config executable""" port = self.getNodeTuple(nodeNumber, clusterName)[self.PORT] #print "%s -a localhost:%d %s" % (self._qpidConfigExec, port, action) ret = os.spawnl(os.P_WAIT, self._qpidConfigExec, self._qpidConfigExec, "-a", "localhost:%d" % port, *action.split()) if ret != 0: raise Exception("_qpidConfig(): cluster=\"%s\" nodeNumber=%d port=%d action=\"%s\" returned %d" % \ (clusterName, nodeNumber, port, action, ret)) def addExchange(self, nodeNumber, clusterName, exchangeType, exchangeName, durable = False, sequence = False, \ ive = False): """Add a named exchange.""" action = "add exchange %s %s" % (exchangeType, exchangeName) action += self._paramBool("durable", durable, True) action += self._paramBool("sequence", sequence, True) action += self._paramBool("ive", ive, True) self._qpidConfig(nodeNumber, clusterName, action) def deleteExchange(self, nodeNumber, clusterName, exchangeName): """Delete a named exchange""" self._qpidConfig(nodeNumber, clusterName, "del exchange %s" % exchangeName) def addQueue(self, nodeNumber, clusterName, queueName, configArgs = None): """Add a queue using qpid-config.""" action = "add queue %s" % queueName if self._storeEnable: action += " --durable" if configArgs != None: action += " %s" % configArgs self._qpidConfig(nodeNumber, clusterName, action) def delQueue(self, nodeNumber, clusterName, queueName): """Delete a named queue using qpid-config.""" self._qpidConfig(nodeNumber, clusterName, "del queue %s" % queueName) def bind(self, nodeNumber, clusterName, exchangeName, queueName, key): """Create an exchange-queue binding using qpid-config.""" self._qpidConfig(nodeNumber, clusterName, "bind %s %s %s" % (exchangeName, queueName, key)) def unbind(self, nodeNumber, clusterName, exchangeName, queueName, key): """Remove an exchange-queue binding using qpid-config.""" self._qpidConfig(nodeNumber, clusterName, "unbind %s %s %s" % (exchangeName, queueName, key)) # --- qpid-route functions (federation) --- def brokerDict(self, nodeNumber, clusterName, host = "localhost", user = None, password = None): """Returns a dictionary containing the broker info to be passed to route functions""" port = self.getNodeTuple(nodeNumber, clusterName)[self.PORT] return {"cluster": clusterName, "node":nodeNumber, "port":port, "host":host, "user":user, "password":password} def _brokerStr(self, brokerDict): """Set up a broker string in the format [user/password@]host:port""" str = "" if brokerDict["user"] !=None and brokerDict["password"] != None: str = "%s@%s" % (brokerDict["user"], brokerDict["password"]) str += "%s:%d" % (brokerDict["host"], brokerDict["port"]) return str def _qpidRoute(self, action): """Set up a route using qpid-route""" #print "%s %s" % (self._qpidRouteExec, action) ret = os.spawnl(os.P_WAIT, self._qpidRouteExec, self._qpidRouteExec, *action.split()) if ret != 0: raise Exception("_qpidRoute(): action=\"%s\" returned %d" % (action, ret)) def routeDynamicAdd(self, destBrokerDict, srcBrokerDict, exchangeName): self._qpidRoute("dynamic add %s %s %s" % (self._brokerStr(destBrokerDict), self._brokerStr(srcBrokerDict), exchangeName)) def routeDynamicDelete(self, destBrokerDict, srcBrokerDict, exchangeName): self._qpidRoute("dynamic del %s %s %s" % (self._brokerStr(destBrokerDict), self._brokerStr(srcBrokerDict), exchangeName)) def routeAdd(self, destBrokerDict, srcBrokerDict, exchangeName, routingKey): self._qpidRoute("route add %s %s %s %s" % (self._brokerStr(destBrokerDict), self._brokerStr(srcBrokerDict), exchangeName, routingKey)) def routeDelete(self, destBrokerDict, srcBrokerDict, exchangeName, routingKey): self._qpidRoute("route del %s %s %s %s" % (self._brokerStr(destBrokerDict), self._brokerStr(srcBrokerDict), exchangeName, routingKey)) def routeQueueAdd(self, destBrokerDict, srcBrokerDict, exchangeName, queueName): self._qpidRoute("queue add %s %s %s %s" % (self._brokerStr(destBrokerDict), self._brokerStr(srcBrokerDict), exchangeName, queueName)) def routeQueueDelete(self, destBrokerDict, srcBrokerDict, exchangeName, queueName): self._qpidRoute("queue del %s %s %s %s" % (self._brokerStr(destBrokerDict), self._brokerStr(srcBrokerDict), exchangeName, queueName)) def routeLinkAdd(self, destBrokerDict, srcBrokerDict): self._qpidRoute("link add %s %s" % (self._brokerStr(destBrokerDict), self._brokerStr(srcBrokerDict))) def routeLinkDelete(self, destBrokerDict, srcBrokerDict): self._qpidRoute("link del %s %s" % (self._brokerStr(destBrokerDict), self._brokerStr(srcBrokerDict))) # --- Message send and receive functions --- def _receiver(self, action): if self._receiverExec == None: raise Exception("Environment variable RECEIVER is not set") cmd = "%s %s" % (self._receiverExec, action) #print cmd return subprocess.Popen(cmd.split(), stdout = subprocess.PIPE) def _sender(self, action): if self._senderExec == None: raise Exception("Environment variable SENDER is not set") cmd = "%s %s" % (self._senderExec, action) #print cmd return subprocess.Popen(cmd.split(), stdin = subprocess.PIPE) def createReciever(self, nodeNumber, clusterName, queueName, numMsgs = None, receiverArgs = None): port = self.getNodeTuple(nodeNumber, clusterName)[self.PORT] action = "--port %d --queue %s" % (port, queueName) if numMsgs != None: action += " --messages %d" % numMsgs if receiverArgs != None: action += " %s" % receiverArgs return self._receiver(action) def createSender(self, nodeNumber, clusterName, exchangeName, routingKey, senderArgs = None): port = self.getNodeTuple(nodeNumber, clusterName)[self.PORT] action = "--port %d --exchange %s" % (port, exchangeName) if routingKey != None and len(routingKey) > 0: action += " --routing-key %s" % routingKey if self._storeEnable: action += " --durable yes" if senderArgs != None: action += " %s" % senderArgs return self._sender(action) def createBindDirectExchangeQueue(self, nodeNumber, clusterName, exchangeName, queueName): self.addExchange(nodeNumber, clusterName, "direct", exchangeName) self.addQueue(nodeNumber, clusterName, queueName) self.bind(nodeNumber, clusterName, exchangeName, queueName, queueName) def createBindTopicExchangeQueues(self, nodeNumber, clusterName, exchangeName, queueNameKeyList): self.addExchange(nodeNumber, clusterName, "topic", exchangeName) for queueName, key in queueNameKeyList.iteritems(): self.addQueue(nodeNumber, clusterName, queueName) self.bind(nodeNumber, clusterName, exchangeName, queueName, key) def createBindFanoutExchangeQueues(self, nodeNumber, clusterName, exchangeName, queueNameList): self.addExchange(nodeNumber, clusterName, "fanout", exchangeName) for queueName in queueNameList: self.addQueue(nodeNumber, clusterName, queueName) self.bind(nodeNumber, clusterName, exchangeName, queueName, "") def sendMsgs(self, nodeNumber, clusterName, exchangeName, routingKey, numMsgs, msgSize = None, wait = True): msgs = self._makeMessageList(numMsgs, msgSize) sender = self.createSender(nodeNumber, clusterName, exchangeName, routingKey) sender.stdin.write(msgs) sender.stdin.close() if wait: sender.wait() return msgs def receiveMsgs(self, nodeNumber, clusterName, queueName, numMsgs, wait = True): receiver = self.createReciever(nodeNumber, clusterName, queueName, numMsgs) cnt = 0 msgs = "" while cnt < numMsgs: rx = receiver.stdout.readline() if rx == "" and receiver.poll() != None: break msgs += rx cnt = cnt + 1 if wait: receiver.wait() return msgs # --- Exchange-specific helper inner classes --- class TestHelper: """ This is a "virtual" superclass for test helpers, and is not useful on its own, but the per-exchange subclasses are designed to keep track of the messages sent to and received from queues which have bindings to that exchange type. """ def __init__(self, testBaseCluster, clusterName, numNodes, exchangeName, queueNameList): """Dictionary of queues and lists of messages sent to them.""" self._txMsgs = {} """Dictionary of queues and lists of messages received from them.""" self._rxMsgs = {} """List of node numbers currently in the cluster""" self._nodes = [] """List of node numbers which have been killed and can therefore be recovered""" self._deadNodes = [] """Last node to be used""" self._lastNode = None self._testBaseCluster = testBaseCluster self._clusterName = clusterName self._exchangeName = exchangeName self._queueNameList = queueNameList self._addQueues(queueNameList) self._testBaseCluster.createCheckCluster(clusterName, numNodes) self._nodes.extend(range(0, numNodes)) def _addQueues(self, queueNameList): for qn in queueNameList: if not qn in self._txMsgs: self._txMsgs[qn] = [] if not qn in self._rxMsgs: self._rxMsgs[qn] = [] def _bindQueue(self, queueName, bindingKey, nodeNumber = None): """Bind a queue to an exchange using a binding key.""" if nodeNumber == None: nodeNumber = self._nodes[0] # first available node self._testBaseCluster.addQueue(nodeNumber, self._clusterName, queueName) self._testBaseCluster.bind(nodeNumber, self._clusterName, self._exchangeName, queueName, bindingKey) def _highestNodeNumber(self): """Find the highest node number used so far between the current nodes and those stopped/killed.""" highestNode = self._nodes[-1] if len(self._deadNodes) == 0: return highestNode highestDeadNode = self._deadNodes[-1] if highestNode > 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)) self._testBaseCluster.waitForNodes(self._clusterName) 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]) self._testBaseCluster.waitForNodes(self._clusterName) 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: while nm > 0: receiver = self._testBaseCluster.createReciever(nodeNumber, self._clusterName, qn, nm) cnt = 0 while cnt < nm: rx = receiver.stdout.readline().strip() if rx == "": if receiver.poll() != None: break elif rx not in self._rxMsgs[qn]: self._rxMsgs[qn].append(rx) cnt = cnt + 1 nm = nm - cnt if wait: receiver.wait() self._rxMsgs[qn].sort() 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.stopAllCheck() if not self.checkMsgs(): self.printMsgs() self._testBaseCluster.fail("Send - receive message mismatch") 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)