<s> from boto . compat import StringIO <EOL> from tests . compat import mock , unittest <EOL> ANY = mock . ANY <EOL> from boto . glacier import vault <EOL> from boto . glacier . job import Job <EOL> from boto . glacier . response import GlacierResponse <EOL> class TestVault ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . size_patch = mock . patch ( '<STR_LIT>' ) <EOL> self . getsize = self . size_patch . start ( ) <EOL> self . api = mock . Mock ( ) <EOL> self . vault = vault . Vault ( self . api , None ) <EOL> self . vault . name = '<STR_LIT>' <EOL> self . mock_open = mock . mock_open ( ) <EOL> stringio = StringIO ( '<STR_LIT:content>' ) <EOL> self . mock_open . return_value . read = stringio . read <EOL> def tearDown ( self ) : <EOL> self . size_patch . stop ( ) <EOL> @ mock . patch ( '<STR_LIT>' , <EOL> return_value = [ b'<STR_LIT:abc>' , b'<STR_LIT>' ] ) <EOL> def test_upload_archive_small_file ( self , compute_hashes ) : <EOL> self . getsize . return_value = <NUM_LIT:1> <EOL> self . api . upload_archive . return_value = { '<STR_LIT>' : '<STR_LIT>' } <EOL> with mock . patch ( '<STR_LIT>' , self . mock_open , <EOL> create = True ) : <EOL> archive_id = self . vault . upload_archive ( <EOL> '<STR_LIT:filename>' , '<STR_LIT>' ) <EOL> self . assertEqual ( archive_id , '<STR_LIT>' ) <EOL> self . api . upload_archive . assert_called_with ( <EOL> '<STR_LIT>' , self . mock_open . return_value , <EOL> mock . ANY , mock . ANY , '<STR_LIT>' ) <EOL> def test_small_part_size_is_obeyed ( self ) : <EOL> self . vault . DefaultPartSize = <NUM_LIT:2> * <NUM_LIT> * <NUM_LIT> <EOL> self . vault . create_archive_writer = mock . Mock ( ) <EOL> self . getsize . return_value = <NUM_LIT:1> <EOL> with mock . patch ( '<STR_LIT>' , self . mock_open , <EOL> create = True ) : <EOL> self . vault . create_archive_from_file ( '<STR_LIT>' ) <EOL> self . vault . create_archive_writer . assert_called_with ( <EOL> description = mock . ANY , part_size = self . vault . DefaultPartSize ) <EOL> def test_large_part_size_is_obeyed ( self ) : <EOL> self . vault . DefaultPartSize = <NUM_LIT:8> * <NUM_LIT> * <NUM_LIT> <EOL> self . vault . create_archive_writer = mock . Mock ( ) <EOL> self . getsize . return_value = <NUM_LIT:1> <EOL> with mock . patch ( '<STR_LIT>' , self . mock_open , <EOL> create = True ) : <EOL> self . vault . create_archive_from_file ( '<STR_LIT>' ) <EOL> self . vault . create_archive_writer . assert_called_with ( <EOL> description = mock . ANY , part_size = self . vault . DefaultPartSize ) <EOL> def test_part_size_needs_to_be_adjusted ( self ) : <EOL> self . getsize . return_value = <NUM_LIT> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> <EOL> self . vault . create_archive_writer = mock . Mock ( ) <EOL> with mock . patch ( '<STR_LIT>' , self . mock_open , <EOL> create = True ) : <EOL> self . vault . create_archive_from_file ( '<STR_LIT>' ) <EOL> expected_part_size = <NUM_LIT:64> * <NUM_LIT> * <NUM_LIT> <EOL> self . vault . create_archive_writer . assert_called_with ( <EOL> description = mock . ANY , part_size = expected_part_size ) <EOL> def test_retrieve_inventory ( self ) : <EOL> class FakeResponse ( object ) : <EOL> status = <NUM_LIT> <EOL> def getheader ( self , key , default = None ) : <EOL> if key == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif key == '<STR_LIT:Content-Type>' : <EOL> return '<STR_LIT:application/json>' <EOL> return '<STR_LIT>' <EOL> def read ( self , amt = None ) : <EOL> return b"""<STR_LIT>""" <EOL> raw_resp = FakeResponse ( ) <EOL> init_resp = GlacierResponse ( raw_resp , [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> raw_resp_2 = FakeResponse ( ) <EOL> desc_resp = GlacierResponse ( raw_resp_2 , [ ] ) <EOL> with mock . patch . object ( self . vault . layer1 , '<STR_LIT>' , <EOL> return_value = init_resp ) : <EOL> with mock . patch . object ( self . vault . layer1 , '<STR_LIT>' , <EOL> return_value = desc_resp ) : <EOL> self . assertEqual ( self . vault . retrieve_inventory ( ) , '<STR_LIT>' ) <EOL> job = self . vault . retrieve_inventory_job ( ) <EOL> self . assertTrue ( isinstance ( job , Job ) ) <EOL> self . assertEqual ( job . id , '<STR_LIT>' ) <EOL> class TestConcurrentUploads ( unittest . TestCase ) : <EOL> def test_concurrent_upload_file ( self ) : <EOL> v = vault . Vault ( None , None ) <EOL> with mock . patch ( '<STR_LIT>' ) as c : <EOL> c . return_value . upload . return_value = '<STR_LIT>' <EOL> archive_id = v . concurrent_create_archive_from_file ( <EOL> '<STR_LIT:filename>' , '<STR_LIT>' ) <EOL> c . return_value . upload . assert_called_with ( '<STR_LIT:filename>' , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( archive_id , '<STR_LIT>' ) <EOL> def test_concurrent_upload_forwards_kwargs ( self ) : <EOL> v = vault . Vault ( None , None ) <EOL> with mock . patch ( '<STR_LIT>' ) as c : <EOL> c . return_value . upload . return_value = '<STR_LIT>' <EOL> archive_id = v . concurrent_create_archive_from_file ( <EOL> '<STR_LIT:filename>' , '<STR_LIT>' , num_threads = <NUM_LIT:10> , <EOL> part_size = <NUM_LIT> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT:8> ) <EOL> c . assert_called_with ( None , None , num_threads = <NUM_LIT:10> , <EOL> part_size = <NUM_LIT> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT:8> ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
