sanitise url when include_password is false (fixes celery#1085) (celery#1095)

* sanitise url when include_password is false  (fixes celery#1085)

* fix flake8 error

Author: shekhar-kamble

kombu/connection.py

@@ -31,7 +31,7 @@
 from .utils.collections import HashedSeq
 from .utils.functional import dictfilter, lazy, retry_over_time, shufflecycle
 from .utils.objects import cached_property
-from .utils.url import as_url, parse_url, quote, urlparse
+from .utils.url import as_url, parse_url, quote, urlparse, maybe_sanitize_url
 
 __all__ = ('Connection', 'ConnectionPool', 'ChannelPool')
 
@@ -656,11 +656,17 @@ def as_uri(self, include_password=False, mask='**',
         """Convert connection parameters to URL form."""
         hostname = self.hostname or 'localhost'
         if self.transport.can_parse_url:
+            connection_as_uri = self.hostname
             if self.uri_prefix:
-                return '%s+%s' % (self.uri_prefix, hostname)
-            return self.hostname
+                connection_as_uri = '%s+%s' % (self.uri_prefix, hostname)
+            if not include_password:
+                connection_as_uri = maybe_sanitize_url(connection_as_uri)
+            return connection_as_uri
         if self.uri_prefix:
-            return '%s+%s' % (self.uri_prefix, hostname)
+            connection_as_uri = '%s+%s' % (self.uri_prefix, hostname)
+            if not include_password:
+                connection_as_uri = maybe_sanitize_url(connection_as_uri)
+            return connection_as_uri
         fields = self.info()
         port, userid, password, vhost, transport = getfields(fields)
 

t/unit/test_connection.py

@@ -30,6 +30,8 @@ def setup(self):
             'port': 5672,
             'virtual_host': 'my/vhost',
         }
+        self.pg_url = 'sqla+postgresql://test:password@yms-pg/yms'
+        self.pg_nopass = 'sqla+postgresql://test:**@yms-pg/yms'
 
     def test_parse_url(self):
         result = parse_url(self.url)
@@ -115,6 +117,11 @@ def test_connection_copy(self):
         clone = deepcopy(conn)
         assert clone.alt == ['amqp://host']
 
+    def test_parse_generated_as_uri_pg(self):
+        conn = Connection(self.pg_url)
+        assert conn.as_uri() == self.pg_nopass
+        assert conn.as_uri(include_password=True) == self.pg_url
+
 
 class test_Connection: