Allow the shadow kwarg and the shadow_name method to set shadow properly (#4381)

* Allow the shadow kwarg and the shadow_name method to set shadow properly 

The shadow_name option in the @app.task() decorator (which overrides the shadow_name method in the Task class) and the shadow keyword argument of Task.apply_async() don't work as advertised.
This moves the shadow=... out of the 'if self.__self__ is not None:' block and allows shadow to be set by the shadow keyword argument of Task.apply_async() or the shadow_name method in the Task class (via, say, the shadow_name option in the @app.task() decorator).

* Added a test to cover calling shadow_name().

* Sort imports.

* Fix missing import.

celery/app/task.py

@@ -528,7 +528,7 @@ class Task(object):
         if self.__self__ is not None:
             args = args if isinstance(args, tuple) else tuple(args or ())
             args = (self.__self__,) + args
-            shadow = shadow or self.shadow_name(args, kwargs, options)
+        shadow = shadow or self.shadow_name(args, kwargs, options)
 
         preopts = self._get_exec_options()
         options = dict(preopts, **options) if options else preopts

t/unit/tasks/test_tasks.py

@@ -5,7 +5,7 @@ import tempfile
 from datetime import datetime, timedelta
 
 import pytest
-from case import ContextMock, MagicMock, Mock, patch
+from case import ANY, ContextMock, MagicMock, Mock, patch
 from kombu import Queue
 
 from celery import Task, group, uuid
@@ -358,6 +358,42 @@ class test_tasks(TasksCase):
 
         add.delay(2, 2)
 
+    def test_shadow_name(self):
+        def shadow_name(task, args, kwargs, options):
+            return 'fooxyz'
+
+        @self.app.task(shadow_name=shadow_name)
+        def shadowed():
+            pass
+
+        old_send_task = self.app.send_task
+        self.app.send_task = Mock()
+
+        shadowed.delay()
+
+        self.app.send_task.assert_called_once_with(ANY, ANY, ANY,
+                                                   compression=ANY,
+                                                   delivery_mode=ANY,
+                                                   exchange=ANY,
+                                                   expires=ANY,
+                                                   immediate=ANY,
+                                                   link=ANY,
+                                                   link_error=ANY,
+                                                   mandatory=ANY,
+                                                   priority=ANY,
+                                                   producer=ANY,
+                                                   queue=ANY,
+                                                   result_cls=ANY,
+                                                   routing_key=ANY,
+                                                   serializer=ANY,
+                                                   soft_time_limit=ANY,
+                                                   task_id=ANY,
+                                                   task_type=ANY,
+                                                   time_limit=ANY,
+                                                   shadow='fooxyz')
+
+        self.app.send_task = old_send_task
+
     def test_typing__disabled(self):
         @self.app.task(typing=False)
         def add(x, y, kw=1):