Test coverage now at 95% (that's without celeryd/worker.py/process/platform)

celery/task.py

@@ -254,16 +254,6 @@ class PeriodicTask(Task):
         super(PeriodicTask, self).__init__()
 
 
-class TestTask(Task):
-    """A simple test task that just logs something."""
-    name = "celery.test_task"
-
-    def run(self, some_arg, **kwargs):
-        logger = self.get_logger(**kwargs)
-        logger.info("TestTask got some_arg=%s" % some_arg)
-tasks.register(TestTask)
-
-
 class DeleteExpiredTaskMetaTask(PeriodicTask):
     """A periodic task that deletes expired task metadata every day.
    

celery/tests/test_conf.py

@@ -2,6 +2,7 @@ import unittest
 from celery import conf
 from django.conf import settings
 
+
 SETTING_VARS = (
     ("CELERY_AMQP_CONSUMER_QUEUE", "AMQP_CONSUMER_QUEUE",
         "DEFAULT_AMQP_CONSUMER_QUEUE"),

celery/tests/test_models.py

@@ -26,6 +26,7 @@ class TestModels(unittest.TestCase):
         m1 = self.createTaskMeta()
         m2 = self.createTaskMeta()
         m3 = self.createTaskMeta()
+        self.assertTrue(unicode(m1).startswith("<Task:"))
         self.assertTrue(m1.task_id)
         self.assertTrue(isinstance(m1.date_done, datetime))
 
@@ -52,11 +53,14 @@ class TestModels(unittest.TestCase):
     def test_periodic_taskmeta(self):
         tasks.register(TestPeriodicTask)
         p = self.createPeriodicTaskMeta(TestPeriodicTask.name)
+        # check that repr works.
+        self.assertTrue(unicode(p).startswith("<PeriodicTask:"))
         self.assertFalse(p in PeriodicTaskMeta.objects.get_waiting_tasks())
         # Have to avoid save() because it applies the auto_now=True.
         PeriodicTaskMeta.objects.filter(name=p.name).update(
                 last_run_at=datetime.now() - TestPeriodicTask.run_every)
         self.assertTrue(p in PeriodicTaskMeta.objects.get_waiting_tasks())
         self.assertTrue(isinstance(p.task, TestPeriodicTask))
-        
+
+        p.delay()
 

celery/tests/test_registry.py

@@ -74,3 +74,7 @@ class TestTaskRegistry(unittest.TestCase):
         self.assertFalse(TestPeriodicTask.name in r)
         r.unregister(FUNC_TASK_NAME)
         self.assertFalse(FUNC_TASK_NAME in r)
+
+        self.assertTrue(func_task())
+        self.assertTrue(TestTask().run())
+        self.assertTrue(TestPeriodicTask().run())

celery/tests/test_task.py

@@ -9,12 +9,6 @@ from celery.log import setup_logger
 from celery import messaging
 
 
-def get_string_io_logger():
-    sio = StringIO()
-    logger = setup_logger(loglevel=logging.INFO, logfile=sio)
-    return logger, sio
-
-
 # Task run functions can't be closures/lambdas, as they're pickled.
 def return_True(self, **kwargs):
     return True
@@ -83,6 +77,7 @@ class TestCeleryTasks(unittest.TestCase):
         registry.tasks.register(T1)
         t1 = T1()
         consumer = t1.get_consumer()
+        self.assertRaises(NotImplementedError, consumer.receive, "foo", "foo")
         consumer.discard_all()
         self.assertTrue(consumer.fetch() is None)
 

testproj/settings.py

@@ -99,7 +99,9 @@ INSTALLED_APPS = (
     'django.contrib.contenttypes',
     'django.contrib.sessions',
     'django.contrib.sites',
-    'celery'
+    'celery',
+    'someapp',
+    'someappwotask',
 )
 
 try:

testproj/someapp/models.py

@@ -0,0 +1,3 @@
+from django.db import models
+
+# Create your models here.

testproj/someapp/tasks.py

@@ -0,0 +1,9 @@
+from celery.task import tasks, Task
+
+
+class SomeAppTask(Task):
+    name = "c.unittest.SomeAppTask"
+
+    def run(self, **kwargs):
+        return 42
+tasks.register(SomeAppTask)

testproj/someapp/views.py

@@ -0,0 +1 @@
+# Create your views here.

testproj/someappwotask/models.py

@@ -0,0 +1,3 @@
+from django.db import models
+
+# Create your models here.

testproj/someappwotask/views.py

@@ -0,0 +1 @@
+# Create your views here.