Fix eager tasks does not populate name field (#8486)

* Add task name to eager request

* Add task name to eager result

* Add tests

* Add an extra check to make sure name is populated in EagerResults

celery/app/task.py

@@ -788,6 +788,7 @@ def apply(self, args=None, kwargs=None,
 
         request = {
             'id': task_id,
+            'task': self.name,
             'retries': retries,
             'is_eager': True,
             'logfile': logfile,
@@ -824,7 +825,7 @@ def apply(self, args=None, kwargs=None,
         if isinstance(retval, Retry) and retval.sig is not None:
             return retval.sig.apply(retries=retries + 1)
         state = states.SUCCESS if ret.info is None else ret.info.state
-        return EagerResult(task_id, retval, state, traceback=tb)
+        return EagerResult(task_id, retval, state, traceback=tb, name=self.name)
 
     def AsyncResult(self, task_id, **kwargs):
         """Get AsyncResult instance for the specified task.

celery/result.py

@@ -984,13 +984,14 @@ def restore(cls, id, backend=None, app=None):
 class EagerResult(AsyncResult):
     """Result that we know has already been executed."""
 
-    def __init__(self, id, ret_value, state, traceback=None):
+    def __init__(self, id, ret_value, state, traceback=None, name=None):
         # pylint: disable=super-init-not-called
         # XXX should really not be inheriting from AsyncResult
         self.id = id
         self._result = ret_value
         self._state = state
         self._traceback = traceback
+        self._name = name
         self.on_ready = promise()
         self.on_ready(self)
 
@@ -1043,6 +1044,7 @@ def _cache(self):
             'result': self._result,
             'status': self._state,
             'traceback': self._traceback,
+            'name': self._name,
         }
 
     @property

t/unit/tasks/test_result.py

@@ -967,6 +967,13 @@ def test_get_sync_subtask_option(self, task_join_will_block):
             res_subtask_async.get()
         res_subtask_async.get(disable_sync_subtasks=False)
 
+    def test_populate_name(self):
+        res = EagerResult('x', 'x', states.SUCCESS, None, 'test_task')
+        assert res.name == 'test_task'
+
+        res = EagerResult('x', 'x', states.SUCCESS, name='test_task_named_argument')
+        assert res.name == 'test_task_named_argument'
+
 
 class test_tuples:
 

t/unit/tasks/test_tasks.py

@@ -1441,6 +1441,7 @@ def test_apply(self):
 
         assert e.successful()
         assert e.ready()
+        assert e.name == 't.unit.tasks.test_tasks.increment_counter'
         assert repr(e).startswith('<EagerResult:')
 
         f = self.raising.apply()
@@ -1450,6 +1451,21 @@ def test_apply(self):
         with pytest.raises(KeyError):
             f.get()
 
+    def test_apply_eager_populates_request_task(self):
+        task_to_apply = self.task_check_request_context
+        with patch.object(
+            task_to_apply.request_stack, "push",
+            wraps=task_to_apply.request_stack.push,
+        ) as mock_push:
+            task_to_apply.apply()
+
+        mock_push.assert_called_once()
+
+        request = mock_push.call_args[0][0]
+
+        assert request.is_eager is True
+        assert request.task == 't.unit.tasks.test_tasks.task_check_request_context'
+
     def test_apply_simulates_delivery_info(self):
         task_to_apply = self.task_check_request_context
         with patch.object(