<s> import os . path <EOL> import random <EOL> from libcloud . common . types import LibcloudError <EOL> from libcloud . storage . base import Object , Container , StorageDriver <EOL> from libcloud . storage . types import ContainerAlreadyExistsError <EOL> from libcloud . storage . types import ContainerDoesNotExistError <EOL> from libcloud . storage . types import ContainerIsNotEmptyError <EOL> from libcloud . storage . types import ObjectDoesNotExistError <EOL> class DummyFileObject ( file ) : <EOL> def __init__ ( self , yield_count = <NUM_LIT:5> , chunk_len = <NUM_LIT:10> ) : <EOL> self . _yield_count = yield_count <EOL> self . _chunk_len = chunk_len <EOL> def read ( self , size ) : <EOL> i = <NUM_LIT:0> <EOL> while i < self . _yield_count : <EOL> yield self . _get_chunk ( self . _chunk_len ) <EOL> i += <NUM_LIT:1> <EOL> raise StopIteration <EOL> def _get_chunk ( self , chunk_len ) : <EOL> chunk = [ str ( x ) for x in random . randint ( <NUM_LIT> , <NUM_LIT> ) ] <EOL> return chunk <EOL> def __len__ ( self ) : <EOL> return self . _yield_count * self . _chunk_len <EOL> class DummyIterator ( object ) : <EOL> def __init__ ( self , data = None ) : <EOL> self . _data = data or [ ] <EOL> self . _current_item = <NUM_LIT:0> <EOL> def next ( self ) : <EOL> if self . _current_item == len ( self . _data ) : <EOL> raise StopIteration <EOL> value = self . _data [ self . _current_item ] <EOL> self . _current_item += <NUM_LIT:1> <EOL> return value <EOL> class DummyStorageDriver ( StorageDriver ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> def __init__ ( self , api_key , api_secret ) : <EOL> self . _containers = { } <EOL> def get_meta_data ( self ) : <EOL> """<STR_LIT>""" <EOL> container_count = len ( self . _containers ) <EOL> object_count = sum ( [ len ( self . _containers [ container ] [ '<STR_LIT>' ] ) for <EOL> container in self . _containers ] ) <EOL> bytes_used = <NUM_LIT:0> <EOL> for container in self . _containers : <EOL> objects = self . _containers [ container ] [ '<STR_LIT>' ] <EOL> for _ , obj in objects . iteritems ( ) : <EOL> bytes_used += obj . size <EOL> return { '<STR_LIT>' : int ( container_count ) , <EOL> '<STR_LIT>' : int ( object_count ) , <EOL> '<STR_LIT>' : int ( bytes_used ) } <EOL> def list_containers ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ container [ '<STR_LIT>' ] for container in <EOL> self . _containers . values ( ) ] <EOL> def list_container_objects ( self , container ) : <EOL> container = self . get_container ( container . name ) <EOL> return container . objects <EOL> def get_container ( self , container_name ) : <EOL> """<STR_LIT>""" <EOL> if container_name not in self . _containers : <EOL> raise ContainerDoesNotExistError ( driver = self , value = None , <EOL> container_name = container_name ) <EOL> return self . _containers [ container_name ] [ '<STR_LIT>' ] <EOL> def get_object ( self , container_name , object_name ) : <EOL> """<STR_LIT>""" <EOL> container = self . get_container ( container_name ) <EOL> container_objects = self . _containers [ container_name ] [ '<STR_LIT>' ] <EOL> if object_name not in container_objects : <EOL> raise ObjectDoesNotExistError ( object_name = object_name , value = None , <EOL> driver = self ) <EOL> return container_objects [ object_name ] <EOL> def create_container ( self , container_name ) : <EOL> """<STR_LIT>""" <EOL> if container_name in self . _containers : <EOL> raise ContainerAlreadyExistsError ( container_name = container_name , <EOL> value = None , driver = self ) <EOL> extra = { '<STR_LIT>' : <NUM_LIT:0> } <EOL> container = Container ( name = container_name , extra = extra , driver = self ) <EOL> self . _containers [ container_name ] = { '<STR_LIT>' : container , <EOL> '<STR_LIT>' : { } <EOL> } <EOL> return container <EOL> def delete_container ( self , container ) : <EOL> """<STR_LIT>""" <EOL> container_name = container . name <EOL> if container_name not in self . _containers : <EOL> raise ContainerDoesNotExistError ( container_name = container_name , <EOL> value = None , driver = self ) <EOL> container = self . _containers [ container_name ] <EOL> if len ( container [ '<STR_LIT>' ] ) > <NUM_LIT:0> : <EOL> raise ContainerIsNotEmptyError ( container_name = container_name , <EOL> value = None , driver = self ) <EOL> del self . _containers [ container_name ] <EOL> return True <EOL> def download_object ( self , obj , destination_path , overwrite_existing = False , <EOL> delete_on_failure = True ) : <EOL> kwargs_dict = { '<STR_LIT>' : obj , <EOL> '<STR_LIT>' : DummyFileObject ( ) , <EOL> '<STR_LIT>' : destination_path , <EOL> '<STR_LIT>' : overwrite_existing , <EOL> '<STR_LIT>' : delete_on_failure } <EOL> return self . _save_object ( ** kwargs_dict ) <EOL> def download_object_as_stream ( self , obj , chunk_size = None ) : <EOL> """<STR_LIT>""" <EOL> return DummyFileObject ( ) <EOL> def upload_object ( self , file_path , container , object_name , extra = None , <EOL> file_hash = None ) : <EOL> """<STR_LIT>""" <EOL> if not os . path . exists ( file_path ) : <EOL> raise LibcloudError ( value = '<STR_LIT>' % ( file_path ) , <EOL> driver = self ) <EOL> size = os . path . getsize ( file_path ) <EOL> return self . _add_object ( container = container , object_name = object_name , <EOL> size = size , extra = extra ) <EOL> def upload_object_via_stream ( self , iterator , container , object_name , extra = None ) : <EOL> """<STR_LIT>""" <EOL> size = len ( iterator ) <EOL> return self . _add_object ( container = container , object_name = object_name , <EOL> size = size , extra = extra ) <EOL> def delete_object ( self , obj ) : <EOL> """<STR_LIT>""" <EOL> container_name = obj . container . name <EOL> object_name = obj . name <EOL> obj = self . get_object ( container_name = container_name , <EOL> object_name = object_name ) <EOL> del self . _containers [ container_name ] [ '<STR_LIT>' ] [ object_name ] <EOL> return True <EOL> def _add_object ( self , container , object_name , size , extra = None ) : <EOL> container = self . get_container ( container . name ) <EOL> extra = extra or { } <EOL> meta_data = extra . get ( '<STR_LIT>' , { } ) <EOL> obj = Object ( name = object_name , size = size , extra = extra , hash = None , <EOL> meta_data = meta_data , container = container , driver = self ) <EOL> self . _containers [ container . name ] [ '<STR_LIT>' ] [ object_name ] = obj <EOL> return obj <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> import doctest <EOL> doctest . testmod ( ) </s>
