#NEWLINE# @ patch ( 'subprocess.Popen' ) #NEWLINE# def test_run__fail( self , mock ) : #NEWLINE# #INDENT# arglist_result = [ ] #NEWLINE# p = subprocess . Popen #NEWLINE# subprocess . Popen = _get_fake_Popen ( arglist_result , 1 ) #NEWLINE# with tempfile . NamedTemporaryFile ( delete = False ) as param_file_mock , tempfile . NamedTemporaryFile ( delete = False ) as prop_file_mock , patch ( 'luigi.contrib.pig.tempfile.NamedTemporaryFile' , side_effect = [ param_file_mock , prop_file_mock ] ) : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# job = ComplexTestJob ( ) #NEWLINE# job . run ( ) #NEWLINE# #UNINDENT# except PigJobError as e : #NEWLINE# #INDENT# p = e #NEWLINE# self . assertEqual ( 'stderr' , p . err ) #NEWLINE# self . assertEqual ( [ [ '/usr/share/pig/bin/pig' , '-x' , 'local' , '-param_file' , arglist_result . name , '-propertyFile' , prop_file_mock . name , '-f' , 'my_complex_pig_script.pig' ] ] , arglist_result ) #NEWLINE# with open ( param_file_mock . name ) as pparams_file : #NEWLINE# #INDENT# pparams = pparams_file . readlines ( ) #NEWLINE# self . assertEqual ( 1 , len ( pparams ) ) #NEWLINE# self . assertEqual ( 'YOUR_PARAM_NAME=Your param value\n' , pparams [ 0 ] ) #NEWLINE# #UNINDENT# with open ( prop_file_mock . name ) as pprops_file : #NEWLINE# #INDENT# pprops = pprops_file . readlines ( ) #NEWLINE# self . assertEqual ( 1 , len ( pprops ) ) #NEWLINE# self . assertEqual ( 'pig.additional.jars=/path/to/your/jar\n' , pprops [ 0 ] ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# self . fail ( 'Should have thrown PigJobError' ) #NEWLINE# #UNINDENT# finally : #NEWLINE# #INDENT# subprocess . Popen = p