diff --git a/gcloud/storage/acl.py b/gcloud/storage/acl.py index 448ad1119c57..9a9c7a091733 100644 --- a/gcloud/storage/acl.py +++ b/gcloud/storage/acl.py @@ -172,10 +172,6 @@ class ACL(object): def __init__(self): self.entities = {} - def clear(self): - """Remove all entities from the ACL.""" - self.entities.clear() - def reset(self): """Remove all entities from the ACL, and clear the ``loaded`` flag.""" self.entities.clear() @@ -338,17 +334,35 @@ def get_entities(self): """ return self.entities.values() - def save(self): + def reload(self): + """Reload the ACL data from Cloud Storage. + + :rtype: :class:`ACL` + :returns: The current ACL. + """ + 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 bucket: :class:`gcloud.storage.bucket.Bucket` @@ -357,19 +371,101 @@ def __init__(self, bucket): super(BucketACL, self).__init__() self.bucket = bucket - def save(self): - """Save this ACL for the current bucket.""" + def reload(self): + """Reload the ACL data from Cloud Storage. + + :rtype: :class:`gcloud.storage.acl.BucketACL` + :returns: The current ACL. + """ + 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) + for entry in found['items']: + self.add_entity(self.entity_from_dict(entry)) + + # Even if we fetch no entries, the ACL is still loaded. + self.loaded = True + + return self + + def save(self, acl=None): + """Save this ACL for the current bucket. + + If called without arguments, this will save the entries + currently stored on this ACL:: - return self.bucket.save_acl(acl=self) + >>> 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 = connection.get_bucket(bucket1_name) + >>> bucket2 = connection.get_bucket(bucket2_name) + >>> bucket2.acl.save(bucket1.get_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. + + :rtype: :class:`gcloud.storage.acl.BucketACL` + :returns: The current ACL. + """ + if acl is None: + acl = self + save_to_backend = acl.loaded + else: + save_to_backend = True + + if save_to_backend: + result = self.bucket.connection.api_request( + method='PATCH', path=self.bucket.path, + data={self._URL_PATH_ELEM: list(acl)}, + query_params={'projection': 'full'}) + self.entities.clear() + for entry in result[self._URL_PATH_ELEM]: + self.add_entity(self.entity_from_dict(entry)) + self.loaded = True + + return self + + def clear(self): + """Remove all ACL entries. + + 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 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() + + Now they work in another part of the company and you want to + 'start fresh' on who has access:: + + >>> acl.clear() + + At this point all the custom rules you created have been removed. + + :rtype: :class:`gcloud.storage.acl.BucketACL` + :returns: The current ACL. + """ + return self.save([]) class DefaultObjectACL(BucketACL): """A class representing the default object ACL for a bucket.""" - def save(self): - """Save this ACL as the default object ACL for the current bucket.""" - - return self.bucket.save_default_object_acl(acl=self) + _URL_PATH_ELEM = 'defaultObjectAcl' class ObjectACL(ACL): @@ -383,7 +479,54 @@ def __init__(self, key): super(ObjectACL, self).__init__() self.key = key - def save(self): - """Save this ACL for the current key.""" + def reload(self): + """Reload the ACL data from Cloud Storage. + + :rtype: :class:`ObjectACL` + :returns: The current ACL. + """ + self.entities.clear() - return self.key.save_acl(acl=self) + url_path = '%s/acl' % self.key.path + found = self.key.connection.api_request(method='GET', path=url_path) + for entry in found['items']: + self.add_entity(self.entity_from_dict(entry)) + + # Even if we fetch no entries, the ACL is still loaded. + self.loaded = True + + return self + + def save(self, acl=None): + """Save the ACL data for this key. + + :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.key.connection.api_request( + method='PATCH', path=self.key.path, data={'acl': list(acl)}, + query_params={'projection': 'full'}) + self.entities.clear() + for entry in result['acl']: + self.add_entity(self.entity_from_dict(entry)) + self.loaded = True + + return self + + def clear(self): + """Remove all ACL rules from the key. + + 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 key that you created even after you clear ACL + rules with this method. + """ + return self.save([]) diff --git a/gcloud/storage/bucket.py b/gcloud/storage/bucket.py index 7e7bd9640049..ffb5c8a09cc8 100644 --- a/gcloud/storage/bucket.py +++ b/gcloud/storage/bucket.py @@ -440,24 +440,6 @@ def disable_website(self): """ return self.configure_website(None, None) - def reload_acl(self): - """Reload the ACL data from Cloud Storage. - - :rtype: :class:`Bucket` - :returns: The current bucket. - """ - self.acl.clear() - - url_path = '%s/acl' % self.path - found = self.connection.api_request(method='GET', path=url_path) - for entry in found['items']: - self.acl.add_entity(self.acl.entity_from_dict(entry)) - - # Even if we fetch no entries, the ACL is still loaded. - self.acl.loaded = True - - return self - def get_acl(self): """Get ACL metadata as a :class:`gcloud.storage.acl.BucketACL` object. @@ -465,112 +447,9 @@ def get_acl(self): :returns: An ACL object for the current bucket. """ if not self.acl.loaded: - self.reload_acl() + self.acl.reload() return self.acl - def save_acl(self, acl=None): - """Save the ACL data for this bucket. - - If called without arguments, this will save the ACL currently - stored on the Bucket object. For example, this will save the - ACL stored in ``some_other_acl``:: - - >>> bucket.acl = some_other_acl - >>> bucket.save_acl() - - You can also provide a specific ACL to save instead of the one - currently set on the Bucket object:: - - >>> bucket.save_acl(acl=my_other_acl) - - You can use this to set access controls to be consistent from - one bucket to another:: - - >>> bucket1 = connection.get_bucket(bucket1_name) - >>> bucket2 = connection.get_bucket(bucket2_name) - >>> bucket2.save_acl(bucket1.get_acl()) - - If you want to **clear** the ACL for the bucket, you must save - an empty list (``[]``) rather than using ``None`` (which is - interpreted as wanting to save the current ACL):: - - >>> bucket.save_acl(None) # Saves the current ACL (self.acl). - >>> bucket.save_acl([]) # Clears the current ACL. - - :type acl: :class:`gcloud.storage.acl.ACL` - :param acl: The ACL object to save. - If left blank, this will save the ACL - set locally on the bucket. - """ - # We do things in this weird way because [] and None - # both evaluate to False, but mean very different things. - if acl is None: - acl = self.acl - dirty = acl.loaded - else: - dirty = True - - if dirty: - result = self.connection.api_request( - method='PATCH', path=self.path, data={'acl': list(acl)}, - query_params={'projection': 'full'}) - self.acl.clear() - for entry in result['acl']: - self.acl.entity(self.acl.entity_from_dict(entry)) - self.acl.loaded = True - - return self - - def clear_acl(self): - """Remove all ACL rules from the bucket. - - 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 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:: - - >>> from gcloud import storage - >>> connection = storage.get_connection(project, email, - private_key_path) - >>> bucket = connection.get_bucket(bucket_name) - >>> acl = bucket.get_acl() - >>> acl.user('coworker1@example.org').grant_read() - >>> acl.user('coworker2@example.org').grant_read() - >>> acl.save() - - Now they work in another part of the company - and you want to 'start fresh' on who has access:: - - >>> acl.clear_acl() - - At this point all the custom rules you created have been removed. - """ - # NOTE: back-end makes some ACL entries sticky (they remain even - # after the PATCH succeeds. - return self.save_acl([]) - - def reload_default_object_acl(self): - """Reload the Default Object ACL rules for this bucket. - - :rtype: :class:`Bucket` - :returns: The current bucket. - """ - doa = self.default_object_acl - doa.clear() - - url_path = '%s/defaultObjectAcl' % self.path - found = self.connection.api_request(method='GET', path=url_path) - for entry in found['items']: - doa.add_entity(doa.entity_from_dict(entry)) - - # Even if we fetch no entries, the ACL is still loaded. - doa.loaded = True - - return self - def get_default_object_acl(self): """Get the current Default Object ACL rules. @@ -581,41 +460,9 @@ def get_default_object_acl(self): :returns: A DefaultObjectACL object for this bucket. """ if not self.default_object_acl.loaded: - self.reload_default_object_acl() + self.default_object_acl.reload() return self.default_object_acl - def save_default_object_acl(self, acl=None): - """Save the Default Object ACL rules for this bucket. - - :type acl: :class:`gcloud.storage.acl.DefaultObjectACL` - :param acl: The DefaultObjectACL object to save. - If not provided, this will look at - the ``default_object_acl`` property - and save that. - """ - if acl is None: - acl = self.default_object_acl - dirty = acl.loaded - else: - dirty = True - - if dirty: - result = self.connection.api_request( - method='PATCH', path=self.path, - data={'defaultObjectAcl': list(acl)}, - query_params={'projection': 'full'}) - doa = self.default_object_acl - doa.clear() - for entry in result['defaultObjectAcl']: - doa.entity(doa.entity_from_dict(entry)) - doa.loaded = True - - return self - - def clear_default_object_acl(self): - """Remove the Default Object ACL from this bucket.""" - return self.save_default_object_acl([]) - def make_public(self, recursive=False, future=False): """Make a bucket public. @@ -628,11 +475,12 @@ def make_public(self, recursive=False, future=False): future public as well. """ self.get_acl().all().grant_read() - self.save_acl() + self.acl.save() if future: - self.get_default_object_acl().all().grant_read() - self.save_default_object_acl() + doa = self.get_default_object_acl() + doa.all().grant_read() + doa.save() if recursive: for key in self: diff --git a/gcloud/storage/key.py b/gcloud/storage/key.py index 217f98ee29e3..73beed32d4d1 100644 --- a/gcloud/storage/key.py +++ b/gcloud/storage/key.py @@ -383,24 +383,6 @@ def patch_metadata(self, metadata): query_params={'projection': 'full'}) return self - def reload_acl(self): - """Reload the ACL data from Cloud Storage. - - :rtype: :class:`Key` - :returns: The current key. - """ - self.acl.clear() - - url_path = '%s/acl' % self.path - found = self.connection.api_request(method='GET', path=url_path) - for entry in found['items']: - self.acl.add_entity(self.acl.entity_from_dict(entry)) - - # Even if we fetch no entries, the ACL is still loaded. - self.acl.loaded = True - - return self - def get_acl(self): """Get ACL metadata as a :class:`gcloud.storage.acl.ObjectACL` object. @@ -408,43 +390,9 @@ def get_acl(self): :returns: An ACL object for the current key. """ if not self.acl.loaded: - self.reload_acl() + self.acl.reload() return self.acl - def save_acl(self, acl=None): - """Save the ACL data for this key. - - :type acl: :class:`gcloud.storage.acl.ACL` - :param acl: The ACL object to save. If left blank, this will - save the ACL set locally on the key. - """ - if acl is None: - acl = self.acl - dirty = acl.loaded - else: - dirty = True - - if dirty: - result = self.connection.api_request( - method='PATCH', path=self.path, data={'acl': list(acl)}, - query_params={'projection': 'full'}) - self.acl.clear() - for entry in result['acl']: - self.acl.entity(self.acl.entity_from_dict(entry)) - self.acl.loaded = True - - return self - - def clear_acl(self): - """Remove all ACL rules from the key. - - 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 key that you created even after you clear ACL - rules with this method. - """ - return self.save_acl([]) - def make_public(self): """Make this key public giving all users read access. @@ -452,7 +400,7 @@ def make_public(self): :returns: The current key. """ self.get_acl().all().grant_read() - self.save_acl() + self.acl.save() return self diff --git a/gcloud/storage/test_acl.py b/gcloud/storage/test_acl.py index 74f9cccd4ffb..67f208759368 100644 --- a/gcloud/storage/test_acl.py +++ b/gcloud/storage/test_acl.py @@ -127,16 +127,6 @@ def test_ctor(self): self.assertEqual(list(acl.get_entities()), []) self.assertFalse(acl.loaded) - def test_clear(self): - TYPE = 'type' - ID = 'id' - acl = self._makeOne() - acl.entity(TYPE, ID) - acl.clear() - self.assertTrue(acl.loaded) - self.assertEqual(acl.entities, {}) - self.assertEqual(list(acl.get_entities()), []) - def test_reset(self): TYPE = 'type' ID = 'id' @@ -405,10 +395,18 @@ 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): + acl = self._makeOne() + self.assertRaises(NotImplementedError, acl.clear) + class Test_BucketACL(unittest2.TestCase): @@ -426,33 +424,114 @@ def test_ctor(self): self.assertEqual(list(acl.get_entities()), []) self.assertTrue(acl.bucket is bucket) - def test_save(self): - class _Bucket(object): - def save_acl(self, acl): - self._saved = acl - bucket = _Bucket() + def test_reload_eager_empty(self): + NAME = 'name' + ROLE = 'role' + connection = _Connection({'items': []}) + bucket = _Bucket(connection, NAME) acl = self._makeOne(bucket) - acl.save() - self.assertTrue(bucket._saved is acl) - - -class Test_DefaultObjectACL(unittest2.TestCase): + acl.entity('allUsers', ROLE) + self.assertTrue(acl.reload() is acl) + 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) - def _getTargetClass(self): - from gcloud.storage.acl import DefaultObjectACL - return DefaultObjectACL + def test_reload_eager_nonempty(self): + NAME = 'name' + ROLE = 'role' + connection = _Connection( + {'items': [{'entity': 'allUsers', 'role': ROLE}]}) + bucket = _Bucket(connection, NAME) + acl = self._makeOne(bucket) + acl.loaded = True + self.assertTrue(acl.reload() is acl) + 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) + + def test_reload_lazy(self): + NAME = 'name' + ROLE = 'role' + connection = _Connection( + {'items': [{'entity': 'allUsers', 'role': ROLE}]}) + bucket = _Bucket(connection, NAME) + acl = self._makeOne(bucket) + self.assertTrue(acl.reload() is acl) + 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) + + def test_save_none_set_none_passed(self): + NAME = 'name' + connection = _Connection() + bucket = _Bucket(connection, NAME) + acl = self._makeOne(bucket) + self.assertTrue(acl.save() is acl) + kw = connection._requested + self.assertEqual(len(kw), 0) - def _makeOne(self, *args, **kw): - return self._getTargetClass()(*args, **kw) + def test_save_no_arg(self): + NAME = 'name' + ROLE = 'role' + AFTER = [{'entity': 'allUsers', 'role': ROLE}] + connection = _Connection({'acl': AFTER}) + bucket = _Bucket(connection, NAME) + acl = self._makeOne(bucket) + acl.entity('allUsers').grant(ROLE) + self.assertTrue(acl.save() is acl) + 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]['data'], {'acl': AFTER}) + self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) + + def test_save_w_arg(self): + NAME = 'name' + ROLE1 = 'role1' + ROLE2 = 'role2' + STICKY = {'entity': 'allUsers', 'role': ROLE2} + new_acl = [{'entity': 'allUsers', 'role': ROLE1}] + connection = _Connection({'acl': [STICKY] + new_acl}) + bucket = _Bucket(connection, NAME) + acl = self._makeOne(bucket) + acl.loaded = True + self.assertTrue(acl.save(new_acl) is acl) + 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'], '/b/%s' % NAME) + self.assertEqual(kw[0]['data'], {'acl': new_acl}) + self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - def test_save(self): - class _Bucket(object): - def save_default_object_acl(self, acl): - self._saved = acl - bucket = _Bucket() + def test_clear(self): + NAME = 'name' + ROLE1 = 'role1' + ROLE2 = 'role2' + STICKY = {'entity': 'allUsers', 'role': ROLE2} + connection = _Connection({'acl': [STICKY]}) + bucket = _Bucket(connection, NAME) acl = self._makeOne(bucket) - acl.save() - self.assertTrue(bucket._saved is acl) + acl.entity('allUsers', ROLE1) + self.assertTrue(acl.clear() is acl) + 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]['data'], {'acl': []}) + self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) class Test_ObjectACL(unittest2.TestCase): @@ -471,11 +550,159 @@ def test_ctor(self): self.assertEqual(list(acl.get_entities()), []) self.assertTrue(acl.key is key) - def test_save(self): - class _Key(object): - def save_acl(self, acl): - self._saved = acl - key = _Key() + def test_reload_eager_empty(self): + NAME = 'name' + KEY = 'key' + ROLE = 'role' + after = {'items': [{'entity': 'allUsers', 'role': ROLE}]} + connection = _Connection(after) + bucket = _Bucket(connection, NAME) + key = _Key(bucket, KEY) acl = self._makeOne(key) - acl.save() - self.assertTrue(key._saved is acl) + acl.loaded = True + self.assertTrue(acl.reload() is acl) + 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' % KEY) + + def test_reload_eager_nonempty(self): + NAME = 'name' + KEY = 'key' + ROLE = 'role' + after = {'items': []} + connection = _Connection(after) + bucket = _Bucket(connection, NAME) + key = _Key(bucket, KEY) + acl = self._makeOne(key) + acl.entity('allUsers', ROLE) + self.assertTrue(acl.reload() is acl) + self.assertEqual(list(acl), []) + + def test_reload_lazy(self): + NAME = 'name' + KEY = 'key' + ROLE = 'role' + after = {'items': [{'entity': 'allUsers', 'role': ROLE}]} + connection = _Connection(after) + bucket = _Bucket(connection, NAME) + key = _Key(bucket, KEY) + acl = self._makeOne(key) + self.assertTrue(acl.reload() is acl) + 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' % KEY) + + def test_save_none_set_none_passed(self): + NAME = 'name' + KEY = 'key' + connection = _Connection() + bucket = _Bucket(connection, NAME) + key = _Key(bucket, KEY) + acl = self._makeOne(key) + self.assertTrue(acl.save() is acl) + kw = connection._requested + self.assertEqual(len(kw), 0) + + def test_save_existing_set_none_passed(self): + NAME = 'name' + KEY = 'key' + connection = _Connection({'foo': 'Foo', 'acl': []}) + bucket = _Bucket(connection, NAME) + key = _Key(bucket, KEY) + acl = self._makeOne(key) + acl.loaded = True + self.assertTrue(acl.save() is acl) + kw = connection._requested + self.assertEqual(len(kw), 1) + self.assertEqual(kw[0]['method'], 'PATCH') + self.assertEqual(kw[0]['path'], '/b/name/o/%s' % KEY) + self.assertEqual(kw[0]['data'], {'acl': []}) + self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) + + def test_save_existing_set_new_passed(self): + NAME = 'name' + KEY = 'key' + ROLE = 'role' + new_acl = [{'entity': 'allUsers', 'role': ROLE}] + connection = _Connection({'foo': 'Foo', 'acl': new_acl}) + bucket = _Bucket(connection, NAME) + key = _Key(bucket, KEY) + acl = self._makeOne(key) + acl.entity('allUsers', 'other-role') + self.assertTrue(acl.save(new_acl) is 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' % KEY) + self.assertEqual(kw[0]['data'], {'acl': new_acl}) + self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) + + def test_clear(self): + NAME = 'name' + KEY = 'key' + ROLE = 'role' + connection = _Connection({'foo': 'Foo', 'acl': []}) + bucket = _Bucket(connection, NAME) + key = _Key(bucket, KEY) + acl = self._makeOne(key) + acl.entity('allUsers', ROLE) + self.assertTrue(acl.clear() is acl) + 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' % KEY) + self.assertEqual(kw[0]['data'], {'acl': []}) + self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) + + +class _Key(object): + + def __init__(self, bucket, key): + self.bucket = bucket + self.key = key + + @property + def connection(self): + return self.bucket.connection + + @property + def path(self): + return '%s/o/%s' % (self.bucket.path, self.key) + + +class _Bucket(object): + + def __init__(self, connection, name): + self.connection = connection + self.name = name + + @property + def path(self): + return '/b/%s' % self.name + + +class _Connection(object): + _delete_ok = False + + def __init__(self, *responses): + self._responses = responses + self._requested = [] + self._deleted = [] + + def api_request(self, **kw): + from gcloud.storage.exceptions import NotFoundError + self._requested.append(kw) + + try: + response, self._responses = self._responses[0], self._responses[1:] + except: # pragma: NO COVER + raise NotFoundError('miss') + else: + return response diff --git a/gcloud/storage/test_bucket.py b/gcloud/storage/test_bucket.py index 6177de01b8ef..478e4922f723 100644 --- a/gcloud/storage/test_bucket.py +++ b/gcloud/storage/test_bucket.py @@ -554,54 +554,6 @@ def test_disable_website(self): self.assertEqual(kw[0]['data'], patched) self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - def test_reload_acl_eager_empty(self): - from gcloud.storage.acl import BucketACL - NAME = 'name' - ROLE = 'role' - connection = _Connection({'items': []}) - bucket = self._makeOne(connection, NAME) - bucket.acl.entity('allUsers', ROLE) - self.assertTrue(bucket.reload_acl() is bucket) - self.assertTrue(isinstance(bucket.acl, BucketACL)) - self.assertEqual(list(bucket.acl), []) - kw = connection._requested - self.assertEqual(len(kw), 1) - self.assertEqual(kw[0]['method'], 'GET') - self.assertEqual(kw[0]['path'], '/b/%s/acl' % NAME) - - def test_reload_acl_eager_nonempty(self): - from gcloud.storage.acl import BucketACL - NAME = 'name' - ROLE = 'role' - connection = _Connection( - {'items': [{'entity': 'allUsers', 'role': ROLE}]}) - bucket = self._makeOne(connection, NAME) - bucket.acl.loaded = True - self.assertTrue(bucket.reload_acl() is bucket) - self.assertTrue(isinstance(bucket.acl, BucketACL)) - self.assertEqual(list(bucket.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) - - def test_reload_acl_lazy(self): - from gcloud.storage.acl import BucketACL - NAME = 'name' - ROLE = 'role' - connection = _Connection( - {'items': [{'entity': 'allUsers', 'role': ROLE}]}) - bucket = self._makeOne(connection, NAME) - self.assertTrue(bucket.reload_acl() is bucket) - self.assertTrue(isinstance(bucket.acl, BucketACL)) - self.assertEqual(list(bucket.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) - def test_get_acl_lazy(self): from gcloud.storage.acl import BucketACL NAME = 'name' @@ -626,105 +578,6 @@ def test_get_acl_eager(self): kw = connection._requested self.assertEqual(len(kw), 0) - def test_save_acl_none_set_none_passed(self): - connection = _Connection() - bucket = self._makeOne() - self.assertTrue(bucket.save_acl() is bucket) - kw = connection._requested - self.assertEqual(len(kw), 0) - - def test_save_acl_existing_set_none_passed(self): - NAME = 'name' - connection = _Connection({'foo': 'Foo', 'acl': []}) - bucket = self._makeOne(connection, NAME) - bucket.acl.loaded = True - self.assertTrue(bucket.save_acl() is bucket) - 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]['data'], {'acl': []}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - - def test_save_acl_existing_set_new_passed(self): - NAME = 'name' - ROLE = 'role' - new_acl = [{'entity': 'allUsers', 'role': ROLE}] - connection = _Connection({'foo': 'Foo', 'acl': new_acl}) - bucket = self._makeOne(connection, NAME) - bucket.acl.loaded = True - self.assertTrue(bucket.save_acl(new_acl) is bucket) - self.assertEqual(list(bucket.acl), new_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]['data'], {'acl': new_acl}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - - def test_clear_acl(self): - NAME = 'name' - ROLE1 = 'role1' - ROLE2 = 'role2' - STICKY = {'entity': 'allUsers', 'role': ROLE2} - connection = _Connection( - # Emulate back-end, which makes some entries "sticky". - {'foo': 'Foo', 'acl': [STICKY]}, - ) - bucket = self._makeOne(connection, NAME) - bucket.acl.entity('allUsers', ROLE1) - self.assertTrue(bucket.clear_acl() is bucket) - self.assertEqual(list(bucket.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]['data'], {'acl': []}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - - def test_reload_default_object_acl_eager_empty(self): - from gcloud.storage.acl import DefaultObjectACL - NAME = 'name' - after = {'items': []} - connection = _Connection(after) - bucket = self._makeOne(connection, NAME) - bucket.default_object_acl.loaded = True - self.assertTrue(bucket.reload_default_object_acl() is bucket) - self.assertTrue( - isinstance(bucket.default_object_acl, DefaultObjectACL)) - self.assertEqual(list(bucket.default_object_acl), []) - - def test_reload_default_object_acl_eager_nonempty(self): - from gcloud.storage.acl import DefaultObjectACL - NAME = 'name' - ROLE = 'role' - after = {'items': [{'entity': 'allUsers', 'role': ROLE}]} - connection = _Connection(after) - bucket = self._makeOne(connection, NAME) - bucket.default_object_acl.entity('allUsers', 'OTHERROLE') - self.assertTrue(bucket.reload_default_object_acl() is bucket) - self.assertTrue( - isinstance(bucket.default_object_acl, DefaultObjectACL)) - self.assertEqual(list(bucket.default_object_acl), - [{'entity': 'allUsers', 'role': ROLE}]) - - def test_reload_default_object_acl_lazy(self): - from gcloud.storage.acl import DefaultObjectACL - NAME = 'name' - ROLE = 'role' - after = {'items': [{'entity': 'allUsers', 'role': ROLE}]} - connection = _Connection(after) - bucket = self._makeOne(connection, NAME) - self.assertTrue(bucket.reload_default_object_acl() is bucket) - self.assertTrue( - isinstance(bucket.default_object_acl, DefaultObjectACL)) - self.assertEqual(list(bucket.default_object_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/defaultObjectAcl' % NAME) - def test_get_default_object_acl_lazy(self): from gcloud.storage.acl import BucketACL NAME = 'name' @@ -749,64 +602,6 @@ def test_get_default_object_acl_eager(self): kw = connection._requested self.assertEqual(len(kw), 0) - def test_save_default_object_acl_none_set_none_passed(self): - connection = _Connection() - bucket = self._makeOne() - self.assertTrue(bucket.save_default_object_acl() is bucket) - kw = connection._requested - self.assertEqual(len(kw), 0) - - def test_save_default_object_acl_existing_set_none_passed(self): - NAME = 'name' - connection = _Connection( - {'foo': 'Foo', 'acl': [], 'defaultObjectAcl': []}, - ) - bucket = self._makeOne(connection, NAME) - bucket.default_object_acl.loaded = True - self.assertTrue(bucket.save_default_object_acl() is bucket) - 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]['data'], {'defaultObjectAcl': []}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - - def test_save_default_object_acl_existing_set_new_passed(self): - NAME = 'name' - ROLE = 'role' - new_acl = [{'entity': 'allUsers', 'role': ROLE}] - connection = _Connection( - {'foo': 'Foo', 'acl': new_acl, 'defaultObjectAcl': new_acl}, - ) - metadata = {'defaultObjectAcl': []} - bucket = self._makeOne(connection, NAME, metadata) - bucket.default_object_acl.loaded = True - self.assertTrue(bucket.save_default_object_acl(new_acl) is bucket) - self.assertEqual(list(bucket.default_object_acl), new_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]['data'], {'defaultObjectAcl': new_acl}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - - def test_clear_default_object_acl(self): - NAME = 'name' - ROLE = 'role' - connection = _Connection( - {'foo': 'Foo', 'acl': [], 'defaultObjectAcl': []}, - ) - bucket = self._makeOne(connection, NAME) - bucket.default_object_acl.entity('allUsers', ROLE) - self.assertTrue(bucket.clear_default_object_acl() is bucket) - self.assertEqual(list(bucket.default_object_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]['data'], {'defaultObjectAcl': []}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - def test_make_public_defaults(self): from gcloud.storage.acl import _ACLEntity NAME = 'name' diff --git a/gcloud/storage/test_key.py b/gcloud/storage/test_key.py index 80ec615c3c5a..7ddd77f49cbc 100644 --- a/gcloud/storage/test_key.py +++ b/gcloud/storage/test_key.py @@ -439,53 +439,6 @@ def test_patch_metadata(self): self.assertEqual(kw[0]['data'], after) self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - def test_reload_acl_eager_empty(self): - from gcloud.storage.acl import ObjectACL - KEY = 'key' - ROLE = 'role' - after = {'items': [{'entity': 'allUsers', 'role': ROLE}]} - connection = _Connection(after) - bucket = _Bucket(connection) - key = self._makeOne(bucket, KEY) - key.acl.loaded = True - self.assertTrue(key.reload_acl() is key) - self.assertTrue(isinstance(key.acl, ObjectACL)) - self.assertEqual(list(key.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' % KEY) - - def test_reload_acl_eager_nonempty(self): - from gcloud.storage.acl import ObjectACL - KEY = 'key' - ROLE = 'role' - after = {'items': []} - connection = _Connection(after) - bucket = _Bucket(connection) - key = self._makeOne(bucket, KEY) - key.acl.entity('allUsers', ROLE) - self.assertTrue(key.reload_acl() is key) - self.assertTrue(isinstance(key.acl, ObjectACL)) - self.assertEqual(list(key.acl), []) - - def test_reload_acl_lazy(self): - from gcloud.storage.acl import ObjectACL - KEY = 'key' - ROLE = 'role' - after = {'items': [{'entity': 'allUsers', 'role': ROLE}]} - connection = _Connection(after) - bucket = _Bucket(connection) - key = self._makeOne(bucket, KEY) - self.assertTrue(key.reload_acl() is key) - self.assertTrue(isinstance(key.acl, ObjectACL)) - self.assertEqual(list(key.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' % KEY) - def test_get_acl_lazy(self): from gcloud.storage.acl import ObjectACL KEY = 'key' @@ -504,62 +457,6 @@ def test_get_acl_eager(self): acl = key.get_acl() self.assertTrue(acl is preset) - def test_save_acl_none_set_none_passed(self): - KEY = 'key' - connection = _Connection() - bucket = _Bucket(connection) - key = self._makeOne(bucket, KEY) - self.assertTrue(key.save_acl() is key) - kw = connection._requested - self.assertEqual(len(kw), 0) - - def test_save_acl_existing_set_none_passed(self): - KEY = 'key' - connection = _Connection({'foo': 'Foo', 'acl': []}) - bucket = _Bucket(connection) - key = self._makeOne(bucket, KEY) - key.acl.loaded = True - self.assertTrue(key.save_acl() is key) - kw = connection._requested - self.assertEqual(len(kw), 1) - self.assertEqual(kw[0]['method'], 'PATCH') - self.assertEqual(kw[0]['path'], '/b/name/o/%s' % KEY) - self.assertEqual(kw[0]['data'], {'acl': []}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - - def test_save_acl_existing_set_new_passed(self): - KEY = 'key' - ROLE = 'role' - new_acl = [{'entity': 'allUsers', 'role': ROLE}] - connection = _Connection({'foo': 'Foo', 'acl': new_acl}) - bucket = _Bucket(connection) - key = self._makeOne(bucket, KEY) - key.acl.entity('allUsers', 'other-role') - self.assertTrue(key.save_acl(new_acl) is key) - self.assertEqual(list(key.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' % KEY) - self.assertEqual(kw[0]['data'], {'acl': new_acl}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - - def test_clear_acl(self): - KEY = 'key' - ROLE = 'role' - connection = _Connection({'foo': 'Foo', 'acl': []}) - bucket = _Bucket(connection) - key = self._makeOne(bucket, KEY) - key.acl.entity('allUsers', ROLE) - self.assertTrue(key.clear_acl() is key) - self.assertEqual(list(key.acl), []) - kw = connection._requested - self.assertEqual(len(kw), 1) - self.assertEqual(kw[0]['method'], 'PATCH') - self.assertEqual(kw[0]['path'], '/b/name/o/%s' % KEY) - self.assertEqual(kw[0]['data'], {'acl': []}) - self.assertEqual(kw[0]['query_params'], {'projection': 'full'}) - def test_make_public(self): from gcloud.storage.acl import _ACLEntity KEY = 'key'