Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Decouple connection from Blob #823

Merged
merged 5 commits into from
Apr 16, 2015
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Removing using of Blob.connection in generate_signed_url.
Allowing the default connection to be used or direct use of
a credentials object.
  • Loading branch information
dhermes committed Apr 16, 2015
commit e3ccb8deb6dc039a5ab24685ab603885931b8812
19 changes: 17 additions & 2 deletions gcloud/storage/blob.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
from gcloud.credentials import generate_signed_url
from gcloud.exceptions import NotFound
from gcloud.storage._helpers import _PropertyMixin
from gcloud.storage._helpers import _require_connection
from gcloud.storage._helpers import _scalar_property
from gcloud.storage import _implicit_environ
from gcloud.storage.acl import ObjectACL
Expand Down Expand Up @@ -164,7 +165,8 @@ def public_url(self):
bucket_name=self.bucket.name,
quoted_name=quote(self.name, safe=''))

def generate_signed_url(self, expiration, method='GET'):
def generate_signed_url(self, expiration, method='GET',
connection=None, credentials=None):
"""Generates a signed URL for this blob.

If you have a blob that you want to allow access to for a set
Expand All @@ -181,6 +183,15 @@ def generate_signed_url(self, expiration, method='GET'):
:type method: string
:param method: The HTTP verb that will be used when requesting the URL.

:type connection: :class:`gcloud.storage.connection.Connection` or
``NoneType``
:param connection: Optional. The connection to use when sending
requests. If not provided, falls back to default.

:type credentials: :class:`oauth2client.client.OAuth2Credentials` or
:class:`NoneType`
:param credentials: The OAuth2 credentials to use to sign the URL.

:rtype: string
:returns: A signed URL you can use to access the resource
until expiration.
Expand All @@ -189,8 +200,12 @@ def generate_signed_url(self, expiration, method='GET'):
bucket_name=self.bucket.name,
quoted_name=quote(self.name, safe=''))

if credentials is None:
connection = _require_connection(connection)
credentials = connection.credentials

return generate_signed_url(
self.connection.credentials, resource=resource,
credentials, resource=resource,
api_access_endpoint=_API_ACCESS_ENDPOINT,
expiration=expiration, method=method)

Expand Down
48 changes: 38 additions & 10 deletions gcloud/storage/test_blob.py
Original file line number Diff line number Diff line change
Expand Up @@ -155,24 +155,42 @@ def test_public_url_w_slash_in_name(self):
blob.public_url,
'http://commondatastorage.googleapis.com/name/parent%2Fchild')

def test_generate_signed_url_w_default_method(self):
def _basic_generate_signed_url_helper(self, credentials=None):
from gcloud._testing import _Monkey
from gcloud.storage import blob as MUT

BLOB_NAME = 'blob-name'
EXPIRATION = '2014-10-16T20:34:37.000Z'
connection = _Connection()
bucket = _Bucket(connection)
bucket = _Bucket(None)
blob = self._makeOne(BLOB_NAME, bucket=bucket)
URI = ('http://example.com/abucket/a-blob-name?Signature=DEADBEEF'
'&Expiration=2014-10-16T20:34:37.000Z')

_called_require = []

def mock_require(connection):
_called_require.append(connection)
return connection

SIGNER = _Signer()
with _Monkey(MUT, generate_signed_url=SIGNER):
self.assertEqual(blob.generate_signed_url(EXPIRATION), URI)
with _Monkey(MUT, generate_signed_url=SIGNER,
_require_connection=mock_require):
signed_uri = blob.generate_signed_url(EXPIRATION,
connection=connection,
credentials=credentials)
self.assertEqual(signed_uri, URI)

if credentials is None:
self.assertEqual(_called_require, [connection])
else:
self.assertEqual(_called_require, [])

PATH = '/name/%s' % (BLOB_NAME,)
EXPECTED_ARGS = (_Connection.credentials,)
if credentials is None:
EXPECTED_ARGS = (_Connection.credentials,)
else:
EXPECTED_ARGS = (credentials,)
EXPECTED_KWARGS = {
'api_access_endpoint': 'https://storage.googleapis.com',
'expiration': EXPIRATION,
Expand All @@ -181,21 +199,30 @@ def test_generate_signed_url_w_default_method(self):
}
self.assertEqual(SIGNER._signed, [(EXPECTED_ARGS, EXPECTED_KWARGS)])

def test_generate_signed_url_w_default_method(self):
self._basic_generate_signed_url_helper()

def test_generate_signed_url_w_credentials(self):
credentials = object()
self._basic_generate_signed_url_helper(credentials=credentials)

def test_generate_signed_url_w_slash_in_name(self):
from gcloud._testing import _Monkey
from gcloud.storage import blob as MUT

BLOB_NAME = 'parent/child'
EXPIRATION = '2014-10-16T20:34:37.000Z'
connection = _Connection()
bucket = _Bucket(connection)
bucket = _Bucket(None)
blob = self._makeOne(BLOB_NAME, bucket=bucket)
URI = ('http://example.com/abucket/a-blob-name?Signature=DEADBEEF'
'&Expiration=2014-10-16T20:34:37.000Z')

SIGNER = _Signer()
with _Monkey(MUT, generate_signed_url=SIGNER):
self.assertEqual(blob.generate_signed_url(EXPIRATION), URI)
signed_url = blob.generate_signed_url(EXPIRATION,
connection=connection)
self.assertEqual(signed_url, URI)

EXPECTED_ARGS = (_Connection.credentials,)
EXPECTED_KWARGS = {
Expand All @@ -213,15 +240,16 @@ def test_generate_signed_url_w_explicit_method(self):
BLOB_NAME = 'blob-name'
EXPIRATION = '2014-10-16T20:34:37.000Z'
connection = _Connection()
bucket = _Bucket(connection)
bucket = _Bucket(None)
blob = self._makeOne(BLOB_NAME, bucket=bucket)
URI = ('http://example.com/abucket/a-blob-name?Signature=DEADBEEF'
'&Expiration=2014-10-16T20:34:37.000Z')

SIGNER = _Signer()
with _Monkey(MUT, generate_signed_url=SIGNER):
self.assertEqual(
blob.generate_signed_url(EXPIRATION, method='POST'), URI)
signed_uri = blob.generate_signed_url(EXPIRATION, method='POST',
connection=connection)
self.assertEqual(signed_uri, URI)

PATH = '/name/%s' % (BLOB_NAME,)
EXPECTED_ARGS = (_Connection.credentials,)
Expand Down