def testThrottling(self):

        # No throttling, no delay for replay
        trManager = TransactionManager(timedelta(seconds=0), MAX_QUEUE_SIZE, THROTTLING_DELAY)
        trManager._flush_without_ioloop = True  # Use blocking API to emulate tornado ioloop

        # Add 3 transactions, make sure no memory limit is in the way
        oneTrSize = MAX_QUEUE_SIZE / 10
        for i in xrange(3):
            tr = memTransaction(oneTrSize, trManager)
            trManager.append(tr)

        # Try to flush them, time it
        before = datetime.now()
        trManager.flush()
        after = datetime.now()
        self.assertTrue((after - before) > 3 * THROTTLING_DELAY - timedelta(microseconds=100000),
                        "before = %s after = %s" % (before, after))
Test throttling while flushing
dataset/ETHPy150Open serverdensity/sd-agent/tests/core/test_transaction.py/TestTransaction.testThrottling