<s> """<STR_LIT>""" <EOL> from cloudcafe . common . tools import randomstring as randstring <EOL> from cloudcafe . objectstorage . objectstorage_api . common . constants import Constants <EOL> from cloudroast . objectstorage . fixtures import ObjectStorageFixture <EOL> CONTENT_TYPE_TEXT = "<STR_LIT>" <EOL> CONTAINER_NAME = "<STR_LIT>" <EOL> BASE_DATA = "<STR_LIT>" <EOL> class ObjectVersioningTest ( ObjectStorageFixture ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( ObjectVersioningTest , cls ) . setUpClass ( ) <EOL> cls . container_name = CONTAINER_NAME <EOL> cls . object_name = "<STR_LIT>" . format ( <EOL> Constants . VALID_OBJECT_NAME , <EOL> randstring . get_random_string ( ) ) <EOL> def setUp ( self ) : <EOL> super ( ObjectVersioningTest , self ) . setUp ( ) <EOL> """<STR_LIT>""" <EOL> self . current_version_container_name = ( <EOL> self . create_temp_container ( <EOL> "<STR_LIT>" . format ( self . container_name ) ) ) <EOL> """<STR_LIT>""" <EOL> self . non_current_version_container_name = ( <EOL> self . create_temp_container ( <EOL> "<STR_LIT>" . format ( self . container_name ) ) ) <EOL> headers = ( <EOL> { "<STR_LIT>" : self . non_current_version_container_name } ) <EOL> self . client . set_container_metadata ( <EOL> self . current_version_container_name , headers = headers ) <EOL> self . num_objects = <NUM_LIT:4> <EOL> self . num_versioned_objects = self . num_objects - <NUM_LIT:1> <EOL> self . object_data_version0 = "<STR_LIT>" . format ( BASE_DATA ) <EOL> self . object_data_version1 = "<STR_LIT>" . format ( BASE_DATA ) <EOL> self . object_data_version2 = "<STR_LIT>" . format ( BASE_DATA ) <EOL> self . object_data_version3 = "<STR_LIT>" . format ( BASE_DATA ) <EOL> """<STR_LIT>""" <EOL> for i in range ( <NUM_LIT:0> , self . num_objects ) : <EOL> data = eval ( "<STR_LIT>" . format ( i ) ) <EOL> headers = { "<STR_LIT>" : data , <EOL> "<STR_LIT>" : str ( len ( data ) ) , <EOL> "<STR_LIT:Content-Type>" : CONTENT_TYPE_TEXT } <EOL> self . client . create_object ( <EOL> self . current_version_container_name , <EOL> self . object_name , <EOL> headers = headers , <EOL> data = eval ( "<STR_LIT>" . format ( i ) ) ) <EOL> def get_prefix ( self , obj_name ) : <EOL> """<STR_LIT>""" <EOL> prefix = None <EOL> padded_hex = None <EOL> temp = str ( hex ( len ( obj_name ) ) ) . split ( "<STR_LIT:x>" ) [ <NUM_LIT:1> ] <EOL> if len ( temp ) == <NUM_LIT:1> : <EOL> padded_hex = "<STR_LIT>" . format ( temp ) <EOL> elif len ( temp ) == <NUM_LIT:2> : <EOL> padded_hex = "<STR_LIT>" . format ( temp ) <EOL> else : <EOL> padded_hex = temp <EOL> prefix = "<STR_LIT>" . format ( padded_hex , obj_name ) <EOL> return prefix <EOL> @ ObjectStorageFixture . required_features ( '<STR_LIT>' ) <EOL> def test_versioned_obj_creation_with_valid_data ( self ) : <EOL> """<STR_LIT>""" <EOL> response = self . client . get_object ( <EOL> self . current_version_container_name , <EOL> self . object_name ) <EOL> expected = self . object_data_version3 <EOL> received = response . content <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" . format ( expected , received ) ) <EOL> """<STR_LIT>""" <EOL> params = { "<STR_LIT>" : "<STR_LIT>" } <EOL> response = self . client . list_objects ( <EOL> self . non_current_version_container_name , <EOL> params = params ) <EOL> resp_obj_count = response . headers . get ( "<STR_LIT>" ) <EOL> expected = self . num_versioned_objects <EOL> received = int ( resp_obj_count ) <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> expected , <EOL> received ) ) <EOL> object_names = [ obj . name for obj in response . entity ] <EOL> expected_prefix = self . get_prefix ( self . object_name ) <EOL> for i in range ( <NUM_LIT:0> , self . num_versioned_objects ) : <EOL> obj_name = object_names [ i ] <EOL> """<STR_LIT>""" <EOL> tokens = obj_name . split ( '<STR_LIT:/>' ) <EOL> expected = <NUM_LIT:2> <EOL> received = len ( tokens ) <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" ) <EOL> """<STR_LIT>""" <EOL> current_name_prefix = tokens [ <NUM_LIT:0> ] <EOL> expected = expected_prefix <EOL> received = current_name_prefix <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" . format ( expected , received ) ) <EOL> """<STR_LIT>""" <EOL> response = self . client . get_object ( <EOL> self . non_current_version_container_name , <EOL> obj_name ) <EOL> expected = eval ( "<STR_LIT>" . format ( i ) ) <EOL> received = response . content <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> expected , <EOL> received ) ) <EOL> """<STR_LIT>""" <EOL> expected = eval ( "<STR_LIT>" . format ( i ) ) <EOL> received = response . headers . get ( "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> expected , <EOL> received ) ) <EOL> """<STR_LIT>""" <EOL> headers = { "<STR_LIT>" : "<STR_LIT>" } <EOL> response = self . client . set_object_metadata ( <EOL> self . current_version_container_name , <EOL> self . object_name , <EOL> headers = headers ) <EOL> self . assertTrue ( <EOL> response . ok , <EOL> msg = "<STR_LIT>" ) <EOL> """<STR_LIT>""" <EOL> response = self . client . list_objects ( <EOL> self . non_current_version_container_name , <EOL> params = params ) <EOL> resp_obj_count = response . headers . get ( "<STR_LIT>" ) <EOL> expected = self . num_versioned_objects <EOL> received = int ( resp_obj_count ) <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> expected , <EOL> received ) ) <EOL> @ ObjectStorageFixture . required_features ( '<STR_LIT>' ) <EOL> def test_versioned_obj_deletion_with_valid_data ( self ) : <EOL> num_deletes = <NUM_LIT:0> <EOL> for i in reversed ( range ( <NUM_LIT:0> , self . num_versioned_objects ) ) : <EOL> self . client . delete_object ( <EOL> self . current_version_container_name , <EOL> self . object_name ) <EOL> """<STR_LIT>""" <EOL> response = self . client . get_object ( <EOL> self . current_version_container_name , <EOL> self . object_name ) <EOL> expected = eval ( "<STR_LIT>" . format ( i ) ) <EOL> received = response . content <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> expected , <EOL> received ) ) <EOL> """<STR_LIT>""" <EOL> expected = eval ( "<STR_LIT>" . format ( i ) ) <EOL> received = response . headers . get ( "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> expected , <EOL> received ) ) <EOL> """<STR_LIT>""" <EOL> response = self . client . list_objects ( <EOL> self . current_version_container_name ) <EOL> resp_obj_count = response . headers . get ( "<STR_LIT>" ) <EOL> expected = <NUM_LIT:1> <EOL> received = int ( resp_obj_count ) <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> expected , <EOL> received ) ) <EOL> num_deletes += <NUM_LIT:1> <EOL> """<STR_LIT>""" <EOL> response = self . client . list_objects ( <EOL> self . non_current_version_container_name ) <EOL> resp_obj_count = response . headers . get ( "<STR_LIT>" ) <EOL> expected = self . num_versioned_objects - num_deletes <EOL> received = int ( resp_obj_count ) <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> expected , <EOL> received ) ) <EOL> """<STR_LIT>""" <EOL> self . client . delete_object ( <EOL> self . current_version_container_name , <EOL> self . object_name ) <EOL> response = self . client . list_objects ( <EOL> self . non_current_version_container_name ) <EOL> resp_obj_count = response . headers . get ( "<STR_LIT>" ) <EOL> expected = <NUM_LIT:0> <EOL> received = int ( resp_obj_count ) <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( expected , received ) ) <EOL> @ ObjectStorageFixture . required_features ( '<STR_LIT>' ) <EOL> def test_disable_object_versioning ( self ) : <EOL> object_data_version4 = "<STR_LIT>" . format ( BASE_DATA ) <EOL> """<STR_LIT>""" <EOL> response = self . client . list_objects ( <EOL> self . non_current_version_container_name ) <EOL> resp_obj_count = response . headers . get ( "<STR_LIT>" ) <EOL> expected = self . num_versioned_objects <EOL> received = int ( resp_obj_count ) <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> expected , <EOL> received ) ) <EOL> """<STR_LIT>""" <EOL> headers = { "<STR_LIT>" : "<STR_LIT>" } <EOL> response = self . client . set_container_metadata ( <EOL> self . current_version_container_name , <EOL> headers = headers ) <EOL> self . assertTrue ( <EOL> response . ok , <EOL> msg = "<STR_LIT>" ) <EOL> headers = { "<STR_LIT>" : object_data_version4 , <EOL> "<STR_LIT>" : str ( len ( object_data_version4 ) ) , <EOL> "<STR_LIT:Content-Type>" : CONTENT_TYPE_TEXT } <EOL> response = self . client . create_object ( <EOL> self . current_version_container_name , <EOL> self . object_name , <EOL> headers = headers , <EOL> data = object_data_version4 ) <EOL> self . assertTrue ( <EOL> response . ok , <EOL> msg = "<STR_LIT>" ) <EOL> """<STR_LIT>""" <EOL> response = self . client . list_objects ( <EOL> self . non_current_version_container_name ) <EOL> resp_obj_count = response . headers . get ( "<STR_LIT>" ) <EOL> expected = self . num_versioned_objects <EOL> received = int ( resp_obj_count ) <EOL> self . assertEqual ( <EOL> expected , <EOL> received , <EOL> msg = "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> expected , <EOL> received ) ) </s>
