<s> import sys <EOL> from io import BytesIO <EOL> from subprocess import PIPE <EOL> from subprocess import Popen <EOL> from mrjob . parse import _find_python_traceback <EOL> from mrjob . parse import is_windows_path <EOL> from mrjob . parse import is_s3_uri <EOL> from mrjob . parse import is_uri <EOL> from mrjob . parse import parse_mr_job_stderr <EOL> from mrjob . parse import parse_port_range_list <EOL> from mrjob . parse import parse_s3_uri <EOL> from mrjob . parse import urlparse <EOL> from mrjob . parse import _parse_progress_from_job_tracker <EOL> from mrjob . parse import _parse_progress_from_resource_manager <EOL> from tests . py2 import TestCase <EOL> class FindPythonTracebackTestCase ( TestCase ) : <EOL> EXAMPLE_TRACEBACK = b"""<STR_LIT>""" <EOL> EXAMPLE_STDERR_TRACEBACK_1 = b"""<STR_LIT>""" <EOL> EXAMPLE_STDERR_TRACEBACK_2 = b"""<STR_LIT>""" <EOL> def test_find_python_traceback ( self ) : <EOL> def run ( * args ) : <EOL> return Popen ( args , stdout = PIPE , stderr = PIPE ) . communicate ( ) <EOL> ok_stdout , ok_stderr = run ( '<STR_LIT>' , '<STR_LIT:-c>' , "<STR_LIT>" ) <EOL> self . assertEqual ( ok_stdout . rstrip ( ) , b"<STR_LIT>" ) <EOL> self . assertEqual ( _find_python_traceback ( BytesIO ( ok_stderr ) ) , None ) <EOL> stdout , stderr = run ( '<STR_LIT>' , '<STR_LIT:-c>' , "<STR_LIT>" ) <EOL> self . assertEqual ( stdout , b'<STR_LIT>' ) <EOL> tb = _find_python_traceback ( BytesIO ( stderr ) ) <EOL> self . assertNotEqual ( tb , None ) <EOL> assert isinstance ( tb , list ) <EOL> self . assertIn ( "<STR_LIT>" , tb [ <NUM_LIT:0> ] ) <EOL> self . assertIn ( "<STR_LIT>" , tb [ - <NUM_LIT:1> ] ) <EOL> if hasattr ( sys , '<STR_LIT>' ) : <EOL> return <EOL> verbose_stdout , verbose_stderr = run ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:-c>' , "<STR_LIT>" ) <EOL> self . assertEqual ( verbose_stdout , b'<STR_LIT>' ) <EOL> self . assertNotEqual ( verbose_stderr , stderr ) <EOL> verbose_tb = _find_python_traceback ( BytesIO ( verbose_stderr ) ) <EOL> self . assertEqual ( verbose_tb , tb ) <EOL> def test_find_multiple_python_tracebacks ( self ) : <EOL> total_traceback = self . EXAMPLE_TRACEBACK + b'<STR_LIT>' <EOL> tb = _find_python_traceback ( BytesIO ( total_traceback ) ) <EOL> self . assertEqual ( '<STR_LIT>' . join ( tb ) , <EOL> self . EXAMPLE_TRACEBACK . decode ( '<STR_LIT:ascii>' ) ) <EOL> def test_find_python_traceback_with_more_stderr ( self ) : <EOL> total_traceback = self . EXAMPLE_STDERR_TRACEBACK_1 + b'<STR_LIT>' <EOL> tb = _find_python_traceback ( BytesIO ( total_traceback ) ) <EOL> self . assertEqual ( '<STR_LIT>' . join ( tb ) , <EOL> self . EXAMPLE_STDERR_TRACEBACK_1 . decode ( '<STR_LIT:ascii>' ) ) <EOL> def test_find_python_traceback_with_more_stderr_2 ( self ) : <EOL> total_traceback = self . EXAMPLE_STDERR_TRACEBACK_2 + b'<STR_LIT>' <EOL> tb = _find_python_traceback ( BytesIO ( total_traceback ) ) <EOL> self . assertEqual ( '<STR_LIT>' . join ( tb ) , <EOL> self . EXAMPLE_STDERR_TRACEBACK_2 . decode ( '<STR_LIT:ascii>' ) ) <EOL> class ParseMRJobStderrTestCase ( TestCase ) : <EOL> def test_empty ( self ) : <EOL> self . assertEqual ( parse_mr_job_stderr ( BytesIO ( ) ) , <EOL> { '<STR_LIT>' : { } , '<STR_LIT>' : [ ] , '<STR_LIT>' : [ ] } ) <EOL> def test_parsing ( self ) : <EOL> INPUT = BytesIO ( <EOL> b'<STR_LIT>' + <EOL> b'<STR_LIT>' + <EOL> b'<STR_LIT>' + <EOL> b'<STR_LIT>' + <EOL> b'<STR_LIT>' + <EOL> b'<STR_LIT>' + <EOL> b'<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> parse_mr_job_stderr ( INPUT ) , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:3> , '<STR_LIT>' : <NUM_LIT:1> } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT> } } , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] <EOL> } ) <EOL> def test_update_counters ( self ) : <EOL> counters = { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:3> , '<STR_LIT>' : <NUM_LIT:1> } } <EOL> parse_mr_job_stderr ( <EOL> BytesIO ( b'<STR_LIT>' ) , counters = counters ) <EOL> self . assertEqual ( counters , { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:3> , '<STR_LIT>' : <NUM_LIT:2> } } ) <EOL> def test_read_single_line ( self ) : <EOL> self . assertEqual ( parse_mr_job_stderr ( b'<STR_LIT>' ) , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:2> } } , <EOL> '<STR_LIT>' : [ ] , '<STR_LIT>' : [ ] } ) <EOL> def test_read_multiple_lines_from_buffer ( self ) : <EOL> self . assertEqual ( <EOL> parse_mr_job_stderr ( b'<STR_LIT>' ) , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:2> } } , <EOL> '<STR_LIT>' : [ ] , '<STR_LIT>' : [ '<STR_LIT>' ] } ) <EOL> def test_negative_counters ( self ) : <EOL> self . assertEqual ( <EOL> parse_mr_job_stderr ( [ b'<STR_LIT>' ] ) , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : - <NUM_LIT:2> } } , <EOL> '<STR_LIT>' : [ ] , '<STR_LIT>' : [ ] } ) <EOL> def test_garbled_counters ( self ) : <EOL> BAD_LINES = [ <EOL> b'<STR_LIT>' , <EOL> b'<STR_LIT>' , <EOL> b'<STR_LIT>' , <EOL> b'<STR_LIT>' , <EOL> b'<STR_LIT>' , <EOL> b'<STR_LIT>' , <EOL> ] <EOL> self . assertEqual ( <EOL> parse_mr_job_stderr ( BAD_LINES ) , <EOL> { '<STR_LIT>' : { } , '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : [ line . decode ( '<STR_LIT:ascii>' ) for line in BAD_LINES ] } ) <EOL> class PortRangeListTestCase ( TestCase ) : <EOL> def test_port_range_list ( self ) : <EOL> self . assertEqual ( parse_port_range_list ( '<STR_LIT>' ) , [ <NUM_LIT> ] ) <EOL> self . assertEqual ( parse_port_range_list ( '<STR_LIT>' ) , [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . assertEqual ( parse_port_range_list ( '<STR_LIT>' ) , [ <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . assertEqual ( parse_port_range_list ( '<STR_LIT>' ) , <EOL> [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . assertEqual ( parse_port_range_list ( '<STR_LIT>' ) , <EOL> [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . assertEqual ( parse_port_range_list ( '<STR_LIT>' ) , <EOL> [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> self . assertEqual ( parse_port_range_list ( '<STR_LIT>' ) , [ <NUM_LIT> ] ) <EOL> self . assertRaises ( ValueError , parse_port_range_list , '<STR_LIT>' ) <EOL> self . assertRaises ( ValueError , parse_port_range_list , <EOL> '<STR_LIT>' ) <EOL> class URITestCase ( TestCase ) : <EOL> def test_uri_parsing ( self ) : <EOL> self . assertEqual ( is_uri ( '<STR_LIT>' ) , False ) <EOL> self . assertEqual ( is_uri ( '<STR_LIT>' ) , True ) <EOL> self . assertEqual ( is_uri ( '<STR_LIT>' ) , False ) <EOL> self . assertEqual ( is_windows_path ( '<STR_LIT>' ) , True ) <EOL> self . assertEqual ( is_windows_path ( '<STR_LIT>' ) , False ) <EOL> self . assertEqual ( is_s3_uri ( '<STR_LIT>' ) , True ) <EOL> self . assertEqual ( is_s3_uri ( '<STR_LIT>' ) , True ) <EOL> self . assertEqual ( is_s3_uri ( '<STR_LIT>' ) , False ) <EOL> self . assertEqual ( parse_s3_uri ( '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_urlparse ( self ) : <EOL> self . assertEqual ( urlparse ( '<STR_LIT>' ) , <EOL> ( '<STR_LIT:http>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertEqual ( urlparse ( '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertEqual ( urlparse ( '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertEqual ( urlparse ( '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertEqual ( urlparse ( '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertEqual ( urlparse ( '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:/>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> class JobTrackerProgressTestCase ( TestCase ) : <EOL> def test_empty ( self ) : <EOL> self . assertEqual ( _parse_progress_from_job_tracker ( b'<STR_LIT>' ) , ( None , None ) ) <EOL> def test_on_html_snippet ( self ) : <EOL> HTML = b"""<STR_LIT>""" <EOL> self . assertEqual ( _parse_progress_from_job_tracker ( HTML ) , <EOL> ( <NUM_LIT> , <NUM_LIT:0> ) ) <EOL> class ResourceManagerProgressTestCase ( TestCase ) : <EOL> def test_empty ( self ) : <EOL> self . assertEqual ( _parse_progress_from_resource_manager ( b'<STR_LIT>' ) , None ) <EOL> def test_on_javascript_snippet ( self ) : <EOL> JS = b"""<STR_LIT>""" <EOL> self . assertEqual ( _parse_progress_from_resource_manager ( JS ) , <NUM_LIT> ) </s>
