add unit test and fix chain.apply (it ignored kwargs) (#4952)

celery/canvas.py

@@ -701,11 +701,11 @@ class _chain(Signature):
         return tasks, results
 
     def apply(self, args=(), kwargs={}, **options):
-        last, fargs = None, args
+        last, (fargs, fkwargs) = None, (args, kwargs)
         for task in self.tasks:
-            res = task.clone(fargs).apply(
+            res = task.clone(fargs, fkwargs).apply(
                 last and (last.get(),), **dict(self.options, **options))
-            res.parent, last, fargs = last, res, None
+            res.parent, last, (fargs, fkwargs) = last, res, (None, None)
         return last
 
     @property

t/unit/tasks/test_canvas.py

@@ -441,6 +441,11 @@ class test_chain(CanvasCase):
         assert res.parent.parent.get() == 8
         assert res.parent.parent.parent is None
 
+    def test_kwargs_apply(self):
+        x = chain(self.add.s(), self.add.s(8), self.add.s(10))
+        res = x.apply(kwargs={'x': 1, 'y': 1}).get()
+        assert res == 20
+
     def test_single_expresion(self):
         x = chain(self.add.s(1, 2)).apply()
         assert x.get() == 3