<s> import py , sys , os , signal , cStringIO , tempfile <EOL> import runner <EOL> class TestRunHelper ( object ) : <EOL> def setup_method ( self , meth ) : <EOL> h , self . fn = tempfile . mkstemp ( ) <EOL> os . close ( h ) <EOL> def teardown_method ( self , meth ) : <EOL> os . unlink ( self . fn ) <EOL> def test_run ( self ) : <EOL> res = runner . run ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , '<STR_LIT:.>' , <EOL> py . path . local ( self . fn ) ) <EOL> assert res == <NUM_LIT:0> <EOL> out = py . path . local ( self . fn ) . read ( '<STR_LIT:r>' ) <EOL> assert out == "<STR_LIT>" <EOL> def test_error ( self ) : <EOL> res = runner . run ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , '<STR_LIT:.>' , py . path . local ( self . fn ) ) <EOL> assert res == <NUM_LIT:3> <EOL> def test_signal ( self ) : <EOL> if sys . platform == '<STR_LIT:win32>' : <EOL> py . test . skip ( "<STR_LIT>" ) <EOL> res = runner . run ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , '<STR_LIT:.>' , py . path . local ( self . fn ) ) <EOL> assert res == - <NUM_LIT:9> <EOL> def test_timeout ( self ) : <EOL> res = runner . run ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , '<STR_LIT:.>' , py . path . local ( self . fn ) , timeout = <NUM_LIT:3> ) <EOL> assert res == - <NUM_LIT> <EOL> def test_timeout_lock ( self ) : <EOL> res = runner . run ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , '<STR_LIT:.>' , py . path . local ( self . fn ) , timeout = <NUM_LIT:3> ) <EOL> assert res == - <NUM_LIT> <EOL> def test_timeout_syscall ( self ) : <EOL> res = runner . run ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , '<STR_LIT:.>' , py . path . local ( self . fn ) , timeout = <NUM_LIT:3> ) <EOL> assert res == - <NUM_LIT> <EOL> def test_timeout_success ( self ) : <EOL> res = runner . run ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , '<STR_LIT:.>' , <EOL> py . path . local ( self . fn ) , timeout = <NUM_LIT:2> ) <EOL> assert res == <NUM_LIT:0> <EOL> out = py . path . local ( self . fn ) . read ( '<STR_LIT:r>' ) <EOL> assert out == "<STR_LIT>" <EOL> class TestExecuteTest ( object ) : <EOL> def setup_class ( cls ) : <EOL> cls . real_run = ( runner . run , ) <EOL> cls . called = [ ] <EOL> cls . exitcode = [ <NUM_LIT:0> ] <EOL> def fake_run ( args , cwd , out , timeout ) : <EOL> cls . called = ( args , cwd , out , timeout ) <EOL> return cls . exitcode [ <NUM_LIT:0> ] <EOL> runner . run = fake_run <EOL> def teardown_class ( cls ) : <EOL> runner . run = cls . real_run [ <NUM_LIT:0> ] <EOL> def test_explicit ( self ) : <EOL> res = runner . execute_test ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> interp = [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> test_driver = [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> timeout = '<STR_LIT>' ) <EOL> expected = [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> assert self . called == ( expected , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert res == <NUM_LIT:0> <EOL> def test_error ( self ) : <EOL> self . exitcode [ : ] = [ <NUM_LIT:1> ] <EOL> res = runner . execute_test ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> interp = [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> test_driver = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> assert res == <NUM_LIT:1> <EOL> self . exitcode [ : ] = [ - signal . SIGSEGV ] <EOL> res = runner . execute_test ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> interp = [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> test_driver = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> assert res == - signal . SIGSEGV <EOL> def test_interpret_exitcode ( self ) : <EOL> failure , extralog = runner . interpret_exitcode ( <NUM_LIT:0> , "<STR_LIT>" ) <EOL> assert not failure <EOL> assert extralog == "<STR_LIT>" <EOL> failure , extralog = runner . interpret_exitcode ( <NUM_LIT:1> , "<STR_LIT>" ) <EOL> assert failure <EOL> assert extralog == "<STR_LIT>" <EOL> failure , extralog = runner . interpret_exitcode ( <NUM_LIT:2> , "<STR_LIT>" ) <EOL> assert failure <EOL> assert extralog == """<STR_LIT>""" <EOL> failure , extralog = runner . interpret_exitcode ( - signal . SIGSEGV , <EOL> "<STR_LIT>" ) <EOL> assert failure <EOL> assert extralog == """<STR_LIT>""" <EOL> class RunnerTests ( object ) : <EOL> with_thread = True <EOL> def setup_class ( cls ) : <EOL> cls . real_invoke_in_thread = ( runner . invoke_in_thread , ) <EOL> if not cls . with_thread : <EOL> runner . invoke_in_thread = lambda func , args : func ( * args ) <EOL> cls . udir = py . path . local . make_numbered_dir ( prefix = '<STR_LIT>' , <EOL> keep = <NUM_LIT:3> ) <EOL> cls . manydir = cls . udir . join ( '<STR_LIT>' ) . ensure ( dir = <NUM_LIT:1> ) <EOL> cls . udir . join ( "<STR_LIT>" ) . write ( "<STR_LIT>" ) <EOL> def fill_test_dir ( test_dir , fromdir = '<STR_LIT>' ) : <EOL> for p in py . path . local ( __file__ ) . dirpath ( <EOL> '<STR_LIT>' , fromdir ) . listdir ( "<STR_LIT>" ) : <EOL> p . copy ( test_dir . join ( '<STR_LIT>' + p . basename ) ) <EOL> test_normal_dir0 = cls . manydir . join ( '<STR_LIT>' , '<STR_LIT>' ) . ensure ( dir = <NUM_LIT:1> ) <EOL> cls . one_test_dir = cls . manydir . join ( '<STR_LIT>' ) <EOL> fill_test_dir ( test_normal_dir0 ) <EOL> test_normal_dir1 = cls . manydir . join ( '<STR_LIT>' , '<STR_LIT>' ) . ensure ( dir = <NUM_LIT:1> ) <EOL> test_normal_dir2 = cls . manydir . join ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) . ensure ( dir = <NUM_LIT:1> ) <EOL> cls . two_test_dir = cls . manydir . join ( '<STR_LIT>' ) <EOL> fill_test_dir ( test_normal_dir1 ) <EOL> fill_test_dir ( test_normal_dir2 ) <EOL> cls . test_stall_dir = cls . udir . join ( '<STR_LIT>' ) . ensure ( dir = <NUM_LIT:1> ) <EOL> test_stall_dir0 = cls . test_stall_dir . join ( '<STR_LIT>' ) . ensure ( dir = <NUM_LIT:1> ) <EOL> fill_test_dir ( test_stall_dir0 , '<STR_LIT>' ) <EOL> def teardown_class ( cls ) : <EOL> runner . invoke_in_thread = cls . real_invoke_in_thread [ <NUM_LIT:0> ] <EOL> def test_one_dir ( self ) : <EOL> test_driver = [ py . path . local ( py . __file__ ) . dirpath ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> log = cStringIO . StringIO ( ) <EOL> out = cStringIO . StringIO ( ) <EOL> run_param = runner . RunParam ( self . one_test_dir ) <EOL> run_param . test_driver = test_driver <EOL> run_param . parallel_runs = <NUM_LIT:3> <EOL> res = runner . execute_tests ( run_param , [ '<STR_LIT>' ] , log , out ) <EOL> assert res <EOL> out = out . getvalue ( ) <EOL> assert out <EOL> assert '<STR_LIT:\r\n>' not in out <EOL> assert '<STR_LIT:\n>' in out <EOL> log = log . getvalue ( ) <EOL> assert '<STR_LIT:\r\n>' not in log <EOL> assert '<STR_LIT:\n>' in log <EOL> log_lines = log . splitlines ( ) <EOL> assert log_lines [ <NUM_LIT:0> ] == "<STR_LIT>" <EOL> nfailures = <NUM_LIT:0> <EOL> noutcomes = <NUM_LIT:0> <EOL> for line in log_lines : <EOL> if line [ <NUM_LIT:0> ] != '<STR_LIT:U+0020>' : <EOL> noutcomes += <NUM_LIT:1> <EOL> if line [ <NUM_LIT:0> ] != '<STR_LIT:.>' : <EOL> nfailures += <NUM_LIT:1> <EOL> assert noutcomes == <NUM_LIT> <EOL> assert nfailures == <NUM_LIT:6> <EOL> def test_one_dir_dry_run ( self ) : <EOL> test_driver = [ py . path . local ( py . __file__ ) . dirpath ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> log = cStringIO . StringIO ( ) <EOL> out = cStringIO . StringIO ( ) <EOL> run_param = runner . RunParam ( self . one_test_dir ) <EOL> run_param . test_driver = test_driver <EOL> run_param . parallel_runs = <NUM_LIT:3> <EOL> run_param . dry_run = True <EOL> res = runner . execute_tests ( run_param , [ '<STR_LIT>' ] , log , out ) <EOL> assert not res <EOL> assert log . getvalue ( ) == "<STR_LIT>" <EOL> out_lines = out . getvalue ( ) . splitlines ( ) <EOL> assert len ( out_lines ) == <NUM_LIT:5> <EOL> assert out_lines [ <NUM_LIT:2> ] . startswith ( "<STR_LIT>" ) <EOL> assert out_lines [ <NUM_LIT:4> ] . startswith ( "<STR_LIT>" ) <EOL> for line in out_lines [ <NUM_LIT:2> : ] : <EOL> assert "<STR_LIT>" in line <EOL> def test_many_dirs ( self ) : <EOL> test_driver = [ py . path . local ( py . __file__ ) . dirpath ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> log = cStringIO . StringIO ( ) <EOL> out = cStringIO . StringIO ( ) <EOL> cleanedup = [ ] <EOL> def cleanup ( testdir ) : <EOL> cleanedup . append ( testdir ) <EOL> run_param = runner . RunParam ( self . manydir ) <EOL> run_param . test_driver = test_driver <EOL> run_param . parallel_runs = <NUM_LIT:3> <EOL> run_param . cleanup = cleanup <EOL> testdirs = [ ] <EOL> run_param . collect_testdirs ( testdirs ) <EOL> alltestdirs = testdirs [ : ] <EOL> res = runner . execute_tests ( run_param , testdirs , log , out ) <EOL> assert res <EOL> assert out . getvalue ( ) <EOL> log_lines = log . getvalue ( ) . splitlines ( ) <EOL> nfailures = <NUM_LIT:0> <EOL> noutcomes = <NUM_LIT:0> <EOL> for line in log_lines : <EOL> if line [ <NUM_LIT:0> ] != '<STR_LIT:U+0020>' : <EOL> noutcomes += <NUM_LIT:1> <EOL> if line [ <NUM_LIT:0> ] != '<STR_LIT:.>' : <EOL> nfailures += <NUM_LIT:1> <EOL> assert noutcomes == <NUM_LIT:3> * <NUM_LIT> <EOL> assert nfailures == <NUM_LIT:3> * <NUM_LIT:6> <EOL> assert set ( cleanedup ) == set ( alltestdirs ) <EOL> def test_timeout ( self ) : <EOL> test_driver = [ py . path . local ( py . __file__ ) . dirpath ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> log = cStringIO . StringIO ( ) <EOL> out = cStringIO . StringIO ( ) <EOL> run_param = runner . RunParam ( self . test_stall_dir ) <EOL> run_param . test_driver = test_driver <EOL> run_param . parallel_runs = <NUM_LIT:3> <EOL> run_param . timeout = <NUM_LIT:3> <EOL> testdirs = [ ] <EOL> run_param . collect_testdirs ( testdirs ) <EOL> res = runner . execute_tests ( run_param , testdirs , log , out ) <EOL> assert res <EOL> log_lines = log . getvalue ( ) . splitlines ( ) <EOL> assert log_lines [ <NUM_LIT:1> ] == '<STR_LIT>' <EOL> def test_run_wrong_interp ( self ) : <EOL> log = cStringIO . StringIO ( ) <EOL> out = cStringIO . StringIO ( ) <EOL> run_param = runner . RunParam ( self . one_test_dir ) <EOL> run_param . interp = [ '<STR_LIT>' ] <EOL> run_param . parallel_runs = <NUM_LIT:3> <EOL> testdirs = [ ] <EOL> run_param . collect_testdirs ( testdirs ) <EOL> res = runner . execute_tests ( run_param , testdirs , log , out ) <EOL> assert res <EOL> log_lines = log . getvalue ( ) . splitlines ( ) <EOL> assert log_lines [ <NUM_LIT:1> ] == '<STR_LIT>' <EOL> def test_run_bad_get_test_driver ( self ) : <EOL> test_driver = [ py . path . local ( py . __file__ ) . dirpath ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> log = cStringIO . StringIO ( ) <EOL> out = cStringIO . StringIO ( ) <EOL> run_param = runner . RunParam ( self . one_test_dir ) <EOL> run_param . parallel_runs = <NUM_LIT:3> <EOL> def boom ( testdir ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> run_param . get_test_driver = boom <EOL> testdirs = [ ] <EOL> run_param . collect_testdirs ( testdirs ) <EOL> res = runner . execute_tests ( run_param , testdirs , log , out ) <EOL> assert res <EOL> log_lines = log . getvalue ( ) . splitlines ( ) <EOL> assert log_lines [ <NUM_LIT:1> ] == '<STR_LIT>' <EOL> class TestRunnerNoThreads ( RunnerTests ) : <EOL> with_thread = False <EOL> def test_collect_testdirs ( self ) : <EOL> res = [ ] <EOL> seen = [ ] <EOL> run_param = runner . RunParam ( self . one_test_dir ) <EOL> real_collect_one_testdir = run_param . collect_one_testdir <EOL> def witness_collect_one_testdir ( testdirs , reldir , tests ) : <EOL> seen . append ( ( reldir , sorted ( map ( str , tests ) ) ) ) <EOL> real_collect_one_testdir ( testdirs , reldir , tests ) <EOL> run_param . collect_one_testdir = witness_collect_one_testdir <EOL> run_param . collect_testdirs ( res ) <EOL> assert res == [ '<STR_LIT>' ] <EOL> assert len ( seen ) == <NUM_LIT:1> <EOL> reldir , tests = seen [ <NUM_LIT:0> ] <EOL> assert reldir == '<STR_LIT>' <EOL> for test in tests : <EOL> assert test . startswith ( '<STR_LIT>' ) <EOL> run_param . collect_one_testdir = real_collect_one_testdir <EOL> res = [ ] <EOL> run_param = runner . RunParam ( self . two_test_dir ) <EOL> run_param . collect_testdirs ( res ) <EOL> assert sorted ( res ) == [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> class TestRunner ( RunnerTests ) : <EOL> pass </s>
