<s> import os <EOL> import sys <EOL> import time <EOL> from circus . process import Process <EOL> from circus . tests . support import ( TestCircus , skipIf , EasyTestSuite , DEBUG , <EOL> poll_for , IS_WINDOWS , PYTHON , SLEEP ) <EOL> import circus . py3compat <EOL> from circus . py3compat import StringIO , PY2 <EOL> RLIMIT = """<STR_LIT>""" <EOL> VERBOSE = """<STR_LIT>""" <EOL> USE_FDS = IS_WINDOWS <EOL> def _nose_no_s ( ) : <EOL> if PY2 : <EOL> return not hasattr ( sys . stdout , '<STR_LIT>' ) <EOL> else : <EOL> return isinstance ( sys . stdout , StringIO ) <EOL> class TestProcess ( TestCircus ) : <EOL> def test_base ( self ) : <EOL> cmd = PYTHON <EOL> args = "<STR_LIT>" <EOL> process = Process ( '<STR_LIT:test>' , <NUM_LIT:1> , cmd , args = args , shell = False , <EOL> use_fds = USE_FDS ) <EOL> try : <EOL> info = process . info ( ) <EOL> self . assertEqual ( process . pid , info [ '<STR_LIT>' ] ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> age = process . age ( ) <EOL> self . assertTrue ( age > <NUM_LIT:0.> ) <EOL> self . assertFalse ( process . is_child ( <NUM_LIT:0> ) ) <EOL> finally : <EOL> process . stop ( ) <EOL> @ skipIf ( DEBUG , '<STR_LIT>' ) <EOL> @ skipIf ( IS_WINDOWS , "<STR_LIT>" ) <EOL> def test_rlimits ( self ) : <EOL> script_file = self . get_tmpfile ( RLIMIT ) <EOL> output_file = self . get_tmpfile ( ) <EOL> cmd = PYTHON <EOL> args = [ script_file , output_file ] <EOL> rlimits = { '<STR_LIT>' : <NUM_LIT:20> , <EOL> '<STR_LIT>' : <NUM_LIT:20> } <EOL> process = Process ( '<STR_LIT:test>' , <NUM_LIT:1> , cmd , args = args , rlimits = rlimits ) <EOL> poll_for ( output_file , '<STR_LIT>' ) <EOL> process . stop ( ) <EOL> with open ( output_file , '<STR_LIT:r>' ) as f : <EOL> output = { } <EOL> for line in f . readlines ( ) : <EOL> line = line . rstrip ( ) <EOL> line = line . split ( '<STR_LIT:=>' , <NUM_LIT:1> ) <EOL> if len ( line ) != <NUM_LIT:2> : <EOL> continue <EOL> limit , value = line <EOL> output [ limit ] = value <EOL> def srt2ints ( val ) : <EOL> return [ circus . py3compat . long ( key ) for key in val [ <NUM_LIT:1> : - <NUM_LIT:1> ] . split ( '<STR_LIT:U+002C>' ) ] <EOL> wanted = [ circus . py3compat . long ( <NUM_LIT:20> ) , circus . py3compat . long ( <NUM_LIT:20> ) ] <EOL> self . assertEqual ( srt2ints ( output [ '<STR_LIT>' ] ) , wanted ) <EOL> self . assertEqual ( srt2ints ( output [ '<STR_LIT>' ] ) , wanted ) <EOL> def test_comparison ( self ) : <EOL> cmd = PYTHON <EOL> args = [ '<STR_LIT>' , ] <EOL> p1 = Process ( '<STR_LIT:test>' , <NUM_LIT:1> , cmd , args = args , use_fds = USE_FDS ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> p2 = Process ( '<STR_LIT:test>' , <NUM_LIT:2> , cmd , args = args , use_fds = USE_FDS ) <EOL> self . assertTrue ( p1 < p2 ) <EOL> self . assertFalse ( p1 == p2 ) <EOL> self . assertTrue ( p1 == p1 ) <EOL> p1 . stop ( ) <EOL> p2 . stop ( ) <EOL> def test_process_parameters ( self ) : <EOL> p1 = Process ( '<STR_LIT:test>' , <NUM_LIT:1> , '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> shell = False , spawn = False , env = { '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> use_fds = USE_FDS ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> p1 . format_args ( ) ) <EOL> p2 = Process ( '<STR_LIT:test>' , <NUM_LIT:1> , '<STR_LIT>' , spawn = False , <EOL> use_fds = USE_FDS ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:1>' ] , p2 . format_args ( ) ) <EOL> os . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> p3 = Process ( '<STR_LIT:test>' , <NUM_LIT:1> , '<STR_LIT>' , shell = False , <EOL> spawn = False , env = { '<STR_LIT:type>' : '<STR_LIT>' } , use_fds = USE_FDS ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT>' ] , p3 . format_args ( ) ) <EOL> os . environ . pop ( '<STR_LIT>' ) <EOL> @ skipIf ( DEBUG , '<STR_LIT>' ) <EOL> @ skipIf ( _nose_no_s ( ) , '<STR_LIT>' ) <EOL> @ skipIf ( IS_WINDOWS , "<STR_LIT>" ) <EOL> def test_streams ( self ) : <EOL> script_file = self . get_tmpfile ( VERBOSE ) <EOL> output_file = self . get_tmpfile ( ) <EOL> cmd = PYTHON <EOL> args = [ script_file , output_file ] <EOL> process = Process ( '<STR_LIT:test>' , <NUM_LIT:1> , cmd , args = args , <EOL> close_child_stdout = True , close_child_stderr = True ) <EOL> try : <EOL> poll_for ( output_file , '<STR_LIT>' ) <EOL> self . assertEqual ( process . stdout . read ( ) , b'<STR_LIT>' ) <EOL> self . assertEqual ( process . stderr . read ( ) , b'<STR_LIT>' ) <EOL> finally : <EOL> process . stop ( ) <EOL> output_file = self . get_tmpfile ( ) <EOL> args [ <NUM_LIT:1> ] = output_file <EOL> process = Process ( '<STR_LIT:test>' , <NUM_LIT:1> , cmd , args = args , <EOL> close_child_stdout = True , close_child_stderr = True , <EOL> pipe_stdout = False , pipe_stderr = False ) <EOL> try : <EOL> poll_for ( output_file , '<STR_LIT>' ) <EOL> self . assertTrue ( process . stdout is None ) <EOL> self . assertTrue ( process . stderr is None ) <EOL> finally : <EOL> process . stop ( ) <EOL> output_file = self . get_tmpfile ( ) <EOL> args [ <NUM_LIT:1> ] = output_file <EOL> process = Process ( '<STR_LIT:test>' , '<STR_LIT:1>' , cmd , args = args ) <EOL> try : <EOL> poll_for ( output_file , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( process . stdout . read ( ) ) , <NUM_LIT> ) <EOL> self . assertEqual ( len ( process . stderr . read ( ) ) , <NUM_LIT> ) <EOL> finally : <EOL> process . stop ( ) <EOL> @ skipIf ( IS_WINDOWS , "<STR_LIT>" ) <EOL> def test_initgroups ( self ) : <EOL> cmd = sys . executable <EOL> args = [ SLEEP % <NUM_LIT:2> ] <EOL> gid = os . getgid ( ) <EOL> uid = os . getuid ( ) <EOL> p1 = Process ( '<STR_LIT:test>' , '<STR_LIT:1>' , cmd , args = args , gid = gid , uid = uid ) <EOL> p1 . stop ( ) <EOL> test_suite = EasyTestSuite ( __name__ ) </s>
