Skip to content
Permalink
Browse files
feat: add helper for bucket bound hostname URLs (#137)
Closes #121
  • Loading branch information
IlyaFaer committed May 28, 2020
1 parent e516ed9 commit b26f9fa8a767b7d5affea8d2c4b87163ce979fd2
@@ -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)
@@ -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
@@ -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)

@@ -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)

@@ -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
@@ -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)
@@ -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))

0 comments on commit b26f9fa

Please sign in to comment.