Now at 96% coverage

Makefile

@@ -4,6 +4,9 @@ pep8:
 	(find . -name "*.py" | xargs pep8 | perl -nle'\
 		print; $$a=1 if $$_}{exit($$a)')
 
+cycomplex:
+	find celery -type f -name "*.py" | xargs pygenie.py complexity
+
 ghdocs:
 	contrib/doc2ghpages
 
@@ -13,17 +16,20 @@ autodoc:
 bump:
 	contrib/bump -c celery
 
-coverage:
+coverage2:
 	[ -d testproj/temp ] || mkdir -p testproj/temp
 	(cd testproj; python manage.py test --figleaf)
 
-coverage2:
+coverage:
 	[ -d testproj/temp ] || mkdir -p testproj/temp
 	(cd testproj; python manage.py test --coverage)
 
 test:
 	(cd testproj; python manage.py test)
 
+testverbose:
+	(cd testproj; python manage.py test --verbosity=2)
+
 releaseok: pep8 autodoc test
 
 removepyc:

celery/backends/base.py

@@ -3,7 +3,7 @@ import time
 import operator
 import threading
 from functools import partial as curry
-from celery.utils import pickle
+from celery.serialization import pickle
 
 
 class TimeoutError(Exception):

celery/fields.py

@@ -5,7 +5,7 @@ Custom Django Model Fields.
 """
 from django.db import models
 from django.conf import settings
-from celery.utils import pickle
+from celery.serialization import pickle
 
 
 class PickledObject(str):

celery/messaging.py

@@ -7,7 +7,7 @@ from carrot.messaging import Publisher, Consumer
 from celery import conf
 from celery.utils import gen_unique_id
 from celery.utils import mitemgetter
-from celery.utils import pickle
+from celery.serialization import pickle
 
 
 MSG_OPTIONS = ("mandatory", "priority",

celery/serialization.py

@@ -0,0 +1,4 @@
+try:
+    import cPickle as pickle
+except ImportError:
+    import pickle

celery/task/__init__.py

@@ -13,7 +13,7 @@ from celery.task.base import ExecuteRemoteTask
 from celery.task.base import AsynchronousMapTask
 from celery.task.builtins import DeleteExpiredTaskMetaTask, PingTask
 from celery.execute import apply_async, delay_task
-from celery.utils import pickle
+from celery.serialization import pickle
 
 
 def discard_all(connect_timeout=AMQP_CONNECTION_TIMEOUT):

celery/task/base.py

@@ -7,7 +7,7 @@ from celery.execute import apply_async, delay_task, apply
 from celery.utils import gen_unique_id
 from datetime import timedelta
 from celery.registry import tasks
-from celery.utils import pickle
+from celery.serialization import pickle
 
 
 class Task(object):

celery/task/builtins.py

@@ -2,7 +2,7 @@ from celery.task.base import Task, TaskSet, PeriodicTask
 from celery.registry import tasks
 from celery.backends import default_backend
 from datetime import timedelta
-from celery.utils import pickle
+from celery.serialization import pickle
 
 
 class DeleteExpiredTaskMetaTask(PeriodicTask):

celery/tests/test_log.py

@@ -1,10 +1,13 @@
-import unittest
-
+from __future__ import with_statement
+import os
 import sys
 import logging
+import unittest
 import multiprocessing
 from StringIO import StringIO
 from celery.log import setup_logger, emergency_error
+from celery.tests.utils import OverrideStdout
+from tempfile import mktemp
 
 
 class TestLog(unittest.TestCase):
@@ -48,3 +51,35 @@ class TestLog(unittest.TestCase):
         emergency_error(sio, "Testing emergency error facility")
         self.assertEquals(sio.getvalue().rpartition(":")[2].strip(),
                              "Testing emergency error facility")
+
+    def test_setup_logger_no_handlers_stream(self):
+        from multiprocessing import get_logger
+        l = get_logger()
+        l.handlers = []
+        with OverrideStdout() as outs:
+            stdout, stderr = outs
+            l = setup_logger(logfile=stderr, loglevel=logging.INFO)
+            l.info("The quick brown fox...")
+            self.assertTrue("The quick brown fox..." in stderr.getvalue())
+
+    def test_setup_logger_no_handlers_file(self):
+        from multiprocessing import get_logger
+        l = get_logger()
+        l.handlers = []
+        tempfile = mktemp(suffix="unittest", prefix="celery")
+        l = setup_logger(logfile=tempfile, loglevel=0)
+        self.assertTrue(isinstance(l.handlers[0], logging.FileHandler))
+
+    def test_emergency_error_stderr(self):
+        with OverrideStdout() as outs:
+            stdout, stderr = outs
+            emergency_error(None, "The lazy dog crawls under the fast fox")
+            self.assertTrue("The lazy dog crawls under the fast fox" in \
+                                stderr.getvalue())
+
+    def test_emergency_error_file(self):
+        tempfile = mktemp(suffix="unittest", prefix="celery")
+        emergency_error(tempfile, "Vandelay Industries")
+        with open(tempfile, "r") as tempfilefh:
+            self.assertTrue("Vandelay Industries" in "".join(tempfilefh))
+        os.unlink(tempfile)

celery/tests/test_monitoring.py

@@ -1,8 +1,10 @@
+from __future__ import with_statement
 import unittest
 import time
 from celery.monitoring import TaskTimerStats, Statistics, StatsCollector
 from carrot.connection import DjangoAMQPConnection
 from celery.messaging import StatsConsumer
+from celery.tests.utils import OverrideStdout
 
 
 class PartialStatistics(Statistics):
@@ -85,16 +87,11 @@ class TestStatsCollector(unittest.TestCase):
         self.assertEquals(self.s.total_tasks_processed, 3)
 
         # Report
-        import sys
-        from StringIO import StringIO
-        out = StringIO()
-        sys.stdout = out
-        self.s.report()
-        sys.stdout = sys.__stdout__
-
-        output = out.getvalue()
-        self.assertTrue(
-                "Total processing time by task type:" in output)
+        with OverrideStdout() as outs:
+            stdout, stderr = outs
+            self.s.report()
+            self.assertTrue(
+                "Total processing time by task type:" in stdout.getvalue())
 
         # Dump to cache
         self.s.dump_to_cache()

celery/tests/test_pickle.py

@@ -1,5 +1,5 @@
 import unittest
-from celery.utils import pickle
+from celery.serialization import pickle
 
 
 class RegularException(Exception):

celery/tests/test_serialization.py

@@ -0,0 +1,13 @@
+import sys
+import unittest
+
+
+class TestAAPickle(unittest.TestCase):
+
+    def test_no_cpickle(self):
+        from celery.tests.utils import mask_modules
+        del(sys.modules["celery.serialization"])
+        with mask_modules("cPickle"):
+            from celery.serialization import pickle
+            import pickle as orig_pickle
+            self.assertTrue(pickle.dumps is orig_pickle.dumps)

celery/tests/test_task_builtins.py

@@ -1,7 +1,7 @@
 import unittest
 from celery.task.builtins import PingTask
 from celery.task.base import ExecuteRemoteTask
-from celery.utils import pickle
+from celery.serialization import pickle
 
 
 def some_func(i):

celery/tests/test_utils.py

@@ -1,3 +1,4 @@
+import sys
 import unittest
 from celery.utils import chunks
 

celery/tests/test_worker.py

@@ -7,7 +7,8 @@ from celery.worker import AMQPListener, WorkController
 from multiprocessing import get_logger
 from carrot.backends.base import BaseMessage
 from celery import registry
-from celery.utils import pickle, gen_unique_id
+from celery.serialization import pickle
+from celery.utils import gen_unique_id
 from datetime import datetime, timedelta
 
 

celery/tests/test_worker_controllers.py

@@ -5,6 +5,35 @@ from Queue import Queue, Empty
 from datetime import datetime, timedelta
 
 from celery.worker.controllers import Mediator, PeriodicWorkController
+from celery.worker.controllers import InfinityThread
+
+
+class MyInfinityThread(InfinityThread):
+
+    def on_iteration(self):
+        import time
+        time.sleep(1)
+
+
+class TestInfinityThread(unittest.TestCase):
+
+    def test_on_iteration(self):
+        self.assertRaises(NotImplementedError, InfinityThread().on_iteration)
+
+    def test_run(self):
+        t = MyInfinityThread()
+        t._shutdown.set()
+        t.run()
+        self.assertTrue(t._stopped.isSet())
+
+    def test_start_stop(self):
+        t = MyInfinityThread()
+        t.start()
+        self.assertFalse(t._shutdown.isSet())
+        self.assertFalse(t._stopped.isSet())
+        t.stop()
+        self.assertTrue(t._shutdown.isSet())
+        self.assertTrue(t._stopped.isSet())
 
 
 class TestMediator(unittest.TestCase):
@@ -20,6 +49,20 @@ class TestMediator(unittest.TestCase):
         self.assertTrue(m._shutdown.isSet())
         self.assertTrue(m._stopped.isSet())
 
+    def test_mediator_on_iteration(self):
+        bucket_queue = Queue()
+        got = {}
+
+        def mycallback(value):
+            got["value"] = value
+
+        m = Mediator(bucket_queue, mycallback)
+        bucket_queue.put("George Constanza")
+
+        m.on_iteration()
+
+        self.assertEquals(got["value"], "George Constanza")
+
 
 class TestPeriodicWorkController(unittest.TestCase):
 
@@ -40,3 +83,9 @@ class TestPeriodicWorkController(unittest.TestCase):
         m.process_hold_queue()
         self.assertRaises(Empty, bucket_queue.get_nowait)
         self.assertEquals(hold_queue.get_nowait(), ("task2", tomorrow))
+
+    def test_run_periodic_tasks(self):
+        bucket_queue = Queue()
+        hold_queue = Queue()
+        m = PeriodicWorkController(bucket_queue, hold_queue)
+        m.run_periodic_tasks()

celery/tests/utils.py

@@ -0,0 +1,55 @@
+from __future__ import with_statement
+from contextlib import contextmanager
+from StringIO import StringIO
+import os
+import sys
+import __builtin__
+ 
+
+@contextmanager
+def mask_modules(*modnames):
+    """Ban some modules from being importable inside the context
+ 
+    For example:
+ 
+        >>> with missing_modules("sys"):
+        ...     try:
+        ...         import sys
+        ...     except ImportError:
+        ...         print "sys not found"
+        sys not found
+ 
+        >>> import sys
+        >>> sys.version
+        (2, 5, 2, 'final', 0)
+ 
+    """
+ 
+    realimport = __builtin__.__import__
+    
+    def myimp(name, *args, **kwargs):
+        if name in modnames:
+            raise ImportError("No module named %s" % name)
+        else:
+            return realimport(name, *args, **kwargs)
+
+    __builtin__.__import__ = myimp
+    yield
+    __builtin__.__import__ = realimport
+
+
+class OverrideStdout(object):
+    """Override ``sys.stdout`` and ``sys.stderr`` with ``StringIO``."""
+
+    def __enter__(self):
+        mystdout = StringIO()
+        mystderr = StringIO()
+        sys.stdout = mystdout
+        sys.stderr = mystderr
+        return mystdout, mystderr
+
+    def __exit__(self, e_type, e_value, e_trace):
+        if e_type:
+            raise e_type(e_value)
+        sys.stdout = sys.__stdout__
+        sys.stderr = sys.__stderr__

celery/utils.py

@@ -5,11 +5,6 @@ Utility functions
 """
 import uuid
 
-try:
-    import cPickle as pickle
-except ImportError:
-    import pickle
-
 
 def chunks(it, n):
     """Split an iterator into chunks with ``n`` elements each.