diff --git a/gcloud/storage/acl.py b/gcloud/storage/acl.py index a2eaf01b2f97..84e3c7911876 100644 --- a/gcloud/storage/acl.py +++ b/gcloud/storage/acl.py @@ -78,6 +78,8 @@ when sending metadata for ACLs to the API. """ +from gcloud.storage._helpers import _require_connection + class _ACLEntity(object): """Class representing a set of roles for an entity. @@ -167,8 +169,14 @@ def revoke_owner(self): class ACL(object): """Container class representing a list of access controls.""" + _URL_PATH_ELEM = 'acl' loaded = False + # Subclasses must override to provide these attributes (typically, + # as properties). + reload_path = None + save_path = None + def __init__(self): self.entities = {} @@ -345,77 +353,33 @@ def get_entities(self): self._ensure_loaded() return list(self.entities.values()) - def reload(self): + def reload(self, connection=None): """Reload the ACL data from Cloud Storage. - This is a virtual method, expected to be implemented by subclasses. - - :raises: :class:`NotImplementedError` - """ - raise NotImplementedError - - def save(self, acl=None): - """A method to be overridden by subclasses. - - :type acl: :class:`gcloud.storage.acl.ACL`, or a compatible list. - :param acl: The ACL object to save. If left blank, this will save - current entries. - - :raises: NotImplementedError - """ - raise NotImplementedError - - def clear(self): - """Remove all entities from the ACL.""" - raise NotImplementedError - - -class BucketACL(ACL): - """An ACL specifically for a bucket.""" - - _URL_PATH_ELEM = 'acl' - - def __init__(self, bucket): + :type connection: :class:`gcloud.storage.connection.Connection` or None + :param connection: explicit connection to use for API request; + defaults to instance property. """ - :type bucket: :class:`gcloud.storage.bucket.Bucket` - :param bucket: The bucket to which this ACL relates. - """ - super(BucketACL, self).__init__() - self.bucket = bucket + path = self.reload_path + connection = _require_connection(connection) - def reload(self): - """Reload the ACL data from Cloud Storage.""" self.entities.clear() - url_path = '%s/%s' % (self.bucket.path, self._URL_PATH_ELEM) - found = self.bucket.connection.api_request(method='GET', path=url_path) + found = connection.api_request(method='GET', path=path) self.loaded = True for entry in found.get('items', ()): self.add_entity(self.entity_from_dict(entry)) - def save(self, acl=None): + def save(self, acl=None, connection=None): """Save this ACL for the current bucket. - If called without arguments, this will save the entries - currently stored on this ACL:: - - >>> acl.save() - - You can also provide a specific ACL to save instead of the one - currently set on the Bucket object:: - - >>> acl.save(acl=my_other_acl) - - You can use this to set access controls to be consistent from - one bucket to another:: - - >>> bucket1 = storage.get_bucket(bucket1_name, connection=connection) - >>> bucket2 = storage.get_bucket(bucket2_name, connection=connection) - >>> bucket2.acl.save(bucket1.acl) - :type acl: :class:`gcloud.storage.acl.ACL`, or a compatible list. :param acl: The ACL object to save. If left blank, this will save current entries. + + :type connection: :class:`gcloud.storage.connection.Connection` or None + :param connection: explicit connection to use for API request; + defaults to instance property. """ if acl is None: acl = self @@ -424,8 +388,11 @@ def save(self, acl=None): save_to_backend = True if save_to_backend: - result = self.bucket.connection.api_request( - method='PATCH', path=self.bucket.path, + path = self.save_path + connection = _require_connection(connection) + result = connection.api_request( + method='PATCH', + path=path, data={self._URL_PATH_ELEM: list(acl)}, query_params={'projection': 'full'}) self.entities.clear() @@ -433,7 +400,7 @@ def save(self, acl=None): self.add_entity(self.entity_from_dict(entry)) self.loaded = True - def clear(self): + def clear(self, connection=None): """Remove all ACL entries. Note that this won't actually remove *ALL* the rules, but it @@ -441,21 +408,33 @@ def clear(self): have access to a bucket that you created even after you clear ACL rules with this method. - For example, imagine that you granted access to this bucket to a - bunch of coworkers:: - - >>> acl.user('coworker1@example.org').grant_read() - >>> acl.user('coworker2@example.org').grant_read() - >>> acl.save() + :type connection: :class:`gcloud.storage.connection.Connection` or None + :param connection: explicit connection to use for API request; + defaults to instance property. + """ + self.save([], connection) - Now they work in another part of the company and you want to - 'start fresh' on who has access:: - >>> acl.clear() +class BucketACL(ACL): + """An ACL specifically for a bucket.""" - At this point all the custom rules you created have been removed. + def __init__(self, bucket): """ - self.save([]) + :type bucket: :class:`gcloud.storage.bucket.Bucket` + :param bucket: The bucket to which this ACL relates. + """ + super(BucketACL, self).__init__() + self.bucket = bucket + + @property + def reload_path(self): + """Compute the path for GET API requests for this ACL.""" + return '%s/%s' % (self.bucket.path, self._URL_PATH_ELEM) + + @property + def save_path(self): + """Compute the path for PATCH API requests for this ACL.""" + return self.bucket.path class DefaultObjectACL(BucketACL): @@ -475,44 +454,12 @@ def __init__(self, blob): super(ObjectACL, self).__init__() self.blob = blob - def reload(self): - """Reload the ACL data from Cloud Storage.""" - self.entities.clear() - - url_path = '%s/acl' % self.blob.path - found = self.blob.connection.api_request(method='GET', path=url_path) - self.loaded = True - for entry in found.get('items', ()): - self.add_entity(self.entity_from_dict(entry)) - - def save(self, acl=None): - """Save the ACL data for this blob. + @property + def reload_path(self): + """Compute the path for GET API requests for this ACL.""" + return '%s/acl' % self.blob.path - :type acl: :class:`gcloud.storage.acl.ACL` - :param acl: The ACL object to save. If left blank, this will - save the entries set locally on the ACL. - """ - if acl is None: - acl = self - save_to_backend = acl.loaded - else: - save_to_backend = True - - if save_to_backend: - result = self.blob.connection.api_request( - method='PATCH', path=self.blob.path, data={'acl': list(acl)}, - query_params={'projection': 'full'}) - self.entities.clear() - for entry in result.get('acl', ()): - self.add_entity(self.entity_from_dict(entry)) - self.loaded = True - - def clear(self): - """Remove all ACL rules from the blob. - - Note that this won't actually remove *ALL* the rules, but it - will remove all the non-default rules. In short, you'll still - have access to a blob that you created even after you clear ACL - rules with this method. - """ - self.save([]) + @property + def save_path(self): + """Compute the path for PATCH API requests for this ACL.""" + return self.blob.path diff --git a/gcloud/storage/test_acl.py b/gcloud/storage/test_acl.py index c998cac895c5..7c0b9fb197e1 100644 --- a/gcloud/storage/test_acl.py +++ b/gcloud/storage/test_acl.py @@ -505,148 +505,199 @@ def test_get_entities_nonempty(self): entity = acl.entity(TYPE, ID) self.assertEqual(acl.get_entities(), [entity]) - def test_reload_raises_NotImplementedError(self): - acl = self._makeOne() - self.assertRaises(NotImplementedError, acl.reload) - - def test_save_raises_NotImplementedError(self): - acl = self._makeOne() - self.assertRaises(NotImplementedError, acl.save) - - def test_clear(self): + def test_reload_missing_w_implicit_connection(self): + # https://github.com/GoogleCloudPlatform/gcloud-python/issues/652 + from gcloud.storage._testing import _monkey_defaults + ROLE = 'role' + connection = _Connection({}) acl = self._makeOne() - self.assertRaises(NotImplementedError, acl.clear) - - -class Test_BucketACL(unittest2.TestCase): - - def _getTargetClass(self): - from gcloud.storage.acl import BucketACL - return BucketACL - - def _makeOne(self, *args, **kw): - return self._getTargetClass()(*args, **kw) - - def test_ctor(self): - bucket = object() - acl = self._makeOne(bucket) - self.assertEqual(acl.entities, {}) - self.assertFalse(acl.loaded) - self.assertTrue(acl.bucket is bucket) + acl.reload_path = '/testing/acl' + acl.loaded = True + acl.entity('allUsers', ROLE) + with _monkey_defaults(connection=connection): + acl.reload() + self.assertEqual(list(acl), []) + kw = connection._requested + self.assertEqual(len(kw), 1) + self.assertEqual(kw[0]['method'], 'GET') + self.assertEqual(kw[0]['path'], '/testing/acl') - def test_reload_eager_missing(self): + def test_reload_missing_w_explicit_connection(self): # https://github.com/GoogleCloudPlatform/gcloud-python/issues/652 - NAME = 'name' ROLE = 'role' connection = _Connection({}) - bucket = _Bucket(NAME, connection) - acl = self._makeOne(bucket) + acl = self._makeOne() + acl.reload_path = '/testing/acl' acl.loaded = True acl.entity('allUsers', ROLE) - acl.reload() + acl.reload(connection=connection) self.assertEqual(list(acl), []) kw = connection._requested self.assertEqual(len(kw), 1) self.assertEqual(kw[0]['method'], 'GET') - self.assertEqual(kw[0]['path'], '/b/%s/acl' % NAME) + self.assertEqual(kw[0]['path'], '/testing/acl') - def test_reload_eager_empty(self): - NAME = 'name' + def test_reload_empty_result_clears_local_w_implicit_connection(self): + from gcloud.storage._testing import _monkey_defaults ROLE = 'role' connection = _Connection({'items': []}) - bucket = _Bucket(NAME, connection) - acl = self._makeOne(bucket) + acl = self._makeOne() + acl.reload_path = '/testing/acl' acl.loaded = True acl.entity('allUsers', ROLE) - acl.reload() + with _monkey_defaults(connection=connection): + acl.reload() + self.assertTrue(acl.loaded) self.assertEqual(list(acl), []) kw = connection._requested self.assertEqual(len(kw), 1) self.assertEqual(kw[0]['method'], 'GET') - self.assertEqual(kw[0]['path'], '/b/%s/acl' % NAME) + self.assertEqual(kw[0]['path'], '/testing/acl') - def test_reload_eager_nonempty(self): - NAME = 'name' + def test_reload_empty_result_clears_local_w_explicit_connection(self): + ROLE = 'role' + connection = _Connection({'items': []}) + acl = self._makeOne() + acl.reload_path = '/testing/acl' + acl.loaded = True + acl.entity('allUsers', ROLE) + acl.reload(connection=connection) + self.assertTrue(acl.loaded) + self.assertEqual(list(acl), []) + kw = connection._requested + self.assertEqual(len(kw), 1) + self.assertEqual(kw[0]['method'], 'GET') + self.assertEqual(kw[0]['path'], '/testing/acl') + + def test_reload_nonempty_result_w_implicit_connection(self): + from gcloud.storage._testing import _monkey_defaults ROLE = 'role' connection = _Connection( {'items': [{'entity': 'allUsers', 'role': ROLE}]}) - bucket = _Bucket(NAME, connection) - acl = self._makeOne(bucket) + acl = self._makeOne() + acl.reload_path = '/testing/acl' acl.loaded = True - acl.reload() + with _monkey_defaults(connection=connection): + acl.reload() + self.assertTrue(acl.loaded) self.assertEqual(list(acl), [{'entity': 'allUsers', 'role': ROLE}]) kw = connection._requested self.assertEqual(len(kw), 1) self.assertEqual(kw[0]['method'], 'GET') - self.assertEqual(kw[0]['path'], '/b/%s/acl' % NAME) + self.assertEqual(kw[0]['path'], '/testing/acl') - def test_reload_lazy(self): - NAME = 'name' + def test_reload_nonempty_result_w_explicit_connection(self): ROLE = 'role' connection = _Connection( {'items': [{'entity': 'allUsers', 'role': ROLE}]}) - bucket = _Bucket(NAME, connection) - acl = self._makeOne(bucket) - acl.reload() + acl = self._makeOne() + acl.reload_path = '/testing/acl' + acl.loaded = True + acl.reload(connection=connection) + self.assertTrue(acl.loaded) self.assertEqual(list(acl), [{'entity': 'allUsers', 'role': ROLE}]) kw = connection._requested self.assertEqual(len(kw), 1) self.assertEqual(kw[0]['method'], 'GET') - self.assertEqual(kw[0]['path'], '/b/%s/acl' % NAME) + self.assertEqual(kw[0]['path'], '/testing/acl') - def test_save_none_set_none_passed(self): - NAME = 'name' + def test_save_none_set_none_passed_w_implicit_connection(self): + from gcloud.storage._testing import _monkey_defaults connection = _Connection() - bucket = _Bucket(NAME, connection) - acl = self._makeOne(bucket) - acl.save() + acl = self._makeOne() + acl._connection = connection + acl.save_path = '/testing' + with _monkey_defaults(connection=connection): + acl.save() kw = connection._requested self.assertEqual(len(kw), 0) - def test_save_existing_missing_none_passed(self): - NAME = 'name' + def test_save_none_set_none_passed_w_explicit_connection(self): + connection = _Connection() + acl = self._makeOne() + acl.save_path = '/testing' + acl.save(connection=connection) + kw = connection._requested + self.assertEqual(len(kw), 0) + + def test_save_existing_missing_none_passed_w_implicit_connection(self): + from gcloud.storage._testing import _monkey_defaults connection = _Connection({}) - bucket = _Bucket(NAME, connection) - acl = self._makeOne(bucket) + acl = self._makeOne() + acl.save_path = '/testing' acl.loaded = True - acl.save() + with _monkey_defaults(connection=connection): + acl.save() self.assertEqual(list(acl), []) kw = connection._requested self.assertEqual(len(kw), 1) self.assertEqual(kw[0]['method'], 'PATCH') - self.assertEqual(kw[0]['path'], '/b/%s' % NAME) + self.assertEqual(kw[0]['path'], '/testing') self.assertEqual(kw[0]['data'], {'acl': []}) self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - def test_save_no_arg(self): - NAME = 'name' + def test_save_existing_missing_none_passed_w_explicit_connection(self): + connection = _Connection({}) + acl = self._makeOne() + acl.save_path = '/testing' + acl.loaded = True + acl.save(connection=connection) + self.assertEqual(list(acl), []) + kw = connection._requested + self.assertEqual(len(kw), 1) + self.assertEqual(kw[0]['method'], 'PATCH') + self.assertEqual(kw[0]['path'], '/testing') + self.assertEqual(kw[0]['data'], {'acl': []}) + self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) + + def test_save_no_arg_w_implicit_connection(self): + from gcloud.storage._testing import _monkey_defaults ROLE = 'role' AFTER = [{'entity': 'allUsers', 'role': ROLE}] connection = _Connection({'acl': AFTER}) - bucket = _Bucket(NAME, connection) - acl = self._makeOne(bucket) + acl = self._makeOne() + acl.save_path = '/testing' acl.loaded = True acl.entity('allUsers').grant(ROLE) - acl.save() + with _monkey_defaults(connection=connection): + acl.save() self.assertEqual(list(acl), AFTER) kw = connection._requested self.assertEqual(len(kw), 1) self.assertEqual(kw[0]['method'], 'PATCH') - self.assertEqual(kw[0]['path'], '/b/%s' % NAME) + self.assertEqual(kw[0]['path'], '/testing') self.assertEqual(kw[0]['data'], {'acl': AFTER}) self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - def test_save_w_arg(self): - NAME = 'name' + def test_save_no_arg_w_explicit_connection(self): + ROLE = 'role' + AFTER = [{'entity': 'allUsers', 'role': ROLE}] + connection = _Connection({'acl': AFTER}) + acl = self._makeOne() + acl.save_path = '/testing' + acl.loaded = True + acl.entity('allUsers').grant(ROLE) + acl.save(connection=connection) + self.assertEqual(list(acl), AFTER) + kw = connection._requested + self.assertEqual(len(kw), 1) + self.assertEqual(kw[0]['method'], 'PATCH') + self.assertEqual(kw[0]['path'], '/testing') + self.assertEqual(kw[0]['data'], {'acl': AFTER}) + self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) + + def test_save_w_arg_w_implicit_connection(self): + from gcloud.storage._testing import _monkey_defaults ROLE1 = 'role1' ROLE2 = 'role2' STICKY = {'entity': 'allUsers', 'role': ROLE2} new_acl = [{'entity': 'allUsers', 'role': ROLE1}] connection = _Connection({'acl': [STICKY] + new_acl}) - bucket = _Bucket(NAME, connection) - acl = self._makeOne(bucket) + acl = self._makeOne() + acl.save_path = '/testing' acl.loaded = True - acl.save(new_acl) + with _monkey_defaults(connection=connection): + acl.save(new_acl) entries = list(acl) self.assertEqual(len(entries), 2) self.assertTrue(STICKY in entries) @@ -654,191 +705,130 @@ def test_save_w_arg(self): kw = connection._requested self.assertEqual(len(kw), 1) self.assertEqual(kw[0]['method'], 'PATCH') - self.assertEqual(kw[0]['path'], '/b/%s' % NAME) + self.assertEqual(kw[0]['path'], '/testing') self.assertEqual(kw[0]['data'], {'acl': new_acl}) self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - def test_clear(self): - NAME = 'name' + def test_save_w_arg_w_explicit_connection(self): + ROLE1 = 'role1' + ROLE2 = 'role2' + STICKY = {'entity': 'allUsers', 'role': ROLE2} + new_acl = [{'entity': 'allUsers', 'role': ROLE1}] + connection = _Connection({'acl': [STICKY] + new_acl}) + acl = self._makeOne() + acl.save_path = '/testing' + acl.loaded = True + acl.save(new_acl, connection) + entries = list(acl) + self.assertEqual(len(entries), 2) + self.assertTrue(STICKY in entries) + self.assertTrue(new_acl[0] in entries) + kw = connection._requested + self.assertEqual(len(kw), 1) + self.assertEqual(kw[0]['method'], 'PATCH') + self.assertEqual(kw[0]['path'], '/testing') + self.assertEqual(kw[0]['data'], {'acl': new_acl}) + self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) + + def test_clear_w_implicit_connection(self): + from gcloud.storage._testing import _monkey_defaults ROLE1 = 'role1' ROLE2 = 'role2' STICKY = {'entity': 'allUsers', 'role': ROLE2} connection = _Connection({'acl': [STICKY]}) - bucket = _Bucket(NAME, connection) - acl = self._makeOne(bucket) + acl = self._makeOne() + acl.save_path = '/testing' acl.loaded = True acl.entity('allUsers', ROLE1) - acl.clear() + with _monkey_defaults(connection=connection): + acl.clear() self.assertEqual(list(acl), [STICKY]) kw = connection._requested self.assertEqual(len(kw), 1) self.assertEqual(kw[0]['method'], 'PATCH') - self.assertEqual(kw[0]['path'], '/b/%s' % NAME) + self.assertEqual(kw[0]['path'], '/testing') self.assertEqual(kw[0]['data'], {'acl': []}) self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) + def test_clear_w_explicit_connection(self): + ROLE1 = 'role1' + ROLE2 = 'role2' + STICKY = {'entity': 'allUsers', 'role': ROLE2} + connection = _Connection({'acl': [STICKY]}) + acl = self._makeOne() + acl.save_path = '/testing' + acl.loaded = True + acl.entity('allUsers', ROLE1) + acl.clear(connection=connection) + self.assertEqual(list(acl), [STICKY]) + kw = connection._requested + self.assertEqual(len(kw), 1) + self.assertEqual(kw[0]['method'], 'PATCH') + self.assertEqual(kw[0]['path'], '/testing') + self.assertEqual(kw[0]['data'], {'acl': []}) + self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) -class Test_ObjectACL(unittest2.TestCase): + +class Test_BucketACL(unittest2.TestCase): def _getTargetClass(self): - from gcloud.storage.acl import ObjectACL - return ObjectACL + from gcloud.storage.acl import BucketACL + return BucketACL def _makeOne(self, *args, **kw): return self._getTargetClass()(*args, **kw) def test_ctor(self): - blob = object() - acl = self._makeOne(blob) + NAME = 'name' + bucket = _Bucket(NAME) + acl = self._makeOne(bucket) self.assertEqual(acl.entities, {}) self.assertFalse(acl.loaded) - self.assertTrue(acl.blob is blob) + self.assertTrue(acl.bucket is bucket) + self.assertEqual(acl.reload_path, '/b/%s/acl' % NAME) + self.assertEqual(acl.save_path, '/b/%s' % NAME) - def test_reload_eager_missing(self): - # https://github.com/GoogleCloudPlatform/gcloud-python/issues/652 - NAME = 'name' - BLOB_NAME = 'blob-name' - ROLE = 'role' - after = {} - connection = _Connection(after) - bucket = _Bucket(NAME, connection) - blob = _Blob(bucket, BLOB_NAME) - acl = self._makeOne(blob) - acl.loaded = True - acl.entity('allUsers', ROLE) - acl.reload() - self.assertEqual(list(acl), []) - def test_reload_eager_empty(self): - NAME = 'name' - BLOB_NAME = 'blob-name' - ROLE = 'role' - after = {'items': []} - connection = _Connection(after) - bucket = _Bucket(NAME, connection) - blob = _Blob(bucket, BLOB_NAME) - acl = self._makeOne(blob) - acl.loaded = True - acl.entity('allUsers', ROLE) - acl.reload() - self.assertEqual(list(acl), []) +class Test_DefaultObjectACL(unittest2.TestCase): - def test_reload_eager_nonempty(self): - NAME = 'name' - BLOB_NAME = 'blob-name' - ROLE = 'role' - after = {'items': [{'entity': 'allUsers', 'role': ROLE}]} - connection = _Connection(after) - bucket = _Bucket(NAME, connection) - blob = _Blob(bucket, BLOB_NAME) - acl = self._makeOne(blob) - acl.loaded = True - acl.reload() - self.assertEqual(list(acl), after['items']) - kw = connection._requested - self.assertEqual(len(kw), 1) - self.assertEqual(kw[0]['method'], 'GET') - self.assertEqual(kw[0]['path'], '/b/name/o/%s/acl' % BLOB_NAME) + def _getTargetClass(self): + from gcloud.storage.acl import DefaultObjectACL + return DefaultObjectACL - def test_reload_lazy(self): - NAME = 'name' - BLOB_NAME = 'blob-name' - ROLE = 'role' - after = {'items': [{'entity': 'allUsers', 'role': ROLE}]} - connection = _Connection(after) - bucket = _Bucket(NAME, connection) - blob = _Blob(bucket, BLOB_NAME) - acl = self._makeOne(blob) - acl.reload() - self.assertEqual(list(acl), - [{'entity': 'allUsers', 'role': ROLE}]) - kw = connection._requested - self.assertEqual(len(kw), 1) - self.assertEqual(kw[0]['method'], 'GET') - self.assertEqual(kw[0]['path'], '/b/name/o/%s/acl' % BLOB_NAME) + def _makeOne(self, *args, **kw): + return self._getTargetClass()(*args, **kw) - def test_save_none_set_none_passed(self): + def test_ctor(self): NAME = 'name' - BLOB_NAME = 'blob-name' - connection = _Connection() - bucket = _Bucket(NAME, connection) - blob = _Blob(bucket, BLOB_NAME) - acl = self._makeOne(blob) - acl.save() - kw = connection._requested - self.assertEqual(len(kw), 0) + bucket = _Bucket(NAME) + acl = self._makeOne(bucket) + self.assertEqual(acl.entities, {}) + self.assertFalse(acl.loaded) + self.assertTrue(acl.bucket is bucket) + self.assertEqual(acl.reload_path, '/b/%s/defaultObjectAcl' % NAME) + self.assertEqual(acl.save_path, '/b/%s' % NAME) - def test_save_existing_missing_none_passed(self): - # https://github.com/GoogleCloudPlatform/gcloud-python/issues/652 - NAME = 'name' - BLOB_NAME = 'blob-name' - connection = _Connection({'foo': 'Foo'}) - bucket = _Bucket(NAME, connection) - blob = _Blob(bucket, BLOB_NAME) - acl = self._makeOne(blob) - acl.loaded = True - acl.save() - kw = connection._requested - self.assertEqual(len(kw), 1) - self.assertEqual(kw[0]['method'], 'PATCH') - self.assertEqual(kw[0]['path'], '/b/name/o/%s' % BLOB_NAME) - self.assertEqual(kw[0]['data'], {'acl': []}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - def test_save_existing_set_none_passed(self): - NAME = 'name' - BLOB_NAME = 'blob-name' - connection = _Connection({'foo': 'Foo', 'acl': []}) - bucket = _Bucket(NAME, connection) - blob = _Blob(bucket, BLOB_NAME) - acl = self._makeOne(blob) - acl.loaded = True - acl.save() - kw = connection._requested - self.assertEqual(len(kw), 1) - self.assertEqual(kw[0]['method'], 'PATCH') - self.assertEqual(kw[0]['path'], '/b/name/o/%s' % BLOB_NAME) - self.assertEqual(kw[0]['data'], {'acl': []}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) +class Test_ObjectACL(unittest2.TestCase): - def test_save_existing_set_new_passed(self): - NAME = 'name' - BLOB_NAME = 'blob-name' - ROLE = 'role' - new_acl = [{'entity': 'allUsers', 'role': ROLE}] - connection = _Connection({'foo': 'Foo', 'acl': new_acl}) - bucket = _Bucket(NAME, connection) - blob = _Blob(bucket, BLOB_NAME) - acl = self._makeOne(blob) - acl.loaded = True - acl.entity('allUsers', 'other-role') - acl.save(new_acl) - self.assertEqual(list(acl), new_acl) - kw = connection._requested - self.assertEqual(len(kw), 1) - self.assertEqual(kw[0]['method'], 'PATCH') - self.assertEqual(kw[0]['path'], '/b/name/o/%s' % BLOB_NAME) - self.assertEqual(kw[0]['data'], {'acl': new_acl}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) + def _getTargetClass(self): + from gcloud.storage.acl import ObjectACL + return ObjectACL + + def _makeOne(self, *args, **kw): + return self._getTargetClass()(*args, **kw) - def test_clear(self): + def test_ctor(self): NAME = 'name' BLOB_NAME = 'blob-name' - ROLE = 'role' - connection = _Connection({'foo': 'Foo', 'acl': []}) - bucket = _Bucket(NAME, connection) + bucket = _Bucket(NAME) blob = _Blob(bucket, BLOB_NAME) acl = self._makeOne(blob) - acl.loaded = True - acl.entity('allUsers', ROLE) - acl.clear() - self.assertEqual(list(acl), []) - kw = connection._requested - self.assertEqual(len(kw), 1) - self.assertEqual(kw[0]['method'], 'PATCH') - self.assertEqual(kw[0]['path'], '/b/name/o/%s' % BLOB_NAME) - self.assertEqual(kw[0]['data'], {'acl': []}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) + self.assertEqual(acl.entities, {}) + self.assertFalse(acl.loaded) + self.assertTrue(acl.blob is blob) + self.assertEqual(acl.reload_path, '/b/%s/o/%s/acl' % (NAME, BLOB_NAME)) + self.assertEqual(acl.save_path, '/b/%s/o/%s' % (NAME, BLOB_NAME)) class _Blob(object): @@ -847,10 +837,6 @@ def __init__(self, bucket, blob): self.bucket = bucket self.blob = blob - @property - def connection(self): - return self.bucket.connection - @property def path(self): return '%s/o/%s' % (self.bucket.path, self.blob) @@ -858,9 +844,8 @@ def path(self): class _Bucket(object): - def __init__(self, name, connection): + def __init__(self, name): self.name = name - self.connection = connection @property def path(self): diff --git a/gcloud/storage/test_blob.py b/gcloud/storage/test_blob.py index 07791862ba53..d5221f330b79 100644 --- a/gcloud/storage/test_blob.py +++ b/gcloud/storage/test_blob.py @@ -704,6 +704,7 @@ def test_upload_from_string_w_text(self): def test_make_public(self): from gcloud.storage.acl import _ACLEntity + from gcloud.storage._testing import _monkey_defaults BLOB_NAME = 'blob-name' permissive = [{'entity': 'allUsers', 'role': _ACLEntity.READER_ROLE}] after = {'acl': permissive} @@ -711,7 +712,8 @@ def test_make_public(self): bucket = _Bucket(connection) blob = self._makeOne(BLOB_NAME, bucket=bucket) blob.acl.loaded = True - blob.make_public() + with _monkey_defaults(connection=connection): + blob.make_public() self.assertEqual(list(blob.acl), permissive) kw = connection._requested self.assertEqual(len(kw), 1) diff --git a/gcloud/storage/test_bucket.py b/gcloud/storage/test_bucket.py index 4f31859c7970..34cc2ec7b6b6 100644 --- a/gcloud/storage/test_bucket.py +++ b/gcloud/storage/test_bucket.py @@ -872,6 +872,7 @@ def test_disable_website(self): def test_make_public_defaults(self): from gcloud.storage.acl import _ACLEntity + from gcloud.storage._testing import _monkey_defaults NAME = 'name' permissive = [{'entity': 'allUsers', 'role': _ACLEntity.READER_ROLE}] after = {'acl': permissive, 'defaultObjectAcl': []} @@ -879,7 +880,8 @@ def test_make_public_defaults(self): bucket = self._makeOne(NAME, connection) bucket.acl.loaded = True bucket.default_object_acl.loaded = True - bucket.make_public() + with _monkey_defaults(connection=connection): + bucket.make_public() self.assertEqual(list(bucket.acl), permissive) self.assertEqual(list(bucket.default_object_acl), []) kw = connection._requested @@ -891,6 +893,7 @@ def test_make_public_defaults(self): def _make_public_w_future_helper(self, default_object_acl_loaded=True): from gcloud.storage.acl import _ACLEntity + from gcloud.storage._testing import _monkey_defaults NAME = 'name' permissive = [{'entity': 'allUsers', 'role': _ACLEntity.READER_ROLE}] after1 = {'acl': permissive, 'defaultObjectAcl': []} @@ -906,7 +909,8 @@ def _make_public_w_future_helper(self, default_object_acl_loaded=True): bucket = self._makeOne(NAME, connection) bucket.acl.loaded = True bucket.default_object_acl.loaded = default_object_acl_loaded - bucket.make_public(future=True) + with _monkey_defaults(connection=connection): + bucket.make_public(future=True) self.assertEqual(list(bucket.acl), permissive) self.assertEqual(list(bucket.default_object_acl), permissive) kw = connection._requested @@ -933,6 +937,7 @@ def test_make_public_w_future_reload_default(self): def test_make_public_recursive(self): from gcloud.storage.acl import _ACLEntity from gcloud.storage.bucket import _BlobIterator + from gcloud.storage._testing import _monkey_defaults _saved = [] class _Blob(object): @@ -969,7 +974,8 @@ def get_items_from_response(self, response): bucket.acl.loaded = True bucket.default_object_acl.loaded = True bucket._iterator_class = _Iterator - bucket.make_public(recursive=True) + with _monkey_defaults(connection=connection): + bucket.make_public(recursive=True) self.assertEqual(list(bucket.acl), permissive) self.assertEqual(list(bucket.default_object_acl), []) self.assertEqual(_saved, [(bucket, BLOB_NAME, True)])