100% coverage for celery.views

celery/tests/test_views.py

@@ -1,10 +1,11 @@
 import sys
 import unittest
 
+from django.http import HttpResponse
 from django.test.client import Client
 from django.test.testcases import TestCase as DjangoTestCase
 from django.core.urlresolvers import reverse
-from django.http import HttpResponse
+from django.template import TemplateDoesNotExist
 
 from anyjson import deserialize as JSON_load
 from billiard.utils.functional import curry
@@ -63,6 +64,17 @@ class TestTaskApply(ViewTestCase):
         finally:
             conf.ALWAYS_EAGER = False
 
+    def test_apply_raises_404_on_unregistered_task(self):
+        conf.ALWAYS_EAGER = True
+        try:
+            name = "xxx.does.not.exist"
+            action = curry(self.client.get, task_apply(kwargs={
+                                                        "task_name": name,
+                                                        "args": "4/4"}))
+            self.assertRaises(TemplateDoesNotExist, action)
+        finally:
+            conf.ALWAYS_EAGER = False
+
 
 class TestTaskStatus(ViewTestCase):
 

celery/tests/test_worker.py

@@ -139,6 +139,24 @@ class TestCarrotListener(unittest.TestCase):
         self.assertEquals(in_bucket.execute(), 2 * 4 * 8)
         self.assertTrue(self.eta_schedule.empty())
 
+    def test_receieve_message_eta_isoformat(self):
+        l = CarrotListener(self.ready_queue, self.eta_schedule, self.logger,
+                           send_events=False)
+        backend = MockBackend()
+        m = create_message(backend, task=foo_task.name,
+                           eta=datetime.now().isoformat(),
+                           args=[2, 4, 8], kwargs={})
+
+        l.event_dispatcher = MockEventDispatcher()
+        l.receive_message(m.decode(), m)
+
+        items = [entry[2] for entry in self.eta_schedule.queue]
+        found = 0
+        for item in items:
+            if item.task_name == foo_task.name:
+                found = True
+        self.assertTrue(found)
+
     def test_revoke(self):
         ready_queue = Queue()
         l = CarrotListener(ready_queue, self.eta_schedule, self.logger,

testproj/urls.py

@@ -1,4 +1,5 @@
-from django.conf.urls.defaults import patterns, url, include, handler500
+from django.conf.urls.defaults import (patterns, url, include,
+                                       handler500, handler404)
 from celery.views import apply
 
 # Uncomment the next two lines to enable the admin: