<s> import StringIO <EOL> import sys <EOL> import time <EOL> from nose import tools <EOL> from docker_registry . core import exceptions <EOL> import docker_registry . testing as testing <EOL> from docker_registry . testing import mock_boto <EOL> from . import mock_s3 <EOL> class StringIOWithError ( StringIO . StringIO ) : <EOL> '''<STR_LIT>''' <EOL> def read ( self , size ) : <EOL> if self . pos == self . len : <EOL> raise IOError ( '<STR_LIT>' ) <EOL> return StringIO . StringIO . read ( self , size ) <EOL> class TestDriver ( testing . Driver ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self ) : <EOL> self . scheme = '<STR_LIT>' <EOL> self . path = '<STR_LIT>' <EOL> self . config = testing . Config ( { } ) <EOL> def tearDown ( self ) : <EOL> self . _storage . _boto_bucket . delete ( ) <EOL> super ( TestDriver , self ) . tearDown ( ) <EOL> @ tools . raises ( exceptions . FileNotFoundError ) <EOL> def test_list_bucket ( self ) : <EOL> filename1 = self . gen_random_string ( ) <EOL> filename2 = self . gen_random_string ( ) <EOL> content = self . gen_random_string ( ) <EOL> self . _storage . put_content ( filename1 , content ) <EOL> self . _storage . put_content ( filename2 + '<STR_LIT:/>' , content ) <EOL> assert sorted ( [ filename1 , filename2 ] ) == sorted ( <EOL> list ( self . _storage . list_directory ( ) ) ) <EOL> self . _storage . remove ( filename1 ) <EOL> self . _storage . remove ( filename2 ) <EOL> s = self . _storage . list_directory ( ) <EOL> s . next ( ) <EOL> def test_stream_write ( self ) : <EOL> self . _storage . buffer_size = <NUM_LIT:7> * <NUM_LIT> * <NUM_LIT> <EOL> filename = self . gen_random_string ( ) <EOL> content = self . gen_random_string ( <NUM_LIT:8> * <NUM_LIT> * <NUM_LIT> ) <EOL> io = StringIOWithError ( content ) <EOL> assert not self . _storage . exists ( filename ) <EOL> try : <EOL> self . _storage . stream_write ( filename , io ) <EOL> except IOError : <EOL> pass <EOL> assert self . _storage . exists ( filename ) <EOL> try : <EOL> self . _storage . stream_write ( filename , io ) <EOL> except IOError : <EOL> pass <EOL> io . close ( ) <EOL> self . _storage . remove ( filename ) <EOL> self . _storage . buffer_size = <NUM_LIT:5> * <NUM_LIT> * <NUM_LIT> <EOL> assert not self . _storage . exists ( filename ) <EOL> def test_init_path ( self ) : <EOL> root_path = self . _storage . _root_path <EOL> if root_path . startswith ( '<STR_LIT:/>' ) : <EOL> self . _storage . _root_path = root_path [ <NUM_LIT:1> : ] <EOL> assert not self . _storage . _init_path ( ) . startswith ( '<STR_LIT:/>' ) <EOL> self . _storage . _root_path = root_path <EOL> def test_debug_key ( self ) : <EOL> filename = self . gen_random_string ( ) <EOL> content = self . gen_random_string ( ) <EOL> self . _storage . put_content ( filename , content ) <EOL> key_path = self . _storage . _init_path ( filename ) <EOL> key = self . _storage . _boto_bucket . lookup ( key_path ) <EOL> self . _storage . _debug_key ( key ) <EOL> saved_stdout = sys . stdout <EOL> output = StringIO . StringIO ( ) <EOL> sys . stdout = output <EOL> dummy = "<STR_LIT>" <EOL> result = self . _storage . _boto_bucket . connection . make_request ( <EOL> '<STR_LIT:d>' , <NUM_LIT:1> , v = <NUM_LIT:2> ) <EOL> assert output . getvalue ( ) == dummy <EOL> assert result == '<STR_LIT>' <EOL> sys . stdout = saved_stdout <EOL> def test_get_tags ( self ) : <EOL> store = self . _storage <EOL> store . _root_path = '<STR_LIT>' <EOL> store . _init_path ( ) <EOL> assert store . _root_path == '<STR_LIT>' <EOL> tag_path = store . tag_path ( '<STR_LIT:test>' , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> store . put_content ( tag_path , '<STR_LIT>' ) <EOL> tags_path = store . tag_path ( '<STR_LIT:test>' , '<STR_LIT:test>' ) <EOL> for fname in store . list_directory ( tags_path ) : <EOL> full_tag_name = fname . split ( '<STR_LIT:/>' ) . pop ( ) <EOL> if not full_tag_name == '<STR_LIT>' : <EOL> continue <EOL> try : <EOL> store . get_content ( fname ) <EOL> except exceptions . FileNotFoundError : <EOL> pass <EOL> except Exception as e : <EOL> raise e <EOL> else : <EOL> assert False <EOL> tag_content = store . get_content ( tag_path ) <EOL> assert tag_content == '<STR_LIT>' <EOL> def test_consistency_latency ( self ) : <EOL> self . testCount = - <NUM_LIT:1> <EOL> mockKey = mock_boto . Key ( ) <EOL> def mockExists ( ) : <EOL> self . testCount += <NUM_LIT:1> <EOL> return self . testCount == <NUM_LIT:1> <EOL> mockKey . exists = mockExists <EOL> mockKey . get_contents_as_string = lambda : "<STR_LIT>" <EOL> self . _storage . makeKey = lambda x : mockKey <EOL> startTime = time . time ( ) <EOL> content = self . _storage . get_content ( "<STR_LIT>" ) <EOL> waitTime = time . time ( ) - startTime <EOL> assert waitTime >= <NUM_LIT:0.1> , ( "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( <NUM_LIT:0.1> * <NUM_LIT:1000> , waitTime * <NUM_LIT:1000> ) ) <EOL> assert content == "<STR_LIT>" , ( "<STR_LIT>" % <EOL> ( "<STR_LIT>" , content ) ) <EOL> @ tools . raises ( exceptions . FileNotFoundError ) <EOL> def test_too_many_read_retries ( self ) : <EOL> self . testCount = - <NUM_LIT:1> <EOL> mockKey = mock_boto . Key ( ) <EOL> def mockExists ( ) : <EOL> self . testCount += <NUM_LIT:1> <EOL> return self . testCount == <NUM_LIT:5> <EOL> mockKey . exists = mockExists <EOL> mockKey . get_contents_as_string = lambda : "<STR_LIT>" <EOL> self . _storage . makeKey = lambda x : mockKey <EOL> self . _storage . get_content ( "<STR_LIT>" ) </s>
