#NEWLINE# def _test_end_to_end( self , args = ( ) ) : #NEWLINE# #INDENT# stdin = BytesIO ( b'foo\nbar\n' ) #NEWLINE# local_input_path = os . path . join ( self . tmp_dir , 'input' ) #NEWLINE# with open ( local_input_path , 'w' ) as local_input_file : #NEWLINE# #INDENT# local_input_file . write ( 'bar\nqux\n' ) #NEWLINE# #UNINDENT# input_to_upload = os . path . join ( self . tmp_dir , 'remote_input' ) #NEWLINE# with open ( input_to_upload , 'w' ) as input_to_upload_file : #NEWLINE# #INDENT# input_to_upload_file . write ( 'foo\n' ) #NEWLINE# #UNINDENT# remote_input_path = 'hdfs:///data/foo' #NEWLINE# check_call ( [ self . hadoop_bin , 'fs' , '-put' , input_to_upload , remote_input_path ] ) #NEWLINE# add_mock_hadoop_counters ( { #NEWLINE# #INDENT# 'foo' : { #NEWLINE# #INDENT# 'bar' : 23 , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } ) #NEWLINE# add_mock_hadoop_counters ( { #NEWLINE# #INDENT# 'baz' : { #NEWLINE# #INDENT# 'qux' : 42 , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } ) #NEWLINE# add_mock_hadoop_output ( [ b'' ] ) #NEWLINE# add_mock_hadoop_output ( [ b'1\t"qux"\n2\t"bar"\n' , b'2\t"foo"\n5\tnull\n' ] ) #NEWLINE# mr_job = MRTwoStepJob ( ( ( ( [ '-r' , 'hadoop' , '-v' , '--no-conf' , '--hadoop-arg' , '-libjar' , '--hadoop-arg' , 'containsJars.jar' ] + list ( input_to_upload_file ) ) + [ '-' , local_input_path , remote_input_path ] ) + [ '--jobconf' , 'x=y' ] ) ) #NEWLINE# mr_job . sandbox ( stdin = stdin ) #NEWLINE# local_tmp_dir = None #NEWLINE# results = [ ] #NEWLINE# with mr_job . make_runner ( ) as runner : #NEWLINE# #INDENT# assert isinstance ( runner , HadoopJobRunner ) #NEWLINE# runner . run ( ) #NEWLINE# for line in runner . stream_output ( ) : #NEWLINE# #INDENT# ( key , value ) = mr_job . parse_output_line ( line ) #NEWLINE# results . append ( ( key , value ) ) #NEWLINE# #UNINDENT# local_tmp_dir = runner . _get_local_tmp_dir ( ) #NEWLINE# assert os . path . exists ( local_tmp_dir ) #NEWLINE# assert any ( runner . fs . ls ( runner . get_output_dir ( ) ) ) #NEWLINE# hdfs_root = get_mock_hdfs_root ( ) #NEWLINE# self . assertEqual ( sorted ( os . listdir ( hdfs_root ) ) , [ 'data' , 'user' ] ) #NEWLINE# home_dir = os . path . join ( hdfs_root , 'user' , getpass . getuser ( ) ) #NEWLINE# self . assertEqual ( os . listdir ( home_dir ) , [ 'tmp' ] ) #NEWLINE# self . assertEqual ( os . listdir ( os . path . join ( home_dir , 'tmp' ) ) , [ 'mrjob' ] ) #NEWLINE# self . assertEqual ( runner . _opts [ 'hadoop_extra_args' ] , [ '-libjar' , 'containsJars.jar' ] ) #NEWLINE# self . assertTrue ( os . path . exists ( runner . _mrjob_tar_gz_path ) ) #NEWLINE# self . assertIn ( runner . _mrjob_tar_gz_path , runner . _upload_mgr . path_to_uri ( ) ) #NEWLINE# self . assertTrue ( os . path . exists ( runner . _setup_wrapper_script_path ) ) #NEWLINE# self . assertIn ( runner . _setup_wrapper_script_path , runner . _upload_mgr . path_to_uri ( ) ) #NEWLINE# mrjob_tar_gz_name = runner . _working_dir_mgr . name ( 'archive' , runner . _mrjob_tar_gz_path ) #NEWLINE# with open ( runner . _setup_wrapper_script_path ) as wrapper : #NEWLINE# #INDENT# self . assertTrue ( any ( ( ( ( 'export PYTHONPATH' in line ) and ( mrjob_tar_gz_name in line ) ) for line in wrapper ) ) ) #NEWLINE# #UNINDENT# #UNINDENT# self . assertEqual ( runner . counters ( ) , [ { #NEWLINE# #INDENT# 'foo' : { #NEWLINE# #INDENT# 'bar' : 23 , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } , { #NEWLINE# #INDENT# 'baz' : { #NEWLINE# #INDENT# 'qux' : 42 , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } ] ) #NEWLINE# self . assertEqual ( sorted ( results ) , [ ( 1 , 'qux' ) , ( 2 , 'bar' ) , ( 2 , 'foo' ) , ( 5 , None ) ] ) #NEWLINE# hadoop_cmd_args = get_mock_hadoop_cmd_args ( ) #NEWLINE# jar_cmd_args = [ cmd_args for cmd_args in hadoop_cmd_args if ( cmd_args [ : 1 ] == [ 'jar' ] ) ] #NEWLINE# self . assertEqual ( len ( jar_cmd_args ) , 2 ) #NEWLINE# ( step_0_args , step_1_args ) = jar_cmd_args #NEWLINE# self . assertIn ( '-inputformat' , step_0_args ) #NEWLINE# self . assertNotIn ( '-outputformat' , step_0_args ) #NEWLINE# self . assertNotIn ( '-inputformat' , step_1_args ) #NEWLINE# self . assertIn ( '-outputformat' , step_1_args ) #NEWLINE# for args in ( step_0_args , step_1_args ) : #NEWLINE# #INDENT# self . assertIn ( '-libjar' , args ) #NEWLINE# self . assertIn ( '-mapper' , args ) #NEWLINE# self . assertLess ( args . index ( '-libjar' ) , args . index ( '-mapper' ) ) #NEWLINE# #UNINDENT# self . assertIn ( '-D' , step_0_args ) #NEWLINE# self . assertIn ( 'x=y' , step_0_args ) #NEWLINE# self . assertIn ( '-D' , step_1_args ) #NEWLINE# self . assertIn ( 'x=z' , step_1_args ) #NEWLINE# assert ( not os . path . exists ( local_tmp_dir ) ) #NEWLINE# assert ( not any ( runner . fs . ls ( runner . get_output_dir ( ) ) ) )