<reponame>spellman/hotqueue
# -*- coding: utf-8 -*-

"""Test suite for the HotQueue library. To run this test suite, execute this
Python program (``python tests.py``). Redis must be running on localhost:6379,
and a list key named 'hotqueue:testqueue' will be created and deleted in db 0
several times while the tests are running.
"""

from time import sleep
import threading
import unittest
try:
    import cPickle as pickle
except ImportError:
    import pickle

from hotqueue import HotQueue


class DummySerializer(object):
    """Dummy serializer that deliberately discards messages on dumps."""
    @staticmethod
    def dumps(s):
        return "foo"
    @staticmethod
    def loads(s):
        return s


class HotQueueTestCase(unittest.TestCase):
    
    def setUp(self):
        """Create the queue instance before the test."""
        self.queue = HotQueue('testqueue')
    
    def tearDown(self):
        """Clear the queue after the test."""
        self.queue.clear()
    
    def test_arguments(self):
        """Test that HotQueue.__init__ accepts arguments correctly, and that
        the Redis key is correctly formed.
        """
        kwargs = {
            'name': "testqueue",
            'serializer': DummySerializer,
            'host': "localhost",
            'port': 6379,
            'db': 0}
        # Instantiate the HotQueue instance:
        self.queue = HotQueue(**kwargs)
        # Ensure that the properties of the instance are as expected:
        self.assertEqual(self.queue.name, kwargs['name'])
        self.assertEqual(self.queue.key, "hotqueue:%s" % kwargs['name'])
        self.assertEqual(self.queue.serializer, kwargs['serializer'])
        # Instantiate a HotQueue instance with only the required args:
        self.queue = HotQueue(kwargs['name'])
        # Ensure that the properties of the instance are as expected:
        self.assertEqual(self.queue.name, kwargs['name'])
        self.assertEqual(self.queue.key, "hotqueue:%s" % kwargs['name'])
        self.assertTrue(self.queue.serializer is pickle) # Defaults to cPickle
                                                         # or pickle, depending
                                                         # on the platform.
    
    def test_consume(self):
        """Test the consume generator method."""
        nums = [1, 2, 3, 4, 5, 6, 7, 8]
        # Test blocking with timeout:
        self.queue.put(*nums)
        msgs = []
        for msg in self.queue.consume(timeout=1):
            msgs.append(msg)
        self.assertEqual(msgs, nums)
        # Test non-blocking:
        self.queue.put(*nums)
        msgs = []
        for msg in self.queue.consume(block=False):
            msgs.append(msg)
        self.assertEqual(msgs, nums)
    
    def test_cleared(self):
        """Test for correct behaviour if the Redis list does not exist."""
        self.assertEqual(len(self.queue), 0)
        self.assertEqual(self.queue.get(), None)
    
    def test_get_order(self):
        """Test that messages are get in the same order they are put."""
        alphabet = ['abc', 'def', 'ghi', 'jkl', 'mno']
        self.queue.put(alphabet[0], alphabet[1], alphabet[2])
        self.queue.put(alphabet[3])
        self.queue.put(alphabet[4])
        msgs = []
        msgs.append(self.queue.get())
        msgs.append(self.queue.get())
        msgs.append(self.queue.get())
        msgs.append(self.queue.get())
        msgs.append(self.queue.get())
        self.assertEqual(msgs, alphabet)
    
    def test_length(self):
        """Test that the length of a queue is returned correctly."""
        self.queue.put('a message')
        self.queue.put('another message')
        self.assertEqual(len(self.queue), 2)
    
    def test_worker(self):
        """Test the worker decorator."""
        colors = ['blue', 'green', 'red', 'pink', 'black']
        # Test blocking with timeout:
        self.queue.put(*colors)
        msgs = []
        @self.queue.worker(timeout=1)
        def appender(msg):
            msgs.append(msg)
        appender()
        self.assertEqual(msgs, colors)
        # Test non-blocking:
        self.queue.put(*colors)
        msgs = []
        @self.queue.worker(block=False)
        def appender(msg):
            msgs.append(msg)
        appender()
        self.assertEqual(msgs, colors)
        # Test decorating a class method:
        self.queue.put(*colors)
        msgs = []
        class MyClass(object):
            @self.queue.worker(block=False)
            def appender(self, msg):
                msgs.append(msg)
        my_instance = MyClass()
        my_instance.appender()
        self.assertEqual(msgs, colors)
    
    def test_threaded(self):
        """Threaded test of put and consume methods."""
        msgs = []
        def put():
            for num in range(3):
                self.queue.put('message %d' % num)
                sleep(0.1)
        def consume():
            for msg in self.queue.consume(timeout=1):
                msgs.append(msg)
        putter = threading.Thread(target=put)
        consumer = threading.Thread(target=consume)
        putter.start()
        consumer.start()
        for thread in [putter, consumer]:
            thread.join()
        self.assertEqual(msgs, ["message 0", "message 1", "message 2"])
    
    def test_custom_serializer(self):
        """Test the use of a custom serializer and None as serializer."""
        msg = "my message"
        # Test using None:
        self.queue.serializer = None
        self.queue.put(msg)
        self.assertEqual(self.queue.get(), msg)
        self.queue.put({"a": 1})
        self.assertEqual(self.queue.get(), "{'a': 1}") # Should be a string
        # Test using DummySerializer:
        self.queue.serializer = DummySerializer
        self.queue.put(msg)
        self.assertEqual(self.queue.get(), "foo")


