<s> """<STR_LIT>""" <EOL> import boto <EOL> import mock <EOL> import logging <EOL> from src import basetest <EOL> from src . clients import s3 <EOL> class S3Test ( basetest . TestCase ) : <EOL> def testList ( self ) : <EOL> objs = [ '<STR_LIT:foo>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> mock_service = mock . MagicMock ( ) <EOL> mock_list_resp = [ { '<STR_LIT:name>' : o } for o in objs ] <EOL> config = { '<STR_LIT>' : mock_list_resp } <EOL> mock_service . configure_mock ( ** config ) <EOL> with mock . patch . object ( boto , '<STR_LIT>' ) as mock_connect_s3 : <EOL> storage = s3 . S3 ( ) <EOL> storage . service = mock_service <EOL> mock_connect_s3 . assert_called_once_with ( None , None ) <EOL> res = storage . ListBucket ( '<STR_LIT>' ) <EOL> for o in objs : <EOL> self . assertIn ( o , res ) <EOL> def testReadSimpleSmall ( self ) : <EOL> s3_url = '<STR_LIT>' <EOL> s3_content_type = '<STR_LIT>' <EOL> payload = '<STR_LIT>' <EOL> def _MockedGetContentsToFile ( buf , <EOL> headers = None , <EOL> cb = None , <EOL> num_cb = <NUM_LIT:10> ) : <EOL> buf . write ( payload ) <EOL> self . assertIsNone ( headers ) <EOL> self . assertIsNone ( cb ) <EOL> self . assertEquals ( num_cb , <NUM_LIT:0> ) <EOL> mock_boto_key = mock . MagicMock ( ) <EOL> mock_boto_key . size . return_value = len ( payload ) <EOL> mock_boto_key . content_type = s3_content_type <EOL> mock_boto_key . get_contents_to_file = _MockedGetContentsToFile <EOL> mock_service = mock . MagicMock ( ) <EOL> config = { '<STR_LIT>' : mock_boto_key } <EOL> mock_service . configure_mock ( ** config ) <EOL> with mock . patch . object ( boto , '<STR_LIT>' ) : <EOL> storage = s3 . S3 ( ) <EOL> storage . service = mock_service <EOL> buf = storage . ReadObject ( url = s3_url ) <EOL> self . assertEquals ( payload , buf . read ( <NUM_LIT> ) ) <EOL> buf . close ( ) <EOL> def testReadCallbackSmall ( self ) : <EOL> s3_url = '<STR_LIT>' <EOL> s3_content_type = '<STR_LIT>' <EOL> payload = '<STR_LIT>' <EOL> def _MockedGetContentsToFile ( buf , <EOL> headers = None , <EOL> cb = None , <EOL> num_cb = <NUM_LIT:10> ) : <EOL> buf . write ( payload ) <EOL> self . assertIsNone ( headers ) <EOL> self . assertIsNotNone ( cb ) <EOL> self . assertEquals ( num_cb , <NUM_LIT:1> ) <EOL> def _Callback ( buf , bytes_read , unused_bytes_left , content_type ) : <EOL> self . assertEquals ( payload , buf . read ( <NUM_LIT> ) ) <EOL> self . assertEquals ( bytes_read , len ( payload ) ) <EOL> self . assertEquals ( content_type , s3_content_type ) <EOL> mock_boto_key = mock . MagicMock ( ) <EOL> mock_boto_key . size = len ( payload ) <EOL> mock_boto_key . content_type = s3_content_type <EOL> mock_boto_key . get_contents_to_file = _MockedGetContentsToFile <EOL> mock_service = mock . MagicMock ( ) <EOL> config = { '<STR_LIT>' : mock_boto_key } <EOL> mock_service . configure_mock ( ** config ) <EOL> with mock . patch . object ( boto , '<STR_LIT>' ) : <EOL> storage = s3 . S3 ( ) <EOL> storage . service = mock_service <EOL> buf = storage . ReadObject ( url = s3_url , handler = _Callback ) <EOL> self . assertIsNone ( buf ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> basetest . main ( ) </s>
