<s> import random , unittest <EOL> from ofs . remote . reststore import RESTOFS <EOL> from ofs import OFSException <EOL> from StringIO import StringIO <EOL> import os <EOL> TEST_TEXT = """<STR_LIT>""" <EOL> BINARY_FILE_NAME = os . path . join ( os . path . dirname ( __file__ ) , '<STR_LIT>' ) <EOL> class TestRESTOFS ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . bucket_name = '<STR_LIT>' <EOL> self . host_name = '<STR_LIT>' <EOL> self . ofs = RESTOFS ( self . host_name ) <EOL> self . ofs . claim_bucket ( self . bucket_name ) <EOL> def tearDown ( self ) : <EOL> self . ofs . _del_bucket ( self . bucket_name ) <EOL> def _makefp ( self ) : <EOL> return StringIO ( TEST_TEXT ) <EOL> def test_exists ( self ) : <EOL> self . assertTrue ( self . ofs . exists ( self . bucket_name ) ) <EOL> def test_claim_bucket ( self ) : <EOL> bucket_name = '<STR_LIT>' <EOL> self . ofs . _del_bucket ( bucket_name ) <EOL> self . assertFalse ( self . ofs . exists ( bucket_name ) ) <EOL> self . assertTrue ( self . ofs . claim_bucket ( bucket_name ) ) <EOL> self . assertTrue ( self . ofs . exists ( bucket_name ) ) <EOL> self . assertFalse ( self . ofs . claim_bucket ( bucket_name ) ) <EOL> self . ofs . _del_bucket ( bucket_name ) <EOL> self . assertFalse ( self . ofs . exists ( bucket_name ) ) <EOL> def test_list_buckets ( self ) : <EOL> buckets = [ b for b in self . ofs . list_buckets ( ) ] <EOL> assert len ( buckets ) > <NUM_LIT:0> , len ( buckets ) <EOL> assert self . bucket_name in buckets , buckets <EOL> def test_stream_write_and_read ( self ) : <EOL> name = "<STR_LIT>" <EOL> self . ofs . put_stream ( self . bucket_name , name , self . _makefp ( ) ) <EOL> text = self . ofs . get_stream ( self . bucket_name , name ) . read ( ) <EOL> assert text == TEST_TEXT , text <EOL> text = self . ofs . get_stream ( self . bucket_name , name , as_stream = False ) <EOL> assert text == TEST_TEXT , text <EOL> def test_binary_write_and_read ( self ) : <EOL> name = "<STR_LIT>" <EOL> fh = file ( BINARY_FILE_NAME , '<STR_LIT:rb>' ) <EOL> self . ofs . put_stream ( self . bucket_name , name , fh ) <EOL> fh . close ( ) <EOL> def test_stream_delete ( self ) : <EOL> name = "<STR_LIT>" <EOL> self . ofs . put_stream ( self . bucket_name , name , self . _makefp ( ) ) <EOL> assert self . ofs . get_stream ( self . bucket_name , name ) != None , name <EOL> self . ofs . del_stream ( self . bucket_name , name ) <EOL> self . assertRaises ( OFSException , self . ofs . get_stream , self . bucket_name , name ) <EOL> def test_meta_save_read ( self ) : <EOL> name = "<STR_LIT>" <EOL> self . ofs . put_stream ( self . bucket_name , name , self . _makefp ( ) , params = { '<STR_LIT:hello>' : '<STR_LIT>' , <EOL> '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> meta = self . ofs . get_metadata ( self . bucket_name , name ) <EOL> assert '<STR_LIT>' in meta , meta <EOL> assert '<STR_LIT>' in meta , meta <EOL> assert '<STR_LIT>' in meta , meta <EOL> assert '<STR_LIT>' in meta , meta <EOL> assert '<STR_LIT>' in meta , meta <EOL> assert '<STR_LIT>' in meta , meta <EOL> assert '<STR_LIT>' in meta , meta <EOL> assert '<STR_LIT>' in meta , meta <EOL> assert meta [ '<STR_LIT:hello>' ] == '<STR_LIT>' , meta [ '<STR_LIT:hello>' ] <EOL> assert meta [ '<STR_LIT:foo>' ] == '<STR_LIT:bar>' , meta [ '<STR_LIT:bar>' ] <EOL> def test_meta_update ( self ) : <EOL> name = "<STR_LIT>" <EOL> self . ofs . put_stream ( self . bucket_name , name , self . _makefp ( ) , params = { '<STR_LIT:hello>' : '<STR_LIT>' , <EOL> '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> meta = self . ofs . get_metadata ( self . bucket_name , name ) <EOL> assert meta [ '<STR_LIT:hello>' ] == '<STR_LIT>' , meta [ '<STR_LIT:hello>' ] <EOL> assert meta [ '<STR_LIT:foo>' ] == '<STR_LIT:bar>' , meta [ '<STR_LIT:bar>' ] <EOL> self . ofs . update_metadata ( self . bucket_name , name , { '<STR_LIT:hello>' : '<STR_LIT>' , <EOL> '<STR_LIT:foo>' : '<STR_LIT>' } ) <EOL> meta = self . ofs . get_metadata ( self . bucket_name , name ) <EOL> assert meta [ '<STR_LIT:hello>' ] == '<STR_LIT>' , meta [ '<STR_LIT:hello>' ] <EOL> assert meta [ '<STR_LIT:foo>' ] == '<STR_LIT>' , meta [ '<STR_LIT:bar>' ] <EOL> def test_meta_special_fields ( self ) : <EOL> name = "<STR_LIT>" <EOL> self . ofs . put_stream ( self . bucket_name , name , self . _makefp ( ) , params = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> meta = self . ofs . get_metadata ( self . bucket_name , name ) <EOL> assert meta [ '<STR_LIT>' ] == '<STR_LIT>' , meta [ '<STR_LIT>' ] <EOL> assert meta [ '<STR_LIT>' ] == len ( TEST_TEXT ) , meta [ '<STR_LIT>' ] <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
