Added a default value for retries in worker.strategy. (celery#5945)

* Added a default value for retries in worker.strategy.

I was facing an issue when adding tasks directly to rabbitmq
using pika instead of calling task.apply_async. The issue was
the self.retry mechanisum was failing. In app/tasks.py the line
`retries = request.retries + 1` was causing the issue. On further
tracing I figured out that it was because the default .get value
(None) was getting passed through this function and was raising
TypeError: unsupported operand type(s) for +: 'NoneType' and 'int'

* Add test cases for default and custom retries value

celery/worker/strategy.py

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

t/unit/worker/test_strategy.py

@@ -13,7 +13,7 @@
 from celery.worker import state
 from celery.worker.request import Request
 from celery.worker.strategy import default as default_strategy
-from celery.worker.strategy import proto1_to_proto2
+from celery.worker.strategy import proto1_to_proto2, hybrid_to_proto2
 
 
 class test_proto1_to_proto2:
@@ -268,3 +268,25 @@ def failed():
             )
             task_message_handler(C.message, None, None, None, None)
             _MyRequest.assert_called()
+
+
+class test_hybrid_to_proto2:
+
+    def setup(self):
+        self.message = Mock(name='message')
+        self.body = {
+            'args': (1,),
+            'kwargs': {'foo': 'baz'},
+            'utc': False,
+            'taskset': '123',
+        }
+
+    def test_retries_default_value(self):
+        _, headers, _, _ = hybrid_to_proto2(self.message, self.body)
+        assert headers.get('retries') == 0
+
+    def test_retries_custom_value(self):
+        _custom_value = 3
+        self.body['retries'] = _custom_value
+        _, headers, _, _ = hybrid_to_proto2(self.message, self.body)
+        assert headers.get('retries') == _custom_value