<s> import unittest2 <EOL> class Test__BlobIterator ( unittest2 . TestCase ) : <EOL> def _getTargetClass ( self ) : <EOL> from gcloud . storage . bucket import _BlobIterator <EOL> return _BlobIterator <EOL> def _makeOne ( self , * args , ** kw ) : <EOL> return self . _getTargetClass ( ) ( * args , ** kw ) <EOL> def test_ctor ( self ) : <EOL> connection = _Connection ( ) <EOL> client = _Client ( connection ) <EOL> bucket = _Bucket ( ) <EOL> iterator = self . _makeOne ( bucket , client = client ) <EOL> self . assertTrue ( iterator . bucket is bucket ) <EOL> self . assertTrue ( iterator . client is client ) <EOL> self . assertEqual ( iterator . path , '<STR_LIT>' % bucket . path ) <EOL> self . assertEqual ( iterator . page_number , <NUM_LIT:0> ) <EOL> self . assertEqual ( iterator . next_page_token , None ) <EOL> self . assertEqual ( iterator . prefixes , set ( ) ) <EOL> def test_get_items_from_response_empty ( self ) : <EOL> connection = _Connection ( ) <EOL> client = _Client ( connection ) <EOL> bucket = _Bucket ( ) <EOL> iterator = self . _makeOne ( bucket , client = client ) <EOL> blobs = list ( iterator . get_items_from_response ( { } ) ) <EOL> self . assertEqual ( blobs , [ ] ) <EOL> self . assertEqual ( iterator . prefixes , set ( ) ) <EOL> def test_get_items_from_response_non_empty ( self ) : <EOL> from gcloud . storage . blob import Blob <EOL> BLOB_NAME = '<STR_LIT>' <EOL> response = { '<STR_LIT>' : [ { '<STR_LIT:name>' : BLOB_NAME } ] , '<STR_LIT>' : [ '<STR_LIT:foo>' ] } <EOL> connection = _Connection ( ) <EOL> client = _Client ( connection ) <EOL> bucket = _Bucket ( ) <EOL> iterator = self . _makeOne ( bucket , client = client ) <EOL> blobs = list ( iterator . get_items_from_response ( response ) ) <EOL> self . assertEqual ( len ( blobs ) , <NUM_LIT:1> ) <EOL> blob = blobs [ <NUM_LIT:0> ] <EOL> self . assertTrue ( isinstance ( blob , Blob ) ) <EOL> self . assertEqual ( blob . name , BLOB_NAME ) <EOL> self . assertEqual ( iterator . prefixes , set ( [ '<STR_LIT:foo>' ] ) ) <EOL> def test_get_items_from_response_cumulative_prefixes ( self ) : <EOL> from gcloud . storage . blob import Blob <EOL> BLOB_NAME = '<STR_LIT>' <EOL> response1 = { '<STR_LIT>' : [ { '<STR_LIT:name>' : BLOB_NAME } ] , '<STR_LIT>' : [ '<STR_LIT:foo>' ] } <EOL> response2 = { <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] , <EOL> } <EOL> connection = _Connection ( ) <EOL> client = _Client ( connection ) <EOL> bucket = _Bucket ( ) <EOL> iterator = self . _makeOne ( bucket , client = client ) <EOL> blobs = list ( iterator . get_items_from_response ( response1 ) ) <EOL> self . assertEqual ( len ( blobs ) , <NUM_LIT:1> ) <EOL> blob = blobs [ <NUM_LIT:0> ] <EOL> self . assertTrue ( isinstance ( blob , Blob ) ) <EOL> self . assertEqual ( blob . name , BLOB_NAME ) <EOL> self . assertEqual ( iterator . prefixes , set ( [ '<STR_LIT:foo>' ] ) ) <EOL> blobs = list ( iterator . get_items_from_response ( response2 ) ) <EOL> self . assertEqual ( len ( blobs ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( iterator . prefixes , set ( [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] ) ) <EOL> class Test_Bucket ( unittest2 . TestCase ) : <EOL> def _makeOne ( self , client = None , name = None , properties = None ) : <EOL> from gcloud . storage . bucket import Bucket <EOL> if client is None : <EOL> connection = _Connection ( ) <EOL> client = _Client ( connection ) <EOL> bucket = Bucket ( client , name = name ) <EOL> bucket . _properties = properties or { } <EOL> return bucket <EOL> def test_ctor ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> properties = { '<STR_LIT:key>' : '<STR_LIT:value>' } <EOL> bucket = self . _makeOne ( name = NAME , properties = properties ) <EOL> self . assertEqual ( bucket . name , NAME ) <EOL> self . assertEqual ( bucket . _properties , properties ) <EOL> self . assertFalse ( bucket . _acl . loaded ) <EOL> self . assertTrue ( bucket . _acl . bucket is bucket ) <EOL> self . assertFalse ( bucket . _default_object_acl . loaded ) <EOL> self . assertTrue ( bucket . _default_object_acl . bucket is bucket ) <EOL> def test_blob ( self ) : <EOL> from gcloud . storage . blob import Blob <EOL> BUCKET_NAME = '<STR_LIT>' <EOL> BLOB_NAME = '<STR_LIT>' <EOL> CHUNK_SIZE = <NUM_LIT> * <NUM_LIT> <EOL> bucket = self . _makeOne ( name = BUCKET_NAME ) <EOL> blob = bucket . blob ( BLOB_NAME , chunk_size = CHUNK_SIZE ) <EOL> self . assertTrue ( isinstance ( blob , Blob ) ) <EOL> self . assertTrue ( blob . bucket is bucket ) <EOL> self . assertTrue ( blob . client is bucket . client ) <EOL> self . assertEqual ( blob . name , BLOB_NAME ) <EOL> self . assertEqual ( blob . chunk_size , CHUNK_SIZE ) <EOL> def test_exists_miss ( self ) : <EOL> from gcloud . exceptions import NotFound <EOL> class _FakeConnection ( object ) : <EOL> _called_with = [ ] <EOL> @ classmethod <EOL> def api_request ( cls , * args , ** kwargs ) : <EOL> cls . _called_with . append ( ( args , kwargs ) ) <EOL> raise NotFound ( args ) <EOL> BUCKET_NAME = '<STR_LIT>' <EOL> bucket = self . _makeOne ( name = BUCKET_NAME ) <EOL> client = _Client ( _FakeConnection ) <EOL> self . assertFalse ( bucket . exists ( client = client ) ) <EOL> expected_called_kwargs = { <EOL> '<STR_LIT>' : '<STR_LIT:GET>' , <EOL> '<STR_LIT:path>' : bucket . path , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT:name>' , <EOL> } , <EOL> '<STR_LIT>' : None , <EOL> } <EOL> expected_cw = [ ( ( ) , expected_called_kwargs ) ] <EOL> self . assertEqual ( _FakeConnection . _called_with , expected_cw ) <EOL> def test_exists_hit ( self ) : <EOL> class _FakeConnection ( object ) : <EOL> _called_with = [ ] <EOL> @ classmethod <EOL> def api_request ( cls , * args , ** kwargs ) : <EOL> cls . _called_with . append ( ( args , kwargs ) ) <EOL> return object ( ) <EOL> BUCKET_NAME = '<STR_LIT>' <EOL> bucket = self . _makeOne ( name = BUCKET_NAME ) <EOL> client = _Client ( _FakeConnection ) <EOL> self . assertTrue ( bucket . exists ( client = client ) ) <EOL> expected_called_kwargs = { <EOL> '<STR_LIT>' : '<STR_LIT:GET>' , <EOL> '<STR_LIT:path>' : bucket . path , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT:name>' , <EOL> } , <EOL> '<STR_LIT>' : None , <EOL> } <EOL> expected_cw = [ ( ( ) , expected_called_kwargs ) ] <EOL> self . assertEqual ( _FakeConnection . _called_with , expected_cw ) <EOL> def test_create_hit ( self ) : <EOL> BUCKET_NAME = '<STR_LIT>' <EOL> DATA = { '<STR_LIT:name>' : BUCKET_NAME } <EOL> connection = _Connection ( DATA ) <EOL> PROJECT = '<STR_LIT>' <EOL> client = _Client ( connection , project = PROJECT ) <EOL> bucket = self . _makeOne ( client = client , name = BUCKET_NAME ) <EOL> bucket . create ( ) <EOL> kw , = connection . _requested <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , '<STR_LIT:POST>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , { '<STR_LIT>' : PROJECT } ) <EOL> self . assertEqual ( kw [ '<STR_LIT:data>' ] , DATA ) <EOL> def test_create_w_extra_properties ( self ) : <EOL> BUCKET_NAME = '<STR_LIT>' <EOL> PROJECT = '<STR_LIT>' <EOL> CORS = [ { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : [ '<STR_LIT:*>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> } ] <EOL> LIFECYCLE_RULES = [ { <EOL> "<STR_LIT:action>" : { "<STR_LIT:type>" : "<STR_LIT>" } , <EOL> "<STR_LIT>" : { "<STR_LIT>" : <NUM_LIT> } <EOL> } ] <EOL> LOCATION = '<STR_LIT>' <EOL> STORAGE_CLASS = '<STR_LIT>' <EOL> DATA = { <EOL> '<STR_LIT:name>' : BUCKET_NAME , <EOL> '<STR_LIT>' : CORS , <EOL> '<STR_LIT>' : { '<STR_LIT>' : LIFECYCLE_RULES } , <EOL> '<STR_LIT:location>' : LOCATION , <EOL> '<STR_LIT>' : STORAGE_CLASS , <EOL> '<STR_LIT>' : { '<STR_LIT>' : True } , <EOL> } <EOL> connection = _Connection ( DATA ) <EOL> client = _Client ( connection , project = PROJECT ) <EOL> bucket = self . _makeOne ( client = client , name = BUCKET_NAME ) <EOL> bucket . cors = CORS <EOL> bucket . lifecycle_rules = LIFECYCLE_RULES <EOL> bucket . location = LOCATION <EOL> bucket . storage_class = STORAGE_CLASS <EOL> bucket . versioning_enabled = True <EOL> bucket . create ( ) <EOL> kw , = connection . _requested <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , '<STR_LIT:POST>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , { '<STR_LIT>' : PROJECT } ) <EOL> self . assertEqual ( kw [ '<STR_LIT:data>' ] , DATA ) <EOL> def test_acl_property ( self ) : <EOL> from gcloud . storage . acl import BucketACL <EOL> bucket = self . _makeOne ( ) <EOL> acl = bucket . acl <EOL> self . assertTrue ( isinstance ( acl , BucketACL ) ) <EOL> self . assertTrue ( acl is bucket . _acl ) <EOL> def test_default_object_acl_property ( self ) : <EOL> from gcloud . storage . acl import DefaultObjectACL <EOL> bucket = self . _makeOne ( ) <EOL> acl = bucket . default_object_acl <EOL> self . assertTrue ( isinstance ( acl , DefaultObjectACL ) ) <EOL> self . assertTrue ( acl is bucket . _default_object_acl ) <EOL> def test_path_no_name ( self ) : <EOL> bucket = self . _makeOne ( ) <EOL> self . assertRaises ( ValueError , getattr , bucket , '<STR_LIT:path>' ) <EOL> def test_path_w_name ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> self . assertEqual ( bucket . path , '<STR_LIT>' % NAME ) <EOL> def test_get_blob_miss ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> NONESUCH = '<STR_LIT>' <EOL> connection = _Connection ( ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> result = bucket . get_blob ( NONESUCH , client = client ) <EOL> self . assertTrue ( result is None ) <EOL> kw , = connection . _requested <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT:path>' ] , '<STR_LIT>' % ( NAME , NONESUCH ) ) <EOL> def test_get_blob_hit ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> BLOB_NAME = '<STR_LIT>' <EOL> connection = _Connection ( { '<STR_LIT:name>' : BLOB_NAME } ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> blob = bucket . get_blob ( BLOB_NAME , client = client ) <EOL> self . assertTrue ( blob . bucket is bucket ) <EOL> self . assertEqual ( blob . name , BLOB_NAME ) <EOL> kw , = connection . _requested <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT:path>' ] , '<STR_LIT>' % ( NAME , BLOB_NAME ) ) <EOL> def test_list_blobs_defaults ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> connection = _Connection ( { '<STR_LIT>' : [ ] } ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> iterator = bucket . list_blobs ( ) <EOL> blobs = list ( iterator ) <EOL> self . assertEqual ( blobs , [ ] ) <EOL> kw , = connection . _requested <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT:path>' ] , '<STR_LIT>' % NAME ) <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_list_blobs_w_all_arguments ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> MAX_RESULTS = <NUM_LIT:10> <EOL> PAGE_TOKEN = '<STR_LIT>' <EOL> PREFIX = '<STR_LIT>' <EOL> DELIMITER = '<STR_LIT:/>' <EOL> VERSIONS = True <EOL> PROJECTION = '<STR_LIT>' <EOL> FIELDS = '<STR_LIT>' <EOL> EXPECTED = { <EOL> '<STR_LIT>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : PAGE_TOKEN , <EOL> '<STR_LIT>' : PREFIX , <EOL> '<STR_LIT>' : DELIMITER , <EOL> '<STR_LIT>' : VERSIONS , <EOL> '<STR_LIT>' : PROJECTION , <EOL> '<STR_LIT>' : FIELDS , <EOL> } <EOL> connection = _Connection ( { '<STR_LIT>' : [ ] } ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> iterator = bucket . list_blobs ( <EOL> max_results = MAX_RESULTS , <EOL> page_token = PAGE_TOKEN , <EOL> prefix = PREFIX , <EOL> delimiter = DELIMITER , <EOL> versions = VERSIONS , <EOL> projection = PROJECTION , <EOL> fields = FIELDS , <EOL> client = client , <EOL> ) <EOL> blobs = list ( iterator ) <EOL> self . assertEqual ( blobs , [ ] ) <EOL> kw , = connection . _requested <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT:path>' ] , '<STR_LIT>' % NAME ) <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , EXPECTED ) <EOL> def test_list_blobs ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> connection = _Connection ( { '<STR_LIT>' : [ ] } ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> iterator = bucket . list_blobs ( ) <EOL> blobs = list ( iterator ) <EOL> self . assertEqual ( blobs , [ ] ) <EOL> kw , = connection . _requested <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT:path>' ] , '<STR_LIT>' % NAME ) <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_delete_miss ( self ) : <EOL> from gcloud . exceptions import NotFound <EOL> NAME = '<STR_LIT:name>' <EOL> connection = _Connection ( ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> self . assertRaises ( NotFound , bucket . delete ) <EOL> expected_cw = [ { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:path>' : bucket . path , <EOL> '<STR_LIT>' : None , <EOL> } ] <EOL> self . assertEqual ( connection . _deleted_buckets , expected_cw ) <EOL> def test_delete_hit ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> GET_BLOBS_RESP = { '<STR_LIT>' : [ ] } <EOL> connection = _Connection ( GET_BLOBS_RESP ) <EOL> connection . _delete_bucket = True <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> result = bucket . delete ( force = True ) <EOL> self . assertTrue ( result is None ) <EOL> expected_cw = [ { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:path>' : bucket . path , <EOL> '<STR_LIT>' : None , <EOL> } ] <EOL> self . assertEqual ( connection . _deleted_buckets , expected_cw ) <EOL> def test_delete_force_delete_blobs ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> BLOB_NAME1 = '<STR_LIT>' <EOL> BLOB_NAME2 = '<STR_LIT>' <EOL> GET_BLOBS_RESP = { <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:name>' : BLOB_NAME1 } , <EOL> { '<STR_LIT:name>' : BLOB_NAME2 } , <EOL> ] , <EOL> } <EOL> DELETE_BLOB1_RESP = DELETE_BLOB2_RESP = { } <EOL> connection = _Connection ( GET_BLOBS_RESP , DELETE_BLOB1_RESP , <EOL> DELETE_BLOB2_RESP ) <EOL> connection . _delete_bucket = True <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> result = bucket . delete ( force = True ) <EOL> self . assertTrue ( result is None ) <EOL> expected_cw = [ { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:path>' : bucket . path , <EOL> '<STR_LIT>' : None , <EOL> } ] <EOL> self . assertEqual ( connection . _deleted_buckets , expected_cw ) <EOL> def test_delete_force_miss_blobs ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> BLOB_NAME = '<STR_LIT>' <EOL> GET_BLOBS_RESP = { '<STR_LIT>' : [ { '<STR_LIT:name>' : BLOB_NAME } ] } <EOL> connection = _Connection ( GET_BLOBS_RESP ) <EOL> connection . _delete_bucket = True <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> result = bucket . delete ( force = True ) <EOL> self . assertTrue ( result is None ) <EOL> expected_cw = [ { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:path>' : bucket . path , <EOL> '<STR_LIT>' : None , <EOL> } ] <EOL> self . assertEqual ( connection . _deleted_buckets , expected_cw ) <EOL> def test_delete_too_many ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> BLOB_NAME1 = '<STR_LIT>' <EOL> BLOB_NAME2 = '<STR_LIT>' <EOL> GET_BLOBS_RESP = { <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:name>' : BLOB_NAME1 } , <EOL> { '<STR_LIT:name>' : BLOB_NAME2 } , <EOL> ] , <EOL> } <EOL> connection = _Connection ( GET_BLOBS_RESP ) <EOL> connection . _delete_bucket = True <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> bucket . _MAX_OBJECTS_FOR_ITERATION = <NUM_LIT:1> <EOL> self . assertRaises ( ValueError , bucket . delete , force = True ) <EOL> self . assertEqual ( connection . _deleted_buckets , [ ] ) <EOL> def test_delete_blob_miss ( self ) : <EOL> from gcloud . exceptions import NotFound <EOL> NAME = '<STR_LIT:name>' <EOL> NONESUCH = '<STR_LIT>' <EOL> connection = _Connection ( ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> self . assertRaises ( NotFound , bucket . delete_blob , NONESUCH ) <EOL> kw , = connection . _requested <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT:path>' ] , '<STR_LIT>' % ( NAME , NONESUCH ) ) <EOL> def test_delete_blob_hit ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> BLOB_NAME = '<STR_LIT>' <EOL> connection = _Connection ( { } ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> result = bucket . delete_blob ( BLOB_NAME ) <EOL> self . assertTrue ( result is None ) <EOL> kw , = connection . _requested <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT:path>' ] , '<STR_LIT>' % ( NAME , BLOB_NAME ) ) <EOL> def test_delete_blobs_empty ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> connection = _Connection ( ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> bucket . delete_blobs ( [ ] ) <EOL> self . assertEqual ( connection . _requested , [ ] ) <EOL> def test_delete_blobs_hit ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> BLOB_NAME = '<STR_LIT>' <EOL> connection = _Connection ( { } ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> bucket . delete_blobs ( [ BLOB_NAME ] ) <EOL> kw = connection . _requested <EOL> self . assertEqual ( len ( kw ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT:path>' ] , '<STR_LIT>' % ( NAME , BLOB_NAME ) ) <EOL> def test_delete_blobs_miss_no_on_error ( self ) : <EOL> from gcloud . exceptions import NotFound <EOL> NAME = '<STR_LIT:name>' <EOL> BLOB_NAME = '<STR_LIT>' <EOL> NONESUCH = '<STR_LIT>' <EOL> connection = _Connection ( { } ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> self . assertRaises ( NotFound , bucket . delete_blobs , [ BLOB_NAME , NONESUCH ] ) <EOL> kw = connection . _requested <EOL> self . assertEqual ( len ( kw ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT:path>' ] , '<STR_LIT>' % ( NAME , BLOB_NAME ) ) <EOL> self . assertEqual ( kw [ <NUM_LIT:1> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ <NUM_LIT:1> ] [ '<STR_LIT:path>' ] , '<STR_LIT>' % ( NAME , NONESUCH ) ) <EOL> def test_delete_blobs_miss_w_on_error ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> BLOB_NAME = '<STR_LIT>' <EOL> NONESUCH = '<STR_LIT>' <EOL> connection = _Connection ( { } ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> errors = [ ] <EOL> bucket . delete_blobs ( [ BLOB_NAME , NONESUCH ] , errors . append ) <EOL> self . assertEqual ( errors , [ NONESUCH ] ) <EOL> kw = connection . _requested <EOL> self . assertEqual ( len ( kw ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT:path>' ] , '<STR_LIT>' % ( NAME , BLOB_NAME ) ) <EOL> self . assertEqual ( kw [ <NUM_LIT:1> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ <NUM_LIT:1> ] [ '<STR_LIT:path>' ] , '<STR_LIT>' % ( NAME , NONESUCH ) ) <EOL> def test_copy_blobs_wo_name ( self ) : <EOL> SOURCE = '<STR_LIT:source>' <EOL> DEST = '<STR_LIT>' <EOL> BLOB_NAME = '<STR_LIT>' <EOL> class _Blob ( object ) : <EOL> name = BLOB_NAME <EOL> path = '<STR_LIT>' % ( SOURCE , BLOB_NAME ) <EOL> connection = _Connection ( { } ) <EOL> client = _Client ( connection ) <EOL> source = self . _makeOne ( client = client , name = SOURCE ) <EOL> dest = self . _makeOne ( client = client , name = DEST ) <EOL> blob = _Blob ( ) <EOL> new_blob = source . copy_blob ( blob , dest ) <EOL> self . assertTrue ( new_blob . bucket is dest ) <EOL> self . assertEqual ( new_blob . name , BLOB_NAME ) <EOL> kw , = connection . _requested <EOL> COPY_PATH = '<STR_LIT>' % ( SOURCE , BLOB_NAME , <EOL> DEST , BLOB_NAME ) <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , '<STR_LIT:POST>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT:path>' ] , COPY_PATH ) <EOL> def test_copy_blobs_w_name ( self ) : <EOL> SOURCE = '<STR_LIT:source>' <EOL> DEST = '<STR_LIT>' <EOL> BLOB_NAME = '<STR_LIT>' <EOL> NEW_NAME = '<STR_LIT>' <EOL> class _Blob ( object ) : <EOL> name = BLOB_NAME <EOL> path = '<STR_LIT>' % ( SOURCE , BLOB_NAME ) <EOL> connection = _Connection ( { } ) <EOL> client = _Client ( connection ) <EOL> source = self . _makeOne ( client = client , name = SOURCE ) <EOL> dest = self . _makeOne ( client = client , name = DEST ) <EOL> blob = _Blob ( ) <EOL> new_blob = source . copy_blob ( blob , dest , NEW_NAME ) <EOL> self . assertTrue ( new_blob . bucket is dest ) <EOL> self . assertEqual ( new_blob . name , NEW_NAME ) <EOL> kw , = connection . _requested <EOL> COPY_PATH = '<STR_LIT>' % ( SOURCE , BLOB_NAME , <EOL> DEST , NEW_NAME ) <EOL> self . assertEqual ( kw [ '<STR_LIT>' ] , '<STR_LIT:POST>' ) <EOL> self . assertEqual ( kw [ '<STR_LIT:path>' ] , COPY_PATH ) <EOL> def test_rename_blob ( self ) : <EOL> BUCKET_NAME = '<STR_LIT>' <EOL> BLOB_NAME = '<STR_LIT>' <EOL> NEW_BLOB_NAME = '<STR_LIT>' <EOL> DATA = { '<STR_LIT:name>' : NEW_BLOB_NAME } <EOL> connection = _Connection ( DATA ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = BUCKET_NAME ) <EOL> class _Blob ( object ) : <EOL> def __init__ ( self , name , bucket_name ) : <EOL> self . name = name <EOL> self . path = '<STR_LIT>' % ( bucket_name , name ) <EOL> self . _deleted = [ ] <EOL> def delete ( self , client = None ) : <EOL> self . _deleted . append ( client ) <EOL> blob = _Blob ( BLOB_NAME , BUCKET_NAME ) <EOL> renamed_blob = bucket . rename_blob ( blob , NEW_BLOB_NAME , client = client ) <EOL> self . assertTrue ( renamed_blob . bucket is bucket ) <EOL> self . assertEqual ( renamed_blob . name , NEW_BLOB_NAME ) <EOL> self . assertEqual ( blob . _deleted , [ client ] ) <EOL> def test_etag ( self ) : <EOL> ETAG = '<STR_LIT>' <EOL> properties = { '<STR_LIT>' : ETAG } <EOL> bucket = self . _makeOne ( properties = properties ) <EOL> self . assertEqual ( bucket . etag , ETAG ) <EOL> def test_id ( self ) : <EOL> ID = '<STR_LIT>' <EOL> properties = { '<STR_LIT:id>' : ID } <EOL> bucket = self . _makeOne ( properties = properties ) <EOL> self . assertEqual ( bucket . id , ID ) <EOL> def test_location_getter ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> before = { '<STR_LIT:location>' : '<STR_LIT>' } <EOL> bucket = self . _makeOne ( name = NAME , properties = before ) <EOL> self . assertEqual ( bucket . location , '<STR_LIT>' ) <EOL> def test_location_setter ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> self . assertEqual ( bucket . location , None ) <EOL> bucket . location = '<STR_LIT>' <EOL> self . assertEqual ( bucket . location , '<STR_LIT>' ) <EOL> self . assertTrue ( '<STR_LIT:location>' in bucket . _changes ) <EOL> def test_lifecycle_rules_getter ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> LC_RULE = { '<STR_LIT:action>' : { '<STR_LIT:type>' : '<STR_LIT>' } , '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT> } } <EOL> rules = [ LC_RULE ] <EOL> properties = { '<STR_LIT>' : { '<STR_LIT>' : rules } } <EOL> bucket = self . _makeOne ( name = NAME , properties = properties ) <EOL> self . assertEqual ( bucket . lifecycle_rules , rules ) <EOL> self . assertFalse ( bucket . lifecycle_rules is rules ) <EOL> def test_lifecycle_rules_setter ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> LC_RULE = { '<STR_LIT:action>' : { '<STR_LIT:type>' : '<STR_LIT>' } , '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT> } } <EOL> rules = [ LC_RULE ] <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> self . assertEqual ( bucket . lifecycle_rules , [ ] ) <EOL> bucket . lifecycle_rules = rules <EOL> self . assertEqual ( bucket . lifecycle_rules , rules ) <EOL> self . assertTrue ( '<STR_LIT>' in bucket . _changes ) <EOL> def test_cors_getter ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> CORS_ENTRY = { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT:GET>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT:127.0.0.1>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT:Content-Type>' ] , <EOL> } <EOL> properties = { '<STR_LIT>' : [ CORS_ENTRY , { } ] } <EOL> bucket = self . _makeOne ( name = NAME , properties = properties ) <EOL> entries = bucket . cors <EOL> self . assertEqual ( len ( entries ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( entries [ <NUM_LIT:0> ] , CORS_ENTRY ) <EOL> self . assertEqual ( entries [ <NUM_LIT:1> ] , { } ) <EOL> self . assertFalse ( entries [ <NUM_LIT:0> ] is CORS_ENTRY ) <EOL> def test_cors_setter ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> CORS_ENTRY = { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT:GET>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT:127.0.0.1>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT:Content-Type>' ] , <EOL> } <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> self . assertEqual ( bucket . cors , [ ] ) <EOL> bucket . cors = [ CORS_ENTRY ] <EOL> self . assertEqual ( bucket . cors , [ CORS_ENTRY ] ) <EOL> self . assertTrue ( '<STR_LIT>' in bucket . _changes ) <EOL> def test_get_logging_w_prefix ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> LOG_BUCKET = '<STR_LIT>' <EOL> LOG_PREFIX = '<STR_LIT>' <EOL> before = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : LOG_BUCKET , <EOL> '<STR_LIT>' : LOG_PREFIX , <EOL> } , <EOL> } <EOL> bucket = self . _makeOne ( name = NAME , properties = before ) <EOL> info = bucket . get_logging ( ) <EOL> self . assertEqual ( info [ '<STR_LIT>' ] , LOG_BUCKET ) <EOL> self . assertEqual ( info [ '<STR_LIT>' ] , LOG_PREFIX ) <EOL> def test_enable_logging_defaults ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> LOG_BUCKET = '<STR_LIT>' <EOL> before = { '<STR_LIT>' : None } <EOL> bucket = self . _makeOne ( name = NAME , properties = before ) <EOL> self . assertTrue ( bucket . get_logging ( ) is None ) <EOL> bucket . enable_logging ( LOG_BUCKET ) <EOL> info = bucket . get_logging ( ) <EOL> self . assertEqual ( info [ '<STR_LIT>' ] , LOG_BUCKET ) <EOL> self . assertEqual ( info [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_enable_logging ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> LOG_BUCKET = '<STR_LIT>' <EOL> LOG_PFX = '<STR_LIT>' <EOL> before = { '<STR_LIT>' : None } <EOL> bucket = self . _makeOne ( name = NAME , properties = before ) <EOL> self . assertTrue ( bucket . get_logging ( ) is None ) <EOL> bucket . enable_logging ( LOG_BUCKET , LOG_PFX ) <EOL> info = bucket . get_logging ( ) <EOL> self . assertEqual ( info [ '<STR_LIT>' ] , LOG_BUCKET ) <EOL> self . assertEqual ( info [ '<STR_LIT>' ] , LOG_PFX ) <EOL> def test_disable_logging ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> before = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } } <EOL> bucket = self . _makeOne ( name = NAME , properties = before ) <EOL> self . assertTrue ( bucket . get_logging ( ) is not None ) <EOL> bucket . disable_logging ( ) <EOL> self . assertTrue ( bucket . get_logging ( ) is None ) <EOL> def test_metageneration ( self ) : <EOL> METAGENERATION = <NUM_LIT> <EOL> properties = { '<STR_LIT>' : METAGENERATION } <EOL> bucket = self . _makeOne ( properties = properties ) <EOL> self . assertEqual ( bucket . metageneration , METAGENERATION ) <EOL> def test_metageneration_unset ( self ) : <EOL> bucket = self . _makeOne ( ) <EOL> self . assertEqual ( bucket . metageneration , None ) <EOL> def test_metageneration_string_val ( self ) : <EOL> METAGENERATION = <NUM_LIT> <EOL> properties = { '<STR_LIT>' : str ( METAGENERATION ) } <EOL> bucket = self . _makeOne ( properties = properties ) <EOL> self . assertEqual ( bucket . metageneration , METAGENERATION ) <EOL> def test_owner ( self ) : <EOL> OWNER = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> properties = { '<STR_LIT>' : OWNER } <EOL> bucket = self . _makeOne ( properties = properties ) <EOL> owner = bucket . owner <EOL> self . assertEqual ( owner [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( owner [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_project_number ( self ) : <EOL> PROJECT_NUMBER = <NUM_LIT> <EOL> properties = { '<STR_LIT>' : PROJECT_NUMBER } <EOL> bucket = self . _makeOne ( properties = properties ) <EOL> self . assertEqual ( bucket . project_number , PROJECT_NUMBER ) <EOL> def test_project_number_unset ( self ) : <EOL> bucket = self . _makeOne ( ) <EOL> self . assertEqual ( bucket . project_number , None ) <EOL> def test_project_number_string_val ( self ) : <EOL> PROJECT_NUMBER = <NUM_LIT> <EOL> properties = { '<STR_LIT>' : str ( PROJECT_NUMBER ) } <EOL> bucket = self . _makeOne ( properties = properties ) <EOL> self . assertEqual ( bucket . project_number , PROJECT_NUMBER ) <EOL> def test_self_link ( self ) : <EOL> SELF_LINK = '<STR_LIT>' <EOL> properties = { '<STR_LIT>' : SELF_LINK } <EOL> bucket = self . _makeOne ( properties = properties ) <EOL> self . assertEqual ( bucket . self_link , SELF_LINK ) <EOL> def test_storage_class_getter ( self ) : <EOL> STORAGE_CLASS = '<STR_LIT>' <EOL> properties = { '<STR_LIT>' : STORAGE_CLASS } <EOL> bucket = self . _makeOne ( properties = properties ) <EOL> self . assertEqual ( bucket . storage_class , STORAGE_CLASS ) <EOL> def test_storage_class_setter_invalid ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> with self . assertRaises ( ValueError ) : <EOL> bucket . storage_class = '<STR_LIT>' <EOL> self . assertFalse ( '<STR_LIT>' in bucket . _changes ) <EOL> def test_storage_class_setter_STANDARD ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> bucket . storage_class = '<STR_LIT>' <EOL> self . assertEqual ( bucket . storage_class , '<STR_LIT>' ) <EOL> self . assertTrue ( '<STR_LIT>' in bucket . _changes ) <EOL> def test_storage_class_setter_NEARLINE ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> bucket . storage_class = '<STR_LIT>' <EOL> self . assertEqual ( bucket . storage_class , '<STR_LIT>' ) <EOL> self . assertTrue ( '<STR_LIT>' in bucket . _changes ) <EOL> def test_storage_class_setter_DURABLE_REDUCED_AVAILABILITY ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> bucket . storage_class = '<STR_LIT>' <EOL> self . assertEqual ( bucket . storage_class , '<STR_LIT>' ) <EOL> self . assertTrue ( '<STR_LIT>' in bucket . _changes ) <EOL> def test_time_created ( self ) : <EOL> import datetime <EOL> from gcloud . _helpers import _RFC3339_MICROS <EOL> from gcloud . _helpers import UTC <EOL> TIMESTAMP = datetime . datetime ( <NUM_LIT> , <NUM_LIT:11> , <NUM_LIT:5> , <NUM_LIT:20> , <NUM_LIT> , <NUM_LIT> , tzinfo = UTC ) <EOL> TIME_CREATED = TIMESTAMP . strftime ( _RFC3339_MICROS ) <EOL> properties = { '<STR_LIT>' : TIME_CREATED } <EOL> bucket = self . _makeOne ( properties = properties ) <EOL> self . assertEqual ( bucket . time_created , TIMESTAMP ) <EOL> def test_time_created_unset ( self ) : <EOL> bucket = self . _makeOne ( ) <EOL> self . assertEqual ( bucket . time_created , None ) <EOL> def test_versioning_enabled_getter_missing ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> self . assertEqual ( bucket . versioning_enabled , False ) <EOL> def test_versioning_enabled_getter ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> before = { '<STR_LIT>' : { '<STR_LIT>' : True } } <EOL> bucket = self . _makeOne ( name = NAME , properties = before ) <EOL> self . assertEqual ( bucket . versioning_enabled , True ) <EOL> def test_versioning_enabled_setter ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> self . assertFalse ( bucket . versioning_enabled ) <EOL> bucket . versioning_enabled = True <EOL> self . assertTrue ( bucket . versioning_enabled ) <EOL> def test_configure_website_defaults ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> UNSET = { '<STR_LIT>' : { '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None } } <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> bucket . configure_website ( ) <EOL> self . assertEqual ( bucket . _properties , UNSET ) <EOL> def test_configure_website ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> WEBSITE_VAL = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT:html>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } } <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> bucket . configure_website ( '<STR_LIT:html>' , '<STR_LIT>' ) <EOL> self . assertEqual ( bucket . _properties , WEBSITE_VAL ) <EOL> def test_disable_website ( self ) : <EOL> NAME = '<STR_LIT:name>' <EOL> UNSET = { '<STR_LIT>' : { '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None } } <EOL> bucket = self . _makeOne ( name = NAME ) <EOL> bucket . disable_website ( ) <EOL> self . assertEqual ( bucket . _properties , UNSET ) <EOL> def test_make_public_defaults ( self ) : <EOL> from gcloud . storage . acl import _ACLEntity <EOL> NAME = '<STR_LIT:name>' <EOL> permissive = [ { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : _ACLEntity . READER_ROLE } ] <EOL> after = { '<STR_LIT>' : permissive , '<STR_LIT>' : [ ] } <EOL> connection = _Connection ( after ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> bucket . acl . loaded = True <EOL> bucket . default_object_acl . loaded = True <EOL> bucket . make_public ( ) <EOL> self . assertEqual ( list ( bucket . acl ) , permissive ) <EOL> self . assertEqual ( list ( bucket . default_object_acl ) , [ ] ) <EOL> kw = connection . _requested <EOL> self . assertEqual ( len ( kw ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT:path>' ] , '<STR_LIT>' % NAME ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT:data>' ] , { '<STR_LIT>' : after [ '<STR_LIT>' ] } ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def _make_public_w_future_helper ( self , default_object_acl_loaded = True ) : <EOL> from gcloud . storage . acl import _ACLEntity <EOL> NAME = '<STR_LIT:name>' <EOL> permissive = [ { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : _ACLEntity . READER_ROLE } ] <EOL> after1 = { '<STR_LIT>' : permissive , '<STR_LIT>' : [ ] } <EOL> after2 = { '<STR_LIT>' : permissive , '<STR_LIT>' : permissive } <EOL> if default_object_acl_loaded : <EOL> num_requests = <NUM_LIT:2> <EOL> connection = _Connection ( after1 , after2 ) <EOL> else : <EOL> num_requests = <NUM_LIT:3> <EOL> connection = _Connection ( after1 , after1 , after2 ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> bucket . acl . loaded = True <EOL> bucket . default_object_acl . loaded = default_object_acl_loaded <EOL> bucket . make_public ( future = True ) <EOL> self . assertEqual ( list ( bucket . acl ) , permissive ) <EOL> self . assertEqual ( list ( bucket . default_object_acl ) , permissive ) <EOL> kw = connection . _requested <EOL> self . assertEqual ( len ( kw ) , num_requests ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT:path>' ] , '<STR_LIT>' % NAME ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT:data>' ] , { '<STR_LIT>' : permissive } ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> if not default_object_acl_loaded : <EOL> self . assertEqual ( kw [ <NUM_LIT:1> ] [ '<STR_LIT>' ] , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( kw [ <NUM_LIT:1> ] [ '<STR_LIT:path>' ] , '<STR_LIT>' % NAME ) <EOL> self . assertEqual ( kw [ - <NUM_LIT:1> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ - <NUM_LIT:1> ] [ '<STR_LIT:path>' ] , '<STR_LIT>' % NAME ) <EOL> self . assertEqual ( kw [ - <NUM_LIT:1> ] [ '<STR_LIT:data>' ] , { '<STR_LIT>' : permissive } ) <EOL> self . assertEqual ( kw [ - <NUM_LIT:1> ] [ '<STR_LIT>' ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_make_public_w_future ( self ) : <EOL> self . _make_public_w_future_helper ( default_object_acl_loaded = True ) <EOL> def test_make_public_w_future_reload_default ( self ) : <EOL> self . _make_public_w_future_helper ( default_object_acl_loaded = False ) <EOL> def test_make_public_recursive ( self ) : <EOL> from gcloud . storage . acl import _ACLEntity <EOL> from gcloud . storage . bucket import _BlobIterator <EOL> _saved = [ ] <EOL> class _Blob ( object ) : <EOL> _granted = False <EOL> def __init__ ( self , bucket , name ) : <EOL> self . _bucket = bucket <EOL> self . _name = name <EOL> @ property <EOL> def acl ( self ) : <EOL> return self <EOL> def all ( self ) : <EOL> return self <EOL> def grant_read ( self ) : <EOL> self . _granted = True <EOL> def save ( self , client = None ) : <EOL> _saved . append ( <EOL> ( self . _bucket , self . _name , self . _granted , client ) ) <EOL> class _Iterator ( _BlobIterator ) : <EOL> def get_items_from_response ( self , response ) : <EOL> for item in response . get ( '<STR_LIT>' , [ ] ) : <EOL> yield _Blob ( self . bucket , item [ '<STR_LIT:name>' ] ) <EOL> NAME = '<STR_LIT:name>' <EOL> BLOB_NAME = '<STR_LIT>' <EOL> permissive = [ { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : _ACLEntity . READER_ROLE } ] <EOL> after = { '<STR_LIT>' : permissive , '<STR_LIT>' : [ ] } <EOL> connection = _Connection ( after , { '<STR_LIT>' : [ { '<STR_LIT:name>' : BLOB_NAME } ] } ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> bucket . acl . loaded = True <EOL> bucket . default_object_acl . loaded = True <EOL> bucket . _iterator_class = _Iterator <EOL> bucket . make_public ( recursive = True ) <EOL> self . assertEqual ( list ( bucket . acl ) , permissive ) <EOL> self . assertEqual ( list ( bucket . default_object_acl ) , [ ] ) <EOL> self . assertEqual ( _saved , [ ( bucket , BLOB_NAME , True , None ) ] ) <EOL> kw = connection . _requested <EOL> self . assertEqual ( len ( kw ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT:path>' ] , '<STR_LIT>' % NAME ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT:data>' ] , { '<STR_LIT>' : permissive } ) <EOL> self . assertEqual ( kw [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( kw [ <NUM_LIT:1> ] [ '<STR_LIT>' ] , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( kw [ <NUM_LIT:1> ] [ '<STR_LIT:path>' ] , '<STR_LIT>' % NAME ) <EOL> max_results = bucket . _MAX_OBJECTS_FOR_ITERATION + <NUM_LIT:1> <EOL> self . assertEqual ( kw [ <NUM_LIT:1> ] [ '<STR_LIT>' ] , <EOL> { '<STR_LIT>' : max_results , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_make_public_recursive_too_many ( self ) : <EOL> from gcloud . storage . acl import _ACLEntity <EOL> PERMISSIVE = [ { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : _ACLEntity . READER_ROLE } ] <EOL> AFTER = { '<STR_LIT>' : PERMISSIVE , '<STR_LIT>' : [ ] } <EOL> NAME = '<STR_LIT:name>' <EOL> BLOB_NAME1 = '<STR_LIT>' <EOL> BLOB_NAME2 = '<STR_LIT>' <EOL> GET_BLOBS_RESP = { <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:name>' : BLOB_NAME1 } , <EOL> { '<STR_LIT:name>' : BLOB_NAME2 } , <EOL> ] , <EOL> } <EOL> connection = _Connection ( AFTER , GET_BLOBS_RESP ) <EOL> client = _Client ( connection ) <EOL> bucket = self . _makeOne ( client = client , name = NAME ) <EOL> bucket . acl . loaded = True <EOL> bucket . default_object_acl . loaded = True <EOL> bucket . _MAX_OBJECTS_FOR_ITERATION = <NUM_LIT:1> <EOL> self . assertRaises ( ValueError , bucket . make_public , recursive = True ) <EOL> class _Connection ( object ) : <EOL> _delete_bucket = False <EOL> def __init__ ( self , * responses ) : <EOL> self . _responses = responses <EOL> self . _requested = [ ] <EOL> self . _deleted_buckets = [ ] <EOL> @ staticmethod <EOL> def _is_bucket_path ( path ) : <EOL> return path . startswith ( '<STR_LIT>' ) and path . count ( '<STR_LIT:/>' ) == <NUM_LIT:2> <EOL> def api_request ( self , ** kw ) : <EOL> from gcloud . exceptions import NotFound <EOL> self . _requested . append ( kw ) <EOL> method = kw . get ( '<STR_LIT>' ) <EOL> path = kw . get ( '<STR_LIT:path>' , '<STR_LIT>' ) <EOL> if method == '<STR_LIT>' and self . _is_bucket_path ( path ) : <EOL> self . _deleted_buckets . append ( kw ) <EOL> if self . _delete_bucket : <EOL> return <EOL> else : <EOL> raise NotFound ( '<STR_LIT>' ) <EOL> try : <EOL> response , self . _responses = self . _responses [ <NUM_LIT:0> ] , self . _responses [ <NUM_LIT:1> : ] <EOL> except : <EOL> raise NotFound ( '<STR_LIT>' ) <EOL> else : <EOL> return response <EOL> class _Bucket ( object ) : <EOL> path = '<STR_LIT>' <EOL> name = '<STR_LIT:name>' <EOL> def __init__ ( self , client = None ) : <EOL> self . client = client <EOL> class _Client ( object ) : <EOL> def __init__ ( self , connection , project = None ) : <EOL> self . connection = connection <EOL> self . project = project </s>
