Tests passing

celery/tests/worker/test_mediator.py

@@ -112,21 +112,3 @@ class test_Mediator(Case):
         m.run()
         self.assertTrue(m._is_shutdown.isSet())
         self.assertTrue(m._is_stopped.isSet())
-
-    def test_mediator_body_revoked(self):
-        ready_queue = Queue()
-        got = {}
-
-        def mycallback(value):
-            got['value'] = value.value
-
-        m = Mediator(ready_queue, mycallback)
-        t = MockTask('Jerry Seinfeld')
-        t.id = uuid()
-        revoked_tasks.add(t.id)
-        ready_queue.put(t)
-
-        m.body()
-
-        self.assertNotIn('value', got)
-        self.assertTrue(t.on_ack.call_count)

celery/tests/worker/test_request.py

@@ -20,8 +20,12 @@ from celery import states
 from celery.app import app_or_default
 from celery.concurrency.base import BasePool
 from celery.datastructures import ExceptionInfo
-from celery.exceptions import (RetryTaskError,
-                               WorkerLostError, InvalidTaskError)
+from celery.exceptions import (
+    RetryTaskError,
+    WorkerLostError,
+    InvalidTaskError,
+    TaskRevokedError,
+)
 from celery.task.trace import (
     trace_task,
     trace_task_ret,
@@ -427,7 +431,8 @@ class test_TaskRequest(Case):
     def test_execute_using_pool_does_not_execute_revoked(self):
         tw = TaskRequest(mytask.name, uuid(), [1], {'f': 'x'})
         revoked.add(tw.id)
-        tw.execute_using_pool(None)
+        with self.assertRaises(TaskRevokedError):
+            tw.execute_using_pool(None)
 
     def test_on_accepted_acks_early(self):
         tw = TaskRequest(mytask.name, uuid(), [1], {'f': 'x'})