Unittests for Tyrant + Cache backends.

celery/backends/tyrant.py

@@ -47,6 +47,7 @@ class Backend(BaseBackend):
                             getattr(settings, "TT_HOST", self.tyrant_host)
         self.tyrant_port = tyrant_port or \
                             getattr(settings, "TT_PORT", self.tyrant_port)
+        self.tyrant_port = int(self.tyrant_port)
         if not self.tyrant_host or not self.tyrant_port:
             raise ImproperlyConfigured(
                 "To use the Tokyo Tyrant backend, you have to "
@@ -57,14 +58,24 @@ class Backend(BaseBackend):
 
     def open(self):
         """Get :class:`pytyrant.PyTyrant`` instance with the current
-        server configuration."""
-        if not self._connection:
+        server configuration.
+        
+        The connection is then cached until you do an
+        explicit :meth:`close`.
+        
+        """
+
+        # connection overrides bool()
+        if self._connection is None:
             self._connection = pytyrant.PyTyrant.open(self.tyrant_host,
                                                       self.tyrant_port)
         return self._connection
 
     def close(self):
-        if self._connection:
+        """Close the tyrant connection and remove the cache."""
+
+        # connection overrides bool()
+        if self._connection is not None:
             self._connection.close()
             self._connection = None
 

celery/tests/test_backends/test_cache.py

@@ -0,0 +1,61 @@
+import sys
+import unittest
+import uuid
+import errno
+import socket
+from celery.backends.cache import Backend as CacheBackend
+from django.conf import settings
+
+
+class SomeClass(object):
+
+    def __init__(self, data):
+        self.data = data
+
+
+class TestCacheBackend(unittest.TestCase):
+
+    def test_mark_as_done(self):
+        cb = CacheBackend()
+
+        tid = str(uuid.uuid4())
+
+        self.assertFalse(cb.is_done(tid))
+        self.assertEquals(cb.get_status(tid), "PENDING")
+        self.assertEquals(cb.get_result(tid), None)
+
+        cb.mark_as_done(tid, 42)
+        self.assertTrue(cb.is_done(tid))
+        self.assertEquals(cb.get_status(tid), "DONE")
+        self.assertEquals(cb.get_result(tid), 42)
+        self.assertTrue(cb._cache.get(tid))
+        self.assertTrue(cb.get_result(tid), 42)
+
+    def test_is_pickled(self):
+        cb = CacheBackend()
+    
+        tid2 = str(uuid.uuid4())
+        result = {"foo": "baz", "bar": SomeClass(12345)}
+        cb.mark_as_done(tid2, result)
+        # is serialized properly.
+        rindb = cb.get_result(tid2)
+        self.assertEquals(rindb.get("foo"), "baz")
+        self.assertEquals(rindb.get("bar").data, 12345)
+
+    def test_mark_as_failure(self):
+        cb = CacheBackend()
+
+        tid3 = str(uuid.uuid4())
+        try:
+            raise KeyError("foo")
+        except KeyError, exception:
+            pass
+        cb.mark_as_failure(tid3, exception)
+        self.assertFalse(cb.is_done(tid3))
+        self.assertEquals(cb.get_status(tid3), "FAILURE")
+        self.assertTrue(isinstance(cb.get_result(tid3), KeyError))
+
+    def test_process_cleanup(self):
+        cb = CacheBackend()
+
+        cb.process_cleanup()

celery/tests/test_backends/test_tyrant.py

@@ -0,0 +1,99 @@
+import sys
+import unittest
+import uuid
+import errno
+import socket
+from celery.backends.tyrant import Backend as TyrantBackend
+from django.conf import settings
+
+_no_tyrant_msg = "* Tokyo Tyrant not running. Will not execute related tests."
+_no_tyrant_msg_emitted = False
+
+
+class SomeClass(object):
+
+    def __init__(self, data):
+        self.data = data
+
+
+def get_tyrant_or_None():
+    tb = TyrantBackend()
+    try:
+        tb.open()
+    except socket.error, exc:
+        if exc.errno == errno.ECONNREFUSED:
+            if not _no_tyrant_msg_emitted:
+                sys.stderr.write("\n" + _no_tyrant_msg + "\n")
+            return None
+        else:
+            raise
+    return tb
+
+
+class TestTyrantBackend(unittest.TestCase):
+
+    def test_cached_connection(self):
+        tb = get_tyrant_or_None()
+        if not tb:
+            return # Skip test 
+
+        self.assertTrue(tb._connection is not None)
+        tb.close()
+        self.assertTrue(tb._connection is None)
+        tb.open()
+        self.assertTrue(tb._connection is not None)
+
+    def test_mark_as_done(self):
+        tb = get_tyrant_or_None()
+        if not tb:
+            return
+
+        tid = str(uuid.uuid4())
+
+        self.assertFalse(tb.is_done(tid))
+        self.assertEquals(tb.get_status(tid), "PENDING")
+        self.assertEquals(tb.get_result(tid), None)
+
+        tb.mark_as_done(tid, 42)
+        self.assertTrue(tb.is_done(tid))
+        self.assertEquals(tb.get_status(tid), "DONE")
+        self.assertEquals(tb.get_result(tid), 42)
+        self.assertTrue(tb._cache.get(tid))
+        self.assertTrue(tb.get_result(tid), 42)
+
+    def test_is_pickled(self):
+        tb = get_tyrant_or_None()
+        if not tb:
+            return
+    
+        tid2 = str(uuid.uuid4())
+        result = {"foo": "baz", "bar": SomeClass(12345)}
+        tb.mark_as_done(tid2, result)
+        # is serialized properly.
+        rindb = tb.get_result(tid2)
+        self.assertEquals(rindb.get("foo"), "baz")
+        self.assertEquals(rindb.get("bar").data, 12345)
+
+    def test_mark_as_failure(self):
+        tb = get_tyrant_or_None()
+        if not tb:
+            return
+
+        tid3 = str(uuid.uuid4())
+        try:
+            raise KeyError("foo")
+        except KeyError, exception:
+            pass
+        tb.mark_as_failure(tid3, exception)
+        self.assertFalse(tb.is_done(tid3))
+        self.assertEquals(tb.get_status(tid3), "FAILURE")
+        self.assertTrue(isinstance(tb.get_result(tid3), KeyError))
+
+    def test_process_cleanup(self):
+        tb = get_tyrant_or_None()
+        if not tb:
+            return
+
+        tb.process_cleanup()
+
+        self.assertTrue(tb._connection is None)

testproj/settings.py

@@ -23,6 +23,9 @@ AMQP_VHOST = "/"
 AMQP_USER = "guest"
 AMQP_PASSWORD = "guest"
 
+TT_HOST = "localhost"
+TT_PORT = 1978
+
 CELERY_AMQP_EXCHANGE = "testcelery"
 CELERY_AMQP_ROUTING_KEY = "testcelery"
 CELERY_AMQP_CONSUMER_QUEUE = "testcelery"