Fix hybrid_to_proto2 with missing timelimit (#4850)

* Fix hybrid_to_proto2 with missing timelimit

If `timelimit` is not defined in `body`, it will default to `None` value. Which will cause result in a crash here:
https://github.com/celery/celery/blob/master/celery/worker/request.py#L188 (`'NoneType' object is not iterable`).

Defaulting to `(None, None)` instead should fix it.

* add testcase

* flake8 and isort

celery/worker/strategy.py

@@ -48,7 +48,7 @@ def hybrid_to_proto2(message, body):
         'eta': body.get('eta'),
         'expires': body.get('expires'),
         'retries': body.get('retries'),
-        'timelimit': body.get('timelimit'),
+        'timelimit': body.get('timelimit', (None, None)),
         'argsrepr': body.get('argsrepr'),
         'kwargsrepr': body.get('kwargsrepr'),
         'origin': body.get('origin'),

t/unit/worker/test_request.py

@@ -25,6 +25,7 @@ from celery.exceptions import (Ignore, InvalidTaskError, Reject, Retry,
 from celery.five import monotonic
 from celery.signals import task_revoked
 from celery.worker import request as module
+from celery.worker import strategy
 from celery.worker.request import Request, create_request_cls
 from celery.worker.request import logger as req_logger
 from celery.worker.state import revoked
@@ -1006,3 +1007,13 @@ class test_create_request_class(RequestCase):
         assert job._apply_result
         weakref_ref.assert_called_with(self.pool.apply_async())
         assert job._apply_result is weakref_ref()
+
+    def test_execute_using_pool__defaults_of_hybrid_to_proto2(self):
+        weakref_ref = Mock(name='weakref.ref')
+        headers = strategy.hybrid_to_proto2('', {'id': uuid(),
+                                                 'task': self.mytask.name})[1]
+        job = self.zRequest(revoked_tasks=set(), ref=weakref_ref, **headers)
+        job.execute_using_pool(self.pool)
+        assert job._apply_result
+        weakref_ref.assert_called_with(self.pool.apply_async())
+        assert job._apply_result is weakref_ref()