Test TaskWrapper on_acknowledgement.

celery/tests/test_worker_job.py

@@ -10,6 +10,12 @@ from celery.utils import gen_unique_id
 from carrot.backends.base import BaseMessage
 import simplejson
 
+scratch = {"ACK": False}
+
+
+def on_ack():
+    scratch["ACK"] = True
+
 
 def mytask(i, **kwargs):
     return i ** i
@@ -120,6 +126,16 @@ class TestTaskWrapper(unittest.TestCase):
         self.assertEquals(meta.result, 256)
         self.assertEquals(meta.status, "DONE")
 
+    def test_execute_ack(self):
+        tid = gen_unique_id()
+        tw = TaskWrapper("cu.mytask", tid, mytask, [4], {"f": "x"},
+                        on_acknowledge=on_ack)
+        self.assertEquals(tw.execute(), 256)
+        meta = TaskMeta.objects.get(task_id=tid)
+        self.assertTrue(scratch["ACK"])
+        self.assertEquals(meta.result, 256)
+        self.assertEquals(meta.status, "DONE")
+
     def test_execute_fail(self):
         tid = gen_unique_id()
         tw = TaskWrapper("cu.mytask-raising", tid, mytask_raising, [4],