Coverage now 96% (Afer adding long needed Tests for AMQP backend)

celery/loaders/default.py

@@ -13,8 +13,7 @@ DEFAULT_SETTINGS = {
 
 
 def wanted_module_item(item):
-    is_private = item.startswith("_")
-    return not is_private
+    return not item.startswith("_")
 
 
 class Loader(BaseLoader):
@@ -40,8 +39,8 @@ class Loader(BaseLoader):
             settings.configure()
         for config_key, config_value in usercfg.items():
             setattr(settings, config_key, config_value)
-        installed_apps = set(DEFAULT_SETTINGS["INSTALLED_APPS"] + \
-                             settings.INSTALLED_APPS)
+        installed_apps = set(list(DEFAULT_SETTINGS["INSTALLED_APPS"]) + \
+                             list(settings.INSTALLED_APPS))
         settings.INSTALLED_APPS = tuple(installed_apps)
         return settings
 

celery/tests/test_backends/test_amqp.py

@@ -0,0 +1,63 @@
+from __future__ import with_statement
+
+import sys
+import unittest
+import errno
+
+from django.core.exceptions import ImproperlyConfigured
+
+from celery.backends.amqp import AMQPBackend
+from celery.utils import gen_unique_id
+from celery.datastructures import ExceptionInfo
+
+
+class SomeClass(object):
+
+    def __init__(self, data):
+        self.data = data
+
+
+class TestRedisBackend(unittest.TestCase):
+
+    def setUp(self):
+        self.backend = AMQPBackend()
+
+    def test_mark_as_done(self):
+        tb = self.backend
+
+        tid = gen_unique_id()
+
+        tb.mark_as_done(tid, 42)
+        self.assertTrue(tb.is_successful(tid))
+        self.assertEquals(tb.get_status(tid), "SUCCESS")
+        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 = self.backend
+
+        tid2 = gen_unique_id()
+        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 = self.backend
+
+        tid3 = gen_unique_id()
+        try:
+            raise KeyError("foo")
+        except KeyError, exception:
+            einfo = ExceptionInfo(sys.exc_info())
+        tb.mark_as_failure(tid3, exception, traceback=einfo.traceback)
+        self.assertFalse(tb.is_successful(tid3))
+        self.assertEquals(tb.get_status(tid3), "FAILURE")
+        self.assertTrue(isinstance(tb.get_result(tid3), KeyError))
+        self.assertEquals(tb.get_traceback(tid3), einfo.traceback)
+
+    def test_process_cleanup(self):
+        self.backend.process_cleanup()

celery/tests/test_loaders.py

@@ -1,4 +1,5 @@
 import os
+import sys
 import unittest
 
 from billiard.utils.functional import wraps
@@ -6,6 +7,7 @@ from billiard.utils.functional import wraps
 from celery import loaders
 from celery.loaders import base
 from celery.loaders import djangoapp
+from celery.loaders import default
 from celery.tests.utils import with_environ
 
 
@@ -85,3 +87,39 @@ class TestDjangoLoader(unittest.TestCase):
 
     def test_find_related_module_no_related(self):
         self.assertFalse(djangoapp.find_related_module("someapp", "frobulators"))
+
+
+class TestDefaultLoader(unittest.TestCase):
+
+    def test_wanted_module_item(self):
+        self.assertTrue(default.wanted_module_item("FOO"))
+        self.assertTrue(default.wanted_module_item("foo"))
+        self.assertFalse(default.wanted_module_item("_foo"))
+        self.assertFalse(default.wanted_module_item("__foo"))
+
+
+    def test_read_configuration(self):
+        from types import ModuleType
+
+        class ConfigModule(ModuleType):
+            pass
+
+        celeryconfig = ConfigModule("celeryconfig")
+        celeryconfig.CELERY_IMPORTS = ("os", "sys")
+
+        sys.modules["celeryconfig"] = celeryconfig
+        try:
+           l = default.Loader()
+           settings = l.read_configuration()
+           self.assertEquals(settings.CELERY_IMPORTS, ("os", "sys"))
+           from django.conf import settings
+           settings.configured = False
+           settings = l.read_configuration()
+           self.assertEquals(settings.CELERY_IMPORTS, ("os", "sys"))
+           self.assertTrue(settings.configured)
+           l.on_worker_init()
+        finally:
+            sys.modules.pop("celeryconfig", None)
+
+
+

testproj/settings.py

@@ -30,7 +30,6 @@ COVERAGE_EXCLUDE_MODULES = ("celery.__init__",
                             "celery.utils.compat",
                             "celery.task.rest",
                             "celery.platform", # FIXME
-                            "celery.loaders.default", # FIXME
                             "celery.backends.mongodb", # FIXME
                             "celery.backends.tyrant", # FIXME
                             )