<s> from __future__ import unicode_literals <EOL> from six . moves . urllib . request import urlopen <EOL> from six . moves . urllib . error import HTTPError <EOL> import boto <EOL> from boto . exception import S3ResponseError <EOL> from boto . s3 . key import Key <EOL> from boto . s3 . connection import OrdinaryCallingFormat <EOL> from freezegun import freeze_time <EOL> import requests <EOL> import sure <EOL> from moto import mock_s3bucket_path <EOL> def create_connection ( key = None , secret = None ) : <EOL> return boto . connect_s3 ( key , secret , calling_format = OrdinaryCallingFormat ( ) ) <EOL> class MyModel ( object ) : <EOL> def __init__ ( self , name , value ) : <EOL> self . name = name <EOL> self . value = value <EOL> def save ( self ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket = conn . get_bucket ( '<STR_LIT>' ) <EOL> k = Key ( bucket ) <EOL> k . key = self . name <EOL> k . set_contents_from_string ( self . value ) <EOL> @ mock_s3bucket_path <EOL> def test_my_model_save ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> conn . create_bucket ( '<STR_LIT>' ) <EOL> model_instance = MyModel ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> model_instance . save ( ) <EOL> conn . get_bucket ( '<STR_LIT>' ) . get_key ( '<STR_LIT>' ) . get_contents_as_string ( ) . should . equal ( b'<STR_LIT>' ) <EOL> @ mock_s3bucket_path <EOL> def test_missing_key ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket = conn . create_bucket ( "<STR_LIT>" ) <EOL> bucket . get_key ( "<STR_LIT>" ) . should . equal ( None ) <EOL> @ mock_s3bucket_path <EOL> def test_missing_key_urllib2 ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> conn . create_bucket ( "<STR_LIT>" ) <EOL> urlopen . when . called_with ( "<STR_LIT>" ) . should . throw ( HTTPError ) <EOL> @ mock_s3bucket_path <EOL> def test_empty_key ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket = conn . create_bucket ( "<STR_LIT>" ) <EOL> key = Key ( bucket ) <EOL> key . key = "<STR_LIT>" <EOL> key . set_contents_from_string ( "<STR_LIT>" ) <EOL> bucket . get_key ( "<STR_LIT>" ) . get_contents_as_string ( ) . should . equal ( b'<STR_LIT>' ) <EOL> @ mock_s3bucket_path <EOL> def test_empty_key_set_on_existing_key ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket = conn . create_bucket ( "<STR_LIT>" ) <EOL> key = Key ( bucket ) <EOL> key . key = "<STR_LIT>" <EOL> key . set_contents_from_string ( "<STR_LIT>" ) <EOL> bucket . get_key ( "<STR_LIT>" ) . get_contents_as_string ( ) . should . equal ( b'<STR_LIT>' ) <EOL> key . set_contents_from_string ( "<STR_LIT>" ) <EOL> bucket . get_key ( "<STR_LIT>" ) . get_contents_as_string ( ) . should . equal ( b'<STR_LIT>' ) <EOL> @ mock_s3bucket_path <EOL> def test_large_key_save ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket = conn . create_bucket ( "<STR_LIT>" ) <EOL> key = Key ( bucket ) <EOL> key . key = "<STR_LIT>" <EOL> key . set_contents_from_string ( "<STR_LIT>" * <NUM_LIT> ) <EOL> bucket . get_key ( "<STR_LIT>" ) . get_contents_as_string ( ) . should . equal ( b'<STR_LIT>' * <NUM_LIT> ) <EOL> @ mock_s3bucket_path <EOL> def test_copy_key ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket = conn . create_bucket ( "<STR_LIT>" ) <EOL> key = Key ( bucket ) <EOL> key . key = "<STR_LIT>" <EOL> key . set_contents_from_string ( "<STR_LIT>" ) <EOL> bucket . copy_key ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket . get_key ( "<STR_LIT>" ) . get_contents_as_string ( ) . should . equal ( b"<STR_LIT>" ) <EOL> bucket . get_key ( "<STR_LIT>" ) . get_contents_as_string ( ) . should . equal ( b"<STR_LIT>" ) <EOL> @ mock_s3bucket_path <EOL> def test_set_metadata ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket = conn . create_bucket ( "<STR_LIT>" ) <EOL> key = Key ( bucket ) <EOL> key . key = '<STR_LIT>' <EOL> key . set_metadata ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> key . set_contents_from_string ( "<STR_LIT>" ) <EOL> bucket . get_key ( '<STR_LIT>' ) . get_metadata ( '<STR_LIT>' ) . should . equal ( '<STR_LIT>' ) <EOL> @ freeze_time ( "<STR_LIT>" ) <EOL> @ mock_s3bucket_path <EOL> def test_last_modified ( ) : <EOL> conn = create_connection ( ) <EOL> bucket = conn . create_bucket ( "<STR_LIT>" ) <EOL> key = Key ( bucket ) <EOL> key . key = "<STR_LIT>" <EOL> key . set_contents_from_string ( "<STR_LIT>" ) <EOL> rs = bucket . get_all_keys ( ) <EOL> rs [ <NUM_LIT:0> ] . last_modified . should . equal ( '<STR_LIT>' ) <EOL> bucket . get_key ( "<STR_LIT>" ) . last_modified . should . equal ( '<STR_LIT>' ) <EOL> @ mock_s3bucket_path <EOL> def test_missing_bucket ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> conn . get_bucket . when . called_with ( '<STR_LIT>' ) . should . throw ( S3ResponseError ) <EOL> @ mock_s3bucket_path <EOL> def test_bucket_with_dash ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> conn . get_bucket . when . called_with ( '<STR_LIT>' ) . should . throw ( S3ResponseError ) <EOL> @ mock_s3bucket_path <EOL> def test_bucket_deletion ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket = conn . create_bucket ( "<STR_LIT>" ) <EOL> key = Key ( bucket ) <EOL> key . key = "<STR_LIT>" <EOL> key . set_contents_from_string ( "<STR_LIT>" ) <EOL> conn . delete_bucket . when . called_with ( "<STR_LIT>" ) . should . throw ( S3ResponseError ) <EOL> bucket . delete_key ( "<STR_LIT>" ) <EOL> conn . delete_bucket ( "<STR_LIT>" ) <EOL> conn . get_bucket . when . called_with ( "<STR_LIT>" ) . should . throw ( S3ResponseError ) <EOL> conn . delete_bucket . when . called_with ( "<STR_LIT>" ) . should . throw ( S3ResponseError ) <EOL> @ mock_s3bucket_path <EOL> def test_get_all_buckets ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> conn . create_bucket ( "<STR_LIT>" ) <EOL> conn . create_bucket ( "<STR_LIT>" ) <EOL> buckets = conn . get_all_buckets ( ) <EOL> buckets . should . have . length_of ( <NUM_LIT:2> ) <EOL> @ mock_s3bucket_path <EOL> def test_post_to_bucket ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket = conn . create_bucket ( "<STR_LIT>" ) <EOL> requests . post ( "<STR_LIT>" , { <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT:file>' : '<STR_LIT>' <EOL> } ) <EOL> bucket . get_key ( '<STR_LIT>' ) . get_contents_as_string ( ) . should . equal ( b'<STR_LIT>' ) <EOL> @ mock_s3bucket_path <EOL> def test_post_with_metadata_to_bucket ( ) : <EOL> conn = create_connection ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket = conn . create_bucket ( "<STR_LIT>" ) <EOL> requests . post ( "<STR_LIT>" , { <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT:file>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> bucket . get_key ( '<STR_LIT>' ) . get_metadata ( '<STR_LIT:test>' ) . should . equal ( '<STR_LIT>' ) <EOL> @ mock_s3bucket_path <EOL> def test_bucket_method_not_implemented ( ) : <EOL> requests . patch . when . called_with ( "<STR_LIT>" ) . should . throw ( NotImplementedError ) <EOL> @ mock_s3bucket_path <EOL> def test_key_method_not_implemented ( ) : <EOL> requests . post . when . called_with ( "<STR_LIT>" ) . should . throw ( NotImplementedError ) <EOL> @ mock_s3bucket_path <EOL> def test_bucket_name_with_dot ( ) : <EOL> conn = create_connection ( ) <EOL> bucket = conn . create_bucket ( '<STR_LIT>' ) <EOL> k = Key ( bucket , '<STR_LIT>' ) <EOL> k . set_contents_from_string ( '<STR_LIT>' ) <EOL> @ mock_s3bucket_path <EOL> def test_key_with_special_characters ( ) : <EOL> conn = create_connection ( ) <EOL> bucket = conn . create_bucket ( '<STR_LIT>' ) <EOL> key = Key ( bucket , '<STR_LIT>' ) <EOL> key . set_contents_from_string ( '<STR_LIT>' ) <EOL> key_list = bucket . list ( '<STR_LIT>' , '<STR_LIT:/>' ) <EOL> keys = [ x for x in key_list ] <EOL> keys [ <NUM_LIT:0> ] . name . should . equal ( "<STR_LIT>" ) <EOL> @ mock_s3bucket_path <EOL> def test_bucket_key_listing_order ( ) : <EOL> conn = create_connection ( ) <EOL> bucket = conn . create_bucket ( '<STR_LIT>' ) <EOL> prefix = '<STR_LIT>' <EOL> def store ( name ) : <EOL> k = Key ( bucket , prefix + name ) <EOL> k . set_contents_from_string ( '<STR_LIT>' ) <EOL> names = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> for name in names : <EOL> store ( name ) <EOL> delimiter = None <EOL> keys = [ x . name for x in bucket . list ( prefix , delimiter ) ] <EOL> keys . should . equal ( [ <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ] ) <EOL> delimiter = '<STR_LIT:/>' <EOL> keys = [ x . name for x in bucket . list ( prefix , delimiter ) ] <EOL> keys . should . equal ( [ <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ] ) <EOL> delimiter = '<STR_LIT:/>' <EOL> keys = [ x . name for x in bucket . list ( prefix = None , delimiter = delimiter ) ] <EOL> keys . should . equal ( [ '<STR_LIT>' ] ) <EOL> delimiter = None <EOL> keys = [ x . name for x in bucket . list ( prefix + '<STR_LIT:x>' , delimiter ) ] <EOL> keys . should . equal ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> delimiter = '<STR_LIT:/>' <EOL> keys = [ x . name for x in bucket . list ( prefix + '<STR_LIT:x>' , delimiter ) ] <EOL> keys . should . equal ( [ '<STR_LIT>' ] ) <EOL> @ mock_s3bucket_path <EOL> def test_delete_keys ( ) : <EOL> conn = create_connection ( ) <EOL> bucket = conn . create_bucket ( '<STR_LIT>' ) <EOL> Key ( bucket = bucket , name = '<STR_LIT>' ) . set_contents_from_string ( '<STR_LIT:abc>' ) <EOL> Key ( bucket = bucket , name = '<STR_LIT>' ) . set_contents_from_string ( '<STR_LIT:abc>' ) <EOL> Key ( bucket = bucket , name = '<STR_LIT>' ) . set_contents_from_string ( '<STR_LIT:abc>' ) <EOL> Key ( bucket = bucket , name = '<STR_LIT>' ) . set_contents_from_string ( '<STR_LIT:abc>' ) <EOL> result = bucket . delete_keys ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> result . deleted . should . have . length_of ( <NUM_LIT:2> ) <EOL> result . errors . should . have . length_of ( <NUM_LIT:0> ) <EOL> keys = bucket . get_all_keys ( ) <EOL> keys . should . have . length_of ( <NUM_LIT:2> ) <EOL> keys [ <NUM_LIT:0> ] . name . should . equal ( '<STR_LIT>' ) <EOL> @ mock_s3bucket_path <EOL> def test_delete_keys_with_invalid ( ) : <EOL> conn = create_connection ( ) <EOL> bucket = conn . create_bucket ( '<STR_LIT>' ) <EOL> Key ( bucket = bucket , name = '<STR_LIT>' ) . set_contents_from_string ( '<STR_LIT:abc>' ) <EOL> Key ( bucket = bucket , name = '<STR_LIT>' ) . set_contents_from_string ( '<STR_LIT:abc>' ) <EOL> Key ( bucket = bucket , name = '<STR_LIT>' ) . set_contents_from_string ( '<STR_LIT:abc>' ) <EOL> Key ( bucket = bucket , name = '<STR_LIT>' ) . set_contents_from_string ( '<STR_LIT:abc>' ) <EOL> result = bucket . delete_keys ( [ '<STR_LIT:abc>' , '<STR_LIT>' ] ) <EOL> result . deleted . should . have . length_of ( <NUM_LIT:1> ) <EOL> result . errors . should . have . length_of ( <NUM_LIT:1> ) <EOL> keys = bucket . get_all_keys ( ) <EOL> keys . should . have . length_of ( <NUM_LIT:3> ) <EOL> keys [ <NUM_LIT:0> ] . name . should . equal ( '<STR_LIT>' ) </s>
