Tests for celery.views.task_is_successful

celery/tests/test_views.py

@@ -0,0 +1,40 @@
+import unittest
+
+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 anyjson import deserialize as JSON_load
+from billiard.utils.functional import curry
+
+from celery.utils import gen_unique_id
+from celery.backends import default_backend
+
+def reversestar(name, **kwargs):
+    return reverse(name, kwargs=kwargs)
+
+is_successful = curry(reversestar, "celery-is_task_successful")
+
+
+class ViewTestCase(DjangoTestCase):
+
+    def assertJSONEquals(self, json, py):
+        json = isinstance(json, HttpResponse) and json.content or json
+        self.assertEquals(JSON_load(json), py)
+
+
+class TestTaskIsSuccessful(ViewTestCase):
+
+    def test_is_successful_success(self):
+        uuid = gen_unique_id()
+        default_backend.mark_as_done(uuid, "Quick")
+        json = self.client.get(is_successful(task_id=uuid))
+        self.assertJSONEquals(json, {"task": {"id": uuid,
+                                              "executed": True}})
+    def test_is_successful_pending(self):
+        uuid = gen_unique_id()
+        default_backend.store_result(uuid, "Quick", "PENDING")
+        json = self.client.get(is_successful(task_id=uuid))
+        self.assertJSONEquals(json, {"task": {"id": uuid,
+                                             "executed": False}})

celery/urls.py

@@ -9,8 +9,8 @@ from celery import views
 
 
 urlpatterns = patterns("",
-    url(r'^(?P<task_id>[\w\d\-]+)/done/?$', views.is_task_done,
-        name="celery-is_task_done"),
+    url(r'^(?P<task_id>[\w\d\-]+)/done/?$', views.is_task_successful,
+        name="celery-is_task_successful"),
     url(r'^(?P<task_id>[\w\d\-]+)/status/?$', views.task_status,
         name="celery-task_status"),
 )

testproj/urls.py

@@ -1,4 +1,4 @@
-from django.conf.urls.defaults import *
+from django.conf.urls.defaults import patterns, url, include
 
 # Uncomment the next two lines to enable the admin:
 # from django.contrib import admin
@@ -14,4 +14,5 @@ urlpatterns = patterns('',
 
     # Uncomment the next line to enable the admin:
     # (r'^admin/(.*)', admin.site.root),
+    url(r"^celery/", include("celery.urls")),
 )