<s> import os <EOL> import re <EOL> import signal <EOL> import sys <EOL> import time <EOL> from subprocess import PIPE <EOL> from subprocess import Popen <EOL> from tests . lib . testing import is_alive <EOL> from tests . lib . testing import pid_tree <EOL> def spawn_and_kill_pipeline ( ) : <EOL> proc = Popen ( ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT:-c>' , <EOL> "<STR_LIT>" <EOL> ) ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> pids = pid_tree ( os . getpid ( ) ) <EOL> assert len ( living_pids ( pids ) ) == <NUM_LIT:6> <EOL> proc . send_signal ( signal . SIGTERM ) <EOL> proc . wait ( ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> return pids <EOL> def living_pids ( pids ) : <EOL> return set ( pid for pid in pids if is_alive ( pid ) ) <EOL> def test_setsid_signals_entire_group ( both_debug_modes , setsid_enabled ) : <EOL> """<STR_LIT>""" <EOL> pids = spawn_and_kill_pipeline ( ) <EOL> assert len ( living_pids ( pids ) ) == <NUM_LIT:0> <EOL> def test_no_setsid_doesnt_signal_entire_group ( <EOL> both_debug_modes , <EOL> setsid_disabled , <EOL> ) : <EOL> """<STR_LIT>""" <EOL> pids = spawn_and_kill_pipeline ( ) <EOL> living = living_pids ( pids ) <EOL> assert len ( living ) == <NUM_LIT:4> <EOL> for pid in living : <EOL> os . kill ( pid , signal . SIGKILL ) <EOL> def spawn_process_which_dies_with_children ( ) : <EOL> """<STR_LIT>""" <EOL> proc = Popen ( <EOL> ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT:-c>' , <EOL> '<STR_LIT>' . format ( <EOL> python = sys . executable , <EOL> ) , <EOL> ) , <EOL> stdout = PIPE , <EOL> ) <EOL> proc . wait ( ) <EOL> assert proc . returncode == <NUM_LIT:0> <EOL> line = proc . stdout . readline ( ) <EOL> match = re . match ( b'<STR_LIT>' , line ) <EOL> assert match , '<STR_LIT>' + str ( line ) <EOL> child_pid = int ( match . group ( <NUM_LIT:1> ) ) <EOL> return child_pid , proc . stdout <EOL> def test_all_processes_receive_term_on_exit_if_setsid ( <EOL> both_debug_modes , <EOL> setsid_enabled , <EOL> ) : <EOL> """<STR_LIT>""" <EOL> child_pid , child_stdout = spawn_process_which_dies_with_children ( ) <EOL> assert child_stdout . readline ( ) == b'<STR_LIT>' <EOL> os . kill ( child_pid , signal . SIGKILL ) <EOL> def test_processes_dont_receive_term_on_exit_if_no_setsid ( <EOL> both_debug_modes , <EOL> setsid_disabled , <EOL> ) : <EOL> """<STR_LIT>""" <EOL> child_pid , child_stdout = spawn_process_which_dies_with_children ( ) <EOL> for signum in [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] : <EOL> os . kill ( child_pid , signum ) <EOL> assert child_stdout . readline ( ) == str ( signum ) . encode ( '<STR_LIT:ascii>' ) + b'<STR_LIT:\n>' <EOL> os . kill ( child_pid , signal . SIGKILL ) <EOL> def test_fails_nonzero_with_bad_exec ( both_debug_modes , both_setsid_modes ) : <EOL> """<STR_LIT>""" <EOL> proc = Popen ( ( '<STR_LIT>' , '<STR_LIT>' ) , stderr = PIPE ) <EOL> proc . wait ( ) <EOL> assert proc . returncode != <NUM_LIT:0> <EOL> assert ( <EOL> b'<STR_LIT>' <EOL> in proc . stderr <EOL> ) </s>
