Tests passing on 2.5

celery/app/__init__.py

@@ -90,7 +90,8 @@ class App(base.BaseApp):
     def TaskSet(self, *args, **kwargs):
         """Create new :class:`~celery.task.sets.TaskSet`."""
         from celery.task.sets import TaskSet
-        return TaskSet(*args, app=self, **kwargs)
+        kwargs["app"] = self
+        return TaskSet(*args, **kwargs)
 
     def worker_main(self, argv=None):
         """Run :program:`celeryd` using `argv`.  Uses :data:`sys.argv`

celery/tests/test_worker/test_worker.py

@@ -86,11 +86,13 @@ def foo_periodic_task():
     return "foo"
 
 
-def create_message(backend, **data):
+def create_message(channel, **data):
     data.setdefault("id", gen_unique_id())
-    return Message(backend, body=pickle.dumps(dict(**data)),
+    channel.no_ack_consumers = set()
+    return Message(channel, body=pickle.dumps(dict(**data)),
                    content_type="application/x-python-serialize",
-                   content_encoding="binary")
+                   content_encoding="binary",
+                   delivery_info={"consumer_tag": "mock"})
 
 
 class test_QoS(unittest.TestCase):

celery/tests/test_worker/test_worker_control.py

@@ -29,6 +29,10 @@ def mytask():
     pass
 
 
+class WorkController(object):
+    autoscaler = None
+
+
 class Consumer(object):
 
     def __init__(self):
@@ -40,6 +44,7 @@ class Consumer(object):
         self.eta_schedule = Timer()
         self.app = current_app
         self.event_dispatcher = Mock()
+        self.controller = WorkController()
 
         from celery.concurrency.base import BasePool
         self.pool = BasePool(10)

celery/worker/autoscale.py

@@ -1,3 +1,5 @@
+from __future__ import absolute_import, with_statement
+
 import os
 import sys
 import threading