diff --git a/storage/google/cloud/storage/blob.py b/storage/google/cloud/storage/blob.py index 0e726c42f0f2..9c89c52b9e24 100644 --- a/storage/google/cloud/storage/blob.py +++ b/storage/google/cloud/storage/blob.py @@ -298,7 +298,7 @@ def public_url(self): return "{storage_base_url}/{bucket_name}/{quoted_name}".format( storage_base_url=_API_ACCESS_ENDPOINT, bucket_name=self.bucket.name, - quoted_name=quote(self.name.encode("utf-8")), + quoted_name=_quote(self.name, safe=b"/~"), ) def generate_signed_url( @@ -416,8 +416,9 @@ def generate_signed_url( elif version not in ("v2", "v4"): raise ValueError("'version' must be either 'v2' or 'v4'") + quoted_name = _quote(self.name, safe=b"/~") resource = "/{bucket_name}/{quoted_name}".format( - bucket_name=self.bucket.name, quoted_name=quote(self.name.encode("utf-8")) + bucket_name=self.bucket.name, quoted_name=quoted_name ) if credentials is None: @@ -1983,7 +1984,7 @@ def _get_encryption_headers(key, source=False): } -def _quote(value): +def _quote(value, safe=b"~"): """URL-quote a string. If the value is unicode, this method first UTF-8 encodes it as bytes and @@ -1994,11 +1995,14 @@ def _quote(value): :type value: str or bytes :param value: The value to be URL-quoted. + :type safe: bytes + :param safe: Bytes *not* to be quoted. By default, includes only ``b'~'``. + :rtype: str :returns: The encoded value (bytes in Python 2, unicode in Python 3). """ value = _to_bytes(value, encoding="utf-8") - return quote(value, safe="") + return quote(value, safe=safe) def _maybe_rewind(stream, rewind=False): diff --git a/storage/tests/unit/test_blob.py b/storage/tests/unit/test_blob.py index aeb38b428f05..b264ddcb8acf 100644 --- a/storage/tests/unit/test_blob.py +++ b/storage/tests/unit/test_blob.py @@ -344,6 +344,14 @@ def test_public_url_w_slash_in_name(self): blob.public_url, "https://storage.googleapis.com/name/parent/child" ) + def test_public_url_w_tilde_in_name(self): + BLOB_NAME = "foo~bar" + bucket = _Bucket() + blob = self._make_one(BLOB_NAME, bucket=bucket) + self.assertEqual( + blob.public_url, "https://storage.googleapis.com/name/foo~bar" + ) + def test_public_url_with_non_ascii(self): blob_name = u"winter \N{snowman}" bucket = _Bucket() @@ -426,7 +434,7 @@ def _generate_signed_url_helper( expected_creds = credentials encoded_name = blob_name.encode("utf-8") - expected_resource = "/name/{}".format(parse.quote(encoded_name)) + expected_resource = "/name/{}".format(parse.quote(encoded_name, safe=b"/~")) expected_kwargs = { "resource": expected_resource, "expiration": expiration, @@ -466,6 +474,10 @@ def test_generate_signed_url_v2_w_slash_in_name(self): BLOB_NAME = "parent/child" self._generate_signed_url_v2_helper(blob_name=BLOB_NAME) + def test_generate_signed_url_v2_w_tilde_in_name(self): + BLOB_NAME = "foo~bar" + self._generate_signed_url_v2_helper(blob_name=BLOB_NAME) + def test_generate_signed_url_v2_w_endpoint(self): self._generate_signed_url_v2_helper( api_access_endpoint="https://api.example.com/v1" @@ -514,6 +526,10 @@ def test_generate_signed_url_v4_w_slash_in_name(self): BLOB_NAME = "parent/child" self._generate_signed_url_v4_helper(blob_name=BLOB_NAME) + def test_generate_signed_url_v4_w_tilde_in_name(self): + BLOB_NAME = "foo~bar" + self._generate_signed_url_v4_helper(blob_name=BLOB_NAME) + def test_generate_signed_url_v4_w_endpoint(self): self._generate_signed_url_v4_helper( api_access_endpoint="https://api.example.com/v1" @@ -3119,10 +3135,10 @@ def test_updated_unset(self): class Test__quote(unittest.TestCase): @staticmethod - def _call_fut(value): + def _call_fut(*args, **kw): from google.cloud.storage.blob import _quote - return _quote(value) + return _quote(*args, **kw) def test_bytes(self): quoted = self._call_fut(b"\xDE\xAD\xBE\xEF") @@ -3137,6 +3153,21 @@ def test_bad_type(self): with self.assertRaises(TypeError): self._call_fut(None) + def test_w_slash_default(self): + with_slash = "foo/bar/baz" + quoted = self._call_fut(with_slash) + self.assertEqual(quoted, "foo%2Fbar%2Fbaz") + + def test_w_slash_w_safe(self): + with_slash = "foo/bar/baz" + quoted_safe = self._call_fut(with_slash, safe=b"/") + self.assertEqual(quoted_safe, with_slash) + + def test_w_tilde(self): + with_tilde = "bam~qux" + quoted = self._call_fut(with_tilde, safe=b"~") + self.assertEqual(quoted, with_tilde) + class Test__maybe_rewind(unittest.TestCase): @staticmethod