<s> from __future__ import with_statement <EOL> import os <EOL> import sys <EOL> import shutil <EOL> import unittest <EOL> import tempfile <EOL> import mock <EOL> from libcloud . common . types import LibcloudError <EOL> from libcloud . storage . base import Container <EOL> from libcloud . storage . types import ContainerDoesNotExistError <EOL> from libcloud . storage . types import ContainerAlreadyExistsError <EOL> from libcloud . storage . types import ContainerIsNotEmptyError <EOL> from libcloud . storage . types import InvalidContainerNameError <EOL> try : <EOL> from libcloud . storage . drivers . local import LocalStorageDriver <EOL> from libcloud . storage . drivers . local import LockLocalStorage <EOL> from lockfile import LockTimeout <EOL> except ImportError : <EOL> print ( '<STR_LIT>' ) <EOL> LocalStorageDriver = None <EOL> LockTimeout = None <EOL> class LocalTests ( unittest . TestCase ) : <EOL> driver_type = LocalStorageDriver <EOL> @ classmethod <EOL> def create_driver ( self ) : <EOL> self . key = tempfile . mkdtemp ( ) <EOL> return self . driver_type ( self . key , None ) <EOL> def setUp ( self ) : <EOL> self . driver = self . create_driver ( ) <EOL> def tearDown ( self ) : <EOL> shutil . rmtree ( self . key ) <EOL> self . key = None <EOL> def make_tmp_file ( self ) : <EOL> _ , tmppath = tempfile . mkstemp ( ) <EOL> with open ( tmppath , '<STR_LIT:wb>' ) as fp : <EOL> fp . write ( b'<STR_LIT>' * <NUM_LIT> ) <EOL> return tmppath <EOL> def remove_tmp_file ( self , tmppath ) : <EOL> os . unlink ( tmppath ) <EOL> def test_list_containers_empty ( self ) : <EOL> containers = self . driver . list_containers ( ) <EOL> self . assertEqual ( len ( containers ) , <NUM_LIT:0> ) <EOL> def test_containers_success ( self ) : <EOL> self . driver . create_container ( '<STR_LIT>' ) <EOL> self . driver . create_container ( '<STR_LIT>' ) <EOL> containers = self . driver . list_containers ( ) <EOL> self . assertEqual ( len ( containers ) , <NUM_LIT:2> ) <EOL> container = containers [ <NUM_LIT:1> ] <EOL> self . assertTrue ( '<STR_LIT>' in container . extra ) <EOL> self . assertTrue ( '<STR_LIT>' in container . extra ) <EOL> self . assertTrue ( '<STR_LIT>' in container . extra ) <EOL> objects = self . driver . list_container_objects ( container = container ) <EOL> self . assertEqual ( len ( objects ) , <NUM_LIT:0> ) <EOL> objects = container . list_objects ( ) <EOL> self . assertEqual ( len ( objects ) , <NUM_LIT:0> ) <EOL> for container in containers : <EOL> self . driver . delete_container ( container ) <EOL> def test_objects_success ( self ) : <EOL> tmppath = self . make_tmp_file ( ) <EOL> container = self . driver . create_container ( '<STR_LIT>' ) <EOL> obj1 = container . upload_object ( tmppath , '<STR_LIT>' ) <EOL> obj2 = container . upload_object ( tmppath , '<STR_LIT>' ) <EOL> obj3 = container . upload_object ( tmppath , '<STR_LIT>' ) <EOL> obj4 = container . upload_object ( tmppath , '<STR_LIT>' ) <EOL> with open ( tmppath , '<STR_LIT:rb>' ) as tmpfile : <EOL> obj5 = container . upload_object_via_stream ( tmpfile , '<STR_LIT>' ) <EOL> objects = self . driver . list_container_objects ( container = container ) <EOL> self . assertEqual ( len ( objects ) , <NUM_LIT:5> ) <EOL> for obj in objects : <EOL> self . assertNotEqual ( obj . hash , None ) <EOL> self . assertEqual ( obj . size , <NUM_LIT> ) <EOL> self . assertEqual ( obj . container . name , '<STR_LIT>' ) <EOL> self . assertTrue ( '<STR_LIT>' in obj . extra ) <EOL> self . assertTrue ( '<STR_LIT>' in obj . extra ) <EOL> self . assertTrue ( '<STR_LIT>' in obj . extra ) <EOL> obj1 . delete ( ) <EOL> obj2 . delete ( ) <EOL> objects = container . list_objects ( ) <EOL> self . assertEqual ( len ( objects ) , <NUM_LIT:3> ) <EOL> container . delete_object ( obj3 ) <EOL> container . delete_object ( obj4 ) <EOL> container . delete_object ( obj5 ) <EOL> objects = container . list_objects ( ) <EOL> self . assertEqual ( len ( objects ) , <NUM_LIT:0> ) <EOL> container . delete ( ) <EOL> self . remove_tmp_file ( tmppath ) <EOL> def test_get_container_doesnt_exist ( self ) : <EOL> try : <EOL> self . driver . get_container ( container_name = '<STR_LIT>' ) <EOL> except ContainerDoesNotExistError : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_get_container_success ( self ) : <EOL> self . driver . create_container ( '<STR_LIT>' ) <EOL> container = self . driver . get_container ( container_name = '<STR_LIT>' ) <EOL> self . assertTrue ( container . name , '<STR_LIT>' ) <EOL> container . delete ( ) <EOL> def test_get_object_container_doesnt_exist ( self ) : <EOL> try : <EOL> self . driver . get_object ( container_name = '<STR_LIT>' , <EOL> object_name = '<STR_LIT:test>' ) <EOL> except ContainerDoesNotExistError : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_get_object_success ( self ) : <EOL> tmppath = self . make_tmp_file ( ) <EOL> container = self . driver . create_container ( '<STR_LIT>' ) <EOL> container . upload_object ( tmppath , '<STR_LIT:test>' ) <EOL> obj = self . driver . get_object ( container_name = '<STR_LIT>' , <EOL> object_name = '<STR_LIT:test>' ) <EOL> self . assertEqual ( obj . name , '<STR_LIT:test>' ) <EOL> self . assertEqual ( obj . container . name , '<STR_LIT>' ) <EOL> self . assertEqual ( obj . size , <NUM_LIT> ) <EOL> self . assertNotEqual ( obj . hash , None ) <EOL> self . assertTrue ( '<STR_LIT>' in obj . extra ) <EOL> self . assertTrue ( '<STR_LIT>' in obj . extra ) <EOL> self . assertTrue ( '<STR_LIT>' in obj . extra ) <EOL> obj . delete ( ) <EOL> container . delete ( ) <EOL> self . remove_tmp_file ( tmppath ) <EOL> def test_create_container_invalid_name ( self ) : <EOL> try : <EOL> self . driver . create_container ( container_name = '<STR_LIT>' ) <EOL> except InvalidContainerNameError : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_create_container_already_exists ( self ) : <EOL> container = self . driver . create_container ( <EOL> container_name = '<STR_LIT>' ) <EOL> try : <EOL> self . driver . create_container ( container_name = '<STR_LIT>' ) <EOL> except ContainerAlreadyExistsError : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> self . driver . delete_container ( container ) <EOL> def test_create_container_success ( self ) : <EOL> name = '<STR_LIT>' <EOL> container = self . driver . create_container ( container_name = name ) <EOL> self . assertEqual ( container . name , name ) <EOL> self . driver . delete_container ( container ) <EOL> def test_delete_container_doesnt_exist ( self ) : <EOL> container = Container ( name = '<STR_LIT>' , extra = None , <EOL> driver = self . driver ) <EOL> try : <EOL> self . driver . delete_container ( container = container ) <EOL> except ContainerDoesNotExistError : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_delete_container_not_empty ( self ) : <EOL> tmppath = self . make_tmp_file ( ) <EOL> container = self . driver . create_container ( '<STR_LIT>' ) <EOL> obj = container . upload_object ( tmppath , '<STR_LIT:test>' ) <EOL> try : <EOL> self . driver . delete_container ( container = container ) <EOL> except ContainerIsNotEmptyError : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> obj . delete ( ) <EOL> self . remove_tmp_file ( tmppath ) <EOL> self . assertTrue ( self . driver . delete_container ( container = container ) ) <EOL> def test_delete_container_not_found ( self ) : <EOL> container = Container ( name = '<STR_LIT>' , extra = { } , <EOL> driver = self . driver ) <EOL> try : <EOL> self . driver . delete_container ( container = container ) <EOL> except ContainerDoesNotExistError : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> def test_delete_container_success ( self ) : <EOL> container = self . driver . create_container ( '<STR_LIT>' ) <EOL> self . assertTrue ( self . driver . delete_container ( container = container ) ) <EOL> def test_download_object_success ( self ) : <EOL> tmppath = self . make_tmp_file ( ) <EOL> container = self . driver . create_container ( '<STR_LIT>' ) <EOL> obj = container . upload_object ( tmppath , '<STR_LIT:test>' ) <EOL> destination_path = tmppath + '<STR_LIT>' <EOL> result = self . driver . download_object ( obj = obj , <EOL> destination_path = destination_path , <EOL> overwrite_existing = False , <EOL> delete_on_failure = True ) <EOL> self . assertTrue ( result ) <EOL> obj . delete ( ) <EOL> container . delete ( ) <EOL> self . remove_tmp_file ( tmppath ) <EOL> os . unlink ( destination_path ) <EOL> def test_download_object_and_overwrite ( self ) : <EOL> tmppath = self . make_tmp_file ( ) <EOL> container = self . driver . create_container ( '<STR_LIT>' ) <EOL> obj = container . upload_object ( tmppath , '<STR_LIT:test>' ) <EOL> destination_path = tmppath + '<STR_LIT>' <EOL> result = self . driver . download_object ( obj = obj , <EOL> destination_path = destination_path , <EOL> overwrite_existing = False , <EOL> delete_on_failure = True ) <EOL> self . assertTrue ( result ) <EOL> try : <EOL> self . driver . download_object ( obj = obj , <EOL> destination_path = destination_path , <EOL> overwrite_existing = False , <EOL> delete_on_failure = True ) <EOL> except LibcloudError : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> result = self . driver . download_object ( obj = obj , <EOL> destination_path = destination_path , <EOL> overwrite_existing = True , <EOL> delete_on_failure = True ) <EOL> self . assertTrue ( result ) <EOL> obj . delete ( ) <EOL> container . delete ( ) <EOL> self . remove_tmp_file ( tmppath ) <EOL> os . unlink ( destination_path ) <EOL> def test_download_object_as_stream_success ( self ) : <EOL> tmppath = self . make_tmp_file ( ) <EOL> container = self . driver . create_container ( '<STR_LIT>' ) <EOL> obj = container . upload_object ( tmppath , '<STR_LIT:test>' ) <EOL> stream = self . driver . download_object_as_stream ( obj = obj , <EOL> chunk_size = <NUM_LIT> ) <EOL> self . assertTrue ( hasattr ( stream , '<STR_LIT>' ) ) <EOL> data = b'<STR_LIT>' . join ( stream ) <EOL> self . assertTrue ( len ( data ) , <NUM_LIT> ) <EOL> obj . delete ( ) <EOL> container . delete ( ) <EOL> self . remove_tmp_file ( tmppath ) <EOL> @ mock . patch ( "<STR_LIT>" , <EOL> mock . MagicMock ( side_effect = LockTimeout ) ) <EOL> def test_proper_lockfile_imports ( self ) : <EOL> lls = LockLocalStorage ( "<STR_LIT>" ) <EOL> self . assertRaises ( LibcloudError , lls . __enter__ ) <EOL> if not LocalStorageDriver : <EOL> class LocalTests ( unittest . TestCase ) : <EOL> pass <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> sys . exit ( unittest . main ( ) ) </s>
