Use Redis coercion mechanism for converting query parameters (#4736)

* #4735 Use redis's corecion mechanism for converting query parameters to the correct types

* Line length

* Add a test for coercing the timeouts in the redis url

celery/backends/redis.py

@@ -29,6 +29,7 @@ except ImportError:
 
 try:
     import redis
+    import redis.connection
     from kombu.transport.redis import get_redis_error_classes
 except ImportError:  # pragma: no cover
     redis = None  # noqa
@@ -249,6 +250,12 @@ class RedisBackend(base.BaseKeyValueStoreBackend, async.AsyncBackendMixin):
         db = db.strip('/') if isinstance(db, string_t) else db
         connparams['db'] = int(db)
 
+        for key, value in query.items():
+            if key in redis.connection.URL_QUERY_ARGUMENT_PARSERS:
+                query[key] = redis.connection.URL_QUERY_ARGUMENT_PARSERS[key](
+                    value
+                )
+
         # Query parameters override other parameters
         connparams.update(query)
         return connparams

t/unit/backends/test_redis.py

@@ -234,6 +234,20 @@ class test_RedisBackend:
         assert x.connparams['socket_timeout'] == 30.0
         assert x.connparams['socket_connect_timeout'] == 100.0
 
+    def test_timeouts_in_url_coerced(self):
+        x = self.Backend(
+            ('redis://:bosco@vandelay.com:123//1?'
+             'socket_timeout=30&socket_connect_timeout=100'),
+            app=self.app,
+        )
+        assert x.connparams
+        assert x.connparams['host'] == 'vandelay.com'
+        assert x.connparams['db'] == 1
+        assert x.connparams['port'] == 123
+        assert x.connparams['password'] == 'bosco'
+        assert x.connparams['socket_timeout'] == 30
+        assert x.connparams['socket_connect_timeout'] == 100
+
     def test_socket_url(self):
         self.app.conf.redis_socket_timeout = 30.0
         self.app.conf.redis_socket_connect_timeout = 100.0