Removes Signature.route_name_for, fixing tests

celery/backends/base.py

@@ -351,9 +351,10 @@ class BaseBackend(object):
             (group_id, body, ), kwargs, countdown=countdown,
         )
 
-    def apply_chord(self, header, partial_args, group_id, body, **options):
-        result = header(*partial_args, task_id=group_id)
-        self.fallback_chord_unlock(group_id, body, **options)
+    def apply_chord(self, header, partial_args, group_id, body,
+                    options={}, **kwargs):
+        result = header(*partial_args, task_id=group_id, **options or {})
+        self.fallback_chord_unlock(group_id, body, **kwargs)
         return result
 
     def current_task_children(self, request=None):
@@ -516,9 +517,9 @@ class KeyValueStoreBackend(BaseBackend):
             return meta
 
     def _apply_chord_incr(self, header, partial_args, group_id, body,
-                          result=None, **options):
+                          result=None, options={}, **kwargs):
         self.save_group(group_id, self.app.GroupResult(group_id, result))
-        return header(*partial_args, task_id=group_id)
+        return header(*partial_args, task_id=group_id, **options or {})
 
     def on_chord_part_return(self, task, state, result, propagate=None):
         if not self.implements_incr:

celery/backends/redis.py

@@ -185,9 +185,9 @@ class RedisBackend(KeyValueStoreBackend):
         return retval
 
     def _new_chord_apply(self, header, partial_args, group_id, body,
-                         result=None, **options):
+                         result=None, options={}, **kwargs):
         # avoids saving the group in the redis db.
-        return header(*partial_args, task_id=group_id)
+        return header(*partial_args, task_id=group_id, **options or {})
 
     def _new_chord_return(self, task, state, result, propagate=None,
                           PROPAGATE_STATES=states.PROPAGATE_STATES):

celery/canvas.py

@@ -244,17 +244,7 @@ class Signature(dict):
             args, kwargs, options = self._merge(args, kwargs, options)
         else:
             args, kwargs, options = self.args, self.kwargs, self.options
-        route_name = route_name or self.route_name_for(args, kwargs, options)
-        return _apply(args, kwargs, route_name=route_name, **options)
-
-    def route_name_for(self, args, kwargs, options):
-        """Can be used to override the name used for routing the task
-        to a queue.
-
-        If this returns :const:`None` the name of the task will be used.
-
-        """
-        pass
+        return _apply(args, kwargs, **options)
 
     def append_to_list_option(self, key, value):
         items = self.options.setdefault(key, [])
@@ -504,15 +494,12 @@ class _basemap(Signature):
             {'task': task, 'it': regen(it)}, immutable=True, **options
         )
 
-    def route_name_for(self, args, kwargs, options):
-        return task_name_from(self.kwargs.get('task'))
-
     def apply_async(self, args=(), kwargs={}, **opts):
         # need to evaluate generators
         task, it = self._unpack_args(self.kwargs)
         return self.type.apply_async(
             (), {'task': task, 'it': list(it)},
-            route_name=self.route_name_for(args, kwargs, opts), **opts
+            route_name=task_name_from(self.kwargs.get('task')), **opts
         )
 
     @classmethod
@@ -555,13 +542,10 @@ class chunks(Signature):
     def from_dict(self, d, app=None):
         return chunks(*self._unpack_args(d['kwargs']), app=app, **d['options'])
 
-    def route_name_for(self, args, kwargs, options):
-        return task_name_from(self.kwargs.get('task'))
-
     def apply_async(self, args=(), kwargs={}, **opts):
         return self.group().apply_async(
             args, kwargs,
-            route_name=self.route_name_for(args, kwargs, opts), **opts
+            route_name=task_name_from(self.kwargs.get('task')), **opts
         )
 
     def __call__(self, **options):
@@ -833,6 +817,9 @@ class chord(Signature):
         root_id = body.options.get('root_id')
         if 'chord_size' not in body:
             body['chord_size'] = self.__length_hint__()
+        options = dict(self.options, **options) if options else self.options
+        if options:
+            body.options.update(options)
 
         results = header.freeze(
             group_id=group_id, chord=body, root_id=root_id).results
@@ -841,7 +828,8 @@ class chord(Signature):
         parent = app.backend.apply_chord(
             header, partial_args, group_id, body,
             interval=interval, countdown=countdown,
-            max_retries=max_retries, propagate=propagate, result=results)
+            options=options, max_retries=max_retries,
+            propagate=propagate, result=results)
         bodyres.parent = parent
         return bodyres
 

celery/tests/app/test_builtins.py

@@ -89,7 +89,6 @@ class test_group(BuiltinsCase):
 
     def test_apply(self):
         x = group([self.add.s(4, 4), self.add.s(8, 8)])
-        x.name = self.task.name
         res = x.apply()
         self.assertEqual(res.get(), [8, 16])
 

celery/tests/tasks/test_canvas.py

@@ -171,6 +171,7 @@ class test_xmap_xstarmap(CanvasCase):
             s.apply_async(foo=1)
             s.type.apply_async.assert_called_with(
                 (), {'task': self.add.s(), 'it': args}, foo=1,
+                route_name=self.add.name,
             )
 
             self.assertEqual(type.from_dict(dict(s)), s)
@@ -192,10 +193,10 @@ class test_chunks(CanvasCase):
         gr = x.group.return_value = Mock()
 
         x.apply_async()
-        gr.apply_async.assert_called_with((), {})
-
+        gr.apply_async.assert_called_with((), {}, route_name=self.add.name)
+        gr.apply_async.reset_mock()
         x()
-        gr.assert_called_with()
+        gr.apply_async.assert_called_with((), {}, route_name=self.add.name)
 
         self.app.conf.CELERY_ALWAYS_EAGER = True
         chunks.apply_chunks(app=self.app, **x['kwargs'])