feat: add helper for bucket bound hostname URLs

google/cloud/storage/_helpers.py

@@ -22,6 +22,7 @@
 from datetime import datetime
 import os
 
+from six.moves.urllib.parse import urlsplit
 from google.cloud.storage.constants import _DEFAULT_TIMEOUT
 
 
@@ -491,3 +492,23 @@ def _raise_if_more_than_one_set(**kwargs):
         )
 
         raise ValueError(msg)
+
+
+def _bucket_bound_hostname_url(host, scheme=None):
+    """Helper to build bucket bound hostname URL.
+
+    :type host: str
+    :param host: Host name.
+
+    :type scheme: str
+    :param scheme: (Optional) Web scheme. If passed, use it
+                   as a scheme in the result URL.
+
+    :rtype: str
+    :returns: A bucket bound hostname URL.
+    """
+    url_parts = urlsplit(host)
+    if url_parts.scheme and url_parts.netloc:
+        return host
+
+    return "{scheme}://{host}/".format(scheme=scheme, host=host)

google/cloud/storage/blob.py

@@ -57,6 +57,7 @@
 from google.cloud.storage._helpers import _add_generation_match_parameters
 from google.cloud.storage._helpers import _PropertyMixin
 from google.cloud.storage._helpers import _scalar_property
+from google.cloud.storage._helpers import _bucket_bound_hostname_url
 from google.cloud.storage._helpers import _convert_to_timestamp
 from google.cloud.storage._helpers import _raise_if_more_than_one_set
 from google.cloud.storage._signing import generate_signed_url_v2
@@ -516,12 +517,9 @@ def generate_signed_url(
                 bucket_name=self.bucket.name
             )
         elif bucket_bound_hostname:
