<s> """<STR_LIT>""" <EOL> from gslib . cloud_api import ServiceException <EOL> from gslib . third_party . storage_apitools import storage_v1_messages as apitools_messages <EOL> from gslib . translation_helper import CreateBucketNotFoundException <EOL> from gslib . translation_helper import CreateObjectNotFoundException <EOL> class MockObject ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , root_object , contents = '<STR_LIT>' ) : <EOL> self . root_object = root_object <EOL> self . contents = contents <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % ( self . root_object . bucket , <EOL> self . root_object . name , <EOL> self . root_object . generation ) <EOL> def __repr__ ( self ) : <EOL> return str ( self ) <EOL> class MockBucket ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , bucket_name , versioned = False ) : <EOL> self . root_object = apitools_messages . Bucket ( <EOL> name = bucket_name , <EOL> versioning = apitools_messages . Bucket . VersioningValue ( enabled = versioned ) ) <EOL> self . objects = { } <EOL> def CreateObject ( self , object_name , contents = '<STR_LIT>' ) : <EOL> return self . CreateObjectWithMetadata ( MockObject ( <EOL> apitools_messages . Object ( name = object_name , contents = contents ) ) ) <EOL> def CreateObjectWithMetadata ( self , apitools_object , contents = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> object_name = apitools_object . name <EOL> if ( self . root_object . versioning and self . root_object . versioning . enabled and <EOL> apitools_object . name in self . objects ) : <EOL> if '<STR_LIT>' in self . objects [ object_name ] : <EOL> apitools_object . generation = ( <EOL> self . objects [ object_name ] [ '<STR_LIT>' ] . root_object . generation + <NUM_LIT:1> ) <EOL> if '<STR_LIT>' not in self . objects [ object_name ] : <EOL> self . objects [ object_name ] [ '<STR_LIT>' ] = [ ] <EOL> self . objects [ object_name ] [ '<STR_LIT>' ] . append ( <EOL> self . objects [ object_name ] [ '<STR_LIT>' ] ) <EOL> elif ( '<STR_LIT>' in self . objects [ object_name ] and <EOL> self . objects [ object_name ] [ '<STR_LIT>' ] ) : <EOL> apitools_object . generation = ( <EOL> self . objects [ object_name ] [ '<STR_LIT>' ] [ - <NUM_LIT:1> ] . root_object . generation <EOL> + <NUM_LIT:1> ) <EOL> else : <EOL> apitools_object . generation = <NUM_LIT:1> <EOL> self . objects [ object_name ] = { } <EOL> new_object = MockObject ( apitools_object , contents = contents ) <EOL> self . objects [ object_name ] [ '<STR_LIT>' ] = new_object <EOL> return new_object <EOL> class MockCloudApi ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , provider = '<STR_LIT>' ) : <EOL> self . buckets = { } <EOL> self . provider = provider <EOL> def MockCreateBucket ( self , bucket_name ) : <EOL> """<STR_LIT>""" <EOL> if bucket_name in self . buckets : <EOL> raise ServiceException ( '<STR_LIT>' % bucket_name , <EOL> status = <NUM_LIT> ) <EOL> self . buckets [ bucket_name ] = MockBucket ( bucket_name ) <EOL> def MockCreateVersionedBucket ( self , bucket_name ) : <EOL> """<STR_LIT>""" <EOL> if bucket_name in self . buckets : <EOL> raise ServiceException ( '<STR_LIT>' % bucket_name , <EOL> status = <NUM_LIT> ) <EOL> self . buckets [ bucket_name ] = MockBucket ( bucket_name , versioned = True ) <EOL> def MockCreateObject ( self , bucket_name , object_name , contents = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if bucket_name not in self . buckets : <EOL> self . MockCreateBucket ( bucket_name ) <EOL> self . buckets [ bucket_name ] . CreateObject ( object_name , contents = contents ) <EOL> def MockCreateObjectWithMetadata ( self , apitools_object , contents = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> assert apitools_object . bucket , '<STR_LIT>' <EOL> assert apitools_object . name , '<STR_LIT>' <EOL> if apitools_object . bucket not in self . buckets : <EOL> self . MockCreateBucket ( apitools_object . bucket ) <EOL> return self . buckets [ apitools_object . bucket ] . CreateObjectWithMetadata ( <EOL> apitools_object , contents = contents ) . root_object <EOL> def GetObjectMetadata ( self , bucket_name , object_name , generation = None , <EOL> provider = None , fields = None ) : <EOL> """<STR_LIT>""" <EOL> if generation : <EOL> generation = long ( generation ) <EOL> if bucket_name in self . buckets : <EOL> bucket = self . buckets [ bucket_name ] <EOL> if object_name in bucket . objects and bucket . objects [ object_name ] : <EOL> if generation : <EOL> if '<STR_LIT>' in bucket . objects [ object_name ] : <EOL> for obj in bucket . objects [ object_name ] [ '<STR_LIT>' ] : <EOL> if obj . root_object . generation == generation : <EOL> return obj . root_object <EOL> if '<STR_LIT>' in bucket . objects [ object_name ] : <EOL> if ( bucket . objects [ object_name ] [ '<STR_LIT>' ] . root_object . generation == <EOL> generation ) : <EOL> return bucket . objects [ object_name ] [ '<STR_LIT>' ] . root_object <EOL> else : <EOL> if '<STR_LIT>' in bucket . objects [ object_name ] : <EOL> return bucket . objects [ object_name ] [ '<STR_LIT>' ] . root_object <EOL> raise CreateObjectNotFoundException ( <NUM_LIT> , self . provider , bucket_name , <EOL> object_name ) <EOL> raise CreateBucketNotFoundException ( <NUM_LIT> , self . provider , bucket_name ) </s>
