<s> from datetime import date <EOL> from eucaops import Eucaops <EOL> import argparse <EOL> import time <EOL> import re <EOL> from eutester . eutestcase import EutesterTestCase <EOL> from eucaops import S3ops <EOL> from boto . exception import S3ResponseError <EOL> from boto . exception import S3CreateError <EOL> import boto <EOL> import boto . s3 <EOL> from boto . s3 . bucket import Bucket <EOL> from boto . s3 . key import Key <EOL> from boto . s3 . acl import ACL , Policy , Grant <EOL> from boto . s3 . connection import Location <EOL> from boto . s3 . lifecycle import Lifecycle , Rule , Expiration <EOL> class BucketTestSuite ( EutesterTestCase ) : <EOL> def __init__ ( self ) : <EOL> self . setuptestcase ( ) <EOL> self . setup_parser ( ) <EOL> self . parser . add_argument ( "<STR_LIT>" , default = None ) <EOL> self . get_args ( ) <EOL> if self . args . endpoint : <EOL> self . tester = S3ops ( credpath = self . args . credpath , endpoint = self . args . endpoint ) <EOL> else : <EOL> self . tester = Eucaops ( credpath = self . args . credpath , config_file = self . args . config , password = self . args . password ) <EOL> self . bucket_prefix = "<STR_LIT>" + str ( int ( time . time ( ) ) ) <EOL> self . buckets_used = set ( ) <EOL> def test_bucket_get_put_delete ( self ) : <EOL> '''<STR_LIT>''' <EOL> test_bucket = self . bucket_prefix + "<STR_LIT>" <EOL> self . buckets_used . add ( test_bucket ) <EOL> self . tester . debug ( "<STR_LIT>" + test_bucket ) <EOL> if self . args . endpoint : <EOL> self . tester . info ( '<STR_LIT>' ) <EOL> self . tester . info ( '<STR_LIT>' ) <EOL> try : <EOL> bucket = self . tester . s3 . create_bucket ( test_bucket ) <EOL> if bucket == None : <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> self . fail ( test_bucket + "<STR_LIT>" ) <EOL> except ( S3ResponseError , S3CreateError ) as e : <EOL> self . fail ( test_bucket + "<STR_LIT>" + str ( e ) ) <EOL> try : <EOL> bucket = self . tester . s3 . get_bucket ( test_bucket ) <EOL> if bucket == None : <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> self . fail ( test_bucket + "<STR_LIT>" ) <EOL> except S3ResponseError as e : <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> self . fail ( "<STR_LIT>" + str ( e ) ) <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> try : <EOL> if self . tester . s3 . get_bucket ( test_bucket ) != None : <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> self . fail ( "<STR_LIT>" + test_bucket + "<STR_LIT>" ) <EOL> except S3ResponseError as e : <EOL> self . tester . debug ( "<STR_LIT>" ) <EOL> self . tester . debug ( "<STR_LIT>" ) <EOL> def test_creating_bucket_invalid_names ( bad_bucket ) : <EOL> should_fail = False <EOL> try : <EOL> bucket = self . tester . create_bucket ( bad_bucket ) <EOL> should_fail = True <EOL> try : <EOL> self . tester . delete_bucket ( bucket ) <EOL> except : <EOL> self . tester . debug ( "<STR_LIT>" ) <EOL> except Exception as e : <EOL> self . tester . debug ( "<STR_LIT>" + bad_bucket + "<STR_LIT>" + e . reason ) <EOL> if should_fail : <EOL> self . fail ( "<STR_LIT>" + bad_bucket ) <EOL> for bad_bucket in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> test_creating_bucket_invalid_names ( self . bucket_prefix + bad_bucket ) <EOL> """<STR_LIT>""" <EOL> try : <EOL> null_bucket_name = "<STR_LIT>" <EOL> bucket_obj = self . tester . create_bucket ( null_bucket_name ) <EOL> self . tester . sleep ( <NUM_LIT:10> ) <EOL> if bucket_obj : <EOL> self . fail ( "<STR_LIT>" ) <EOL> except S3ResponseError as e : <EOL> assert ( e . status == <NUM_LIT> ) , '<STR_LIT>' + str ( e . status ) <EOL> assert ( re . search ( "<STR_LIT>" , e . code ) ) , "<STR_LIT>" <EOL> except Exception , e : <EOL> self . tester . debug ( "<STR_LIT>" + str ( e ) ) <EOL> def test_bucket_acl ( self ) : <EOL> '''<STR_LIT>''' <EOL> test_bucket = self . bucket_prefix + '<STR_LIT>' <EOL> self . buckets_used . add ( test_bucket ) <EOL> test_user_id = self . tester . s3 . get_canonical_user_id ( ) <EOL> self . tester . debug ( '<STR_LIT>' + test_bucket + '<STR_LIT>' + test_user_id ) <EOL> try : <EOL> acl_bucket = self . tester . s3 . create_bucket ( test_bucket ) <EOL> self . tester . debug ( '<STR_LIT>' + test_bucket ) <EOL> except S3CreateError : <EOL> self . tester . debug ( "<STR_LIT>" ) <EOL> try : <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> acl_bucket = self . tester . s3 . create_bucket ( test_bucket ) <EOL> except : <EOL> self . tester . debug ( "<STR_LIT>" ) <EOL> self . fail ( "<STR_LIT>" + test_bucket ) <EOL> policy = acl_bucket . get_acl ( ) <EOL> if policy == None : <EOL> self . fail ( '<STR_LIT>' ) <EOL> self . tester . debug ( policy ) <EOL> if len ( policy . acl . grants ) > <NUM_LIT:1> : <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> self . fail ( '<STR_LIT>' + policy . acl . grants . grants . __len__ ( ) ) <EOL> if policy . acl . grants [ <NUM_LIT:0> ] . id != test_user_id or policy . acl . grants [ <NUM_LIT:0> ] . permission != '<STR_LIT>' : <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> self . fail ( '<STR_LIT>' + policy . acl . grants [ <NUM_LIT:0> ] . display_name + '<STR_LIT:U+0020>' + policy . acl . grants [ <NUM_LIT:0> ] . permission + '<STR_LIT:U+0020>' + policy . acl . grants [ <NUM_LIT:0> ] . id ) <EOL> owner_display_name = policy . acl . grants [ <NUM_LIT:0> ] . display_name <EOL> owner_id = policy . acl . grants [ <NUM_LIT:0> ] . id <EOL> for acl in boto . s3 . acl . CannedACLStrings : <EOL> if acl == "<STR_LIT>" or acl == "<STR_LIT>" : <EOL> continue <EOL> self . tester . info ( '<STR_LIT>' + acl ) <EOL> try : <EOL> acl_bucket . set_acl ( acl ) <EOL> acl_check = acl_bucket . get_acl ( ) <EOL> except Exception as e : <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> self . fail ( "<STR_LIT>" + acl + "<STR_LIT>" + str ( e ) ) <EOL> self . tester . info ( "<STR_LIT>" + acl + "<STR_LIT>" + acl_check . acl . to_xml ( ) ) <EOL> expected_acl = self . tester . get_canned_acl ( bucket_owner_id = owner_id , canned_acl = acl , bucket_owner_display_name = owner_display_name ) <EOL> if expected_acl == None : <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> self . fail ( "<STR_LIT>" + acl ) <EOL> if not self . tester . check_acl_equivalence ( acl1 = expected_acl , acl2 = acl_check . acl ) : <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> self . fail ( "<STR_LIT>" + acl + "<STR_LIT>" + acl_check . acl . to_xml ( ) + "<STR_LIT>" + expected_acl . to_xml ( ) ) <EOL> else : <EOL> self . tester . debug ( "<STR_LIT>" + acl ) <EOL> try : <EOL> acl_bucket . set_acl ( '<STR_LIT>' ) <EOL> self . fail ( '<STR_LIT>' ) <EOL> except : <EOL> self . tester . debug ( "<STR_LIT>" ) <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> self . buckets_used . remove ( test_bucket ) <EOL> self . tester . debug ( "<STR_LIT>" ) <EOL> pass <EOL> def test_bucket_key_list_delim_prefix ( self ) : <EOL> """<STR_LIT>""" <EOL> test_bucket_name = self . bucket_prefix + "<STR_LIT>" <EOL> self . buckets_used . add ( test_bucket_name ) <EOL> self . tester . debug ( '<STR_LIT>' + test_bucket_name ) <EOL> try : <EOL> testbucket = self . tester . s3 . create_bucket ( bucket_name = test_bucket_name ) <EOL> except S3CreateError : <EOL> self . tester . debug ( "<STR_LIT>" ) <EOL> try : <EOL> testbucket = self . tester . s3 . get_bucket ( bucket_name = test_bucket_name ) <EOL> except S3ResponseError as err : <EOL> self . tester . debug ( "<STR_LIT>" ) <EOL> for b in self . tester . s3 . get_all_buckets ( ) : <EOL> self . tester . debug ( "<STR_LIT>" + b . name ) <EOL> self . fail ( "<STR_LIT>" + test_bucket_name + "<STR_LIT>" + err . error_message ) <EOL> prefix = "<STR_LIT>" <EOL> delim = "<STR_LIT:/>" <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> tmp = str ( i ) <EOL> self . tester . debug ( "<STR_LIT>" + tmp ) <EOL> key = testbucket . new_key ( "<STR_LIT>" + tmp ) <EOL> key . set_contents_from_string ( "<STR_LIT>" ) <EOL> key = testbucket . new_key ( prefix + "<STR_LIT>" + tmp ) <EOL> key . set_contents_from_string ( "<STR_LIT>" ) <EOL> key = testbucket . new_key ( prefix + delim + "<STR_LIT:object>" + tmp ) <EOL> key . set_contents_from_string ( "<STR_LIT>" ) <EOL> key = testbucket . new_key ( prefix + delim + "<STR_LIT>" + delim + "<STR_LIT>" + tmp + "<STR_LIT>" ) <EOL> key . set_contents_from_string ( "<STR_LIT>" ) <EOL> keys = testbucket . get_all_keys ( prefix = prefix , delimiter = delim , max_keys = <NUM_LIT:10> ) <EOL> self . tester . debug ( "<STR_LIT>" + str ( len ( keys ) ) + "<STR_LIT>" ) <EOL> for k in keys : <EOL> self . tester . debug ( k ) <EOL> keys = testbucket . get_all_keys ( prefix = prefix , delimiter = delim , max_keys = <NUM_LIT:20> ) <EOL> self . tester . debug ( "<STR_LIT>" + str ( len ( keys ) ) + "<STR_LIT>" ) <EOL> for k in keys : <EOL> self . tester . debug ( k ) <EOL> print "<STR_LIT>" <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> testbucket . delete_key ( "<STR_LIT>" + str ( i ) ) <EOL> testbucket . delete_key ( prefix + "<STR_LIT>" + str ( i ) ) <EOL> testbucket . delete_key ( prefix + delim + "<STR_LIT:object>" + str ( i ) ) <EOL> testbucket . delete_key ( prefix + delim + "<STR_LIT>" + delim + "<STR_LIT>" + str ( i ) + "<STR_LIT>" ) <EOL> print "<STR_LIT>" <EOL> self . tester . s3 . delete_bucket ( testbucket ) <EOL> def test_bucket_location ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . args . endpoint : <EOL> self . tester . info ( '<STR_LIT>' ) <EOL> self . tester . info ( '<STR_LIT>' ) <EOL> test_bucket = self . bucket_prefix + "<STR_LIT>" <EOL> self . tester . debug ( '<STR_LIT>' + test_bucket ) <EOL> self . tester . s3 . create_bucket ( test_bucket ) <EOL> bucket = self . tester . s3 . get_bucket ( test_bucket ) <EOL> if bucket != None and ( bucket . get_location ( ) == Location . DEFAULT ) : <EOL> self . tester . debug ( '<STR_LIT>' ) <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> else : <EOL> bucket . delete ( ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> test_bucket = self . bucket_prefix + "<STR_LIT>" <EOL> self . tester . debug ( '<STR_LIT>' + test_bucket ) <EOL> self . tester . s3 . create_bucket ( test_bucket , location = Location . DEFAULT ) <EOL> bucket = self . tester . s3 . get_bucket ( test_bucket ) <EOL> if bucket != None and ( bucket . get_location ( ) == Location . DEFAULT ) : <EOL> self . tester . debug ( '<STR_LIT>' ) <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> else : <EOL> bucket . delete ( ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> if self . args . endpoint : <EOL> self . tester . info ( '<STR_LIT>' ) <EOL> else : <EOL> test_bucket = self . bucket_prefix + "<STR_LIT>" <EOL> self . tester . debug ( '<STR_LIT>' + test_bucket ) <EOL> bucket = self . tester . s3 . create_bucket ( test_bucket , location = Location . EU ) <EOL> self . buckets_used . add ( test_bucket ) <EOL> if bucket == None : <EOL> self . fail ( "<STR_LIT>" ) <EOL> else : <EOL> loc = bucket . get_location ( ) <EOL> if loc == Location . EU : <EOL> self . tester . debug ( '<STR_LIT>' ) <EOL> bucket . delete ( ) <EOL> else : <EOL> bucket . delete ( ) <EOL> self . fail ( "<STR_LIT>" + loc ) <EOL> self . buckets_used . remove ( test_bucket ) <EOL> self . tester . debug ( "<STR_LIT>" ) <EOL> pass <EOL> def test_bucket_logging ( self ) : <EOL> """<STR_LIT>""" <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> test_bucket = self . bucket_prefix + "<STR_LIT>" <EOL> log_dest_bucket = self . bucket_prefix + "<STR_LIT>" <EOL> self . buckets_used . add ( test_bucket ) <EOL> self . buckets_used . add ( log_dest_bucket ) <EOL> log_prefix = "<STR_LIT>" <EOL> try : <EOL> bucket = self . tester . s3 . create_bucket ( test_bucket ) <EOL> except S3CreateError : <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> try : <EOL> self . tester . s3 . delete_bucket ( test_bucket ) <EOL> bucket = self . tester . s3 . create_bucket ( test_bucket ) <EOL> except : <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> try : <EOL> dest_bucket = self . tester . s3 . create_bucket ( log_dest_bucket ) <EOL> except S3CreateError : <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> try : <EOL> self . tester . s3 . delete_bucket ( log_dest_bucket ) <EOL> dest_bucket = self . tester . s3 . create_bucket ( log_dest_bucket ) <EOL> except : <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> log_delivery_policy = dest_bucket . get_acl ( ) <EOL> log_delivery_policy . acl . add_grant ( Grant ( type = "<STR_LIT>" , uri = "<STR_LIT>" , permission = "<STR_LIT>" ) ) <EOL> log_delivery_policy . acl . add_grant ( Grant ( type = "<STR_LIT>" , uri = "<STR_LIT>" , permission = "<STR_LIT>" ) ) <EOL> dest_bucket . set_acl ( log_delivery_policy ) <EOL> bucket . enable_logging ( log_dest_bucket , target_prefix = log_prefix ) <EOL> k = bucket . new_key ( '<STR_LIT>' ) <EOL> k . set_contents_from_string ( '<STR_LIT>' ) <EOL> k = bucket . new_key ( '<STR_LIT>' ) <EOL> k . set_contents_from_string ( "<STR_LIT>" ) <EOL> k = bucket . get_key ( '<STR_LIT>' ) <EOL> result1 = k . get_contents_as_string ( ) <EOL> self . tester . info ( "<STR_LIT>" + result1 ) <EOL> k = bucket . get_key ( '<STR_LIT>' ) <EOL> result2 = k . get_contents_as_string ( ) <EOL> self . tester . info ( "<STR_LIT>" + result2 ) <EOL> keylist = bucket . list ( ) <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> for k in keylist : <EOL> if isinstance ( k , boto . s3 . prefix . Prefix ) : <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> else : <EOL> self . tester . info ( '<STR_LIT>' + k . name ) <EOL> time . sleep ( <NUM_LIT:15> ) <EOL> log_bucket = self . tester . s3 . get_bucket ( log_dest_bucket ) <EOL> for k in log_bucket . list ( prefix = log_prefix ) : <EOL> self . tester . info ( '<STR_LIT>' + k . name ) <EOL> log_obj = log_bucket . get_key ( k ) <EOL> self . tester . info ( "<STR_LIT>" + k . get_contents_as_string ( ) ) <EOL> log_data = log_obj . get_content_as_string ( ) <EOL> self . tester . info ( '<STR_LIT>' + log_data ) <EOL> self . tester . info ( '<STR_LIT>' ) <EOL> def test_bucket_versioning ( self ) : <EOL> test_bucket = self . bucket_prefix + "<STR_LIT>" <EOL> self . tester . info ( '<STR_LIT>' + test_bucket ) <EOL> version_bucket = self . tester . s3 . create_bucket ( test_bucket ) <EOL> self . buckets_used . add ( test_bucket ) <EOL> version_status = version_bucket . get_versioning_status ( ) . get ( "<STR_LIT>" ) <EOL> if version_status != None : <EOL> version_bucket . delete ( ) <EOL> self . fail ( "<STR_LIT>" + str ( version_status ) ) <EOL> elif version_status == None : <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> version_bucket . configure_versioning ( True ) <EOL> if self . args . endpoint : <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> version_status = version_bucket . get_versioning_status ( ) . get ( "<STR_LIT>" ) <EOL> if version_status == None or version_status != "<STR_LIT>" : <EOL> version_bucket . delete ( ) <EOL> self . fail ( "<STR_LIT>" + str ( version_status ) ) <EOL> elif version_status == None : <EOL> version_bucket . delete ( ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> self . tester . info ( "<STR_LIT>" + version_status ) <EOL> version_bucket . configure_versioning ( False ) <EOL> if self . args . endpoint : <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> version_status = version_bucket . get_versioning_status ( ) . get ( "<STR_LIT>" ) <EOL> if version_status == None or version_status != "<STR_LIT>" : <EOL> version_bucket . delete ( ) <EOL> self . fail ( "<STR_LIT>" + str ( version_status ) ) <EOL> elif version_status == None : <EOL> version_bucket . delete ( ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> self . tester . info ( "<STR_LIT>" + version_status ) <EOL> version_bucket . configure_versioning ( True ) <EOL> if self . args . endpoint : <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> version_status = version_bucket . get_versioning_status ( ) . get ( "<STR_LIT>" ) <EOL> if version_status == None or version_status != "<STR_LIT>" : <EOL> version_bucket . delete ( ) <EOL> self . fail ( "<STR_LIT>" + str ( version_status ) ) <EOL> elif version_status == None : <EOL> version_bucket . delete ( ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> self . tester . info ( "<STR_LIT>" + version_status ) <EOL> version_bucket . delete ( ) <EOL> self . buckets_used . remove ( test_bucket ) <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> def test_bucket_key_listing_paging ( self ) : <EOL> """<STR_LIT>""" <EOL> test_bucket_name = self . bucket_prefix + "<STR_LIT>" <EOL> self . buckets_used . add ( test_bucket_name ) <EOL> self . tester . info ( '<STR_LIT>' + test_bucket_name ) <EOL> try : <EOL> testbucket = self . tester . s3 . create_bucket ( bucket_name = test_bucket_name ) <EOL> except S3CreateError : <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> try : <EOL> testbucket = self . tester . s3 . get_bucket ( bucket_name = test_bucket_name ) <EOL> except S3ResponseError as err : <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> for b in self . tester . s3 . get_all_buckets ( ) : <EOL> self . tester . info ( '<STR_LIT>' + b . name ) <EOL> self . fail ( "<STR_LIT>" + test_bucket_name + "<STR_LIT>" + err . error_message ) <EOL> key_name_prefix = "<STR_LIT>" <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> key_name = key_name_prefix + str ( i ) <EOL> self . tester . info ( "<STR_LIT>" + key_name ) <EOL> testbucket . new_key ( key_name ) . set_contents_from_string ( "<STR_LIT>" ) <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> key_name = key_name_prefix + "<STR_LIT>" + str ( i ) <EOL> self . tester . info ( "<STR_LIT>" + key_name ) <EOL> testbucket . new_key ( key_name ) . set_contents_from_string ( "<STR_LIT>" ) <EOL> key_list = testbucket . get_all_keys ( max_keys = <NUM_LIT:50> ) <EOL> self . tester . info ( "<STR_LIT>" + str ( len ( key_list ) ) + "<STR_LIT>" ) <EOL> for k in key_list : <EOL> assert isinstance ( k , Key ) <EOL> self . tester . info ( '<STR_LIT>' + k . name ) <EOL> if len ( key_list ) != <NUM_LIT:50> : <EOL> self . fail ( "<STR_LIT>" + str ( len ( key_list ) ) ) <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> key_name = key_name_prefix + str ( i ) <EOL> self . tester . info ( "<STR_LIT>" + key_name ) <EOL> testbucket . delete_key ( key_name ) <EOL> key_name = key_name_prefix + "<STR_LIT>" + str ( i ) <EOL> self . tester . info ( "<STR_LIT>" + key_name ) <EOL> testbucket . delete_key ( key_name ) <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> key_list = testbucket . get_all_keys ( ) <EOL> for k in key_list : <EOL> self . tester . info ( "<STR_LIT>" + k . name ) <EOL> testbucket . delete_key ( k ) <EOL> self . tester . info ( "<STR_LIT>" ) <EOL> self . tester . s3 . delete_bucket ( testbucket ) <EOL> self . buckets_used . add ( test_bucket_name ) <EOL> def test_list_multipart_uploads ( self ) : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def test_bucket_lifecycle ( self ) : <EOL> lifecycle_id = '<STR_LIT>' <EOL> lifecycle_prefix = '<STR_LIT>' <EOL> lifecycle_status = '<STR_LIT>' <EOL> lifecycle_expiration = <NUM_LIT:1> <EOL> bucket_name = self . bucket_prefix + "<STR_LIT>" <EOL> self . buckets_used . add ( bucket_name ) <EOL> bucket = self . tester . create_bucket ( bucket_name ) <EOL> lifecycle = Lifecycle ( ) <EOL> lifecycle . add_rule ( lifecycle_id , lifecycle_prefix , lifecycle_status , lifecycle_expiration ) <EOL> bucket . configure_lifecycle ( lifecycle ) <EOL> responses = bucket . get_lifecycle_config ( ) <EOL> assert ( len ( responses ) == <NUM_LIT:1> ) , '<STR_LIT>' <EOL> lifecycle_response = responses [ <NUM_LIT:0> ] <EOL> assert ( lifecycle_response . id == lifecycle_id ) , "<STR_LIT>" + lifecycle_id + "<STR_LIT>" + lifecycle_response . id <EOL> assert ( lifecycle_response . prefix == lifecycle_prefix ) , "<STR_LIT>" + lifecycle_prefix + "<STR_LIT>" + lifecycle_response . prefix <EOL> assert ( lifecycle_response . status == lifecycle_status ) , "<STR_LIT>" + lifecycle_status + "<STR_LIT>" + lifecycle_response . status <EOL> assert ( lifecycle_response . expiration . days == lifecycle_expiration ) , "<STR_LIT>" + str ( lifecycle_expiration ) + "<STR_LIT>" + str ( lifecycle_response . expiration . days ) <EOL> bucket . delete_lifecycle_configuration ( ) <EOL> assert ( len ( responses ) == <NUM_LIT:1> ) , "<STR_LIT>" + len ( responses ) + "<STR_LIT>" <EOL> bucket_name = self . bucket_prefix + "<STR_LIT>" <EOL> bucket = self . tester . create_bucket ( bucket_name ) <EOL> self . buckets_used . add ( bucket_name ) <EOL> date = '<STR_LIT>' <EOL> lifecycle = Lifecycle ( ) <EOL> lifecycle . add_rule ( "<STR_LIT:1>" , "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT:1> ) <EOL> lifecycle . add_rule ( "<STR_LIT:2>" , "<STR_LIT>" , "<STR_LIT>" , Expiration ( days = <NUM_LIT:2> ) ) <EOL> lifecycle . add_rule ( "<STR_LIT:3>" , "<STR_LIT>" , "<STR_LIT>" , Expiration ( date = date ) ) <EOL> lifecycle . add_rule ( "<STR_LIT:4>" , "<STR_LIT>" , "<STR_LIT>" , Expiration ( date = date ) ) <EOL> bucket . configure_lifecycle ( lifecycle ) <EOL> lifecycle_responses = bucket . get_lifecycle_config ( ) <EOL> if lifecycle_responses < <NUM_LIT:0> : <EOL> self . fail ( "<STR_LIT>" ) <EOL> for response in lifecycle_responses : <EOL> if response . id == "<STR_LIT:1>" : <EOL> assert ( response . prefix == "<STR_LIT>" ) , "<STR_LIT>" + "<STR_LIT>" + "<STR_LIT>" + response . prefix <EOL> assert ( response . status == "<STR_LIT>" ) , "<STR_LIT>" + "<STR_LIT>" + "<STR_LIT>" + response . status <EOL> assert ( response . expiration . days == <NUM_LIT:1> ) , "<STR_LIT>" + str ( <NUM_LIT:1> ) + "<STR_LIT>" + str ( response . expiration . days ) <EOL> elif response . id == "<STR_LIT:2>" : <EOL> assert ( response . prefix == "<STR_LIT>" ) , "<STR_LIT>" + "<STR_LIT>" + "<STR_LIT>" + response . prefix <EOL> assert ( response . status == "<STR_LIT>" ) , "<STR_LIT>" + "<STR_LIT>" + "<STR_LIT>" + response . status <EOL> assert ( response . expiration . days == <NUM_LIT:2> ) , "<STR_LIT>" + str ( <NUM_LIT:2> ) + "<STR_LIT>" + str ( response . expiration . days ) <EOL> elif response . id == "<STR_LIT:3>" : <EOL> assert ( response . prefix == "<STR_LIT>" ) , "<STR_LIT>" + "<STR_LIT>" + "<STR_LIT>" + response . prefix <EOL> assert ( response . status == "<STR_LIT>" ) , "<STR_LIT>" + "<STR_LIT>" + "<STR_LIT>" + response . status <EOL> assert ( response . expiration . date == date ) , "<STR_LIT>" + date + "<STR_LIT>" + str ( response . expiration . date ) <EOL> elif response . id == "<STR_LIT:4>" : <EOL> assert ( response . prefix == "<STR_LIT>" ) , "<STR_LIT>" + "<STR_LIT>" + "<STR_LIT>" + response . prefix <EOL> assert ( response . status == "<STR_LIT>" ) , "<STR_LIT>" + "<STR_LIT>" + "<STR_LIT>" + response . status <EOL> assert ( response . expiration . date == date ) , "<STR_LIT>" + date + "<STR_LIT>" + str ( response . expiration . date ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> self . debug ( "<STR_LIT>" ) <EOL> for bucket in self . buckets_used : <EOL> self . tester . clear_bucket ( bucket ) <EOL> def test_bucket_policy ( self ) : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def test_bucket_website ( self ) : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def clean_method ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . tester . info ( '<STR_LIT>' ) <EOL> for bucket in self . buckets_used : <EOL> try : <EOL> self . tester . info ( '<STR_LIT>' + bucket + '<STR_LIT>' ) <EOL> if self . tester . s3 . bucket_exists ( bucket ) : <EOL> self . tester . info ( '<STR_LIT>' ) <EOL> self . tester . clear_bucket ( bucket ) <EOL> self . buckets_used . remove ( bucket ) <EOL> else : <EOL> self . tester . info ( '<STR_LIT>' + bucket + '<STR_LIT>' ) <EOL> except : <EOL> self . tester . info ( '<STR_LIT>' + str ( bucket ) ) <EOL> return <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> testcase = BucketTestSuite ( ) <EOL> list = testcase . args . tests or [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> unit_list = [ ] <EOL> for test in list : <EOL> unit_list . append ( testcase . create_testunit_by_name ( test ) ) <EOL> result = testcase . run_test_case_list ( unit_list , clean_on_exit = True ) <EOL> exit ( result ) </s>
