[rhmessaging-commits] rhmessaging commits: r3623 - in store/trunk/cpp: tests and 1 other directories.

Author: kpvdr Date: 2009-09-14 11:21:33 -0400 (Mon, 14 Sep 2009) New Revision: 3623 Modified: store/trunk/cpp/lib/JournalImpl.cpp store/trunk/cpp/lib/MessageStoreImpl.cpp store/trunk/cpp/tests/SimpleTest.cpp store/trunk/cpp/tests/python_tests/flow_to_disk.py store/trunk/cpp/tests/run_python_tests Log: Reversed checkin 3622/813825 until its problems can be resolved. Modified: store/trunk/cpp/lib/JournalImpl.cpp =================================================================== --- store/trunk/cpp/lib/JournalImpl.cpp 2009-09-11 13:35:04 UTC (rev 3622) +++ store/trunk/cpp/lib/JournalImpl.cpp 2009-09-14 15:21:33 UTC (rev 3623) @@ -252,8 +252,6 @@ #define MAX_AIO_SLEEPS 1000 // 10 sec #define AIO_SLEEP_TIME 10000 // 10 ms -// Return true if content is recovered from store; false if content is external and must be recovered from an external store. -// Throw exception for all errors. bool JournalImpl::loadMsgContent(u_int64_t rid, std::string& data, size_t length, size_t offset) { Modified: store/trunk/cpp/lib/MessageStoreImpl.cpp =================================================================== --- store/trunk/cpp/lib/MessageStoreImpl.cpp 2009-09-11 13:35:04 UTC (rev 3622) +++ store/trunk/cpp/lib/MessageStoreImpl.cpp 2009-09-14 15:21:33 UTC (rev 3623) @@ -1474,15 +1474,15 @@ JournalImpl* jc = static_cast<JournalImpl*>(queue->getExternalQueueStore()); if (txn->getXid().empty()) { if (message->isContentReleased()) { - jc->enqueue_extern_data_record(size, dtokp.get(), !message->isPersistent()); + jc->enqueue_extern_data_record(size, dtokp.get(), false); } else { - jc->enqueue_data_record(buff, size, size, dtokp.get(), !message->isPersistent()); + jc->enqueue_data_record(buff, size, size, dtokp.get(), false); } } else { if (message->isContentReleased()) { - jc->enqueue_extern_txn_data_record(size, dtokp.get(), txn->getXid(), !message->isPersistent()); + jc->enqueue_extern_txn_data_record(size, dtokp.get(), txn->getXid(), false); } else { - jc->enqueue_txn_data_record(buff, size, size, dtokp.get(), txn->getXid(), !message->isPersistent()); + jc->enqueue_txn_data_record(buff, size, size, dtokp.get(), txn->getXid(), false); } } } else { Modified: store/trunk/cpp/tests/SimpleTest.cpp =================================================================== --- store/trunk/cpp/tests/SimpleTest.cpp 2009-09-11 13:35:04 UTC (rev 3622) +++ store/trunk/cpp/tests/SimpleTest.cpp 2009-09-14 15:21:33 UTC (rev 3623) @@ -377,8 +377,7 @@ store.stage(pmsg); //append to it - msg->setStore(&store); - msg->releaseContent(true);//ensure that data is not held in memory but is appended to disk when added + msg->releaseContent(&store);//ensure that data is not held in memory but is appended to disk when added store.appendContent(cpmsg, data1); store.appendContent(cpmsg, data2); Modified: store/trunk/cpp/tests/python_tests/flow_to_disk.py =================================================================== --- store/trunk/cpp/tests/python_tests/flow_to_disk.py 2009-09-11 13:35:04 UTC (rev 3622) +++ store/trunk/cpp/tests/python_tests/flow_to_disk.py 2009-09-14 15:21:33 UTC (rev 3623) @@ -34,16 +34,16 @@ # --- Helper functions --- - def _browse(self, queueName, destinationTag, acquireMode, numMsgs, msgSize, txnConsume): + def _browse(self, qn, dt, am, numMsgs, msgSize, txnConsume): txid = None if txnConsume: - txid = self._makeXid("consumer-xid-%s" % queueName) + txid = self._makeXid("consumer-xid-%s" % qn) self.session.dtx_select() self.assertEqual(self.XA_OK, self.session.dtx_start(xid=txid).status) - self.session.message_subscribe(queue=queueName, destination=destinationTag, acquire_mode=acquireMode) - self.session.message_flow(destination=destinationTag, unit=self.session.credit_unit.message, value=0xFFFFFFFF) - self.session.message_flow(destination=destinationTag, unit=self.session.credit_unit.byte, value=0xFFFFFFFF) - queue = self.session.incoming(destinationTag) + self.session.message_subscribe(queue=qn, destination=dt, acquire_mode=am) + self.session.message_flow(destination=dt, unit=self.session.credit_unit.message, value=0xFFFFFFFF) + self.session.message_flow(destination=dt, unit=self.session.credit_unit.byte, value=0xFFFFFFFF) + queue = self.session.incoming(dt) ids = RangedSet() for msgNum in range(0, numMsgs): expectedStr = self._makeMessage(msgNum, msgSize) @@ -52,16 +52,16 @@ ids.add(msg.id) return ids, txid - def _checkCancel(self, queueName, destinationTag, numMsgs, ids): + def _checkCancel(self, qn, dt, numMsgs, ids): self.session.message_release(ids) - self.session.queue_declare(queue=queueName) - self.assertEqual(numMsgs, self.session.queue_query(queue=queueName).message_count) - self.session.message_cancel(destination=destinationTag) + self.session.queue_declare(queue=qn) + self.assertEqual(numMsgs, self.session.queue_query(queue=qn).message_count) + self.session.message_cancel(destination=dt) - def _checkConsume(self, queueName, acquireMode, numMsgs, ids, txid, txnConsume): - if acquireMode == self.session.acquire_mode.not_acquired: - self.session.queue_declare(queue=queueName) - self.assertEqual(numMsgs, self.session.queue_query(queue=queueName).message_count) + def _checkConsume(self, qn, am, numMsgs, ids, txid, txnConsume): + if am == self.session.acquire_mode.not_acquired: + self.session.queue_declare(queue=qn) + self.assertEqual(numMsgs, self.session.queue_query(queue=qn).message_count) response = self.session.message_acquire(ids) for range in ids: for msgId in range: @@ -73,15 +73,10 @@ self.assertEqual(self.XA_OK, self.session.dtx_commit(xid=txid, one_phase=False).status) self._resetChannel() - def _checkEmpty(self, queueName): - self.session.queue_declare(queue=queueName) - self.assertEqual(0, self.session.queue_query(queue=queueName).message_count) + def _checkEmpty(self, qn): + self.session.queue_declare(queue=qn) + self.assertEqual(0, self.session.queue_query(queue=qn).message_count) - def _handleSessionClose(self): - time.sleep(1) - self.tearDown() - self.setUp() - def _makeMessage(self, msgCnt, msgSize): msg = "Message-%04d" % msgCnt msgLen = len(msg) @@ -98,45 +93,19 @@ branchqual = "v%s" % self.txCounter return self.session.xid(format=0, global_id=txid, branch_id=branchqual) - def _produceDirect(self, queueName, deliveryMode, numMsgs, msgSize, txnProduce): + def _produce(self, qn, dm, numMsgs, msgSize, txnProduce): if txnProduce: - txid = self._makeXid("producer-xid-%s" % queueName) + txid = self._makeXid("producer-xid-%s" % qn) self.session.dtx_select() self.assertEqual(self.XA_OK, self.session.dtx_start(xid=txid).status) for msgNum in range(0, numMsgs): msg_str = self._makeMessage(msgNum, msgSize) - self.session.message_transfer(message=Message(self.session.delivery_properties(routing_key=queueName, delivery_mode=deliveryMode), msg_str)) + self.session.message_transfer(message=Message(self.session.delivery_properties(routing_key=qn, delivery_mode=dm), msg_str)) if txnProduce: self.assertEqual(self.XA_OK, self.session.dtx_end(xid=txid).status) self.assertEqual(self.XA_OK, self.session.dtx_prepare(xid=txid).status) self.assertEqual(self.XA_OK, self.session.dtx_commit(xid=txid, one_phase=False).status) self._resetChannel() - - def _produceFanout(self, exchangeName, deliveryMode, numMsgs, msgSize, txnProduce, fail_on_msg): - if txnProduce: - txid = self._makeXid("producer-xid-%s" % exchangeName) - self.session.dtx_select() - self.assertEqual(self.XA_OK, self.session.dtx_start(xid=txid).status) - try: - for msgNum in range(0, numMsgs): - msg_str = self._makeMessage(msgNum, msgSize) - self.session.message_transfer(destination=exchangeName, message=Message(self.session.delivery_properties(delivery_mode=deliveryMode), msg_str)) - if fail_on_msg != None and msgNum > fail_on_msg and not txnProduce: - self.fail("Failed to throw SessionException on message_transfer %s" % fail_on_msg) - except SessionException: - if fail_on_msg != None and msgNum == fail_on_msg and not txnProduce: - self._handleSessionClose() - if txnProduce: - self.assertEqual(self.XA_OK, self.session.dtx_end(xid=txid).status) - self.assertEqual(self.XA_OK, self.session.dtx_prepare(xid=txid).status) - try: - self.assertEqual(self.XA_OK, self.session.dtx_commit(xid=txid, one_phase=False).status) - except SessionException: - if fail_on_msg != None and txnProduce: - self._handleSessionClose() - else: - self.fail("Failed to throw expected SessionException on dtx_commit") - self._resetChannel() def _randomBool(self): if random.randint(0, 1) > 0: @@ -147,66 +116,109 @@ self.session.close() self.session = self.conn.session("test-session", 1) - # --- Simple test framework (single queue) --- + # --- Simple tests --- - def simpleLimit(self, queueName, max_count = None, max_size = None, persistent = False, pre_acquired = True, numMsgs = 15, msgSize = None, browse = True, policy = "flow_to_disk"): - if max_count != None or max_size != None: - queueArgs = {'qpid.policy_type':policy} + def test_FlowToDisk_00_SimpleMaxCount(self): + """Flow-to-disk tests based on setting max_count""" + self.simpleLimit("test_FlowToDisk_00a", max_count = 10) + self.simpleLimit("test_FlowToDisk_00b", max_count = 10, persistent = True) + self.simpleLimit("test_FlowToDisk_00c", max_count = 10, max_size = 10000000, numMsgs = 100, msgSize = 10000) + self.simpleLimit("test_FlowToDisk_00d", max_count = 10, max_size = 10000000, persistent = True, numMsgs = 100, msgSize = 10000) + + def test_FlowToDisk_01_SimpleMaxSize(self): + """Flow-to-disk tests based on setting max_size""" + self.simpleLimit("test_FlowToDisk_01a", max_size = 100) + self.simpleLimit("test_FlowToDisk_01b", max_size = 100, persistent = True) + self.simpleLimit("test_FlowToDisk_01c", max_size = 100000, numMsgs = 100, msgSize = 10000) + self.simpleLimit("test_FlowToDisk_01d", max_size = 100000, persistent = True, numMsgs = 100, msgSize = 10000) + + def test_FlowToDisk_02_SimpleMaxCountNotAcquired(self): + """Flow-to-disk tests based on setting max_count, but not using pre-acquired mode (ie browsing)""" + self.simpleLimit("test_FlowToDisk_02a", max_count = 10, pre_acquired = False) + self.simpleLimit("test_FlowToDisk_02b", max_count = 10, persistent = True, pre_acquired = False) + self.simpleLimit("test_FlowToDisk_02c", max_count = 10, max_size = 10000000, pre_acquired = False, numMsgs = 100, msgSize = 10000) + self.simpleLimit("test_FlowToDisk_02d", max_count = 10, max_size = 10000000, persistent = True, pre_acquired = False, numMsgs = 100, msgSize = 10000) + + def test_FlowToDisk_03_SimpleMaxSizeNotAcquired(self): + """Flow-to-disk tests based on setting max_size, but not using pre-acquired mode (ie browsing)""" + self.simpleLimit("test_FlowToDisk_03a", max_size = 100, pre_acquired = False) + self.simpleLimit("test_FlowToDisk_03b", max_size = 100, persistent = True, pre_acquired = False) + self.simpleLimit("test_FlowToDisk_03c", max_size = 100, pre_acquired = False, numMsgs = 100, msgSize = 10000) + self.simpleLimit("test_FlowToDisk_03d", max_size = 100, persistent = True, pre_acquired = False, numMsgs = 100, msgSize = 10000) + + def test_FlowToDisk_04_MaxSizeMaxCount(self): + """Flow-to-disk tests based on setting both max-count and max-size at the same time""" + self.simpleLimit("test_FlowToDisk_04a", max_count = 10, max_size = 1000) + self.simpleLimit("test_FlowToDisk_04b", max_count = 10, max_size = 1000, msgSize = 250) + self.simpleLimit("test_FlowToDisk_04c", max_count = 10, max_size = 1000, persistent = True) + self.simpleLimit("test_FlowToDisk_04d", max_count = 10, max_size = 1000, msgSize = 250, persistent = True) + + def test_FlowToDisk_05_Randomized(self): + """Randomized flow-to-disk tests""" + seed = long(1000.0 * time.time()) + print "seed=0x%x" % seed + random.seed(seed) + for i in range(0, 10): + self.randomLimit(i) + + def simpleLimit(self, qn, max_count = None, max_size = None, persistent = False, pre_acquired = True, numMsgs = 15, msgSize = None, browse = True): + qa = {'qpid.policy_type':'flow_to_disk'} if max_count != None: - queueArgs['qpid.max_count'] = max_count + qa['qpid.max_count'] = max_count if max_size != None: - queueArgs['qpid.max_size'] = max_size + qa['qpid.max_size'] = max_size if persistent: - deliveryMode = self.session.delivery_mode.persistent + dm = self.session.delivery_mode.persistent else: - deliveryMode = self.session.delivery_mode.non_persistent + dm = self.session.delivery_mode.non_persistent if pre_acquired: - acquireMode = self.session.acquire_mode.pre_acquired + am = self.session.acquire_mode.pre_acquired else: - acquireMode = self.session.acquire_mode.not_acquired + am = self.session.acquire_mode.not_acquired # Cycle through the produce/consume block transaction combinations for i in range(0, 4): - txnProduce = i & 1 != 0 # Transactional produce - txnConsume = i & 2 != 0 # Transactional consume - self._txSimpleLimit(queueName, queueArgs, deliveryMode, acquireMode, numMsgs, msgSize, txnProduce, txnConsume, browse) + tp = i & 1 != 0 # Transactional produce + tc = i & 2 != 0 # Transactional consume + self.txSimpleLimit(qn, qa, dm, am, numMsgs, msgSize, tp, tc, browse) - def randomLimit(self, count, policy = "flow_to_disk"): - queueArgs = {'qpid.policy_type':policy} - queueName = "randomized_test_%04d" % count + def randomLimit(self, count): + qa = {'qpid.policy_type':'flow_to_disk'} + qn = "randomized_test_%04d" % count # Flow to disk policy maxCount = None if self._randomBool(): maxCount = random.randint(0,10000) - queueArgs['qpid.max_count'] = maxCount + qa['qpid.max_count'] = maxCount maxSize = None if self._randomBool(): maxSize = random.randint(0, 1000000) - queueArgs['qpid.max_size'] = maxSize + qa['qpid.max_size'] = maxSize # Persistence if self._randomBool(): - deliveryMode = self.session.delivery_mode.persistent + dm = self.session.delivery_mode.persistent else: - deliveryMode = self.session.delivery_mode.non_persistent + dm = self.session.delivery_mode.non_persistent # Acquired mode if self._randomBool(): - acquireMode = self.session.acquire_mode.pre_acquired + am = self.session.acquire_mode.pre_acquired browse = False else: - acquireMode = self.session.acquire_mode.not_acquired + am = self.session.acquire_mode.not_acquired browse = self._randomBool() numMsgs = random.randint(1, 10000) sizeLimit = int(1000000 / numMsgs) msgSize = random.randint(1, sizeLimit) - txnProduce = self._randomBool() - txnConsume = self._randomBool() + tp = self._randomBool() + tc = self._randomBool() - self._txSimpleLimit(queueName, queueArgs, deliveryMode, acquireMode, numMsgs, msgSize, txnProduce, txnConsume, browse) + #print " qn=%s, qa=%s, dm=%s, am=%s, numMsgs=%d, msgSize=%d, tp=%s, tc=%s, browse=%s" % (qn, qa, dm, am, numMsgs, msgSize, tp, tc, browse) + self.txSimpleLimit(qn, qa, dm, am, numMsgs, msgSize, tp, tc, browse) - def _txSimpleLimit(self, queueName, queueArgs, deliveryMode, acquireMode, numMsgs, msgSize, txnProduce, txnConsume, browse): + def txSimpleLimit(self, qn, qa, dm, am, numMsgs, msgSize, tp, tc, browse): """ Test a simple case of message limits which will force flow-to-disk. * queue_args sets a limit - either max_count and/or max_size @@ -215,323 +227,19 @@ * Check the broker has no messages left. """ - self.session.queue_declare(queue=queueName, durable=True, arguments=queueArgs) + self.session.queue_declare(queue=qn, durable=True, arguments=qa) - # Add messages - self._produceDirect(queueName, deliveryMode, numMsgs, msgSize, txnProduce) + # --- Add messages --- + self._produce(qn, dm, numMsgs, msgSize, tp) - # Browse messages (if possible) - if acquireMode == self.session.acquire_mode.not_acquired and browse: - dtA = "tagA-%d-%d" % (txnProduce, txnConsume) - ids, txid = self._browse(queueName, dtA, acquireMode, numMsgs, msgSize, False) - self._checkCancel(queueName, dtA, numMsgs, ids) + # --- Browse messages (if possible) --- + if am == self.session.acquire_mode.not_acquired and browse: + dtA = "tagA-%d-%d" % (tp, tc) + ids, txid = self._browse(qn, dtA, am, numMsgs, msgSize, False) + self._checkCancel(qn, dtA, numMsgs, ids) - # Consume messages - dtB = "tagB-%d-%d" % (txnProduce, txnConsume) - ids, txid = self._browse(queueName, dtB, acquireMode, numMsgs, msgSize, txnConsume) - self._checkConsume(queueName, acquireMode, numMsgs, ids, txid, txnConsume) - self._checkEmpty(queueName) - - self.session.queue_delete(queue=queueName) - - # --- Multi-queue test framework --- - - def multiQueueLimit(self, queueMap, exchangeName = "amq.fanout", persistent = False, pre_acquired = True, numMsgs = 15, msgSize = None, browse = True, policy = "flow_to_disk", fail_on_msg = None): - if persistent: - deliveryMode = self.session.delivery_mode.persistent - else: - deliveryMode = self.session.delivery_mode.non_persistent - if pre_acquired: - acquireMode = self.session.acquire_mode.pre_acquired - else: - acquireMode = self.session.acquire_mode.not_acquired - # Cycle through the produce/consume block transaction combinations -# TODO - re-enable full transactional testing when BZ 522499 (which crashes the broker) is fixed. -# for i in range(0, 4): -# txnProduce = i & 1 != 0 # Transactional produce -# txnConsume = i & 2 != 0 # Transactional consume -# self._txMultiQueueLimit(queueMap, exchangeName, deliveryMode, acquireMode, numMsgs, msgSize, txnProduce, txnConsume, browse, fail_on_msg) - for i in range(0, 2): - txnConsume = i & 1 != 0 # Transactional consume - self._txMultiQueueLimit(queueMap, exchangeName, deliveryMode, acquireMode, numMsgs, msgSize, False, txnConsume, browse, policy, fail_on_msg) - - def _txMultiQueueLimit(self, queueMap, exchangeName, deliveryMode, acquireMode, numMsgs, msgSize, txnProduce, txnConsume, browse, policy, fail_on_msg): - """ - Test a multi-queue case - queueMap = queue map where map is queue name (key) against queue args (value) - """ - self._multiQueueSetup(queueMap, exchangeName, txnProduce, txnConsume, policy) - self._produceFanout(exchangeName, deliveryMode, numMsgs, msgSize, txnProduce, fail_on_msg) - if fail_on_msg == None: - # Browse messages (ie get without acquiring) - if acquireMode == self.session.acquire_mode.not_acquired and browse: - for queue in queueMap.iterkeys(): - deliveryTagA = "tagA-%d-%d" % (txnProduce, txnConsume) - queueName = "%s-%d-%d" % (queue, txnProduce, txnConsume) - ids, txid = self._browse(queueName, deliveryTagA, acquireMode, numMsgs, msgSize, False) - self._checkCancel(queueName, deliveryTagA, numMsgs, ids) - # Consume messages - for queue in queueMap.iterkeys(): - deliveryTagB = "tagB-%d-%d" % (txnProduce, txnConsume) - queueName = "%s-%d-%d" % (queue, txnProduce, txnConsume) - ids, txid = self._browse(queueName, deliveryTagB, acquireMode, numMsgs, msgSize, txnConsume) - self._checkConsume(queueName, acquireMode, numMsgs, ids, txid, txnConsume) - self._checkEmpty(queueName) - self._multiQueueTeardown(queueMap, exchangeName, txnProduce, txnConsume) - - def _multiQueueSetup(self, queueMap, exchangeName, txnProduce, txnConsume, policy, exclusive = True): - for queue, qp in queueMap.iteritems(): - queueArgs = {} - ftd = False - for p,v in qp.iteritems(): - if p == "persistent": - d = v == True - elif p == "qpid.max_count": - if v != None: - queueArgs[p] = v - ftd = True - elif p == "qpid.max_size": - if v != None: - queueArgs[p] = v - ftd = True - if ftd: - queueArgs["qpid.policy_type"] = policy - queueName = "%s-%d-%d" % (queue, txnProduce, txnConsume) - self.session.queue_declare(queue = queueName, durable = d, arguments = queueArgs, exclusive = exclusive) - self.session.exchange_bind(exchange = exchangeName, queue = queueName) - - def _multiQueueTeardown(self, queueMap, exchangeName, txnProduce, txnConsume): - for queue in queueMap.iterkeys(): - queueName = "%s-%d-%d" % (queue, txnProduce, txnConsume) - self.session.exchange_unbind(exchange = exchangeName, queue = queueName) - self.session.queue_delete(queue = queueName) - - # --- Simple tests --- - - def test_00_SimpleMaxCount(self): - """Flow-to-disk tests based on setting max_count""" - # Small msgs - self.simpleLimit("test_00a", max_count = 10) - self.simpleLimit("test_00b", max_count = 10, persistent = True) - # Large msgs - set max_size to high number to make sure it does not interfere - self.simpleLimit("test_00c", max_count = 10, max_size = 10000000, numMsgs = 100, msgSize = 10000) - self.simpleLimit("test_00d", max_count = 10, max_size = 10000000, persistent = True, numMsgs = 100, msgSize = 10000) - - def test_01_SimpleMaxSize(self): - """Flow-to-disk tests based on setting max_size""" - # Small msgs - self.simpleLimit("test_01a", max_size = 100) - self.simpleLimit("test_01b", max_size = 100, persistent = True) - # Large msgs - set max_size to high number to make sure it does not interfere - self.simpleLimit("test_01c", max_size = 100000, numMsgs = 100, msgSize = 10000) - self.simpleLimit("test_01d", max_size = 100000, persistent = True, numMsgs = 100, msgSize = 10000) - - def test_02_SimpleMaxCountNotAcquired(self): - """Flow-to-disk tests based on setting max_count, but not using pre-acquired mode (ie browsing)""" - # Small msgs - self.simpleLimit("test_02a", max_count = 10, pre_acquired = False) - self.simpleLimit("test_02b", max_count = 10, persistent = True, pre_acquired = False) - # Large msgs - set max_size to high number to make sure it does not interfere - self.simpleLimit("test_02c", max_count = 10, max_size = 10000000, pre_acquired = False, numMsgs = 100, msgSize = 10000) - self.simpleLimit("test_02d", max_count = 10, max_size = 10000000, persistent = True, pre_acquired = False, numMsgs = 100, msgSize = 10000) - - def test_03_SimpleMaxSizeNotAcquired(self): - """Flow-to-disk tests based on setting max_size, but not using pre-acquired mode (ie browsing)""" - # Small msgs - self.simpleLimit("test_03a", max_size = 100, pre_acquired = False) - self.simpleLimit("test_03b", max_size = 100, persistent = True, pre_acquired = False) - # Large msgs - set max_size to high number to make sure it does not interfere - self.simpleLimit("test_03c", max_size = 100, pre_acquired = False, numMsgs = 100, msgSize = 10000) - self.simpleLimit("test_03d", max_size = 100, persistent = True, pre_acquired = False, numMsgs = 100, msgSize = 10000) - - def test_04_MaxSizeMaxCount(self): - """Flow-to-disk tests based on setting both max-count and max-size at the same time""" - # Small msgs - self.simpleLimit("test_04a", max_count = 10, max_size = 1000) - self.simpleLimit("test_04b", max_count = 10, max_size = 1000, msgSize = 250) - # Large msgs - set max_size to high number to make sure it does not interfere - self.simpleLimit("test_04c", max_count = 10, max_size = 1000, persistent = True) - self.simpleLimit("test_04d", max_count = 10, max_size = 1000, msgSize = 250, persistent = True) - - # --- Multi-queue tests --- - - def test_05_MultiQueueTransQueueTransMsg(self): - """Flow-to-disk tests where both queues and messages are transient and messages are routed to more than one queue""" - queueMap1 = {"test_05a" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_05b" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None} } - queueMap2 = {"test_05c" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_05d" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None} } - queueMap3 = {"test_05e" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_05f" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000} } - queueMap4 = {"test_05g" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None}, - "test_05h" : {"persistent":False, "qpid.max_count":8, "qpid.max_size": None} } - queueMap5 = {"test_05i" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_05j" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 800} } - queueMap6 = {"test_05k" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_05l" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None}, - "test_05m" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_05n" : {"persistent":False, "qpid.max_count":8, "qpid.max_size": 800} } - self.multiQueueLimit(queueMap1) - self.multiQueueLimit(queueMap2, fail_on_msg = 10) - self.multiQueueLimit(queueMap3, msgSize = 100, fail_on_msg = 10) - self.multiQueueLimit(queueMap4, fail_on_msg = 8) - self.multiQueueLimit(queueMap5, msgSize = 100, fail_on_msg = 8) - self.multiQueueLimit(queueMap6, msgSize = 100, fail_on_msg = 8) - - def test_06_MultiQueueDurableQueueTransMsg(self): - """Flow-to-disk tests where both queues are durable but messages are transient and messages are routed to more than one queue""" - queueMap1 = {"test_06a" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_06b" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None} } - queueMap2 = {"test_06c" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_06d" : {"persistent":True, "qpid.max_count":10, "qpid.max_size": None} } - queueMap3 = {"test_06e" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_06f" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1000} } - queueMap4 = {"test_06g" : {"persistent":True, "qpid.max_count":10, "qpid.max_size": None}, - "test_06h" : {"persistent":True, "qpid.max_count":8, "qpid.max_size": None} } - queueMap5 = {"test_06i" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_06j" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 800} } - queueMap6 = {"test_06k" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_06l" : {"persistent":True, "qpid.max_count":10, "qpid.max_size": None}, - "test_06m" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_06n" : {"persistent":True, "qpid.max_count":8, "qpid.max_size": 800} } - self.multiQueueLimit(queueMap1) - self.multiQueueLimit(queueMap2, fail_on_msg = 10) - self.multiQueueLimit(queueMap3, msgSize = 100, fail_on_msg = 10) - self.multiQueueLimit(queueMap4, fail_on_msg = 8) - self.multiQueueLimit(queueMap5, msgSize = 100, fail_on_msg = 8) - self.multiQueueLimit(queueMap6, msgSize = 100, fail_on_msg = 8) - - - def test_07_MultiQueueTransQueueDurableMsg(self): - """Flow-to-disk tests where both queues and messages are transient and messages are routed to more than one queue""" - queueMap1 = {"test_07a" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_07b" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None} } - queueMap2 = {"test_07c" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_07d" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None} } - queueMap3 = {"test_07e" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_07f" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000} } - queueMap4 = {"test_07g" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None}, - "test_07h" : {"persistent":False, "qpid.max_count":8, "qpid.max_size": None} } - queueMap5 = {"test_07i" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_07j" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 800} } - queueMap6 = {"test_07k" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_07l" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None}, - "test_07m" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_07n" : {"persistent":False, "qpid.max_count":8, "qpid.max_size": 800} } - self.multiQueueLimit(queueMap1, persistent = True) - self.multiQueueLimit(queueMap2, persistent = True, fail_on_msg = 10) - self.multiQueueLimit(queueMap3, msgSize = 100, persistent = True, fail_on_msg = 10) - self.multiQueueLimit(queueMap4, persistent = True, fail_on_msg = 8) - self.multiQueueLimit(queueMap5, msgSize = 100, persistent = True, fail_on_msg = 8) - self.multiQueueLimit(queueMap6, msgSize = 100, persistent = True, fail_on_msg = 8) - - def test_08_MultiQueueDurableQueueDurableMsg(self): - """Flow-to-disk tests where both queues are durable but messages are transient and messages are routed to more than one queue""" - queueMap1 = {"test_08a" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_08b" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None} } - queueMap2 = {"test_08c" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_08d" : {"persistent":True, "qpid.max_count":10, "qpid.max_size": None} } - queueMap3 = {"test_08e" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_08f" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1000} } - queueMap4 = {"test_08g" : {"persistent":True, "qpid.max_count":10, "qpid.max_size": None}, - "test_08h" : {"persistent":True, "qpid.max_count":8, "qpid.max_size": None} } - queueMap5 = {"test_08i" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_08j" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 800} } - queueMap6 = {"test_08k" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_08l" : {"persistent":True, "qpid.max_count":10, "qpid.max_size": None}, - "test_08m" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_08n" : {"persistent":True, "qpid.max_count":8, "qpid.max_size": 800} } - self.multiQueueLimit(queueMap1, persistent = True) - self.multiQueueLimit(queueMap2, persistent = True) - self.multiQueueLimit(queueMap3, msgSize = 100, persistent = True) - self.multiQueueLimit(queueMap4, persistent = True) - self.multiQueueLimit(queueMap5, msgSize = 100, persistent = True) - self.multiQueueLimit(queueMap6, msgSize = 100, persistent = True) - - def test_09_MultiQueueMixedQueueTransMsg(self): - """Flow-to-disk tests where both queues are durable but messages are transient and messages are routed to more than one queue""" - queueMap1 = {"test_09a" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_09b" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None} } - queueMap2 = {"test_09c" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_09d" : {"persistent":True, "qpid.max_count":10, "qpid.max_size": None} } - queueMap3 = {"test_09e" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_09f" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None} } - queueMap4 = {"test_09g" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_09h" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1000} } - queueMap5 = {"test_09i" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_09j" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000} } - queueMap6 = {"test_09k" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None}, - "test_09l" : {"persistent":True, "qpid.max_count":8, "qpid.max_size": None} } - queueMap7 = {"test_09m" : {"persistent":True, "qpid.max_count":10, "qpid.max_size": None}, - "test_09n" : {"persistent":False, "qpid.max_count":8, "qpid.max_size": None} } - queueMap8 = {"test_09o" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_09p" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 800} } - queueMap9 = {"test_09q" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_09r" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 800} } - queueMap10 = {"test_09s" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_09t" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None}, - "test_09u" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_09v" : {"persistent":False, "qpid.max_count":8, "qpid.max_size": 800}, - "test_09w" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_09x" : {"persistent":True, "qpid.max_count":12, "qpid.max_size": None}, - "test_09y" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1100}, - "test_09z" : {"persistent":True, "qpid.max_count":7, "qpid.max_size": 900} } - self.multiQueueLimit(queueMap1) - self.multiQueueLimit(queueMap2, fail_on_msg = 10) - self.multiQueueLimit(queueMap3, fail_on_msg = 10) - self.multiQueueLimit(queueMap4, msgSize = 100, fail_on_msg = 10) - self.multiQueueLimit(queueMap5, msgSize = 100, fail_on_msg = 10) - self.multiQueueLimit(queueMap6, fail_on_msg = 8) - self.multiQueueLimit(queueMap7, fail_on_msg = 8) - self.multiQueueLimit(queueMap8, msgSize = 100, fail_on_msg = 8) - self.multiQueueLimit(queueMap9, msgSize = 100, fail_on_msg = 8) - self.multiQueueLimit(queueMap10, msgSize = 100, fail_on_msg = 7) - - def test_10_MultiQueueMixedQueueDurableMsg(self): - """Flow-to-disk tests where both queues are durable but messages are transient and messages are routed to more than one queue""" - queueMap1 = {"test_10a" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_10b" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None} } - queueMap2 = {"test_10c" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_10d" : {"persistent":True, "qpid.max_count":10, "qpid.max_size": None} } - queueMap3 = {"test_10e" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_10f" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None} } - queueMap4 = {"test_10g" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_10h" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1000} } - queueMap5 = {"test_10i" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_10j" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000} } - queueMap6 = {"test_10k" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None}, - "test_10l" : {"persistent":True, "qpid.max_count":8, "qpid.max_size": None} } - queueMap7 = {"test_10m" : {"persistent":True, "qpid.max_count":10, "qpid.max_size": None}, - "test_10n" : {"persistent":False, "qpid.max_count":8, "qpid.max_size": None} } - queueMap8 = {"test_10o" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_10p" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 800} } - queueMap9 = {"test_10q" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_10r" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 800} } - queueMap10 = {"test_10s" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": None}, - "test_10t" : {"persistent":False, "qpid.max_count":10, "qpid.max_size": None}, - "test_10u" : {"persistent":False, "qpid.max_count":None, "qpid.max_size": 1000}, - "test_10v" : {"persistent":False, "qpid.max_count":8, "qpid.max_size": 800}, - "test_10w" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": None}, - "test_10x" : {"persistent":True, "qpid.max_count":12, "qpid.max_size": None}, - "test_10y" : {"persistent":True, "qpid.max_count":None, "qpid.max_size": 1100}, - "test_10z" : {"persistent":True, "qpid.max_count":7, "qpid.max_size": 900} } - self.multiQueueLimit(queueMap1, persistent = True) - self.multiQueueLimit(queueMap2, persistent = True, fail_on_msg = 10) - self.multiQueueLimit(queueMap3, persistent = True, fail_on_msg = 10) - self.multiQueueLimit(queueMap4, msgSize = 100, persistent = True, fail_on_msg = 10) - self.multiQueueLimit(queueMap5, msgSize = 100, persistent = True, fail_on_msg = 10) - self.multiQueueLimit(queueMap6, persistent = True, fail_on_msg = 8) - self.multiQueueLimit(queueMap7, persistent = True, fail_on_msg = 8) - self.multiQueueLimit(queueMap8, msgSize = 100, persistent = True, fail_on_msg = 8) - self.multiQueueLimit(queueMap9, msgSize = 100, persistent = True, fail_on_msg = 8) - self.multiQueueLimit(queueMap10, msgSize = 100, persistent = True, fail_on_msg = 7) - - # --- Long and randomized tests --- - - def test_11_Randomized(self): - """Randomized flow-to-disk tests""" - seed = long(1000.0 * time.time()) - print "seed=0x%x" % seed - random.seed(seed) - for i in range(0, 10): - self.randomLimit(i) + # --- Consume messages --- + dtB = "tagB-%d-%d" % (tp, tc) + ids, txid = self._browse(qn, dtB, am, numMsgs, msgSize, tc) + self._checkConsume(qn, am, numMsgs, ids, txid, tc) + self._checkEmpty(qn) Modified: store/trunk/cpp/tests/run_python_tests =================================================================== --- store/trunk/cpp/tests/run_python_tests 2009-09-11 13:35:04 UTC (rev 3622) +++ store/trunk/cpp/tests/run_python_tests 2009-09-14 15:21:33 UTC (rev 3623) @@ -78,12 +78,10 @@ start_broker() { ${QPIDD} --daemon --port 0 --no-module-dir --load-module=${STORE_LIB} --data-dir=${STORE_DIR} --auth=no --log-enable info+ --log-to-file ${STORE_DIR}/broker.python-test.log > qpidd.port LOCAL_PORT=`cat qpidd.port` - echo "run_python_tests: Started qpidd on port ${LOCAL_PORT}" } stop_broker() { - echo "run_python_tests: Stopping broker on port ${LOCAL_PORT}" - ${QPIDD} -q --port ${LOCAL_PORT} + ${QPIDD} -q --port $LOCAL_PORT } fail=0