Tests passing

celery/app/amqp.py

@@ -229,7 +229,7 @@ class TaskPublisher(messaging.Publisher):
         send(body, exchange=exchange, mandatory=mandatory,
              immediate=immediate, routing_key=routing_key,
              serializer=serializer or self.serializer,
-             delivery_mode=delivery_mode or self.delivery_mode,
+             delivery_mode=delivery_mode,
              compression=compression or self.compression)
         signals.task_sent.send(sender=task_name, **body)
         if event_dispatcher:

celery/app/builtins.py

@@ -120,8 +120,8 @@ def add_group_task(app):
             return result
 
         def apply(self, args=(), kwargs={}, **options):
-            tasks, result = self.prepare(options, **kwargs)
-            return super(Group, self).apply((tasks, result), **options)
+            tasks, result, gid = self.prepare(options, **kwargs)
+            return super(Group, self).apply((tasks, result, gid), **options)
 
     return Group
 

celery/tests/app/test_amqp.py

@@ -3,22 +3,9 @@ from __future__ import with_statement
 
 from mock import Mock
 
-from celery.app.amqp import MSG_OPTIONS, extract_msg_options
 from celery.tests.utils import AppCase
 
 
-class test_extract_message_options(AppCase):
-
-    def test_MSG_OPTIONS(self):
-        self.assertTrue(MSG_OPTIONS)
-
-    def test_extract_msg_options(self):
-        testing = {"mandatory": True, "routing_key": "foo.xuzzy"}
-        result = extract_msg_options(testing)
-        self.assertEqual(result["mandatory"], True)
-        self.assertEqual(result["routing_key"], "foo.xuzzy")
-
-
 class test_TaskPublisher(AppCase):
 
     def test__exit__(self):

celery/tests/worker/test_autoscale.py

@@ -67,20 +67,20 @@ class test_Autoscaler(Case):
         self.assertFalse(x.joined)
 
     @sleepdeprived(autoscale)
-    def test_scale(self):
+    def test_body(self):
         x = autoscale.Autoscaler(self.pool, 10, 3)
-        x.scale()
+        x.body()
         self.assertEqual(x.pool.num_processes, 3)
         for i in range(20):
             state.reserved_requests.add(i)
-        x.scale()
-        x.scale()
+        x.body()
+        x.body()
         self.assertEqual(x.pool.num_processes, 10)
         state.reserved_requests.clear()
-        x.scale()
+        x.body()
         self.assertEqual(x.pool.num_processes, 10)
         x._last_action = time() - 10000
-        x.scale()
+        x.body()
         self.assertEqual(x.pool.num_processes, 3)
 
     def test_run(self):

celery/tests/worker/test_worker.py

@@ -22,7 +22,7 @@ from celery.exceptions import SystemTerminate
 from celery.task import task as task_dec
 from celery.task import periodic_task as periodic_task_dec
 from celery.utils import uuid
-from celery.worker import WorkController, Queues
+from celery.worker import WorkController, Queues, Timers
 from celery.worker.buckets import FastQueue
 from celery.worker.job import Request
 from celery.worker.consumer import Consumer as MainConsumer
@@ -805,7 +805,9 @@ class test_WorkController(AppCase):
         signals.worker_process_init.connect(on_worker_process_init)
 
         loader = Mock()
+        loader.override_backends = {}
         app = Celery(loader=loader, set_as_current=False)
+        app.loader = loader
         app.conf = AttributeDict(DEFAULTS)
         process_initializer(app, "awesome.worker.com")
         _signals.ignore.assert_any_call(*WORKER_SIGIGNORE)
@@ -865,14 +867,14 @@ class test_WorkController(AppCase):
         except KeyError:
             exc_info = sys.exc_info()
 
-        worker.on_timer_error(exc_info)
+        Timers(worker).on_timer_error(exc_info)
         msg, args = self.logger.error.call_args[0]
         self.assertIn("KeyError", msg % args)
 
     def test_on_timer_tick(self):
         worker = WorkController(concurrency=1, loglevel=10)
 
-        worker.on_timer_tick(30.0)
+        Timers(worker).on_timer_tick(30.0)
         xargs = self.logger.debug.call_args[0]
         fmt, arg = xargs[0], xargs[1]
         self.assertEqual(30.0, arg)