<s> import sys <EOL> import os <EOL> import errno <EOL> import greentest <EOL> import gevent <EOL> from gevent import subprocess <EOL> import time <EOL> import gc <EOL> PYPY = hasattr ( sys , '<STR_LIT>' ) <EOL> PY3 = sys . version_info [ <NUM_LIT:0> ] >= <NUM_LIT:3> <EOL> if subprocess . mswindows : <EOL> SETBINARY = '<STR_LIT>' <EOL> else : <EOL> SETBINARY = '<STR_LIT>' <EOL> python_universal_newlines = hasattr ( sys . stdout , '<STR_LIT>' ) <EOL> if PY3 and subprocess . mswindows : <EOL> python_universal_newlines_broken = True <EOL> else : <EOL> python_universal_newlines_broken = False <EOL> class Test ( greentest . TestCase ) : <EOL> def setUp ( self ) : <EOL> gc . collect ( ) <EOL> gc . collect ( ) <EOL> def test_exit ( self ) : <EOL> popen = subprocess . Popen ( [ sys . executable , '<STR_LIT:-c>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( popen . wait ( ) , <NUM_LIT:10> ) <EOL> def test_wait ( self ) : <EOL> popen = subprocess . Popen ( [ sys . executable , '<STR_LIT:-c>' , '<STR_LIT>' ] ) <EOL> gevent . wait ( [ popen ] ) <EOL> self . assertEqual ( popen . poll ( ) , <NUM_LIT:11> ) <EOL> def test_child_exception ( self ) : <EOL> try : <EOL> subprocess . Popen ( [ '<STR_LIT:*>' ] ) . wait ( ) <EOL> except OSError as ex : <EOL> assert ex . errno == <NUM_LIT:2> , ex <EOL> else : <EOL> raise AssertionError ( '<STR_LIT>' ) <EOL> def test_leak ( self ) : <EOL> num_before = greentest . get_number_open_files ( ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> stdout = subprocess . PIPE ) <EOL> p . wait ( ) <EOL> del p <EOL> if PYPY : <EOL> gc . collect ( ) <EOL> gc . collect ( ) <EOL> num_after = greentest . get_number_open_files ( ) <EOL> self . assertEqual ( num_before , num_after ) <EOL> def test_communicate ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> ( stdout , stderr ) = p . communicate ( b"<STR_LIT>" ) <EOL> self . assertEqual ( stdout , b"<STR_LIT>" ) <EOL> if sys . executable . endswith ( '<STR_LIT>' ) : <EOL> assert stderr . startswith ( b'<STR_LIT>' ) <EOL> else : <EOL> self . assertEqual ( stderr , b"<STR_LIT>" ) <EOL> def test_universal1 ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' + SETBINARY + <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> universal_newlines = <NUM_LIT:1> , <EOL> bufsize = <NUM_LIT:1> ) <EOL> try : <EOL> stdout = p . stdout . read ( ) <EOL> if python_universal_newlines : <EOL> if not python_universal_newlines_broken : <EOL> self . assertEqual ( stdout , <EOL> "<STR_LIT>" ) <EOL> else : <EOL> self . assertEqual ( stdout , <EOL> '<STR_LIT>' ) <EOL> else : <EOL> self . assertEqual ( stdout , <EOL> "<STR_LIT>" ) <EOL> finally : <EOL> p . stdout . close ( ) <EOL> def test_universal2 ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' + SETBINARY + <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> universal_newlines = <NUM_LIT:1> , <EOL> bufsize = <NUM_LIT:1> ) <EOL> try : <EOL> stdout = p . stdout . read ( ) <EOL> if python_universal_newlines : <EOL> if not python_universal_newlines_broken : <EOL> self . assertEqual ( stdout , <EOL> "<STR_LIT>" ) <EOL> else : <EOL> self . assertEqual ( stdout , <EOL> '<STR_LIT>' ) <EOL> else : <EOL> self . assertEqual ( stdout , <EOL> "<STR_LIT>" ) <EOL> finally : <EOL> p . stdout . close ( ) <EOL> if sys . platform != '<STR_LIT:win32>' : <EOL> def test_nonblock_removed ( self ) : <EOL> r , w = os . pipe ( ) <EOL> stdin = subprocess . FileObject ( r ) <EOL> p = subprocess . Popen ( [ '<STR_LIT>' , '<STR_LIT:text>' ] , stdin = stdin ) <EOL> try : <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> self . assertEqual ( p . poll ( ) , None ) <EOL> finally : <EOL> if p . poll ( ) is None : <EOL> p . kill ( ) <EOL> stdin . close ( ) <EOL> os . close ( w ) <EOL> def test_issue148 ( self ) : <EOL> for i in range ( <NUM_LIT:7> ) : <EOL> try : <EOL> subprocess . Popen ( '<STR_LIT>' ) <EOL> except OSError as ex : <EOL> if ex . errno != errno . ENOENT : <EOL> raise <EOL> else : <EOL> raise AssertionError ( '<STR_LIT>' ) <EOL> def test_check_output_keyword_error ( self ) : <EOL> try : <EOL> subprocess . check_output ( [ sys . executable , '<STR_LIT:-c>' , '<STR_LIT>' ] ) <EOL> except subprocess . CalledProcessError as e : <EOL> self . assertEqual ( e . returncode , <NUM_LIT> ) <EOL> else : <EOL> raise AssertionError ( '<STR_LIT>' ) <EOL> def test_popen_bufsize ( self ) : <EOL> if PY3 : <EOL> return <EOL> p = subprocess . Popen ( [ sys . executable , '<STR_LIT>' , '<STR_LIT:-c>' , <EOL> '<STR_LIT>' ] , <EOL> stdin = subprocess . PIPE , stdout = subprocess . PIPE ) <EOL> p . stdin . write ( b'<STR_LIT>' ) <EOL> r = p . stdout . readline ( ) <EOL> self . assertEqual ( r , b'<STR_LIT>' ) <EOL> if sys . platform != '<STR_LIT:win32>' : <EOL> def test_subprocess_in_native_thread ( self ) : <EOL> from gevent import monkey <EOL> ex = [ ] <EOL> Thread = monkey . get_original ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def fn ( ) : <EOL> try : <EOL> gevent . subprocess . Popen ( '<STR_LIT>' , shell = True ) <EOL> raise AssertionError ( "<STR_LIT>" ) <EOL> except Exception as e : <EOL> ex . append ( e ) <EOL> thread = Thread ( target = fn ) <EOL> thread . start ( ) <EOL> thread . join ( ) <EOL> self . assertEqual ( len ( ex ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( isinstance ( ex [ <NUM_LIT:0> ] , TypeError ) , ex ) <EOL> self . assertEqual ( ex [ <NUM_LIT:0> ] . args [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> test_subprocess_in_native_thread . ignore_leakcheck = True <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> greentest . main ( ) </s>