if __name__ == "__main__":
    unittest.main()


# -*- coding: utf-8 -*-

"""Test suite for the HotQueue library. To run this test suite, execute this
Python program (``python tests.py``). Redis must be running on localhost:6379,
and a list key named 'hotqueue:testqueue' will be created and deleted in db 0
several times while the tests are running.
"""

from time import sleep
import threading
import unittest
try:
    import cPickle as pickle
except ImportError:
    import pickle

from hotqueue import HotQueue


class DummySerializer(object):
    """Dummy serializer that deliberately discards messages on dumps."""
    @staticmethod
    def dumps(s):
        return "foo"
    @staticmethod
    def loads(s):
        return s


class HotQueueTestCase(unittest.TestCase):
    
    def setUp(self):
        """Create the queue instance before the test."""
        self.queue = HotQueue('testqueue')
    
    def tearDown(self):
        """Clear the queue after the test."""
        self.queue.clear()
    
    def test_arguments(self):
        """Test that HotQueue.__init__ accepts arguments correctly, and that
        the Redis key is correctly formed.
        """
        kwargs = {
            'name': "testqueue",
            'serializer': DummySerializer,
            'host': "localhost",
            'port': 6379,
            'db': 0}
        # Instantiate the HotQueue instance:
        self.queue = HotQueue(**kwargs)
        # Ensure that the properties of the instance are as expected:
        self.assertEqual(self.queue.name, kwargs['name'])
        self.assertEqual(self.queue.key, "hotqueue:%s" % kwargs['name'])
        self.assertEqual(self.queue.serializer, kwargs['serializer'])
        # Instantiate a HotQueue instance with only the required args:
        self.queue = HotQueue(kwargs['name'])
        # Ensure that the properties of the instance are as expected:
        self.assertEqual(self.queue.name, kwargs['name'])
        self.assertEqual(self.queue.key, "hotqueue:%s" % kwargs['name'])
        self.assertTrue(self.queue.serializer is pickle) # Defaults to cPickle
                                                         # or pickle, depending
                                                         # on the platform.
    
    def test_consume(self):
        """Test the consume generator method."""
        nums = [1, 2, 3, 4, 5, 6, 7, 8]
        # Test blocking with timeout:
        self.queue.put(*nums)
        msgs = []
        for msg in self.queue.consume(timeout=1):
            msgs.append(msg)
        self.assertEqual(msgs, nums)
        # Test non-blocking:
        self.queue.put(*nums)
        msgs = []
        for msg in self.queue.consume(block=False):
            msgs.append(msg)
        self.assertEqual(msgs, nums)
    
    def test_cleared(self):
        """Test for correct behaviour if the Redis list does not exist."""
        self.assertEqual(len(self.queue), 0)
        self.assertEqual(self.queue.get(), None)
    
    def test_get_order(self):
        """Test that messages are get in the same order they are put."""
        alphabet = ['abc', 'def', 'ghi', 'jkl', 'mno']
        self.queue.put(alphabet[0], alphabet[1], alphabet[2])
        self.queue.put(alphabet[3])
        self.queue.put(alphabet[4])
        msgs = []
        msgs.append(self.queue.get())
        msgs.append(self.queue.get())
        msgs.append(self.queue.get())
        msgs.append(self.queue.get())
        msgs.append(self.queue.get())
        self.assertEqual(msgs, alphabet)
    
    def test_length(self):
        """Test that the length of a queue is returned correctly."""
        self.queue.put('a message')
        self.queue.put('another message')
        self.assertEqual(len(self.queue), 2)
    
    def test_worker(self):
        """Test the worker decorator."""
        colors = ['blue', 'green', 'red', 'pink', 'black']
        # Test blocking with timeout:
        self.queue.put(*colors)
        msgs = []
        @self.queue.worker(timeout=1)
        def appender(msg):
            msgs.append(msg)
        appender()
        self.assertEqual(msgs, colors)
        # Test non-blocking:
        self.queue.put(*colors)
        msgs = []
        @self.queue.worker(block=False)
        def appender(msg):
            msgs.append(msg)
        appender()
        self.assertEqual(msgs, colors)
        # Test decorating a class method:
        self.queue.put(*colors)
        msgs = []
        class MyClass(object):
            @self.queue.worker(block=False)
            def appender(self, msg):
                msgs.append(msg)
        my_instance = MyClass()
        my_instance.appender()
        self.assertEqual(msgs, colors)
    
    def test_threaded(self):
        """Threaded test of put and consume methods."""
        msgs = []
        def put():
            for num in range(3):
                self.queue.put('message %d' % num)
                sleep(0.1)
        def consume():
            for msg in self.queue.consume(timeout=1):
                msgs.append(msg)
        putter = threading.Thread(target=put)
        consumer = threading.Thread(target=consume)
        putter.start()
        consumer.start()
        for thread in [putter, consumer]:
            thread.join()
        self.assertEqual(msgs, ["message 0", "message 1", "message 2"])
    
    def test_custom_serializer(self):
        """Test the use of a custom serializer and None as serializer."""
        msg = "my message"
        # Test using None:
        self.queue.serializer = None
        self.queue.put(msg)
        self.assertEqual(self.queue.get(), msg)
        self.queue.put({"a": 1})
        self.assertEqual(self.queue.get(), "{'a': 1}") # Should be a string
        # Test using DummySerializer:
        self.queue.serializer = DummySerializer
        self.queue.put(msg)
        self.assertEqual(self.queue.get(), "foo")


if __name__ == "__main__":
    unittest.main()
# -*- coding: utf-8 -*- Test suite for the HotQueue library. To run this test suite, execute this Python program (``python tests.py``). Redis must be running on localhost:6379, and a list key named 'hotqueue:testqueue' will be created and deleted in db 0 several times while the tests are running. Dummy serializer that deliberately discards messages on dumps. Create the queue instance before the test. Clear the queue after the test. Test that HotQueue.__init__ accepts arguments correctly, and that         the Redis key is correctly formed. # Instantiate the HotQueue instance: # Ensure that the properties of the instance are as expected: # Instantiate a HotQueue instance with only the required args: # Ensure that the properties of the instance are as expected: # Defaults to cPickle # or pickle, depending # on the platform. Test the consume generator method. # Test blocking with timeout: # Test non-blocking: Test for correct behaviour if the Redis list does not exist. Test that messages are get in the same order they are put. Test that the length of a queue is returned correctly. Test the worker decorator. # Test blocking with timeout: # Test non-blocking: # Test decorating a class method: Threaded test of put and consume methods. Test the use of a custom serializer and None as serializer. # Test using None: # Should be a string # Test using DummySerializer: