<s> """<STR_LIT>""" <EOL> import datetime <EOL> import getpass <EOL> import os <EOL> import os . path <EOL> import shutil <EOL> import signal <EOL> import stat <EOL> import sys <EOL> import tarfile <EOL> import tempfile <EOL> from io import BytesIO <EOL> from subprocess import CalledProcessError <EOL> from mrjob . hadoop import HadoopJobRunner <EOL> from mrjob . inline import InlineMRJobRunner <EOL> from mrjob . job import MRJob <EOL> from mrjob . local import LocalMRJobRunner <EOL> from mrjob . py2 import PY2 <EOL> from mrjob . py2 import StringIO <EOL> from mrjob . runner import MRJobRunner <EOL> from mrjob . tools . emr . audit_usage import _JOB_KEY_RE <EOL> from mrjob . util import log_to_stream <EOL> from mrjob . util import tar_and_gzip <EOL> from tests . mr_os_walk_job import MROSWalkJob <EOL> from tests . mr_two_step_job import MRTwoStepJob <EOL> from tests . mr_word_count import MRWordCount <EOL> from tests . py2 import TestCase <EOL> from tests . py2 import patch <EOL> from tests . quiet import no_handlers_for_logger <EOL> from tests . sandbox import EmptyMrjobConfTestCase <EOL> from tests . sandbox import SandboxedTestCase <EOL> from tests . sandbox import mrjob_conf_patcher <EOL> class WithStatementTestCase ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . local_tmp_dir = None <EOL> def tearDown ( self ) : <EOL> if self . local_tmp_dir : <EOL> shutil . rmtree ( self . local_tmp_dir ) <EOL> self . local_tmp_dir = None <EOL> def _test_cleanup_after_with_statement ( self , mode , should_exist ) : <EOL> with InlineMRJobRunner ( cleanup = mode , conf_paths = [ ] ) as runner : <EOL> self . local_tmp_dir = runner . _get_local_tmp_dir ( ) <EOL> self . assertTrue ( os . path . exists ( self . local_tmp_dir ) ) <EOL> self . assertEqual ( os . path . exists ( self . local_tmp_dir ) , should_exist ) <EOL> if not should_exist : <EOL> self . local_tmp_dir = None <EOL> def test_cleanup_all ( self ) : <EOL> self . _test_cleanup_after_with_statement ( [ '<STR_LIT>' ] , False ) <EOL> def test_cleanup_tmp ( self ) : <EOL> self . _test_cleanup_after_with_statement ( [ '<STR_LIT>' ] , False ) <EOL> def test_cleanup_local_tmp ( self ) : <EOL> self . _test_cleanup_after_with_statement ( [ '<STR_LIT>' ] , False ) <EOL> def test_cleanup_cloud_tmp ( self ) : <EOL> self . _test_cleanup_after_with_statement ( [ '<STR_LIT>' ] , True ) <EOL> def test_cleanup_none ( self ) : <EOL> self . _test_cleanup_after_with_statement ( [ '<STR_LIT>' ] , True ) <EOL> def test_cleanup_error ( self ) : <EOL> self . assertRaises ( ValueError , self . _test_cleanup_after_with_statement , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , True ) <EOL> self . assertRaises ( ValueError , self . _test_cleanup_after_with_statement , <EOL> [ '<STR_LIT>' ] , True ) <EOL> def test_double_none_okay ( self ) : <EOL> self . _test_cleanup_after_with_statement ( [ '<STR_LIT>' , '<STR_LIT>' ] , True ) <EOL> class TestJobName ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . blank_out_environment ( ) <EOL> self . monkey_patch_getuser ( ) <EOL> def tearDown ( self ) : <EOL> self . restore_getuser ( ) <EOL> self . restore_environment ( ) <EOL> def blank_out_environment ( self ) : <EOL> self . _old_environ = os . environ . copy ( ) <EOL> os . environ . clear ( ) <EOL> def restore_environment ( self ) : <EOL> os . environ . clear ( ) <EOL> os . environ . update ( self . _old_environ ) <EOL> def monkey_patch_getuser ( self ) : <EOL> self . _real_getuser = getpass . getuser <EOL> self . getuser_should_fail = False <EOL> def fake_getuser ( ) : <EOL> if self . getuser_should_fail : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> else : <EOL> return self . _real_getuser ( ) <EOL> getpass . getuser = fake_getuser <EOL> def restore_getuser ( self ) : <EOL> getpass . getuser = self . _real_getuser <EOL> def test_empty ( self ) : <EOL> runner = InlineMRJobRunner ( conf_paths = [ ] ) <EOL> match = _JOB_KEY_RE . match ( runner . get_job_key ( ) ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:1> ) , '<STR_LIT>' ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:2> ) , getpass . getuser ( ) ) <EOL> def test_empty_no_user ( self ) : <EOL> self . getuser_should_fail = True <EOL> runner = InlineMRJobRunner ( conf_paths = [ ] ) <EOL> match = _JOB_KEY_RE . match ( runner . get_job_key ( ) ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:1> ) , '<STR_LIT>' ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:2> ) , '<STR_LIT>' ) <EOL> def test_auto_label ( self ) : <EOL> runner = MRTwoStepJob ( [ '<STR_LIT>' ] ) . make_runner ( ) <EOL> match = _JOB_KEY_RE . match ( runner . get_job_key ( ) ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:1> ) , '<STR_LIT>' ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:2> ) , getpass . getuser ( ) ) <EOL> def test_auto_owner ( self ) : <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> runner = InlineMRJobRunner ( conf_paths = [ ] ) <EOL> match = _JOB_KEY_RE . match ( runner . get_job_key ( ) ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:1> ) , '<STR_LIT>' ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:2> ) , '<STR_LIT>' ) <EOL> def test_auto_everything ( self ) : <EOL> test_start = datetime . datetime . utcnow ( ) <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> runner = MRTwoStepJob ( [ '<STR_LIT>' ] ) . make_runner ( ) <EOL> match = _JOB_KEY_RE . match ( runner . get_job_key ( ) ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:1> ) , '<STR_LIT>' ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:2> ) , '<STR_LIT>' ) <EOL> job_start = datetime . datetime . strptime ( <EOL> match . group ( <NUM_LIT:3> ) + match . group ( <NUM_LIT:4> ) , '<STR_LIT>' ) <EOL> job_start = job_start . replace ( microsecond = int ( match . group ( <NUM_LIT:5> ) ) ) <EOL> self . assertGreaterEqual ( job_start , test_start ) <EOL> self . assertLessEqual ( job_start - test_start , <EOL> datetime . timedelta ( seconds = <NUM_LIT:5> ) ) <EOL> def test_owner_and_label_switches ( self ) : <EOL> runner_opts = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> runner = MRTwoStepJob ( runner_opts ) . make_runner ( ) <EOL> match = _JOB_KEY_RE . match ( runner . get_job_key ( ) ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:1> ) , '<STR_LIT>' ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:2> ) , '<STR_LIT>' ) <EOL> def test_owner_and_label_kwargs ( self ) : <EOL> runner = InlineMRJobRunner ( conf_paths = [ ] , <EOL> owner = '<STR_LIT>' , label = '<STR_LIT>' ) <EOL> match = _JOB_KEY_RE . match ( runner . get_job_key ( ) ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:1> ) , '<STR_LIT>' ) <EOL> self . assertEqual ( match . group ( <NUM_LIT:2> ) , '<STR_LIT>' ) <EOL> class CreateMrjobTarGzTestCase ( TestCase ) : <EOL> def test_create_mrjob_tar_gz ( self ) : <EOL> with no_handlers_for_logger ( '<STR_LIT>' ) : <EOL> with InlineMRJobRunner ( conf_paths = [ ] ) as runner : <EOL> mrjob_tar_gz_path = runner . _create_mrjob_tar_gz ( ) <EOL> mrjob_tar_gz = tarfile . open ( mrjob_tar_gz_path ) <EOL> contents = mrjob_tar_gz . getnames ( ) <EOL> for path in contents : <EOL> self . assertEqual ( path [ : <NUM_LIT:6> ] , '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , contents ) <EOL> class TestStreamingOutput ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . make_tmp_dir ( ) <EOL> def tearDown ( self ) : <EOL> self . rm_tmp_dir ( ) <EOL> def make_tmp_dir ( self ) : <EOL> self . tmp_dir = tempfile . mkdtemp ( prefix = '<STR_LIT>' ) <EOL> def rm_tmp_dir ( self ) : <EOL> shutil . rmtree ( self . tmp_dir ) <EOL> def test_stream_output ( self ) : <EOL> a_dir_path = os . path . join ( self . tmp_dir , '<STR_LIT:a>' ) <EOL> b_dir_path = os . path . join ( self . tmp_dir , '<STR_LIT:b>' ) <EOL> l_dir_path = os . path . join ( self . tmp_dir , '<STR_LIT>' ) <EOL> os . mkdir ( a_dir_path ) <EOL> os . mkdir ( b_dir_path ) <EOL> os . mkdir ( l_dir_path ) <EOL> a_file_path = os . path . join ( a_dir_path , '<STR_LIT>' ) <EOL> b_file_path = os . path . join ( b_dir_path , '<STR_LIT>' ) <EOL> c_file_path = os . path . join ( self . tmp_dir , '<STR_LIT>' ) <EOL> x_file_path = os . path . join ( l_dir_path , '<STR_LIT>' ) <EOL> y_file_path = os . path . join ( self . tmp_dir , '<STR_LIT>' ) <EOL> with open ( a_file_path , '<STR_LIT:w>' ) as f : <EOL> f . write ( '<STR_LIT:A>' ) <EOL> with open ( b_file_path , '<STR_LIT:w>' ) as f : <EOL> f . write ( '<STR_LIT:B>' ) <EOL> with open ( c_file_path , '<STR_LIT:w>' ) as f : <EOL> f . write ( '<STR_LIT:C>' ) <EOL> with open ( x_file_path , '<STR_LIT:w>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> with open ( y_file_path , '<STR_LIT:w>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> runner = InlineMRJobRunner ( conf_paths = [ ] , output_dir = self . tmp_dir ) <EOL> self . assertEqual ( sorted ( runner . stream_output ( ) ) , <EOL> [ b'<STR_LIT:A>' , b'<STR_LIT:B>' , b'<STR_LIT:C>' ] ) <EOL> class TestInvokeSort ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . make_tmp_dir_and_set_up_files ( ) <EOL> self . save_environment ( ) <EOL> def tearDown ( self ) : <EOL> self . restore_environment ( ) <EOL> self . rm_tmp_dir ( ) <EOL> def make_tmp_dir_and_set_up_files ( self ) : <EOL> self . tmp_dir = tempfile . mkdtemp ( ) <EOL> self . a = os . path . join ( self . tmp_dir , '<STR_LIT:a>' ) <EOL> with open ( self . a , '<STR_LIT:w>' ) as a : <EOL> a . write ( '<STR_LIT>' ) <EOL> a . write ( '<STR_LIT>' ) <EOL> a . write ( '<STR_LIT>' ) <EOL> self . b = os . path . join ( self . tmp_dir , '<STR_LIT:b>' ) <EOL> with open ( self . b , '<STR_LIT:w>' ) as b : <EOL> b . write ( '<STR_LIT>' ) <EOL> b . write ( '<STR_LIT>' ) <EOL> b . write ( '<STR_LIT>' ) <EOL> self . out = os . path . join ( self . tmp_dir , '<STR_LIT>' ) <EOL> def rm_tmp_dir ( self ) : <EOL> shutil . rmtree ( self . tmp_dir ) <EOL> def save_environment ( self ) : <EOL> self . _old_environ = os . environ . copy ( ) <EOL> def restore_environment ( self ) : <EOL> os . environ . clear ( ) <EOL> os . environ . update ( self . _old_environ ) <EOL> def find_real_sort_bin ( self ) : <EOL> for path in os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) . split ( os . pathsep ) or ( ) : <EOL> for sort_path in [ os . path . join ( path , '<STR_LIT>' ) , <EOL> os . path . join ( path , '<STR_LIT>' ) ] : <EOL> if os . path . exists ( sort_path ) : <EOL> return os . path . abspath ( sort_path ) <EOL> raise Exception ( "<STR_LIT>" ) <EOL> def use_alternate_sort ( self , script_contents ) : <EOL> sort_bin = os . path . join ( self . tmp_dir , '<STR_LIT>' ) <EOL> with open ( sort_bin , '<STR_LIT:w>' ) as f : <EOL> f . write ( '<STR_LIT>' % sys . executable ) <EOL> f . write ( script_contents ) <EOL> os . chmod ( sort_bin , stat . S_IREAD | stat . S_IEXEC ) <EOL> os . environ [ '<STR_LIT>' ] = self . tmp_dir <EOL> def use_simulated_windows_sort ( self ) : <EOL> script_contents = """<STR_LIT>""" % ( self . find_real_sort_bin ( ) ) <EOL> self . use_alternate_sort ( script_contents ) <EOL> def use_bad_sort ( self ) : <EOL> script_contents = """<STR_LIT>""" <EOL> self . use_alternate_sort ( script_contents ) <EOL> def environment_variable_checks ( self , runner , environment_check_list ) : <EOL> environment_vars = { } <EOL> def check_call_se ( * args , ** kwargs ) : <EOL> for key in kwargs [ '<STR_LIT>' ] . keys ( ) : <EOL> environment_vars [ key ] = kwargs [ '<STR_LIT>' ] [ key ] <EOL> with patch ( '<STR_LIT>' , side_effect = check_call_se ) : <EOL> runner . _invoke_sort ( [ self . a ] , self . out ) <EOL> for key in environment_check_list : <EOL> self . assertEqual ( environment_vars . get ( key , None ) , <EOL> runner . _opts [ '<STR_LIT>' ] ) <EOL> def test_no_files ( self ) : <EOL> runner = MRJobRunner ( conf_paths = [ ] ) <EOL> self . assertRaises ( ValueError , <EOL> runner . _invoke_sort , [ ] , self . out ) <EOL> def test_one_file ( self ) : <EOL> runner = MRJobRunner ( conf_paths = [ ] ) <EOL> self . addCleanup ( runner . cleanup ) <EOL> runner . _invoke_sort ( [ self . a ] , self . out ) <EOL> with open ( self . out ) as out_f : <EOL> self . assertEqual ( list ( out_f ) , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> def test_two_files ( self ) : <EOL> runner = MRJobRunner ( conf_paths = [ ] ) <EOL> self . addCleanup ( runner . cleanup ) <EOL> runner . _invoke_sort ( [ self . a , self . b ] , self . out ) <EOL> with open ( self . out ) as out_f : <EOL> self . assertEqual ( list ( out_f ) , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> def test_windows_sort_on_one_file ( self ) : <EOL> self . use_simulated_windows_sort ( ) <EOL> self . test_one_file ( ) <EOL> def test_windows_sort_on_two_files ( self ) : <EOL> self . use_simulated_windows_sort ( ) <EOL> self . test_two_files ( ) <EOL> def test_bad_sort ( self ) : <EOL> self . use_bad_sort ( ) <EOL> runner = MRJobRunner ( conf_paths = [ ] ) <EOL> self . addCleanup ( runner . cleanup ) <EOL> with no_handlers_for_logger ( ) : <EOL> self . assertRaises ( ( CalledProcessError , IOError ) , <EOL> runner . _invoke_sort , [ self . a , self . b ] , self . out ) <EOL> def test_environment_variables_non_windows ( self ) : <EOL> runner = MRJobRunner ( conf_paths = [ ] ) <EOL> self . addCleanup ( runner . cleanup ) <EOL> self . environment_variable_checks ( runner , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_environment_variables_windows ( self ) : <EOL> runner = MRJobRunner ( conf_paths = [ ] ) <EOL> self . addCleanup ( runner . cleanup ) <EOL> runner . _sort_is_windows_sort = True <EOL> self . environment_variable_checks ( runner , [ '<STR_LIT>' ] ) <EOL> class HadoopArgsForStepTestCase ( EmptyMrjobConfTestCase ) : <EOL> def test_empty ( self ) : <EOL> job = MRWordCount ( ) <EOL> with job . make_runner ( ) as runner : <EOL> self . assertEqual ( runner . _hadoop_args_for_step ( <NUM_LIT:0> ) , [ ] ) <EOL> def test_cmdenv ( self ) : <EOL> job = MRWordCount ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> with job . make_runner ( ) as runner : <EOL> self . assertEqual ( runner . _hadoop_args_for_step ( <NUM_LIT:0> ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> ] ) <EOL> def test_hadoop_input_format ( self ) : <EOL> input_format = '<STR_LIT>' <EOL> job1 = MRWordCount ( ) <EOL> job1 . HADOOP_INPUT_FORMAT = input_format <EOL> with job1 . make_runner ( ) as runner1 : <EOL> self . assertEqual ( runner1 . _hadoop_args_for_step ( <NUM_LIT:0> ) , <EOL> [ '<STR_LIT>' , input_format ] ) <EOL> job2 = MRTwoStepJob ( ) <EOL> job2 . HADOOP_INPUT_FORMAT = input_format <EOL> with job2 . make_runner ( ) as runner2 : <EOL> self . assertEqual ( runner2 . _hadoop_args_for_step ( <NUM_LIT:0> ) , <EOL> [ '<STR_LIT>' , input_format ] ) <EOL> self . assertEqual ( runner2 . _hadoop_args_for_step ( <NUM_LIT:1> ) , [ ] ) <EOL> def test_hadoop_output_format ( self ) : <EOL> output_format = '<STR_LIT>' <EOL> job1 = MRWordCount ( ) <EOL> job1 . HADOOP_OUTPUT_FORMAT = output_format <EOL> with job1 . make_runner ( ) as runner1 : <EOL> self . assertEqual ( runner1 . _hadoop_args_for_step ( <NUM_LIT:0> ) , <EOL> [ '<STR_LIT>' , output_format ] ) <EOL> job2 = MRTwoStepJob ( ) <EOL> job2 . HADOOP_OUTPUT_FORMAT = output_format <EOL> with job2 . make_runner ( ) as runner2 : <EOL> self . assertEqual ( runner2 . _hadoop_args_for_step ( <NUM_LIT:0> ) , [ ] ) <EOL> self . assertEqual ( runner2 . _hadoop_args_for_step ( <NUM_LIT:1> ) , <EOL> [ '<STR_LIT>' , output_format ] ) <EOL> def test_jobconf ( self ) : <EOL> jobconf_args = [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> job = MRWordCount ( jobconf_args ) <EOL> with job . make_runner ( ) as runner : <EOL> self . assertEqual ( runner . _hadoop_args_for_step ( <NUM_LIT:0> ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> ] ) <EOL> def test_empty_jobconf_values ( self ) : <EOL> job = MRWordCount ( ) <EOL> job . JOBCONF = { '<STR_LIT:foo>' : '<STR_LIT>' , '<STR_LIT:bar>' : None } <EOL> with job . make_runner ( ) as runner : <EOL> self . assertEqual ( runner . _hadoop_args_for_step ( <NUM_LIT:0> ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_configuration_translation ( self ) : <EOL> job = MRWordCount ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> with job . make_runner ( ) as runner : <EOL> with no_handlers_for_logger ( '<STR_LIT>' ) : <EOL> with patch . object ( runner , <EOL> '<STR_LIT>' , return_value = '<STR_LIT>' ) : <EOL> self . assertEqual ( <EOL> runner . _hadoop_args_for_step ( <NUM_LIT:0> ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' <EOL> ] ) <EOL> def test_jobconf_from_step ( self ) : <EOL> jobconf = { '<STR_LIT>' : '<STR_LIT:bar>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> runner = LocalMRJobRunner ( jobconf = jobconf ) <EOL> runner . _steps = [ { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } } ] <EOL> self . assertEqual ( runner . _hadoop_args_for_step ( <NUM_LIT:0> ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> ] ) <EOL> def test_partitioner ( self ) : <EOL> partitioner = '<STR_LIT>' <EOL> job = MRWordCount ( [ '<STR_LIT>' , partitioner ] ) <EOL> with job . make_runner ( ) as runner : <EOL> self . assertEqual ( runner . _hadoop_args_for_step ( <NUM_LIT:0> ) , <EOL> [ '<STR_LIT>' , partitioner ] ) <EOL> class StrictProtocolsInConfTestCase ( TestCase ) : <EOL> STRICT_MRJOB_CONF = { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : True } } } <EOL> LOOSE_MRJOB_CONF = { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : False } } } <EOL> def test_default ( self ) : <EOL> job = MRJob ( ) <EOL> with job . make_runner ( ) as runner : <EOL> self . assertEqual ( runner . _opts [ '<STR_LIT>' ] , True ) <EOL> def test_strict_mrjob_conf ( self ) : <EOL> job = MRJob ( ) <EOL> with mrjob_conf_patcher ( self . STRICT_MRJOB_CONF ) : <EOL> with job . make_runner ( ) as runner : <EOL> self . assertEqual ( runner . _opts [ '<STR_LIT>' ] , True ) <EOL> def test_loose_mrjob_conf ( self ) : <EOL> job = MRJob ( ) <EOL> with mrjob_conf_patcher ( self . LOOSE_MRJOB_CONF ) : <EOL> with job . make_runner ( ) as runner : <EOL> self . assertEqual ( runner . _opts [ '<STR_LIT>' ] , False ) <EOL> class CheckInputPathsTestCase ( TestCase ) : <EOL> def test_check_input_paths_enabled_by_default ( self ) : <EOL> job = MRWordCount ( ) <EOL> with job . make_runner ( ) as runner : <EOL> self . assertTrue ( runner . _opts [ '<STR_LIT>' ] ) <EOL> def test_check_input_paths_disabled ( self ) : <EOL> job = MRWordCount ( [ '<STR_LIT>' ] ) <EOL> with job . make_runner ( ) as runner : <EOL> self . assertFalse ( runner . _opts [ '<STR_LIT>' ] ) <EOL> def test_can_disable_check_input_paths_in_config ( self ) : <EOL> job = MRWordCount ( ) <EOL> with mrjob_conf_patcher ( <EOL> { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : False } } } ) : <EOL> with job . make_runner ( ) as runner : <EOL> self . assertFalse ( runner . _opts [ '<STR_LIT>' ] ) <EOL> class SetupTestCase ( SandboxedTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( SetupTestCase , self ) . setUp ( ) <EOL> os . mkdir ( os . path . join ( self . tmp_dir , '<STR_LIT:foo>' ) ) <EOL> self . foo_py = os . path . join ( self . tmp_dir , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> with open ( self . foo_py , '<STR_LIT:w>' ) as foo_py : <EOL> foo_py . write ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . foo_sh = os . path . join ( self . tmp_dir , '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> with open ( self . foo_sh , '<STR_LIT:w>' ) as foo_sh : <EOL> foo_sh . write ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> os . chmod ( self . foo_sh , stat . S_IRWXU ) <EOL> self . foo_tar_gz = os . path . join ( self . tmp_dir , '<STR_LIT>' ) <EOL> tar_and_gzip ( os . path . join ( self . tmp_dir , '<STR_LIT:foo>' ) , self . foo_tar_gz ) <EOL> self . foo_py_size = os . path . getsize ( self . foo_py ) <EOL> self . foo_sh_size = os . path . getsize ( self . foo_sh ) <EOL> self . foo_tar_gz_size = os . path . getsize ( self . foo_tar_gz ) <EOL> def test_file_upload ( self ) : <EOL> job = MROSWalkJob ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , self . foo_sh , <EOL> '<STR_LIT>' , self . foo_sh + '<STR_LIT>' , <EOL> ] ) <EOL> job . sandbox ( ) <EOL> with job . make_runner ( ) as r : <EOL> r . run ( ) <EOL> path_to_size = dict ( job . parse_output_line ( line ) <EOL> for line in r . stream_output ( ) ) <EOL> self . assertEqual ( path_to_size . get ( '<STR_LIT>' ) , self . foo_sh_size ) <EOL> self . assertEqual ( path_to_size . get ( '<STR_LIT>' ) , self . foo_sh_size ) <EOL> def test_archive_upload ( self ) : <EOL> job = MROSWalkJob ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , self . foo_tar_gz , <EOL> '<STR_LIT>' , self . foo_tar_gz + '<STR_LIT>' , <EOL> ] ) <EOL> job . sandbox ( ) <EOL> with job . make_runner ( ) as r : <EOL> with no_handlers_for_logger ( '<STR_LIT>' ) : <EOL> r . run ( ) <EOL> path_to_size = dict ( job . parse_output_line ( line ) <EOL> for line in r . stream_output ( ) ) <EOL> self . assertEqual ( path_to_size . get ( '<STR_LIT>' ) , <EOL> self . foo_py_size ) <EOL> self . assertEqual ( path_to_size . get ( '<STR_LIT>' ) , <EOL> self . foo_py_size ) <EOL> def test_deprecated_python_archive_option ( self ) : <EOL> job = MROSWalkJob ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , self . foo_tar_gz ] ) <EOL> job . sandbox ( ) <EOL> with job . make_runner ( ) as r : <EOL> r . run ( ) <EOL> path_to_size = dict ( job . parse_output_line ( line ) <EOL> for line in r . stream_output ( ) ) <EOL> self . assertEqual ( path_to_size . get ( '<STR_LIT>' ) , <EOL> self . foo_py_size * <NUM_LIT:2> ) <EOL> def test_deprecated_setup_cmd_option ( self ) : <EOL> job = MROSWalkJob ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> job . sandbox ( ) <EOL> with job . make_runner ( ) as r : <EOL> r . run ( ) <EOL> path_to_size = dict ( job . parse_output_line ( line ) <EOL> for line in r . stream_output ( ) ) <EOL> self . assertIn ( '<STR_LIT>' , path_to_size ) <EOL> def test_deprecated_setup_script_option ( self ) : <EOL> job = MROSWalkJob ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , self . foo_sh ] ) <EOL> job . sandbox ( ) <EOL> with job . make_runner ( ) as r : <EOL> r . run ( ) <EOL> path_to_size = dict ( job . parse_output_line ( line ) <EOL> for line in r . stream_output ( ) ) <EOL> self . assertEqual ( path_to_size . get ( '<STR_LIT>' ) , self . foo_sh_size ) <EOL> self . assertIn ( '<STR_LIT>' , path_to_size ) <EOL> def test_python_archive ( self ) : <EOL> job = MROSWalkJob ( [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' % self . foo_tar_gz <EOL> ] ) <EOL> job . sandbox ( ) <EOL> with job . make_runner ( ) as r : <EOL> r . run ( ) <EOL> path_to_size = dict ( job . parse_output_line ( line ) <EOL> for line in r . stream_output ( ) ) <EOL> self . assertEqual ( path_to_size . get ( '<STR_LIT>' ) , <EOL> self . foo_py_size * <NUM_LIT:2> ) <EOL> def test_setup_command ( self ) : <EOL> job = MROSWalkJob ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> job . sandbox ( ) <EOL> with job . make_runner ( ) as r : <EOL> r . run ( ) <EOL> path_to_size = dict ( job . parse_output_line ( line ) <EOL> for line in r . stream_output ( ) ) <EOL> self . assertIn ( '<STR_LIT>' , path_to_size ) <EOL> def test_setup_script ( self ) : <EOL> job = MROSWalkJob ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , self . foo_sh + '<STR_LIT:#>' ] ) <EOL> job . sandbox ( ) <EOL> with job . make_runner ( ) as r : <EOL> r . run ( ) <EOL> path_to_size = dict ( job . parse_output_line ( line ) <EOL> for line in r . stream_output ( ) ) <EOL> self . assertEqual ( path_to_size . get ( '<STR_LIT>' ) , self . foo_sh_size ) <EOL> self . assertIn ( '<STR_LIT>' , path_to_size ) <EOL> def test_bad_setup_command ( self ) : <EOL> bar_path = os . path . join ( self . tmp_dir , '<STR_LIT:bar>' ) <EOL> baz_path = os . path . join ( self . tmp_dir , '<STR_LIT>' ) <EOL> job = MROSWalkJob ( [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' % bar_path , <EOL> '<STR_LIT>' , '<STR_LIT:false>' , <EOL> '<STR_LIT>' , '<STR_LIT>' % baz_path , <EOL> '<STR_LIT>' , <EOL> ] ) <EOL> job . sandbox ( ) <EOL> with job . make_runner ( ) as r : <EOL> self . assertRaises ( Exception , r . run ) <EOL> self . assertTrue ( os . path . exists ( bar_path ) ) <EOL> self . assertFalse ( os . path . exists ( baz_path ) ) <EOL> def test_stdin_bypasses_wrapper_script ( self ) : <EOL> job = MROSWalkJob ( [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> ] ) <EOL> job . sandbox ( stdin = BytesIO ( b'<STR_LIT>' ) ) <EOL> def alarm_handler ( * args , ** kwargs ) : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> try : <EOL> self . _old_alarm_handler = signal . signal ( <EOL> signal . SIGALRM , alarm_handler ) <EOL> signal . alarm ( <NUM_LIT:10> ) <EOL> with job . make_runner ( ) as r : <EOL> r . run ( ) <EOL> path_to_size = dict ( job . parse_output_line ( line ) <EOL> for line in r . stream_output ( ) ) <EOL> self . assertEqual ( path_to_size . get ( '<STR_LIT>' ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( path_to_size . get ( None ) , '<STR_LIT>' ) <EOL> finally : <EOL> signal . alarm ( <NUM_LIT:0> ) <EOL> signal . signal ( signal . SIGALRM , self . _old_alarm_handler ) <EOL> def test_wrapper_script_only_writes_to_stderr ( self ) : <EOL> job = MROSWalkJob ( [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> ] ) <EOL> job . sandbox ( ) <EOL> with no_handlers_for_logger ( '<STR_LIT>' ) : <EOL> stderr = StringIO ( ) <EOL> log_to_stream ( '<STR_LIT>' , stderr , debug = True ) <EOL> with job . make_runner ( ) as r : <EOL> r . run ( ) <EOL> output = b'<STR_LIT>' . join ( r . stream_output ( ) ) <EOL> self . assertIn ( '<STR_LIT>' , stderr . getvalue ( ) ) <EOL> self . assertNotIn ( b'<STR_LIT>' , output ) <EOL> class ClosedRunnerTestCase ( EmptyMrjobConfTestCase ) : <EOL> def test_job_closed_on_cleanup ( self ) : <EOL> job = MRWordCount ( ) <EOL> with job . make_runner ( ) as runner : <EOL> self . assertFalse ( runner . _closed ) <EOL> self . assertTrue ( runner . _closed ) <EOL> class InterpreterTestCase ( TestCase ) : <EOL> def default_python_bin ( self ) : <EOL> return [ '<STR_LIT>' ] if PY2 else [ '<STR_LIT>' ] <EOL> def test_default ( self ) : <EOL> runner = MRJobRunner ( ) <EOL> self . assertEqual ( runner . _interpreter ( ) , <EOL> self . default_python_bin ( ) ) <EOL> self . assertEqual ( runner . _interpreter ( steps = True ) , <EOL> [ sys . executable ] ) <EOL> def test_python_bin ( self ) : <EOL> runner = MRJobRunner ( python_bin = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _interpreter ( ) , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _interpreter ( steps = True ) , [ sys . executable ] ) <EOL> def test_steps_python_bin ( self ) : <EOL> runner = MRJobRunner ( steps_python_bin = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _interpreter ( ) , <EOL> self . default_python_bin ( ) ) <EOL> self . assertEqual ( runner . _interpreter ( steps = True ) , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_interpreter ( self ) : <EOL> runner = MRJobRunner ( interpreter = [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _interpreter ( ) , [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _interpreter ( steps = True ) , [ '<STR_LIT>' ] ) <EOL> def test_steps_interpreter ( self ) : <EOL> runner = MRJobRunner ( interpreter = [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> steps_interpreter = [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _interpreter ( ) , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _interpreter ( steps = True ) , [ '<STR_LIT>' ] ) <EOL> def test_interpreter_overrides_python_bin ( self ) : <EOL> runner = MRJobRunner ( interpreter = [ '<STR_LIT>' ] , <EOL> python_bin = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _interpreter ( ) , [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _interpreter ( steps = True ) , [ '<STR_LIT>' ] ) <EOL> def test_interpreter_overrides_steps_python_bin ( self ) : <EOL> runner = MRJobRunner ( interpreter = [ '<STR_LIT>' ] , <EOL> steps_python_bin = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _interpreter ( ) , [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _interpreter ( steps = True ) , [ '<STR_LIT>' ] ) <EOL> class BootstrapMRJobTestCase ( TestCase ) : <EOL> def test_default ( self ) : <EOL> runner = MRJobRunner ( conf_paths = [ ] ) <EOL> self . assertEqual ( runner . _bootstrap_mrjob ( ) , True ) <EOL> def test_no_bootstrap_mrjob ( self ) : <EOL> runner = MRJobRunner ( conf_paths = [ ] , bootstrap_mrjob = False ) <EOL> self . assertEqual ( runner . _bootstrap_mrjob ( ) , False ) <EOL> def test_interpreter ( self ) : <EOL> runner = MRJobRunner ( conf_paths = [ ] , interpreter = [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( runner . _bootstrap_mrjob ( ) , False ) <EOL> def test_bootstrap_mrjob_overrides_interpreter ( self ) : <EOL> runner = MRJobRunner ( <EOL> conf_paths = [ ] , interpreter = [ '<STR_LIT>' ] , bootstrap_mrjob = True ) <EOL> self . assertEqual ( runner . _bootstrap_mrjob ( ) , True ) <EOL> class FSPassthroughTestCase ( TestCase ) : <EOL> def test_passthrough ( self ) : <EOL> runner = InlineMRJobRunner ( ) <EOL> with no_handlers_for_logger ( '<STR_LIT>' ) : <EOL> stderr = StringIO ( ) <EOL> log_to_stream ( '<STR_LIT>' , stderr ) <EOL> self . assertEqual ( runner . ls , runner . fs . ls ) <EOL> self . assertEqual ( runner . _cat_file , runner . fs . _cat_file ) <EOL> self . assertIn ( <EOL> '<STR_LIT>' , <EOL> stderr . getvalue ( ) ) <EOL> self . assertIn ( <EOL> '<STR_LIT>' , <EOL> stderr . getvalue ( ) ) <EOL> def test_prefer_own_methods ( self ) : <EOL> runner = HadoopJobRunner ( <EOL> hadoop_bin = '<STR_LIT>' , <EOL> hadoop_home = '<STR_LIT>' , <EOL> hadoop_streaming_jar = '<STR_LIT>' ) <EOL> with no_handlers_for_logger ( '<STR_LIT>' ) : <EOL> stderr = StringIO ( ) <EOL> log_to_stream ( '<STR_LIT>' , stderr ) <EOL> self . assertEqual ( runner . ls , runner . fs . ls ) <EOL> self . assertNotEqual ( runner . get_hadoop_version , <EOL> runner . fs . get_hadoop_version ) <EOL> self . assertIn ( <EOL> '<STR_LIT>' , <EOL> stderr . getvalue ( ) ) <EOL> self . assertNotIn ( '<STR_LIT>' , stderr . getvalue ( ) ) <EOL> def test_pass_through_fields ( self ) : <EOL> runner = HadoopJobRunner ( <EOL> hadoop_bin = '<STR_LIT>' , <EOL> hadoop_home = '<STR_LIT>' , <EOL> hadoop_streaming_jar = '<STR_LIT>' ) <EOL> with no_handlers_for_logger ( '<STR_LIT>' ) : <EOL> stderr = StringIO ( ) <EOL> log_to_stream ( '<STR_LIT>' , stderr ) <EOL> self . assertEqual ( runner . _hadoop_bin , runner . fs . _hadoop_bin ) <EOL> self . assertIn ( <EOL> '<STR_LIT>' , <EOL> stderr . getvalue ( ) ) </s>
