<s> """<STR_LIT>""" <EOL> import os <EOL> import types <EOL> import mock <EOL> import swiftclient <EOL> import testtools <EOL> from os_loganalyze . tests import base <EOL> import os_loganalyze . util <EOL> import os_loganalyze . wsgi as log_wsgi <EOL> SEVS = { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:3> , <EOL> '<STR_LIT>' : <NUM_LIT:4> , <EOL> '<STR_LIT>' : <NUM_LIT:5> , <EOL> '<STR_LIT>' : <NUM_LIT:6> <EOL> } <EOL> SEVS_SEQ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ISO8601RE = r'<STR_LIT>' <EOL> def count_types ( gen ) : <EOL> counts = { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> } <EOL> laststatus = None <EOL> for line in gen : <EOL> counts [ '<STR_LIT>' ] = counts [ '<STR_LIT>' ] + <NUM_LIT:1> <EOL> for key in counts : <EOL> if '<STR_LIT>' % key in line : <EOL> laststatus = key <EOL> continue <EOL> if laststatus : <EOL> counts [ laststatus ] = counts [ laststatus ] + <NUM_LIT:1> <EOL> return counts <EOL> def compute_total ( level , counts ) : <EOL> total = <NUM_LIT:0> <EOL> for l in SEVS_SEQ [ SEVS [ level ] : ] : <EOL> if counts . get ( l ) : <EOL> total = total + counts [ l ] <EOL> return total <EOL> def fake_get_object ( self , container , name , resp_chunk_size = None ) : <EOL> name = name [ len ( '<STR_LIT>' ) : ] <EOL> if not os . path . isfile ( base . samples_path ( '<STR_LIT>' ) + name ) : <EOL> return { } , None <EOL> if resp_chunk_size : <EOL> def _object_body ( ) : <EOL> with open ( base . samples_path ( '<STR_LIT>' ) + name ) as f : <EOL> buf = f . read ( resp_chunk_size ) <EOL> while buf : <EOL> yield buf <EOL> buf = f . read ( resp_chunk_size ) <EOL> object_body = _object_body ( ) <EOL> else : <EOL> with open ( base . samples_path ( '<STR_LIT>' ) + name ) as f : <EOL> object_body = f . read ( ) <EOL> resp_headers = os_loganalyze . util . get_headers_for_file ( <EOL> base . samples_path ( '<STR_LIT>' ) + name ) <EOL> return resp_headers , object_body <EOL> def fake_get_container_factory ( _swift_index_items = None ) : <EOL> def fake_get_container ( self , container , prefix = None , delimiter = None ) : <EOL> index_items = [ ] <EOL> if _swift_index_items : <EOL> for i in _swift_index_items : <EOL> if i [ - <NUM_LIT:1> ] == '<STR_LIT:/>' : <EOL> index_items . append ( { '<STR_LIT>' : os . path . join ( prefix , i ) } ) <EOL> else : <EOL> index_items . append ( { '<STR_LIT:name>' : os . path . join ( prefix , i ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } ) <EOL> elif _swift_index_items == [ ] : <EOL> name = prefix [ len ( '<STR_LIT>' ) : ] <EOL> p = os . path . join ( base . samples_path ( '<STR_LIT>' ) , name ) <EOL> for i in os . listdir ( p ) : <EOL> if os . path . isdir ( os . path . join ( p , i ) ) : <EOL> index_items . append ( <EOL> { '<STR_LIT>' : os . path . join ( prefix , i + '<STR_LIT:/>' ) } ) <EOL> else : <EOL> index_items . append ( { '<STR_LIT:name>' : os . path . join ( prefix , i ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } ) <EOL> else : <EOL> pass <EOL> return { } , index_items <EOL> return fake_get_container <EOL> class TestWsgiDisk ( base . TestCase ) : <EOL> """<STR_LIT>""" <EOL> files = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:50> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:6> , <EOL> '<STR_LIT>' : <NUM_LIT:5> <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } , <EOL> } <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_pass_through_all ( self ) : <EOL> for fname in self . files : <EOL> gen = self . get_generator ( fname , html = False ) <EOL> counts = count_types ( gen ) <EOL> self . assertEqual ( counts [ '<STR_LIT>' ] , self . files [ fname ] [ '<STR_LIT>' ] ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_pass_through_at_levels ( self ) : <EOL> for fname in self . files : <EOL> for level in self . files [ fname ] : <EOL> if level == '<STR_LIT>' : <EOL> continue <EOL> gen = self . get_generator ( fname , level = level , html = False ) <EOL> counts = count_types ( gen ) <EOL> total = compute_total ( level , self . files [ fname ] ) <EOL> print ( fname , counts ) <EOL> self . assertEqual ( counts [ '<STR_LIT>' ] , total ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_invalid_file ( self ) : <EOL> gen = log_wsgi . application ( <EOL> self . fake_env ( PATH_INFO = '<STR_LIT>' ) , self . _start_response ) <EOL> self . assertEqual ( gen , [ '<STR_LIT>' ] ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_file_not_found ( self ) : <EOL> gen = log_wsgi . application ( <EOL> self . fake_env ( PATH_INFO = '<STR_LIT>' ) , <EOL> self . _start_response ) <EOL> self . assertEqual ( gen , [ '<STR_LIT>' ] ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_plain_text ( self ) : <EOL> gen = self . get_generator ( '<STR_LIT>' , html = False ) <EOL> self . assertEqual ( type ( gen ) , types . GeneratorType ) <EOL> first = gen . next ( ) <EOL> self . assertIn ( <EOL> '<STR_LIT>' , <EOL> first ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_html_gen ( self ) : <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> first = gen . next ( ) <EOL> self . assertIn ( '<STR_LIT>' , first ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_plain_non_compressed ( self ) : <EOL> gen = self . get_generator ( '<STR_LIT>' , html = False ) <EOL> self . assertEqual ( type ( gen ) , types . GeneratorType ) <EOL> first = gen . next ( ) <EOL> self . assertIn ( <EOL> '<STR_LIT>' , <EOL> first ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_passthrough_filter ( self ) : <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> first = gen . next ( ) <EOL> self . assertNotIn ( '<STR_LIT:html>' , first ) <EOL> with open ( base . samples_path ( '<STR_LIT>' ) + '<STR_LIT>' ) as f : <EOL> self . assertEqual ( first , f . readline ( ) ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_config_no_filter ( self ) : <EOL> self . wsgi_config_file = ( base . samples_path ( '<STR_LIT>' ) + <EOL> '<STR_LIT>' ) <EOL> gen = self . get_generator ( '<STR_LIT>' , limit = <NUM_LIT:10> ) <EOL> lines = <NUM_LIT:0> <EOL> for line in gen : <EOL> lines += <NUM_LIT:1> <EOL> self . assertNotEqual ( <NUM_LIT:12> , lines ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_config_passthrough_view ( self ) : <EOL> self . wsgi_config_file = ( base . samples_path ( '<STR_LIT>' ) + <EOL> '<STR_LIT>' ) <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> first = gen . next ( ) <EOL> self . assertNotIn ( '<STR_LIT>' , first ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_file_conditions ( self ) : <EOL> self . wsgi_config_file = ( base . samples_path ( '<STR_LIT>' ) + <EOL> '<STR_LIT>' ) <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> first = gen . next ( ) <EOL> self . assertIn ( '<STR_LIT>' , first ) <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> first = gen . next ( ) <EOL> self . assertIn ( '<STR_LIT>' , first ) <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> first = gen . next ( ) <EOL> self . assertNotIn ( '<STR_LIT:html>' , first ) <EOL> with open ( base . samples_path ( '<STR_LIT>' ) + '<STR_LIT>' ) as f : <EOL> self . assertEqual ( first , f . readline ( ) ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_accept_range ( self ) : <EOL> gen = self . get_generator ( '<STR_LIT>' , range_bytes = '<STR_LIT>' ) <EOL> body = gen . next ( ) <EOL> self . assertEqual ( '<STR_LIT>' , body ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_accept_range_seek ( self ) : <EOL> gen = self . get_generator ( '<STR_LIT>' , range_bytes = '<STR_LIT>' ) <EOL> body = gen . next ( ) <EOL> self . assertEqual ( '<STR_LIT>' , body ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_accept_range_no_start ( self ) : <EOL> gen = self . get_generator ( '<STR_LIT>' , range_bytes = '<STR_LIT>' ) <EOL> body = gen . next ( ) <EOL> self . assertEqual ( '<STR_LIT>' , body ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_accept_range_no_end ( self ) : <EOL> gen = self . get_generator ( '<STR_LIT>' , range_bytes = '<STR_LIT>' ) <EOL> body = gen . next ( ) <EOL> self . assertNotIn ( '<STR_LIT>' , body ) <EOL> self . assertIn ( '<STR_LIT>' , body ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_accept_range_no_start_no_end ( self ) : <EOL> gen = self . get_generator ( '<STR_LIT>' , range_bytes = '<STR_LIT:->' ) <EOL> body = gen . next ( ) <EOL> self . assertEqual ( '<STR_LIT>' , body ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_accept_range_no_hyphen ( self ) : <EOL> gen = self . get_generator ( '<STR_LIT>' , range_bytes = '<STR_LIT>' ) <EOL> body = gen . next ( ) <EOL> self . assertEqual ( '<STR_LIT>' , body ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_container_factory ( ) ) <EOL> def test_folder_index ( self ) : <EOL> self . wsgi_config_file = ( base . samples_path ( '<STR_LIT>' ) + <EOL> '<STR_LIT>' ) <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> full = '<STR_LIT>' <EOL> for line in gen : <EOL> full += line <EOL> full_lines = full . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( '<STR_LIT>' , full_lines [ <NUM_LIT:0> ] ) <EOL> self . assertIn ( '<STR_LIT>' , full_lines [ <NUM_LIT:3> ] ) <EOL> self . assertThat ( <EOL> full_lines [ <NUM_LIT:9> ] , <EOL> testtools . matchers . MatchesRegex ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' + ISO8601RE + r'<STR_LIT>' <EOL> r'<STR_LIT>' ) ) <EOL> self . assertThat ( <EOL> full_lines [ - <NUM_LIT:5> ] , <EOL> testtools . matchers . MatchesRegex ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' + ISO8601RE + r'<STR_LIT>' <EOL> r'<STR_LIT>' ) ) <EOL> self . assertEqual ( '<STR_LIT>' , full_lines [ - <NUM_LIT:1> ] ) <EOL> class TestWsgiSwift ( TestWsgiDisk ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( TestWsgiSwift , self ) . setUp ( ) <EOL> self . samples_directory = '<STR_LIT>' <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_compare_disk_to_swift_html ( self ) : <EOL> """<STR_LIT>""" <EOL> self . samples_directory = '<STR_LIT>' <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> result_disk = '<STR_LIT>' <EOL> for line in gen : <EOL> result_disk += line <EOL> self . samples_directory = '<STR_LIT>' <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> result_swift = '<STR_LIT>' <EOL> for line in gen : <EOL> result_swift += line <EOL> self . assertEqual ( result_disk , result_swift ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_compare_disk_to_swift_plain ( self ) : <EOL> """<STR_LIT>""" <EOL> self . samples_directory = '<STR_LIT>' <EOL> gen = self . get_generator ( '<STR_LIT>' , html = False ) <EOL> result_disk = '<STR_LIT>' <EOL> for line in gen : <EOL> result_disk += line <EOL> self . samples_directory = '<STR_LIT>' <EOL> gen = self . get_generator ( '<STR_LIT>' , html = False ) <EOL> result_swift = '<STR_LIT>' <EOL> for line in gen : <EOL> result_swift += line <EOL> self . assertEqual ( result_disk , result_swift ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_skip_file ( self ) : <EOL> self . assertRaises ( <EOL> TypeError , <EOL> self . get_generator ( '<STR_LIT>' , source = '<STR_LIT>' ) ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_compare_disk_to_swift_no_compression ( self ) : <EOL> """<STR_LIT>""" <EOL> self . samples_directory = '<STR_LIT>' <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> result_disk = '<STR_LIT>' <EOL> for line in gen : <EOL> result_disk += line <EOL> self . samples_directory = '<STR_LIT>' <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> result_swift = '<STR_LIT>' <EOL> for line in gen : <EOL> result_swift += line <EOL> self . assertEqual ( result_disk , result_swift ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> def test_compare_disk_to_swift_no_chunks ( self ) : <EOL> self . wsgi_config_file = ( base . samples_path ( '<STR_LIT>' ) + <EOL> '<STR_LIT>' ) <EOL> self . test_compare_disk_to_swift_no_compression ( ) <EOL> self . test_compare_disk_to_swift_plain ( ) <EOL> self . test_compare_disk_to_swift_html ( ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_object ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_container_factory ( [ ] ) ) <EOL> def test_folder_index ( self ) : <EOL> self . wsgi_config_file = ( base . samples_path ( '<STR_LIT>' ) + <EOL> '<STR_LIT>' ) <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> full = '<STR_LIT>' <EOL> for line in gen : <EOL> full += line <EOL> full_lines = full . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( '<STR_LIT>' , full_lines [ <NUM_LIT:0> ] ) <EOL> self . assertIn ( '<STR_LIT>' , full_lines [ <NUM_LIT:3> ] ) <EOL> self . assertThat ( <EOL> full_lines [ <NUM_LIT:9> ] , <EOL> testtools . matchers . MatchesRegex ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' + ISO8601RE + r'<STR_LIT>' <EOL> r'<STR_LIT>' ) ) <EOL> self . assertThat ( <EOL> full_lines [ - <NUM_LIT:5> ] , <EOL> testtools . matchers . MatchesRegex ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' + ISO8601RE + r'<STR_LIT>' <EOL> r'<STR_LIT>' ) ) <EOL> self . assertEqual ( '<STR_LIT>' , full_lines [ - <NUM_LIT:1> ] ) <EOL> @ mock . patch . object ( swiftclient . client . Connection , '<STR_LIT>' , <EOL> fake_get_container_factory ( [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT>' , '<STR_LIT:z>' ] ) ) <EOL> def test_folder_index_dual ( self ) : <EOL> self . samples_directory = '<STR_LIT>' <EOL> self . wsgi_config_file = ( base . samples_path ( '<STR_LIT>' ) + <EOL> '<STR_LIT>' ) <EOL> gen = self . get_generator ( '<STR_LIT>' ) <EOL> full = '<STR_LIT>' <EOL> for line in gen : <EOL> full += line <EOL> full_lines = full . split ( '<STR_LIT:\n>' ) <EOL> self . assertEqual ( '<STR_LIT>' , full_lines [ <NUM_LIT:0> ] ) <EOL> self . assertIn ( '<STR_LIT>' , full_lines [ <NUM_LIT:3> ] ) <EOL> self . assertThat ( <EOL> full_lines [ <NUM_LIT:9> ] , <EOL> testtools . matchers . MatchesRegex ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' + ISO8601RE + r'<STR_LIT>' <EOL> r'<STR_LIT>' ) ) <EOL> self . assertThat ( <EOL> full_lines [ <NUM_LIT:11> ] , <EOL> testtools . matchers . MatchesRegex ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' + ISO8601RE + r'<STR_LIT>' <EOL> r'<STR_LIT>' ) ) <EOL> self . assertThat ( <EOL> full_lines [ <NUM_LIT> ] , <EOL> testtools . matchers . MatchesRegex ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' + ISO8601RE + r'<STR_LIT>' <EOL> r'<STR_LIT>' ) ) <EOL> self . assertEqual ( <EOL> full_lines [ <NUM_LIT> ] , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertThat ( <EOL> full_lines [ - <NUM_LIT:7> ] , <EOL> testtools . matchers . MatchesRegex ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' + ISO8601RE + r'<STR_LIT>' <EOL> r'<STR_LIT>' ) ) <EOL> self . assertThat ( <EOL> full_lines [ - <NUM_LIT:5> ] , <EOL> testtools . matchers . MatchesRegex ( <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' + ISO8601RE + r'<STR_LIT>' <EOL> r'<STR_LIT>' ) ) <EOL> self . assertEqual ( '<STR_LIT>' , full_lines [ - <NUM_LIT:1> ] ) </s>
