<s> from __future__ import absolute_import , division , print_function , with_statement <EOL> import logging <EOL> import os <EOL> import signal <EOL> import subprocess <EOL> import sys <EOL> from tornado . httpclient import HTTPClient , HTTPError <EOL> from tornado . httpserver import HTTPServer <EOL> from tornado . ioloop import IOLoop <EOL> from tornado . log import gen_log <EOL> from tornado . process import fork_processes , task_id , Subprocess <EOL> from tornado . simple_httpclient import SimpleAsyncHTTPClient <EOL> from tornado . testing import bind_unused_port , ExpectLog , AsyncTestCase <EOL> from tornado . test . util import unittest , skipIfNonUnix <EOL> from tornado . web import RequestHandler , Application <EOL> def skip_if_twisted ( ) : <EOL> if IOLoop . configured_class ( ) . __name__ . endswith ( '<STR_LIT>' ) : <EOL> raise unittest . SkipTest ( "<STR_LIT>" ) <EOL> @ skipIfNonUnix <EOL> class ProcessTest ( unittest . TestCase ) : <EOL> def get_app ( self ) : <EOL> class ProcessHandler ( RequestHandler ) : <EOL> def get ( self ) : <EOL> if self . get_argument ( "<STR_LIT>" , None ) : <EOL> os . _exit ( int ( self . get_argument ( "<STR_LIT>" ) ) ) <EOL> if self . get_argument ( "<STR_LIT>" , None ) : <EOL> os . kill ( os . getpid ( ) , <EOL> int ( self . get_argument ( "<STR_LIT>" ) ) ) <EOL> self . write ( str ( os . getpid ( ) ) ) <EOL> return Application ( [ ( "<STR_LIT:/>" , ProcessHandler ) ] ) <EOL> def tearDown ( self ) : <EOL> if task_id ( ) is not None : <EOL> logging . error ( "<STR_LIT>" ) <EOL> logging . shutdown ( ) <EOL> os . _exit ( <NUM_LIT:1> ) <EOL> signal . alarm ( <NUM_LIT:0> ) <EOL> super ( ProcessTest , self ) . tearDown ( ) <EOL> def test_multi_process ( self ) : <EOL> skip_if_twisted ( ) <EOL> with ExpectLog ( gen_log , "<STR_LIT>" ) : <EOL> self . assertFalse ( IOLoop . initialized ( ) ) <EOL> sock , port = bind_unused_port ( ) <EOL> def get_url ( path ) : <EOL> return "<STR_LIT>" % ( port , path ) <EOL> signal . alarm ( <NUM_LIT:5> ) <EOL> try : <EOL> id = fork_processes ( <NUM_LIT:3> , max_restarts = <NUM_LIT:3> ) <EOL> self . assertTrue ( id is not None ) <EOL> signal . alarm ( <NUM_LIT:5> ) <EOL> except SystemExit as e : <EOL> self . assertEqual ( e . code , <NUM_LIT:0> ) <EOL> self . assertTrue ( task_id ( ) is None ) <EOL> sock . close ( ) <EOL> return <EOL> try : <EOL> if id in ( <NUM_LIT:0> , <NUM_LIT:1> ) : <EOL> self . assertEqual ( id , task_id ( ) ) <EOL> server = HTTPServer ( self . get_app ( ) ) <EOL> server . add_sockets ( [ sock ] ) <EOL> IOLoop . instance ( ) . start ( ) <EOL> elif id == <NUM_LIT:2> : <EOL> self . assertEqual ( id , task_id ( ) ) <EOL> sock . close ( ) <EOL> client = HTTPClient ( SimpleAsyncHTTPClient ) <EOL> def fetch ( url , fail_ok = False ) : <EOL> try : <EOL> return client . fetch ( get_url ( url ) ) <EOL> except HTTPError as e : <EOL> if not ( fail_ok and e . code == <NUM_LIT> ) : <EOL> raise <EOL> fetch ( "<STR_LIT>" , fail_ok = True ) <EOL> fetch ( "<STR_LIT>" , fail_ok = True ) <EOL> int ( fetch ( "<STR_LIT:/>" ) . body ) <EOL> fetch ( "<STR_LIT>" , fail_ok = True ) <EOL> pid = int ( fetch ( "<STR_LIT:/>" ) . body ) <EOL> fetch ( "<STR_LIT>" , fail_ok = True ) <EOL> pid2 = int ( fetch ( "<STR_LIT:/>" ) . body ) <EOL> self . assertNotEqual ( pid , pid2 ) <EOL> fetch ( "<STR_LIT>" , fail_ok = True ) <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> except Exception : <EOL> logging . error ( "<STR_LIT>" , id , exc_info = True ) <EOL> raise <EOL> @ skipIfNonUnix <EOL> class SubprocessTest ( AsyncTestCase ) : <EOL> def test_subprocess ( self ) : <EOL> subproc = Subprocess ( [ sys . executable , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> stdin = Subprocess . STREAM , <EOL> stdout = Subprocess . STREAM , stderr = subprocess . STDOUT , <EOL> io_loop = self . io_loop ) <EOL> self . addCleanup ( lambda : os . kill ( subproc . pid , signal . SIGTERM ) ) <EOL> subproc . stdout . read_until ( b'<STR_LIT>' , self . stop ) <EOL> self . wait ( ) <EOL> subproc . stdin . write ( b"<STR_LIT>" ) <EOL> subproc . stdout . read_until ( b'<STR_LIT:\n>' , self . stop ) <EOL> data = self . wait ( ) <EOL> self . assertEqual ( data , b"<STR_LIT>" ) <EOL> subproc . stdout . read_until ( b"<STR_LIT>" , self . stop ) <EOL> self . wait ( ) <EOL> subproc . stdin . write ( b"<STR_LIT>" ) <EOL> subproc . stdout . read_until_close ( self . stop ) <EOL> data = self . wait ( ) <EOL> self . assertEqual ( data , b"<STR_LIT>" ) <EOL> def test_close_stdin ( self ) : <EOL> subproc = Subprocess ( [ sys . executable , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> stdin = Subprocess . STREAM , <EOL> stdout = Subprocess . STREAM , stderr = subprocess . STDOUT , <EOL> io_loop = self . io_loop ) <EOL> self . addCleanup ( lambda : os . kill ( subproc . pid , signal . SIGTERM ) ) <EOL> subproc . stdout . read_until ( b'<STR_LIT>' , self . stop ) <EOL> self . wait ( ) <EOL> subproc . stdin . close ( ) <EOL> subproc . stdout . read_until_close ( self . stop ) <EOL> data = self . wait ( ) <EOL> self . assertEqual ( data , b"<STR_LIT:\n>" ) <EOL> def test_stderr ( self ) : <EOL> subproc = Subprocess ( [ sys . executable , '<STR_LIT>' , '<STR_LIT:-c>' , <EOL> r"<STR_LIT>" ] , <EOL> stderr = Subprocess . STREAM , <EOL> io_loop = self . io_loop ) <EOL> self . addCleanup ( lambda : os . kill ( subproc . pid , signal . SIGTERM ) ) <EOL> subproc . stderr . read_until ( b'<STR_LIT:\n>' , self . stop ) <EOL> data = self . wait ( ) <EOL> self . assertEqual ( data , b'<STR_LIT>' ) <EOL> def test_sigchild ( self ) : <EOL> skip_if_twisted ( ) <EOL> Subprocess . initialize ( io_loop = self . io_loop ) <EOL> self . addCleanup ( Subprocess . uninitialize ) <EOL> subproc = Subprocess ( [ sys . executable , '<STR_LIT:-c>' , '<STR_LIT>' ] , <EOL> io_loop = self . io_loop ) <EOL> subproc . set_exit_callback ( self . stop ) <EOL> ret = self . wait ( ) <EOL> self . assertEqual ( ret , <NUM_LIT:0> ) <EOL> self . assertEqual ( subproc . returncode , ret ) <EOL> def test_sigchild_signal ( self ) : <EOL> skip_if_twisted ( ) <EOL> Subprocess . initialize ( io_loop = self . io_loop ) <EOL> self . addCleanup ( Subprocess . uninitialize ) <EOL> subproc = Subprocess ( [ sys . executable , '<STR_LIT:-c>' , <EOL> '<STR_LIT>' ] , <EOL> io_loop = self . io_loop ) <EOL> subproc . set_exit_callback ( self . stop ) <EOL> os . kill ( subproc . pid , signal . SIGTERM ) <EOL> ret = self . wait ( ) <EOL> self . assertEqual ( subproc . returncode , ret ) <EOL> self . assertEqual ( ret , - signal . SIGTERM ) </s>