-            if ":" in bucket_bound_hostname:
-                api_access_endpoint = bucket_bound_hostname
-            else:
-                api_access_endpoint = "{scheme}://{bucket_bound_hostname}".format(
-                    scheme=scheme, bucket_bound_hostname=bucket_bound_hostname
-                )
+            api_access_endpoint = _bucket_bound_hostname_url(
+                bucket_bound_hostname, scheme
+            )
         else:
             resource = "/{bucket_name}/{quoted_name}".format(
                 bucket_name=self.bucket.name, quoted_name=quoted_name

google/cloud/storage/bucket.py

@@ -38,6 +38,7 @@
 from google.cloud.storage._helpers import _validate_name
 from google.cloud.storage._signing import generate_signed_url_v2
 from google.cloud.storage._signing import generate_signed_url_v4
+from google.cloud.storage._helpers import _bucket_bound_hostname_url
 from google.cloud.storage.acl import BucketACL
 from google.cloud.storage.acl import DefaultObjectACL
 from google.cloud.storage.blob import Blob
@@ -2861,12 +2862,9 @@ def generate_signed_url(
                 bucket_name=self.name
             )
         elif bucket_bound_hostname:
-            if ":" in bucket_bound_hostname:
-                api_access_endpoint = bucket_bound_hostname
-            else:
-                api_access_endpoint = "{scheme}://{bucket_bound_hostname}".format(
-                    scheme=scheme, bucket_bound_hostname=bucket_bound_hostname
-                )
+            api_access_endpoint = _bucket_bound_hostname_url(
+                bucket_bound_hostname, scheme
+            )
         else:
             resource = "/{bucket_name}".format(bucket_name=self.name)
 

google/cloud/storage/client.py

@@ -30,6 +30,7 @@
 from google.cloud.client import ClientWithProject
 from google.cloud.exceptions import NotFound
 from google.cloud.storage._helpers import _get_storage_host
+from google.cloud.storage._helpers import _bucket_bound_hostname_url
 from google.cloud.storage._http import Connection
 from google.cloud.storage._signing import (
     get_expiration_seconds_v4,
@@ -1079,15 +1080,8 @@ def generate_signed_post_policy_v4(
         # designate URL
         if virtual_hosted_style:
             url = "https://{}.storage.googleapis.com/".format(bucket_name)
-
         elif bucket_bound_hostname:
-            if ":" in bucket_bound_hostname:  # URL includes scheme
-                url = bucket_bound_hostname
-
-            else:  # scheme is given separately
-                url = "{scheme}://{host}/".format(
-                    scheme=scheme, host=bucket_bound_hostname
-                )
+            url = _bucket_bound_hostname_url(bucket_bound_hostname, scheme)
         else:
             url = "https://storage.googleapis.com/{}/".format(bucket_name)
 

tests/unit/test__helpers.py

@@ -525,6 +525,24 @@ def test_add_generation_match_parameters_tuple(self):
             )
 
 
+class Test__bucket_bound_hostname_url(unittest.TestCase):
+    def _call_fut(self, **args):
+        from google.cloud.storage._helpers import _bucket_bound_hostname_url
+
+        return _bucket_bound_hostname_url(**args)
+
+    def test_full_hostname(self):
+        HOST = "scheme://domain.tcl/"
+        self.assertEqual(self._call_fut(host=HOST), HOST)
+
+    def test_hostname_and_scheme(self):
+        HOST = "domain.tcl"
+        SCHEME = "scheme"
+        EXPECTED_URL = SCHEME + "://" + HOST + "/"
+
+        self.assertEqual(self._call_fut(host=HOST, scheme=SCHEME), EXPECTED_URL)
+
+
 class _Connection(object):
     def __init__(self, *responses):
         self._responses = responses

tests/unit/test_blob.py

@@ -405,6 +405,7 @@ def _generate_signed_url_helper(
     ):
         from six.moves.urllib import parse
         from google.cloud._helpers import UTC
+        from google.cloud.storage._helpers import _bucket_bound_hostname_url
         from google.cloud.storage.blob import _API_ACCESS_ENDPOINT
         from google.cloud.storage.blob import _get_encryption_headers
 
@@ -464,12 +465,9 @@ def _generate_signed_url_helper(
                 bucket.name
             )
         elif bucket_bound_hostname:
-            if ":" in bucket_bound_hostname:
-                expected_api_access_endpoint = bucket_bound_hostname
-            else:
-                expected_api_access_endpoint = "{scheme}://{bucket_bound_hostname}".format(
-                    scheme=scheme, bucket_bound_hostname=bucket_bound_hostname
-                )
+            expected_api_access_endpoint = _bucket_bound_hostname_url(
+                bucket_bound_hostname, scheme
+            )
         else:
             expected_api_access_endpoint = api_access_endpoint
             expected_resource = "/{}/{}".format(bucket.name, quoted_name)

tests/unit/test_bucket.py

@@ -2990,6 +2990,7 @@ def _generate_signed_url_helper(
     ):
         from six.moves.urllib import parse
         from google.cloud._helpers import UTC
+        from google.cloud.storage._helpers import _bucket_bound_hostname_url
         from google.cloud.storage.blob import _API_ACCESS_ENDPOINT
 
         api_access_endpoint = api_access_endpoint or _API_ACCESS_ENDPOINT
@@ -3037,12 +3038,9 @@ def _generate_signed_url_helper(
                 bucket_name
             )
         elif bucket_bound_hostname:
-            if ":" in bucket_bound_hostname:
-                expected_api_access_endpoint = bucket_bound_hostname
-            else:
-                expected_api_access_endpoint = "{scheme}://{bucket_bound_hostname}".format(
-                    scheme=scheme, bucket_bound_hostname=bucket_bound_hostname
-                )
+            expected_api_access_endpoint = _bucket_bound_hostname_url(
+                bucket_bound_hostname, scheme
+            )
         else:
             expected_api_access_endpoint = api_access_endpoint
             expected_resource = "/{}".format(parse.quote(bucket_name))