<s> import unittest <EOL> from test import script_helper <EOL> from test import support <EOL> import subprocess <EOL> import sys <EOL> import signal <EOL> import io <EOL> import locale <EOL> import os <EOL> import errno <EOL> import tempfile <EOL> import time <EOL> import re <EOL> import selectors <EOL> import sysconfig <EOL> import warnings <EOL> import select <EOL> import shutil <EOL> import gc <EOL> import textwrap <EOL> try : <EOL> import threading <EOL> except ImportError : <EOL> threading = None <EOL> mswindows = ( sys . platform == "<STR_LIT:win32>" ) <EOL> if mswindows : <EOL> SETBINARY = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> SETBINARY = '<STR_LIT>' <EOL> try : <EOL> mkstemp = tempfile . mkstemp <EOL> except AttributeError : <EOL> def mkstemp ( ) : <EOL> """<STR_LIT>""" <EOL> fname = tempfile . mktemp ( ) <EOL> return os . open ( fname , os . O_RDWR | os . O_CREAT ) , fname <EOL> class BaseTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> support . reap_children ( ) <EOL> def tearDown ( self ) : <EOL> for inst in subprocess . _active : <EOL> inst . wait ( ) <EOL> subprocess . _cleanup ( ) <EOL> self . assertFalse ( subprocess . _active , "<STR_LIT>" ) <EOL> def assertStderrEqual ( self , stderr , expected , msg = None ) : <EOL> actual = support . strip_python_stderr ( stderr ) <EOL> expected = expected . strip ( ) <EOL> self . assertEqual ( actual , expected , msg ) <EOL> class PopenTestException ( Exception ) : <EOL> pass <EOL> class PopenExecuteChildRaises ( subprocess . Popen ) : <EOL> """<STR_LIT>""" <EOL> def _execute_child ( self , * args , ** kwargs ) : <EOL> raise PopenTestException ( "<STR_LIT>" ) <EOL> class ProcessTestCase ( BaseTestCase ) : <EOL> def test_io_buffered_by_default ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> stdin = subprocess . PIPE , stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> try : <EOL> self . assertIsInstance ( p . stdin , io . BufferedIOBase ) <EOL> self . assertIsInstance ( p . stdout , io . BufferedIOBase ) <EOL> self . assertIsInstance ( p . stderr , io . BufferedIOBase ) <EOL> finally : <EOL> p . stdin . close ( ) <EOL> p . stdout . close ( ) <EOL> p . stderr . close ( ) <EOL> p . wait ( ) <EOL> def test_io_unbuffered_works ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> stdin = subprocess . PIPE , stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE , bufsize = <NUM_LIT:0> ) <EOL> try : <EOL> self . assertIsInstance ( p . stdin , io . RawIOBase ) <EOL> self . assertIsInstance ( p . stdout , io . RawIOBase ) <EOL> self . assertIsInstance ( p . stderr , io . RawIOBase ) <EOL> finally : <EOL> p . stdin . close ( ) <EOL> p . stdout . close ( ) <EOL> p . stderr . close ( ) <EOL> p . wait ( ) <EOL> def test_call_seq ( self ) : <EOL> rc = subprocess . call ( [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] ) <EOL> self . assertEqual ( rc , <NUM_LIT> ) <EOL> def test_call_timeout ( self ) : <EOL> self . assertRaises ( subprocess . TimeoutExpired , subprocess . call , <EOL> [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> timeout = <NUM_LIT:0.1> ) <EOL> def test_check_call_zero ( self ) : <EOL> rc = subprocess . check_call ( [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] ) <EOL> self . assertEqual ( rc , <NUM_LIT:0> ) <EOL> def test_check_call_nonzero ( self ) : <EOL> with self . assertRaises ( subprocess . CalledProcessError ) as c : <EOL> subprocess . check_call ( [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] ) <EOL> self . assertEqual ( c . exception . returncode , <NUM_LIT> ) <EOL> def test_check_output ( self ) : <EOL> output = subprocess . check_output ( <EOL> [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] ) <EOL> self . assertIn ( b'<STR_LIT>' , output ) <EOL> def test_check_output_nonzero ( self ) : <EOL> with self . assertRaises ( subprocess . CalledProcessError ) as c : <EOL> subprocess . check_output ( <EOL> [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] ) <EOL> self . assertEqual ( c . exception . returncode , <NUM_LIT:5> ) <EOL> def test_check_output_stderr ( self ) : <EOL> output = subprocess . check_output ( <EOL> [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> stderr = subprocess . STDOUT ) <EOL> self . assertIn ( b'<STR_LIT>' , output ) <EOL> def test_check_output_stdin_arg ( self ) : <EOL> tf = tempfile . TemporaryFile ( ) <EOL> self . addCleanup ( tf . close ) <EOL> tf . write ( b'<STR_LIT>' ) <EOL> tf . seek ( <NUM_LIT:0> ) <EOL> output = subprocess . check_output ( <EOL> [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] , <EOL> stdin = tf ) <EOL> self . assertIn ( b'<STR_LIT>' , output ) <EOL> def test_check_output_input_arg ( self ) : <EOL> output = subprocess . check_output ( <EOL> [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] , <EOL> input = b'<STR_LIT>' ) <EOL> self . assertIn ( b'<STR_LIT>' , output ) <EOL> def test_check_output_stdout_arg ( self ) : <EOL> with self . assertRaises ( ValueError ) as c : <EOL> output = subprocess . check_output ( <EOL> [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> stdout = sys . stdout ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> self . assertIn ( '<STR_LIT>' , c . exception . args [ <NUM_LIT:0> ] ) <EOL> def test_check_output_stdin_with_input_arg ( self ) : <EOL> tf = tempfile . TemporaryFile ( ) <EOL> self . addCleanup ( tf . close ) <EOL> tf . write ( b'<STR_LIT>' ) <EOL> tf . seek ( <NUM_LIT:0> ) <EOL> with self . assertRaises ( ValueError ) as c : <EOL> output = subprocess . check_output ( <EOL> [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> stdin = tf , input = b'<STR_LIT>' ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> self . assertIn ( '<STR_LIT>' , c . exception . args [ <NUM_LIT:0> ] ) <EOL> self . assertIn ( '<STR_LIT:input>' , c . exception . args [ <NUM_LIT:0> ] ) <EOL> def test_check_output_timeout ( self ) : <EOL> with self . assertRaises ( subprocess . TimeoutExpired ) as c : <EOL> output = subprocess . check_output ( <EOL> [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] , <EOL> timeout = <NUM_LIT:3> ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> self . assertEqual ( c . exception . output , b'<STR_LIT>' ) <EOL> def test_call_kwargs ( self ) : <EOL> newenv = os . environ . copy ( ) <EOL> newenv [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> rc = subprocess . call ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> env = newenv ) <EOL> self . assertEqual ( rc , <NUM_LIT:1> ) <EOL> def test_invalid_args ( self ) : <EOL> with support . captured_stderr ( ) as s : <EOL> self . assertRaises ( TypeError , subprocess . Popen , invalid_arg_name = <NUM_LIT:1> ) <EOL> argcount = subprocess . Popen . __init__ . __code__ . co_argcount <EOL> too_many_args = [ <NUM_LIT:0> ] * ( argcount + <NUM_LIT:1> ) <EOL> self . assertRaises ( TypeError , subprocess . Popen , * too_many_args ) <EOL> self . assertEqual ( s . getvalue ( ) , '<STR_LIT>' ) <EOL> def test_stdin_none ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> p . wait ( ) <EOL> self . assertEqual ( p . stdin , None ) <EOL> def test_stdout_none ( self ) : <EOL> code = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , code ] , <EOL> stdout = subprocess . PIPE , stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> out , err = p . communicate ( ) <EOL> self . assertEqual ( p . returncode , <NUM_LIT:0> , err ) <EOL> self . assertEqual ( out . rstrip ( ) , b'<STR_LIT>' ) <EOL> def test_stderr_none ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , '<STR_LIT>' ] , <EOL> stdin = subprocess . PIPE , stdout = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stdin . close ) <EOL> p . wait ( ) <EOL> self . assertEqual ( p . stderr , None ) <EOL> def _assert_python ( self , pre_args , ** kwargs ) : <EOL> args = pre_args + [ "<STR_LIT>" ] <EOL> p = subprocess . Popen ( args , ** kwargs ) <EOL> p . wait ( ) <EOL> self . assertEqual ( <NUM_LIT> , p . returncode ) <EOL> def test_executable ( self ) : <EOL> doesnotexist = os . path . join ( os . path . dirname ( sys . executable ) , <EOL> "<STR_LIT>" ) <EOL> self . _assert_python ( [ doesnotexist , "<STR_LIT:-c>" ] , executable = sys . executable ) <EOL> def test_executable_takes_precedence ( self ) : <EOL> pre_args = [ sys . executable , "<STR_LIT:-c>" ] <EOL> self . _assert_python ( pre_args ) <EOL> self . assertRaises ( FileNotFoundError , self . _assert_python , pre_args , <EOL> executable = "<STR_LIT>" ) <EOL> @ unittest . skipIf ( mswindows , "<STR_LIT>" ) <EOL> def test_executable_replaces_shell ( self ) : <EOL> self . _assert_python ( [ ] , executable = sys . executable , shell = True ) <EOL> def _normalize_cwd ( self , cwd ) : <EOL> original_cwd = os . getcwd ( ) <EOL> os . chdir ( cwd ) <EOL> cwd = os . getcwd ( ) <EOL> os . chdir ( original_cwd ) <EOL> return cwd <EOL> def _split_python_path ( self ) : <EOL> python_path = os . path . realpath ( sys . executable ) <EOL> return os . path . split ( python_path ) <EOL> def _assert_cwd ( self , expected_cwd , python_arg , ** kwargs ) : <EOL> p = subprocess . Popen ( [ python_arg , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] , <EOL> stdout = subprocess . PIPE , <EOL> ** kwargs ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> p . wait ( ) <EOL> self . assertEqual ( <NUM_LIT> , p . returncode ) <EOL> normcase = os . path . normcase <EOL> self . assertEqual ( normcase ( expected_cwd ) , <EOL> normcase ( p . stdout . read ( ) . decode ( "<STR_LIT:utf-8>" ) ) ) <EOL> def test_cwd ( self ) : <EOL> temp_dir = tempfile . gettempdir ( ) <EOL> temp_dir = self . _normalize_cwd ( temp_dir ) <EOL> self . _assert_cwd ( temp_dir , sys . executable , cwd = temp_dir ) <EOL> @ unittest . skipIf ( mswindows , "<STR_LIT>" ) <EOL> def test_cwd_with_relative_arg ( self ) : <EOL> python_dir , python_base = self . _split_python_path ( ) <EOL> rel_python = os . path . join ( os . curdir , python_base ) <EOL> with support . temp_cwd ( '<STR_LIT>' , quiet = True ) as wrong_dir : <EOL> self . assertRaises ( FileNotFoundError , subprocess . Popen , <EOL> [ rel_python ] ) <EOL> self . assertRaises ( FileNotFoundError , subprocess . Popen , <EOL> [ rel_python ] , cwd = wrong_dir ) <EOL> python_dir = self . _normalize_cwd ( python_dir ) <EOL> self . _assert_cwd ( python_dir , rel_python , cwd = python_dir ) <EOL> @ unittest . skipIf ( mswindows , "<STR_LIT>" ) <EOL> def test_cwd_with_relative_executable ( self ) : <EOL> python_dir , python_base = self . _split_python_path ( ) <EOL> rel_python = os . path . join ( os . curdir , python_base ) <EOL> doesntexist = "<STR_LIT>" <EOL> with support . temp_cwd ( '<STR_LIT>' , quiet = True ) as wrong_dir : <EOL> self . assertRaises ( FileNotFoundError , subprocess . Popen , <EOL> [ doesntexist ] , executable = rel_python ) <EOL> self . assertRaises ( FileNotFoundError , subprocess . Popen , <EOL> [ doesntexist ] , executable = rel_python , <EOL> cwd = wrong_dir ) <EOL> python_dir = self . _normalize_cwd ( python_dir ) <EOL> self . _assert_cwd ( python_dir , doesntexist , executable = rel_python , <EOL> cwd = python_dir ) <EOL> def test_cwd_with_absolute_arg ( self ) : <EOL> python_dir , python_base = self . _split_python_path ( ) <EOL> abs_python = os . path . join ( python_dir , python_base ) <EOL> rel_python = os . path . join ( os . curdir , python_base ) <EOL> with script_helper . temp_dir ( ) as wrong_dir : <EOL> self . assertRaises ( FileNotFoundError , subprocess . Popen , <EOL> [ rel_python ] , cwd = wrong_dir ) <EOL> wrong_dir = self . _normalize_cwd ( wrong_dir ) <EOL> self . _assert_cwd ( wrong_dir , abs_python , cwd = wrong_dir ) <EOL> @ unittest . skipIf ( sys . base_prefix != sys . prefix , <EOL> '<STR_LIT>' ) <EOL> def test_executable_with_cwd ( self ) : <EOL> python_dir , python_base = self . _split_python_path ( ) <EOL> python_dir = self . _normalize_cwd ( python_dir ) <EOL> self . _assert_cwd ( python_dir , "<STR_LIT>" , <EOL> executable = sys . executable , cwd = python_dir ) <EOL> @ unittest . skipIf ( sys . base_prefix != sys . prefix , <EOL> '<STR_LIT>' ) <EOL> @ unittest . skipIf ( sysconfig . is_python_build ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_executable_without_cwd ( self ) : <EOL> self . _assert_cwd ( os . getcwd ( ) , "<STR_LIT>" , <EOL> executable = sys . executable ) <EOL> def test_stdin_pipe ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] , <EOL> stdin = subprocess . PIPE ) <EOL> p . stdin . write ( b"<STR_LIT>" ) <EOL> p . stdin . close ( ) <EOL> p . wait ( ) <EOL> self . assertEqual ( p . returncode , <NUM_LIT:1> ) <EOL> def test_stdin_filedes ( self ) : <EOL> tf = tempfile . TemporaryFile ( ) <EOL> self . addCleanup ( tf . close ) <EOL> d = tf . fileno ( ) <EOL> os . write ( d , b"<STR_LIT>" ) <EOL> os . lseek ( d , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] , <EOL> stdin = d ) <EOL> p . wait ( ) <EOL> self . assertEqual ( p . returncode , <NUM_LIT:1> ) <EOL> def test_stdin_fileobj ( self ) : <EOL> tf = tempfile . TemporaryFile ( ) <EOL> self . addCleanup ( tf . close ) <EOL> tf . write ( b"<STR_LIT>" ) <EOL> tf . seek ( <NUM_LIT:0> ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] , <EOL> stdin = tf ) <EOL> p . wait ( ) <EOL> self . assertEqual ( p . returncode , <NUM_LIT:1> ) <EOL> def test_stdout_pipe ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . assertEqual ( p . stdout . read ( ) , b"<STR_LIT>" ) <EOL> def test_stdout_filedes ( self ) : <EOL> tf = tempfile . TemporaryFile ( ) <EOL> self . addCleanup ( tf . close ) <EOL> d = tf . fileno ( ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] , <EOL> stdout = d ) <EOL> p . wait ( ) <EOL> os . lseek ( d , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> self . assertEqual ( os . read ( d , <NUM_LIT> ) , b"<STR_LIT>" ) <EOL> def test_stdout_fileobj ( self ) : <EOL> tf = tempfile . TemporaryFile ( ) <EOL> self . addCleanup ( tf . close ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] , <EOL> stdout = tf ) <EOL> p . wait ( ) <EOL> tf . seek ( <NUM_LIT:0> ) <EOL> self . assertEqual ( tf . read ( ) , b"<STR_LIT>" ) <EOL> def test_stderr_pipe ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] , <EOL> stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> self . assertStderrEqual ( p . stderr . read ( ) , b"<STR_LIT>" ) <EOL> def test_stderr_filedes ( self ) : <EOL> tf = tempfile . TemporaryFile ( ) <EOL> self . addCleanup ( tf . close ) <EOL> d = tf . fileno ( ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] , <EOL> stderr = d ) <EOL> p . wait ( ) <EOL> os . lseek ( d , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> self . assertStderrEqual ( os . read ( d , <NUM_LIT> ) , b"<STR_LIT>" ) <EOL> def test_stderr_fileobj ( self ) : <EOL> tf = tempfile . TemporaryFile ( ) <EOL> self . addCleanup ( tf . close ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] , <EOL> stderr = tf ) <EOL> p . wait ( ) <EOL> tf . seek ( <NUM_LIT:0> ) <EOL> self . assertStderrEqual ( tf . read ( ) , b"<STR_LIT>" ) <EOL> def test_stdout_stderr_pipe ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . STDOUT ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . assertStderrEqual ( p . stdout . read ( ) , b"<STR_LIT>" ) <EOL> def test_stdout_stderr_file ( self ) : <EOL> tf = tempfile . TemporaryFile ( ) <EOL> self . addCleanup ( tf . close ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdout = tf , <EOL> stderr = tf ) <EOL> p . wait ( ) <EOL> tf . seek ( <NUM_LIT:0> ) <EOL> self . assertStderrEqual ( tf . read ( ) , b"<STR_LIT>" ) <EOL> def test_stdout_filedes_of_stdout ( self ) : <EOL> code = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , code ] , <EOL> stdout = subprocess . PIPE , stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> out , err = p . communicate ( ) <EOL> self . assertEqual ( p . returncode , <NUM_LIT:0> , err ) <EOL> self . assertEqual ( out . rstrip ( ) , b'<STR_LIT>' ) <EOL> def test_stdout_devnull ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . DEVNULL ) <EOL> p . wait ( ) <EOL> self . assertEqual ( p . stdout , None ) <EOL> def test_stderr_devnull ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stderr = subprocess . DEVNULL ) <EOL> p . wait ( ) <EOL> self . assertEqual ( p . stderr , None ) <EOL> def test_stdin_devnull ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdin = subprocess . DEVNULL ) <EOL> p . wait ( ) <EOL> self . assertEqual ( p . stdin , None ) <EOL> def test_env ( self ) : <EOL> newenv = os . environ . copy ( ) <EOL> newenv [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> with subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> env = newenv ) as p : <EOL> stdout , stderr = p . communicate ( ) <EOL> self . assertEqual ( stdout , b"<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . platform == '<STR_LIT:win32>' , <EOL> '<STR_LIT>' ) <EOL> @ unittest . skipIf ( sysconfig . get_config_var ( '<STR_LIT>' ) is not None , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_empty_env ( self ) : <EOL> with subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> env = { } ) as p : <EOL> stdout , stderr = p . communicate ( ) <EOL> self . assertIn ( stdout . strip ( ) , <EOL> ( b"<STR_LIT>" , <EOL> b"<STR_LIT>" ) ) <EOL> def test_communicate_stdin ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdin = subprocess . PIPE ) <EOL> p . communicate ( b"<STR_LIT>" ) <EOL> self . assertEqual ( p . returncode , <NUM_LIT:1> ) <EOL> def test_communicate_stdout ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE ) <EOL> ( stdout , stderr ) = p . communicate ( ) <EOL> self . assertEqual ( stdout , b"<STR_LIT>" ) <EOL> self . assertEqual ( stderr , None ) <EOL> def test_communicate_stderr ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] , <EOL> stderr = subprocess . PIPE ) <EOL> ( stdout , stderr ) = p . communicate ( ) <EOL> self . assertEqual ( stdout , None ) <EOL> self . assertStderrEqual ( stderr , b"<STR_LIT>" ) <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> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> self . addCleanup ( p . stdin . close ) <EOL> ( stdout , stderr ) = p . communicate ( b"<STR_LIT>" ) <EOL> self . assertEqual ( stdout , b"<STR_LIT>" ) <EOL> self . assertStderrEqual ( stderr , b"<STR_LIT>" ) <EOL> def test_communicate_timeout ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> universal_newlines = True , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> self . assertRaises ( subprocess . TimeoutExpired , p . communicate , "<STR_LIT>" , <EOL> timeout = <NUM_LIT> ) <EOL> ( stdout , stderr ) = p . communicate ( ) <EOL> self . assertEqual ( stdout , "<STR_LIT>" ) <EOL> self . assertStderrEqual ( stderr . encode ( ) , b"<STR_LIT>" ) <EOL> def test_communicate_timeout_large_ouput ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <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> self . assertRaises ( subprocess . TimeoutExpired , p . communicate , timeout = <NUM_LIT> ) <EOL> ( stdout , _ ) = p . communicate ( ) <EOL> self . assertEqual ( len ( stdout ) , <NUM_LIT:4> * <NUM_LIT:64> * <NUM_LIT> ) <EOL> def test_communicate_pipe_fd_leak ( self ) : <EOL> for stdin_pipe in ( False , True ) : <EOL> for stdout_pipe in ( False , True ) : <EOL> for stderr_pipe in ( False , True ) : <EOL> options = { } <EOL> if stdin_pipe : <EOL> options [ '<STR_LIT>' ] = subprocess . PIPE <EOL> if stdout_pipe : <EOL> options [ '<STR_LIT>' ] = subprocess . PIPE <EOL> if stderr_pipe : <EOL> options [ '<STR_LIT>' ] = subprocess . PIPE <EOL> if not options : <EOL> continue <EOL> p = subprocess . Popen ( ( sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ) , ** options ) <EOL> p . communicate ( ) <EOL> if p . stdin is not None : <EOL> self . assertTrue ( p . stdin . closed ) <EOL> if p . stdout is not None : <EOL> self . assertTrue ( p . stdout . closed ) <EOL> if p . stderr is not None : <EOL> self . assertTrue ( p . stderr . closed ) <EOL> def test_communicate_returns ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] ) <EOL> ( stdout , stderr ) = p . communicate ( ) <EOL> self . assertEqual ( stdout , None ) <EOL> self . assertEqual ( stderr , None ) <EOL> def test_communicate_pipe_buf ( self ) : <EOL> x , y = os . pipe ( ) <EOL> os . close ( x ) <EOL> os . close ( y ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % <EOL> support . PIPE_MAX_SIZE ] , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> self . addCleanup ( p . stdin . close ) <EOL> string_to_write = b"<STR_LIT:a>" * support . PIPE_MAX_SIZE <EOL> ( stdout , stderr ) = p . communicate ( string_to_write ) <EOL> self . assertEqual ( stdout , string_to_write ) <EOL> def test_writes_before_communicate ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> self . addCleanup ( p . stdin . close ) <EOL> p . stdin . write ( b"<STR_LIT>" ) <EOL> ( stdout , stderr ) = p . communicate ( b"<STR_LIT>" ) <EOL> self . assertEqual ( stdout , b"<STR_LIT>" ) <EOL> self . assertStderrEqual ( stderr , b"<STR_LIT>" ) <EOL> def test_universal_newlines ( 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> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> universal_newlines = <NUM_LIT:1> ) <EOL> p . stdin . write ( "<STR_LIT>" ) <EOL> p . stdin . flush ( ) <EOL> self . assertEqual ( p . stdout . readline ( ) , "<STR_LIT>" ) <EOL> p . stdin . write ( "<STR_LIT>" ) <EOL> p . stdin . close ( ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . assertEqual ( p . stdout . readline ( ) , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( p . stdout . read ( <NUM_LIT:6> ) , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( p . stdout . read ( ) , <EOL> "<STR_LIT>" ) <EOL> def test_universal_newlines_communicate ( 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> '<STR_LIT>' ] , <EOL> stderr = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> universal_newlines = <NUM_LIT:1> ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> ( stdout , stderr ) = p . communicate ( ) <EOL> self . assertEqual ( stdout , <EOL> "<STR_LIT>" ) <EOL> def test_universal_newlines_communicate_stdin ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' + SETBINARY + textwrap . dedent ( '''<STR_LIT>''' ) ] , <EOL> stdin = subprocess . PIPE , <EOL> universal_newlines = <NUM_LIT:1> ) <EOL> ( stdout , stderr ) = p . communicate ( "<STR_LIT>" ) <EOL> self . assertEqual ( p . returncode , <NUM_LIT:0> ) <EOL> def test_universal_newlines_communicate_input_none ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> universal_newlines = True ) <EOL> p . communicate ( ) <EOL> self . assertEqual ( p . returncode , <NUM_LIT:0> ) <EOL> def test_universal_newlines_communicate_stdin_stdout_stderr ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' + SETBINARY + textwrap . dedent ( '''<STR_LIT>''' ) ] , <EOL> stdin = subprocess . PIPE , <EOL> stderr = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> universal_newlines = True ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> ( stdout , stderr ) = p . communicate ( "<STR_LIT>" ) <EOL> self . assertEqual ( p . returncode , <NUM_LIT:0> ) <EOL> self . assertEqual ( "<STR_LIT>" , stdout ) <EOL> self . assertTrue ( stderr . startswith ( "<STR_LIT>" ) ) <EOL> def test_universal_newlines_communicate_encodings ( self ) : <EOL> import _bootlocale <EOL> for encoding in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> old_getpreferredencoding = _bootlocale . getpreferredencoding <EOL> def getpreferredencoding ( do_setlocale = True ) : <EOL> return encoding <EOL> code = ( "<STR_LIT>" <EOL> r"<STR_LIT>" % <EOL> encoding ) <EOL> args = [ sys . executable , '<STR_LIT:-c>' , code ] <EOL> try : <EOL> _bootlocale . getpreferredencoding = getpreferredencoding <EOL> popen = subprocess . Popen ( args , universal_newlines = True , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE ) <EOL> stdout , stderr = popen . communicate ( input = '<STR_LIT>' ) <EOL> finally : <EOL> _bootlocale . getpreferredencoding = old_getpreferredencoding <EOL> self . assertEqual ( stdout , '<STR_LIT>' ) <EOL> def test_no_leaking ( self ) : <EOL> if not mswindows : <EOL> max_handles = <NUM_LIT> <EOL> else : <EOL> max_handles = <NUM_LIT> <EOL> handles = [ ] <EOL> tmpdir = tempfile . mkdtemp ( ) <EOL> try : <EOL> for i in range ( max_handles ) : <EOL> try : <EOL> tmpfile = os . path . join ( tmpdir , support . TESTFN ) <EOL> handles . append ( os . open ( tmpfile , os . O_WRONLY | os . O_CREAT ) ) <EOL> except OSError as e : <EOL> if e . errno != errno . EMFILE : <EOL> raise <EOL> break <EOL> else : <EOL> self . skipTest ( "<STR_LIT>" <EOL> "<STR_LIT>" % max_handles ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> os . close ( handles . pop ( ) ) <EOL> for i in range ( <NUM_LIT:15> ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> data = p . communicate ( b"<STR_LIT>" ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( data , b"<STR_LIT>" ) <EOL> finally : <EOL> for h in handles : <EOL> os . close ( h ) <EOL> shutil . rmtree ( tmpdir ) <EOL> def test_list2cmdline ( self ) : <EOL> self . assertEqual ( subprocess . list2cmdline ( [ '<STR_LIT>' , '<STR_LIT:d>' , '<STR_LIT:e>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( subprocess . list2cmdline ( [ '<STR_LIT>' , '<STR_LIT:\\>' , '<STR_LIT:d>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( subprocess . list2cmdline ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:d>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( subprocess . list2cmdline ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:h>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( subprocess . list2cmdline ( [ '<STR_LIT>' , '<STR_LIT:c>' , '<STR_LIT:d>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( subprocess . list2cmdline ( [ '<STR_LIT>' , '<STR_LIT:d>' , '<STR_LIT:e>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( subprocess . list2cmdline ( [ '<STR_LIT>' , '<STR_LIT:d>' , '<STR_LIT:e>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( subprocess . list2cmdline ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) <EOL> def test_poll ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] , <EOL> stdin = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdin . close ) <EOL> self . assertIsNone ( p . poll ( ) ) <EOL> os . write ( p . stdin . fileno ( ) , b'<STR_LIT:A>' ) <EOL> p . wait ( ) <EOL> self . assertEqual ( p . poll ( ) , <NUM_LIT:0> ) <EOL> def test_wait ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] ) <EOL> self . assertEqual ( p . wait ( ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( p . wait ( ) , <NUM_LIT:0> ) <EOL> def test_wait_timeout ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , <EOL> "<STR_LIT:-c>" , "<STR_LIT>" ] ) <EOL> with self . assertRaises ( subprocess . TimeoutExpired ) as c : <EOL> p . wait ( timeout = <NUM_LIT> ) <EOL> self . assertIn ( "<STR_LIT>" , str ( c . exception ) ) <EOL> self . assertEqual ( p . wait ( timeout = <NUM_LIT:3> ) , <NUM_LIT:0> ) <EOL> def test_invalid_bufsize ( self ) : <EOL> with self . assertRaises ( TypeError ) : <EOL> subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> def test_bufsize_is_none ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , None ) <EOL> self . assertEqual ( p . wait ( ) , <NUM_LIT:0> ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , bufsize = None ) <EOL> self . assertEqual ( p . wait ( ) , <NUM_LIT:0> ) <EOL> def _test_bufsize_equal_one ( self , line , expected , universal_newlines ) : <EOL> with subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . DEVNULL , <EOL> bufsize = <NUM_LIT:1> , <EOL> universal_newlines = universal_newlines ) as p : <EOL> p . stdin . write ( line ) <EOL> os . close ( p . stdin . fileno ( ) ) <EOL> read_line = p . stdout . readline ( ) <EOL> try : <EOL> p . stdin . close ( ) <EOL> except OSError : <EOL> pass <EOL> p . stdin = None <EOL> self . assertEqual ( p . returncode , <NUM_LIT:0> ) <EOL> self . assertEqual ( read_line , expected ) <EOL> def test_bufsize_equal_one_text_mode ( self ) : <EOL> line = "<STR_LIT>" <EOL> self . _test_bufsize_equal_one ( line , line , universal_newlines = True ) <EOL> def test_bufsize_equal_one_binary_mode ( self ) : <EOL> line = b'<STR_LIT>' + os . linesep . encode ( ) <EOL> self . _test_bufsize_equal_one ( line , b'<STR_LIT>' , universal_newlines = False ) <EOL> def test_leaking_fds_on_error ( self ) : <EOL> for i in range ( <NUM_LIT> ) : <EOL> with self . assertRaises ( OSError ) as c : <EOL> subprocess . Popen ( [ '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> if c . exception . errno not in ( errno . ENOENT , errno . EACCES ) : <EOL> raise c . exception <EOL> @ unittest . skipIf ( threading is None , "<STR_LIT>" ) <EOL> def test_double_close_on_error ( self ) : <EOL> fds = [ ] <EOL> def open_fds ( ) : <EOL> for i in range ( <NUM_LIT:20> ) : <EOL> fds . extend ( os . pipe ( ) ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> t = threading . Thread ( target = open_fds ) <EOL> t . start ( ) <EOL> try : <EOL> with self . assertRaises ( EnvironmentError ) : <EOL> subprocess . Popen ( [ '<STR_LIT>' ] , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> finally : <EOL> t . join ( ) <EOL> exc = None <EOL> for fd in fds : <EOL> try : <EOL> os . close ( fd ) <EOL> except OSError as e : <EOL> exc = e <EOL> if exc is not None : <EOL> raise exc <EOL> @ unittest . skipIf ( threading is None , "<STR_LIT>" ) <EOL> def test_threadsafe_wait ( self ) : <EOL> """<STR_LIT>""" <EOL> proc = subprocess . Popen ( [ sys . executable , '<STR_LIT:-c>' , <EOL> '<STR_LIT>' ] ) <EOL> self . assertEqual ( proc . returncode , None ) <EOL> results = [ ] <EOL> def kill_proc_timer_thread ( ) : <EOL> results . append ( ( '<STR_LIT>' , proc . poll ( ) ) ) <EOL> proc . kill ( ) <EOL> proc . wait ( ) <EOL> results . append ( ( '<STR_LIT>' , proc . returncode ) ) <EOL> proc . wait ( ) <EOL> results . append ( ( '<STR_LIT>' , proc . returncode ) ) <EOL> t = threading . Timer ( <NUM_LIT> , kill_proc_timer_thread ) <EOL> t . start ( ) <EOL> if mswindows : <EOL> expected_errorcode = <NUM_LIT:1> <EOL> else : <EOL> expected_errorcode = - <NUM_LIT:9> <EOL> proc . wait ( timeout = <NUM_LIT:20> ) <EOL> self . assertEqual ( proc . returncode , expected_errorcode , <EOL> msg = "<STR_LIT>" ) <EOL> proc . wait ( ) <EOL> self . assertEqual ( proc . returncode , expected_errorcode , <EOL> msg = "<STR_LIT>" ) <EOL> t . join ( ) <EOL> self . assertEqual ( [ ( '<STR_LIT>' , None ) , <EOL> ( '<STR_LIT>' , expected_errorcode ) , <EOL> ( '<STR_LIT>' , expected_errorcode ) ] , <EOL> results ) <EOL> def test_issue8780 ( self ) : <EOL> code = '<STR_LIT:;>' . join ( ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> "<STR_LIT>" , <EOL> '<STR_LIT>' ) ) <EOL> output = subprocess . check_output ( [ sys . executable , '<STR_LIT:-c>' , code ] ) <EOL> self . assertTrue ( output . startswith ( b'<STR_LIT>' ) , ascii ( output ) ) <EOL> def test_handles_closed_on_exception ( self ) : <EOL> ifhandle , ifname = mkstemp ( ) <EOL> ofhandle , ofname = mkstemp ( ) <EOL> efhandle , efname = mkstemp ( ) <EOL> try : <EOL> subprocess . Popen ( [ "<STR_LIT:*>" ] , stdin = ifhandle , stdout = ofhandle , <EOL> stderr = efhandle ) <EOL> except OSError : <EOL> os . close ( ifhandle ) <EOL> os . remove ( ifname ) <EOL> os . close ( ofhandle ) <EOL> os . remove ( ofname ) <EOL> os . close ( efhandle ) <EOL> os . remove ( efname ) <EOL> self . assertFalse ( os . path . exists ( ifname ) ) <EOL> self . assertFalse ( os . path . exists ( ofname ) ) <EOL> self . assertFalse ( os . path . exists ( efname ) ) <EOL> def test_communicate_epipe ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , '<STR_LIT>' ] , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> self . addCleanup ( p . stdin . close ) <EOL> p . communicate ( b"<STR_LIT:x>" * <NUM_LIT:2> ** <NUM_LIT:20> ) <EOL> def test_communicate_epipe_only_stdin ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , '<STR_LIT>' ] , <EOL> stdin = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdin . close ) <EOL> p . wait ( ) <EOL> p . communicate ( b"<STR_LIT:x>" * <NUM_LIT:2> ** <NUM_LIT:20> ) <EOL> @ unittest . skipUnless ( hasattr ( signal , '<STR_LIT>' ) , <EOL> "<STR_LIT>" ) <EOL> @ unittest . skipUnless ( hasattr ( os , '<STR_LIT>' ) , <EOL> "<STR_LIT>" ) <EOL> @ unittest . skipUnless ( hasattr ( os , '<STR_LIT>' ) , <EOL> "<STR_LIT>" ) <EOL> def test_communicate_eintr ( self ) : <EOL> def handler ( signum , frame ) : <EOL> pass <EOL> old_handler = signal . signal ( signal . SIGUSR1 , handler ) <EOL> self . addCleanup ( signal . signal , signal . SIGUSR1 , old_handler ) <EOL> args = [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] <EOL> for stream in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> kw = { stream : subprocess . PIPE } <EOL> with subprocess . Popen ( args , ** kw ) as process : <EOL> process . communicate ( ) <EOL> @ unittest . skipUnless ( os . path . isdir ( '<STR_LIT>' % os . getpid ( ) ) , <EOL> "<STR_LIT>" ) <EOL> def test_failed_child_execute_fd_leak ( self ) : <EOL> """<STR_LIT>""" <EOL> fd_directory = '<STR_LIT>' % os . getpid ( ) <EOL> fds_before_popen = os . listdir ( fd_directory ) <EOL> with self . assertRaises ( PopenTestException ) : <EOL> PopenExecuteChildRaises ( <EOL> [ sys . executable , '<STR_LIT:-c>' , '<STR_LIT>' ] , stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , stderr = subprocess . PIPE ) <EOL> fds_after_exception = os . listdir ( fd_directory ) <EOL> self . assertEqual ( fds_before_popen , fds_after_exception ) <EOL> @ unittest . skipIf ( mswindows , "<STR_LIT>" ) <EOL> class POSIXProcessTestCase ( BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ) . setUp ( ) <EOL> self . _nonexistent_dir = "<STR_LIT>" <EOL> def _get_chdir_exception ( self ) : <EOL> try : <EOL> os . chdir ( self . _nonexistent_dir ) <EOL> except OSError as e : <EOL> desired_exception = e <EOL> desired_exception . strerror += '<STR_LIT>' + repr ( self . _nonexistent_dir ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" % <EOL> self . _nonexistent_dir ) <EOL> return desired_exception <EOL> def test_exception_cwd ( self ) : <EOL> """<STR_LIT>""" <EOL> desired_exception = self . _get_chdir_exception ( ) <EOL> try : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> cwd = self . _nonexistent_dir ) <EOL> except OSError as e : <EOL> self . assertEqual ( desired_exception . errno , e . errno ) <EOL> self . assertEqual ( desired_exception . strerror , e . strerror ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" % desired_exception ) <EOL> def test_exception_bad_executable ( self ) : <EOL> """<STR_LIT>""" <EOL> desired_exception = self . _get_chdir_exception ( ) <EOL> try : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> executable = self . _nonexistent_dir ) <EOL> except OSError as e : <EOL> self . assertEqual ( desired_exception . errno , e . errno ) <EOL> self . assertEqual ( desired_exception . strerror , e . strerror ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" % desired_exception ) <EOL> def test_exception_bad_args_0 ( self ) : <EOL> """<STR_LIT>""" <EOL> desired_exception = self . _get_chdir_exception ( ) <EOL> try : <EOL> p = subprocess . Popen ( [ self . _nonexistent_dir , "<STR_LIT:-c>" , "<STR_LIT>" ] ) <EOL> except OSError as e : <EOL> self . assertEqual ( desired_exception . errno , e . errno ) <EOL> self . assertEqual ( desired_exception . strerror , e . strerror ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" % desired_exception ) <EOL> def test_restore_signals ( self ) : <EOL> subprocess . call ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , restore_signals = True ) <EOL> subprocess . call ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , restore_signals = False ) <EOL> def test_start_new_session ( self ) : <EOL> try : <EOL> output = subprocess . check_output ( <EOL> [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] , <EOL> start_new_session = True ) <EOL> except OSError as e : <EOL> if e . errno != errno . EPERM : <EOL> raise <EOL> else : <EOL> parent_pgid = os . getpgid ( os . getpid ( ) ) <EOL> child_pgid = int ( output ) <EOL> self . assertNotEqual ( parent_pgid , child_pgid ) <EOL> def test_run_abort ( self ) : <EOL> with support . SuppressCrashReport ( ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' ] ) <EOL> p . wait ( ) <EOL> self . assertEqual ( - p . returncode , signal . SIGABRT ) <EOL> def test_preexec ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> preexec_fn = lambda : os . putenv ( "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . assertEqual ( p . stdout . read ( ) , b"<STR_LIT>" ) <EOL> def test_preexec_exception ( self ) : <EOL> def raise_it ( ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> try : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> preexec_fn = raise_it ) <EOL> except subprocess . SubprocessError as e : <EOL> self . assertTrue ( <EOL> subprocess . _posixsubprocess , <EOL> "<STR_LIT>" ) <EOL> except ValueError as e : <EOL> self . assertIn ( "<STR_LIT>" , e . args [ <NUM_LIT:0> ] ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class _TestExecuteChildPopen ( subprocess . Popen ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , testcase , * args , ** kwargs ) : <EOL> self . _testcase = testcase <EOL> subprocess . Popen . __init__ ( self , * args , ** kwargs ) <EOL> def _execute_child ( self , * args , ** kwargs ) : <EOL> try : <EOL> subprocess . Popen . _execute_child ( self , * args , ** kwargs ) <EOL> finally : <EOL> devzero_fds = [ os . open ( "<STR_LIT>" , os . O_RDONLY ) <EOL> for _ in range ( <NUM_LIT:8> ) ] <EOL> try : <EOL> for fd in devzero_fds : <EOL> self . _testcase . assertNotIn ( <EOL> fd , ( self . stdin . fileno ( ) , self . stdout . fileno ( ) , <EOL> self . stderr . fileno ( ) ) , <EOL> msg = "<STR_LIT>" ) <EOL> finally : <EOL> for fd in devzero_fds : <EOL> os . close ( fd ) <EOL> @ unittest . skipIf ( not os . path . exists ( "<STR_LIT>" ) , "<STR_LIT>" ) <EOL> def test_preexec_errpipe_does_not_double_close_pipes ( self ) : <EOL> """<STR_LIT>""" <EOL> def raise_it ( ) : <EOL> raise subprocess . SubprocessError ( <EOL> "<STR_LIT>" ) <EOL> with self . assertRaises ( subprocess . SubprocessError ) : <EOL> self . _TestExecuteChildPopen ( <EOL> self , [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> stdin = subprocess . PIPE , stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE , preexec_fn = raise_it ) <EOL> def test_preexec_gc_module_failure ( self ) : <EOL> def raise_runtime_error ( ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> enabled = gc . isenabled ( ) <EOL> orig_gc_disable = gc . disable <EOL> orig_gc_isenabled = gc . isenabled <EOL> try : <EOL> gc . disable ( ) <EOL> self . assertFalse ( gc . isenabled ( ) ) <EOL> subprocess . call ( [ sys . executable , '<STR_LIT:-c>' , '<STR_LIT>' ] , <EOL> preexec_fn = lambda : None ) <EOL> self . assertFalse ( gc . isenabled ( ) , <EOL> "<STR_LIT>" ) <EOL> gc . enable ( ) <EOL> self . assertTrue ( gc . isenabled ( ) ) <EOL> subprocess . call ( [ sys . executable , '<STR_LIT:-c>' , '<STR_LIT>' ] , <EOL> preexec_fn = lambda : None ) <EOL> self . assertTrue ( gc . isenabled ( ) , "<STR_LIT>" ) <EOL> gc . disable = raise_runtime_error <EOL> self . assertRaises ( RuntimeError , subprocess . Popen , <EOL> [ sys . executable , '<STR_LIT:-c>' , '<STR_LIT>' ] , <EOL> preexec_fn = lambda : None ) <EOL> del gc . isenabled <EOL> self . assertRaises ( AttributeError , subprocess . Popen , <EOL> [ sys . executable , '<STR_LIT:-c>' , '<STR_LIT>' ] , <EOL> preexec_fn = lambda : None ) <EOL> finally : <EOL> gc . disable = orig_gc_disable <EOL> gc . isenabled = orig_gc_isenabled <EOL> if not enabled : <EOL> gc . disable ( ) <EOL> def test_args_string ( self ) : <EOL> fd , fname = mkstemp ( ) <EOL> with open ( fd , "<STR_LIT:w>" , errors = "<STR_LIT>" ) as fobj : <EOL> fobj . write ( "<STR_LIT>" ) <EOL> fobj . write ( "<STR_LIT>" % <EOL> sys . executable ) <EOL> os . chmod ( fname , <NUM_LIT> ) <EOL> p = subprocess . Popen ( fname ) <EOL> p . wait ( ) <EOL> os . remove ( fname ) <EOL> self . assertEqual ( p . returncode , <NUM_LIT> ) <EOL> def test_invalid_args ( self ) : <EOL> self . assertRaises ( ValueError , subprocess . call , <EOL> [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] , <EOL> startupinfo = <NUM_LIT> ) <EOL> self . assertRaises ( ValueError , subprocess . call , <EOL> [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] , <EOL> creationflags = <NUM_LIT> ) <EOL> def test_shell_sequence ( self ) : <EOL> newenv = os . environ . copy ( ) <EOL> newenv [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> p = subprocess . Popen ( [ "<STR_LIT>" ] , shell = <NUM_LIT:1> , <EOL> stdout = subprocess . PIPE , <EOL> env = newenv ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . assertEqual ( p . stdout . read ( ) . strip ( b"<STR_LIT>" ) , b"<STR_LIT>" ) <EOL> def test_shell_string ( self ) : <EOL> newenv = os . environ . copy ( ) <EOL> newenv [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> p = subprocess . Popen ( "<STR_LIT>" , shell = <NUM_LIT:1> , <EOL> stdout = subprocess . PIPE , <EOL> env = newenv ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . assertEqual ( p . stdout . read ( ) . strip ( b"<STR_LIT>" ) , b"<STR_LIT>" ) <EOL> def test_call_string ( self ) : <EOL> fd , fname = mkstemp ( ) <EOL> with open ( fd , "<STR_LIT:w>" , errors = "<STR_LIT>" ) as fobj : <EOL> fobj . write ( "<STR_LIT>" ) <EOL> fobj . write ( "<STR_LIT>" % <EOL> sys . executable ) <EOL> os . chmod ( fname , <NUM_LIT> ) <EOL> rc = subprocess . call ( fname ) <EOL> os . remove ( fname ) <EOL> self . assertEqual ( rc , <NUM_LIT> ) <EOL> def test_specific_shell ( self ) : <EOL> shells = [ ] <EOL> for prefix in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> for name in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> sh = os . path . join ( prefix , name ) <EOL> if os . path . isfile ( sh ) : <EOL> shells . append ( sh ) <EOL> if not shells : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> sh = '<STR_LIT>' <EOL> if os . path . isfile ( sh ) and not os . path . islink ( sh ) : <EOL> shells . append ( sh ) <EOL> for sh in shells : <EOL> p = subprocess . Popen ( "<STR_LIT>" , executable = sh , shell = True , <EOL> stdout = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . assertEqual ( p . stdout . read ( ) . strip ( ) , bytes ( sh , '<STR_LIT:ascii>' ) ) <EOL> def _kill_process ( self , method , * args ) : <EOL> old_handler = signal . signal ( signal . SIGINT , signal . default_int_handler ) <EOL> try : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , """<STR_LIT>""" ] , <EOL> close_fds = True , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> finally : <EOL> signal . signal ( signal . SIGINT , old_handler ) <EOL> p . stdout . read ( <NUM_LIT:1> ) <EOL> getattr ( p , method ) ( * args ) <EOL> return p <EOL> @ unittest . skipIf ( sys . platform . startswith ( ( '<STR_LIT>' , '<STR_LIT>' ) ) , <EOL> "<STR_LIT>" ) <EOL> def _kill_dead_process ( self , method , * args ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , """<STR_LIT>""" ] , <EOL> close_fds = True , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> p . stdout . read ( <NUM_LIT:1> ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> getattr ( p , method ) ( * args ) <EOL> p . communicate ( ) <EOL> def test_send_signal ( self ) : <EOL> p = self . _kill_process ( '<STR_LIT>' , signal . SIGINT ) <EOL> _ , stderr = p . communicate ( ) <EOL> self . assertIn ( b'<STR_LIT>' , stderr ) <EOL> self . assertNotEqual ( p . wait ( ) , <NUM_LIT:0> ) <EOL> def test_kill ( self ) : <EOL> p = self . _kill_process ( '<STR_LIT>' ) <EOL> _ , stderr = p . communicate ( ) <EOL> self . assertStderrEqual ( stderr , b'<STR_LIT>' ) <EOL> self . assertEqual ( p . wait ( ) , - signal . SIGKILL ) <EOL> def test_terminate ( self ) : <EOL> p = self . _kill_process ( '<STR_LIT>' ) <EOL> _ , stderr = p . communicate ( ) <EOL> self . assertStderrEqual ( stderr , b'<STR_LIT>' ) <EOL> self . assertEqual ( p . wait ( ) , - signal . SIGTERM ) <EOL> def test_send_signal_dead ( self ) : <EOL> self . _kill_dead_process ( '<STR_LIT>' , signal . SIGINT ) <EOL> def test_kill_dead ( self ) : <EOL> self . _kill_dead_process ( '<STR_LIT>' ) <EOL> def test_terminate_dead ( self ) : <EOL> self . _kill_dead_process ( '<STR_LIT>' ) <EOL> def _save_fds ( self , save_fds ) : <EOL> fds = [ ] <EOL> for fd in save_fds : <EOL> inheritable = os . get_inheritable ( fd ) <EOL> saved = os . dup ( fd ) <EOL> fds . append ( ( fd , saved , inheritable ) ) <EOL> return fds <EOL> def _restore_fds ( self , fds ) : <EOL> for fd , saved , inheritable in fds : <EOL> os . dup2 ( saved , fd , inheritable = inheritable ) <EOL> os . close ( saved ) <EOL> def check_close_std_fds ( self , fds ) : <EOL> stdin = <NUM_LIT:0> <EOL> saved_fds = self . _save_fds ( fds ) <EOL> for fd , saved , inheritable in saved_fds : <EOL> if fd == <NUM_LIT:0> : <EOL> stdin = saved <EOL> break <EOL> try : <EOL> for fd in fds : <EOL> os . close ( fd ) <EOL> out , err = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdin = stdin , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) . communicate ( ) <EOL> err = support . strip_python_stderr ( err ) <EOL> self . assertEqual ( ( out , err ) , ( b'<STR_LIT>' , b'<STR_LIT>' ) ) <EOL> finally : <EOL> self . _restore_fds ( saved_fds ) <EOL> def test_close_fd_0 ( self ) : <EOL> self . check_close_std_fds ( [ <NUM_LIT:0> ] ) <EOL> def test_close_fd_1 ( self ) : <EOL> self . check_close_std_fds ( [ <NUM_LIT:1> ] ) <EOL> def test_close_fd_2 ( self ) : <EOL> self . check_close_std_fds ( [ <NUM_LIT:2> ] ) <EOL> def test_close_fds_0_1 ( self ) : <EOL> self . check_close_std_fds ( [ <NUM_LIT:0> , <NUM_LIT:1> ] ) <EOL> def test_close_fds_0_2 ( self ) : <EOL> self . check_close_std_fds ( [ <NUM_LIT:0> , <NUM_LIT:2> ] ) <EOL> def test_close_fds_1_2 ( self ) : <EOL> self . check_close_std_fds ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> def test_close_fds_0_1_2 ( self ) : <EOL> self . check_close_std_fds ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> def test_small_errpipe_write_fd ( self ) : <EOL> """<STR_LIT>""" <EOL> new_stdin = os . dup ( <NUM_LIT:0> ) <EOL> new_stdout = os . dup ( <NUM_LIT:1> ) <EOL> try : <EOL> os . close ( <NUM_LIT:0> ) <EOL> os . close ( <NUM_LIT:1> ) <EOL> subprocess . Popen ( [ <EOL> sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] ) . wait ( ) <EOL> finally : <EOL> os . dup2 ( new_stdin , <NUM_LIT:0> ) <EOL> os . dup2 ( new_stdout , <NUM_LIT:1> ) <EOL> os . close ( new_stdin ) <EOL> os . close ( new_stdout ) <EOL> def test_remapping_std_fds ( self ) : <EOL> temps = [ mkstemp ( ) for i in range ( <NUM_LIT:3> ) ] <EOL> try : <EOL> temp_fds = [ fd for fd , fname in temps ] <EOL> for fd , fname in temps : <EOL> os . unlink ( fname ) <EOL> os . write ( temp_fds [ <NUM_LIT:1> ] , b"<STR_LIT>" ) <EOL> os . lseek ( temp_fds [ <NUM_LIT:1> ] , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> saved_fds = self . _save_fds ( range ( <NUM_LIT:3> ) ) <EOL> try : <EOL> for fd , temp_fd in enumerate ( temp_fds ) : <EOL> os . dup2 ( temp_fd , fd ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdin = temp_fds [ <NUM_LIT:1> ] , <EOL> stdout = temp_fds [ <NUM_LIT:2> ] , <EOL> stderr = temp_fds [ <NUM_LIT:0> ] ) <EOL> p . wait ( ) <EOL> finally : <EOL> self . _restore_fds ( saved_fds ) <EOL> for fd in temp_fds : <EOL> os . lseek ( fd , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> out = os . read ( temp_fds [ <NUM_LIT:2> ] , <NUM_LIT> ) <EOL> err = support . strip_python_stderr ( os . read ( temp_fds [ <NUM_LIT:0> ] , <NUM_LIT> ) ) <EOL> self . assertEqual ( out , b"<STR_LIT>" ) <EOL> self . assertEqual ( err , b"<STR_LIT>" ) <EOL> finally : <EOL> for fd in temp_fds : <EOL> os . close ( fd ) <EOL> def check_swap_fds ( self , stdin_no , stdout_no , stderr_no ) : <EOL> temps = [ mkstemp ( ) for i in range ( <NUM_LIT:3> ) ] <EOL> temp_fds = [ fd for fd , fname in temps ] <EOL> try : <EOL> for fd , fname in temps : <EOL> os . unlink ( fname ) <EOL> saved_fds = self . _save_fds ( range ( <NUM_LIT:3> ) ) <EOL> try : <EOL> for fd , temp_fd in enumerate ( temp_fds ) : <EOL> os . dup2 ( temp_fd , fd ) <EOL> os . write ( stdin_no , b"<STR_LIT>" ) <EOL> os . lseek ( stdin_no , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdin = stdin_no , <EOL> stdout = stdout_no , <EOL> stderr = stderr_no ) <EOL> p . wait ( ) <EOL> for fd in temp_fds : <EOL> os . lseek ( fd , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> out = os . read ( stdout_no , <NUM_LIT> ) <EOL> err = support . strip_python_stderr ( os . read ( stderr_no , <NUM_LIT> ) ) <EOL> finally : <EOL> self . _restore_fds ( saved_fds ) <EOL> self . assertEqual ( out , b"<STR_LIT>" ) <EOL> self . assertEqual ( err , b"<STR_LIT>" ) <EOL> finally : <EOL> for fd in temp_fds : <EOL> os . close ( fd ) <EOL> def test_swap_fds ( self ) : <EOL> self . check_swap_fds ( <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . check_swap_fds ( <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:1> ) <EOL> self . check_swap_fds ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:2> ) <EOL> self . check_swap_fds ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:0> ) <EOL> self . check_swap_fds ( <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> self . check_swap_fds ( <NUM_LIT:2> , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> def test_surrogates_error_message ( self ) : <EOL> def prepare ( ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> try : <EOL> subprocess . call ( <EOL> [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> preexec_fn = prepare ) <EOL> except ValueError as err : <EOL> self . assertIsNone ( subprocess . _posixsubprocess ) <EOL> self . assertEqual ( str ( err ) , "<STR_LIT>" ) <EOL> except subprocess . SubprocessError as err : <EOL> self . assertIsNotNone ( subprocess . _posixsubprocess ) <EOL> self . assertEqual ( str ( err ) , "<STR_LIT>" ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def test_undecodable_env ( self ) : <EOL> for key , value in ( ( '<STR_LIT:test>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) ) : <EOL> encoded_value = value . encode ( "<STR_LIT:ascii>" , "<STR_LIT>" ) <EOL> script = "<STR_LIT>" % repr ( key ) <EOL> env = os . environ . copy ( ) <EOL> env [ key ] = value <EOL> env [ '<STR_LIT>' ] = '<STR_LIT:C>' <EOL> if sys . platform . startswith ( "<STR_LIT>" ) : <EOL> decoded_value = encoded_value . decode ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> else : <EOL> decoded_value = value <EOL> stdout = subprocess . check_output ( <EOL> [ sys . executable , "<STR_LIT:-c>" , script ] , <EOL> env = env ) <EOL> stdout = stdout . rstrip ( b'<STR_LIT>' ) <EOL> self . assertEqual ( stdout . decode ( '<STR_LIT:ascii>' ) , ascii ( decoded_value ) ) <EOL> key = key . encode ( "<STR_LIT:ascii>" , "<STR_LIT>" ) <EOL> script = "<STR_LIT>" % repr ( key ) <EOL> env = os . environ . copy ( ) <EOL> env [ key ] = encoded_value <EOL> stdout = subprocess . check_output ( <EOL> [ sys . executable , "<STR_LIT:-c>" , script ] , <EOL> env = env ) <EOL> stdout = stdout . rstrip ( b'<STR_LIT>' ) <EOL> self . assertEqual ( stdout . decode ( '<STR_LIT:ascii>' ) , ascii ( encoded_value ) ) <EOL> def test_bytes_program ( self ) : <EOL> abs_program = os . fsencode ( sys . executable ) <EOL> path , program = os . path . split ( sys . executable ) <EOL> program = os . fsencode ( program ) <EOL> exitcode = subprocess . call ( [ abs_program , "<STR_LIT:-c>" , "<STR_LIT>" ] ) <EOL> self . assertEqual ( exitcode , <NUM_LIT:0> ) <EOL> cmd = b"<STR_LIT:'>" + abs_program + b"<STR_LIT>" <EOL> exitcode = subprocess . call ( cmd , shell = True ) <EOL> self . assertEqual ( exitcode , <NUM_LIT:0> ) <EOL> env = os . environ . copy ( ) <EOL> env [ "<STR_LIT>" ] = path <EOL> exitcode = subprocess . call ( [ program , "<STR_LIT:-c>" , "<STR_LIT>" ] , env = env ) <EOL> self . assertEqual ( exitcode , <NUM_LIT:0> ) <EOL> envb = os . environb . copy ( ) <EOL> envb [ b"<STR_LIT>" ] = os . fsencode ( path ) <EOL> exitcode = subprocess . call ( [ program , "<STR_LIT:-c>" , "<STR_LIT>" ] , env = envb ) <EOL> self . assertEqual ( exitcode , <NUM_LIT:0> ) <EOL> def test_pipe_cloexec ( self ) : <EOL> sleeper = support . findfile ( "<STR_LIT>" , subdir = "<STR_LIT>" ) <EOL> fd_status = support . findfile ( "<STR_LIT>" , subdir = "<STR_LIT>" ) <EOL> p1 = subprocess . Popen ( [ sys . executable , sleeper ] , <EOL> stdin = subprocess . PIPE , stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE , close_fds = False ) <EOL> self . addCleanup ( p1 . communicate , b'<STR_LIT>' ) <EOL> p2 = subprocess . Popen ( [ sys . executable , fd_status ] , <EOL> stdout = subprocess . PIPE , close_fds = False ) <EOL> output , error = p2 . communicate ( ) <EOL> result_fds = set ( map ( int , output . split ( b'<STR_LIT:U+002C>' ) ) ) <EOL> unwanted_fds = set ( [ p1 . stdin . fileno ( ) , p1 . stdout . fileno ( ) , <EOL> p1 . stderr . fileno ( ) ] ) <EOL> self . assertFalse ( result_fds & unwanted_fds , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( unwanted_fds , result_fds & unwanted_fds ) ) <EOL> def test_pipe_cloexec_real_tools ( self ) : <EOL> qcat = support . findfile ( "<STR_LIT>" , subdir = "<STR_LIT>" ) <EOL> qgrep = support . findfile ( "<STR_LIT>" , subdir = "<STR_LIT>" ) <EOL> subdata = b'<STR_LIT>' <EOL> data = subdata * <NUM_LIT:4> + b'<STR_LIT:\n>' <EOL> p1 = subprocess . Popen ( [ sys . executable , qcat ] , <EOL> stdin = subprocess . PIPE , stdout = subprocess . PIPE , <EOL> close_fds = False ) <EOL> p2 = subprocess . Popen ( [ sys . executable , qgrep , subdata ] , <EOL> stdin = p1 . stdout , stdout = subprocess . PIPE , <EOL> close_fds = False ) <EOL> self . addCleanup ( p1 . wait ) <EOL> self . addCleanup ( p2 . wait ) <EOL> def kill_p1 ( ) : <EOL> try : <EOL> p1 . terminate ( ) <EOL> except ProcessLookupError : <EOL> pass <EOL> def kill_p2 ( ) : <EOL> try : <EOL> p2 . terminate ( ) <EOL> except ProcessLookupError : <EOL> pass <EOL> self . addCleanup ( kill_p1 ) <EOL> self . addCleanup ( kill_p2 ) <EOL> p1 . stdin . write ( data ) <EOL> p1 . stdin . close ( ) <EOL> readfiles , ignored1 , ignored2 = select . select ( [ p2 . stdout ] , [ ] , [ ] , <NUM_LIT:10> ) <EOL> self . assertTrue ( readfiles , "<STR_LIT>" ) <EOL> self . assertEqual ( p2 . stdout . read ( ) , data ) <EOL> p1 . stdout . close ( ) <EOL> p2 . stdout . close ( ) <EOL> def test_close_fds ( self ) : <EOL> fd_status = support . findfile ( "<STR_LIT>" , subdir = "<STR_LIT>" ) <EOL> fds = os . pipe ( ) <EOL> self . addCleanup ( os . close , fds [ <NUM_LIT:0> ] ) <EOL> self . addCleanup ( os . close , fds [ <NUM_LIT:1> ] ) <EOL> open_fds = set ( fds ) <EOL> for _ in range ( <NUM_LIT:9> ) : <EOL> fd = os . open ( os . devnull , os . O_RDONLY ) <EOL> self . addCleanup ( os . close , fd ) <EOL> open_fds . add ( fd ) <EOL> for fd in open_fds : <EOL> os . set_inheritable ( fd , True ) <EOL> p = subprocess . Popen ( [ sys . executable , fd_status ] , <EOL> stdout = subprocess . PIPE , close_fds = False ) <EOL> output , ignored = p . communicate ( ) <EOL> remaining_fds = set ( map ( int , output . split ( b'<STR_LIT:U+002C>' ) ) ) <EOL> self . assertEqual ( remaining_fds & open_fds , open_fds , <EOL> "<STR_LIT>" ) <EOL> p = subprocess . Popen ( [ sys . executable , fd_status ] , <EOL> stdout = subprocess . PIPE , close_fds = True ) <EOL> output , ignored = p . communicate ( ) <EOL> remaining_fds = set ( map ( int , output . split ( b'<STR_LIT:U+002C>' ) ) ) <EOL> self . assertFalse ( remaining_fds & open_fds , <EOL> "<STR_LIT>" ) <EOL> self . assertIn ( <NUM_LIT:1> , remaining_fds , "<STR_LIT>" ) <EOL> fds_to_keep = set ( open_fds . pop ( ) for _ in range ( <NUM_LIT:8> ) ) <EOL> p = subprocess . Popen ( [ sys . executable , fd_status ] , <EOL> stdout = subprocess . PIPE , close_fds = True , <EOL> pass_fds = ( ) ) <EOL> output , ignored = p . communicate ( ) <EOL> remaining_fds = set ( map ( int , output . split ( b'<STR_LIT:U+002C>' ) ) ) <EOL> self . assertFalse ( remaining_fds & fds_to_keep & open_fds , <EOL> "<STR_LIT>" ) <EOL> self . assertIn ( <NUM_LIT:1> , remaining_fds , "<STR_LIT>" ) <EOL> @ unittest . skipIf ( sys . platform . startswith ( "<STR_LIT>" ) and <EOL> os . stat ( "<STR_LIT>" ) . st_dev == os . stat ( "<STR_LIT>" ) . st_dev , <EOL> "<STR_LIT>" ) <EOL> def test_close_fds_when_max_fd_is_lowered ( self ) : <EOL> """<STR_LIT>""" <EOL> fd_status = support . findfile ( "<STR_LIT>" , subdir = "<STR_LIT>" ) <EOL> p = subprocess . Popen ( [ sys . executable , '<STR_LIT:-c>' , textwrap . dedent ( <EOL> '''<STR_LIT>''' % fd_status ) ] , stdout = subprocess . PIPE ) <EOL> output , unused_stderr = p . communicate ( ) <EOL> output_lines = output . splitlines ( ) <EOL> self . assertEqual ( len ( output_lines ) , <NUM_LIT:2> , <EOL> msg = "<STR_LIT>" % output ) <EOL> opened_fds = set ( map ( int , output_lines [ <NUM_LIT:0> ] . strip ( ) . split ( b'<STR_LIT:U+002C>' ) ) ) <EOL> remaining_fds = set ( map ( int , output_lines [ <NUM_LIT:1> ] . strip ( ) . split ( b'<STR_LIT:U+002C>' ) ) ) <EOL> self . assertFalse ( remaining_fds & opened_fds , <EOL> msg = "<STR_LIT>" ) <EOL> @ support . requires_mac_ver ( <NUM_LIT:10> , <NUM_LIT:5> ) <EOL> def test_pass_fds ( self ) : <EOL> fd_status = support . findfile ( "<STR_LIT>" , subdir = "<STR_LIT>" ) <EOL> open_fds = set ( ) <EOL> for x in range ( <NUM_LIT:5> ) : <EOL> fds = os . pipe ( ) <EOL> self . addCleanup ( os . close , fds [ <NUM_LIT:0> ] ) <EOL> self . addCleanup ( os . close , fds [ <NUM_LIT:1> ] ) <EOL> os . set_inheritable ( fds [ <NUM_LIT:0> ] , True ) <EOL> os . set_inheritable ( fds [ <NUM_LIT:1> ] , True ) <EOL> open_fds . update ( fds ) <EOL> for fd in open_fds : <EOL> p = subprocess . Popen ( [ sys . executable , fd_status ] , <EOL> stdout = subprocess . PIPE , close_fds = True , <EOL> pass_fds = ( fd , ) ) <EOL> output , ignored = p . communicate ( ) <EOL> remaining_fds = set ( map ( int , output . split ( b'<STR_LIT:U+002C>' ) ) ) <EOL> to_be_closed = open_fds - { fd } <EOL> self . assertIn ( fd , remaining_fds , "<STR_LIT>" ) <EOL> self . assertFalse ( remaining_fds & to_be_closed , <EOL> "<STR_LIT>" ) <EOL> with self . assertWarns ( RuntimeWarning ) as context : <EOL> self . assertFalse ( subprocess . call ( <EOL> [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> close_fds = False , pass_fds = ( fd , ) ) ) <EOL> self . assertIn ( '<STR_LIT>' , str ( context . warning ) ) <EOL> def test_pass_fds_inheritable ( self ) : <EOL> script = support . findfile ( "<STR_LIT>" , subdir = "<STR_LIT>" ) <EOL> inheritable , non_inheritable = os . pipe ( ) <EOL> self . addCleanup ( os . close , inheritable ) <EOL> self . addCleanup ( os . close , non_inheritable ) <EOL> os . set_inheritable ( inheritable , True ) <EOL> os . set_inheritable ( non_inheritable , False ) <EOL> pass_fds = ( inheritable , non_inheritable ) <EOL> args = [ sys . executable , script ] <EOL> args += list ( map ( str , pass_fds ) ) <EOL> p = subprocess . Popen ( args , <EOL> stdout = subprocess . PIPE , close_fds = True , <EOL> pass_fds = pass_fds ) <EOL> output , ignored = p . communicate ( ) <EOL> fds = set ( map ( int , output . split ( b'<STR_LIT:U+002C>' ) ) ) <EOL> self . assertEqual ( fds , set ( pass_fds ) , "<STR_LIT>" % output ) <EOL> self . assertEqual ( os . get_inheritable ( inheritable ) , True ) <EOL> self . assertEqual ( os . get_inheritable ( non_inheritable ) , False ) <EOL> def test_stdout_stdin_are_single_inout_fd ( self ) : <EOL> with io . open ( os . devnull , "<STR_LIT>" ) as inout : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> stdout = inout , stdin = inout ) <EOL> p . wait ( ) <EOL> def test_stdout_stderr_are_single_inout_fd ( self ) : <EOL> with io . open ( os . devnull , "<STR_LIT>" ) as inout : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> stdout = inout , stderr = inout ) <EOL> p . wait ( ) <EOL> def test_stderr_stdin_are_single_inout_fd ( self ) : <EOL> with io . open ( os . devnull , "<STR_LIT>" ) as inout : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> stderr = inout , stdin = inout ) <EOL> p . wait ( ) <EOL> def test_wait_when_sigchild_ignored ( self ) : <EOL> sigchild_ignore = support . findfile ( "<STR_LIT>" , <EOL> subdir = "<STR_LIT>" ) <EOL> p = subprocess . Popen ( [ sys . executable , sigchild_ignore ] , <EOL> stdout = subprocess . PIPE , stderr = subprocess . PIPE ) <EOL> stdout , stderr = p . communicate ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , p . returncode , "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> stderr . decode ( '<STR_LIT:utf-8>' ) ) <EOL> def test_select_unbuffered ( self ) : <EOL> select = support . import_module ( "<STR_LIT>" ) <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> bufsize = <NUM_LIT:0> ) <EOL> f = p . stdout <EOL> self . addCleanup ( f . close ) <EOL> try : <EOL> self . assertEqual ( f . read ( <NUM_LIT:4> ) , b"<STR_LIT>" ) <EOL> self . assertIn ( f , select . select ( [ f ] , [ ] , [ ] , <NUM_LIT:0.0> ) [ <NUM_LIT:0> ] ) <EOL> finally : <EOL> p . wait ( ) <EOL> def test_zombie_fast_process_del ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> ident = id ( p ) <EOL> pid = p . pid <EOL> del p <EOL> self . assertIn ( ident , [ id ( o ) for o in subprocess . _active ] ) <EOL> def test_leak_fast_process_del_killed ( self ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> ident = id ( p ) <EOL> pid = p . pid <EOL> del p <EOL> os . kill ( pid , signal . SIGKILL ) <EOL> self . assertIn ( ident , [ id ( o ) for o in subprocess . _active ] ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> with self . assertRaises ( OSError ) as c : <EOL> with subprocess . Popen ( [ '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) as proc : <EOL> pass <EOL> self . assertRaises ( OSError , os . waitpid , pid , <NUM_LIT:0> ) <EOL> self . assertNotIn ( ident , [ id ( o ) for o in subprocess . _active ] ) <EOL> def test_close_fds_after_preexec ( self ) : <EOL> fd_status = support . findfile ( "<STR_LIT>" , subdir = "<STR_LIT>" ) <EOL> fd = os . dup ( <NUM_LIT:1> ) <EOL> self . addCleanup ( os . close , fd ) <EOL> p = subprocess . Popen ( [ sys . executable , fd_status ] , <EOL> stdout = subprocess . PIPE , close_fds = True , <EOL> preexec_fn = lambda : os . dup2 ( <NUM_LIT:1> , fd ) ) <EOL> output , ignored = p . communicate ( ) <EOL> remaining_fds = set ( map ( int , output . split ( b'<STR_LIT:U+002C>' ) ) ) <EOL> self . assertNotIn ( fd , remaining_fds ) <EOL> @ support . cpython_only <EOL> def test_fork_exec ( self ) : <EOL> import _posixsubprocess <EOL> gc_enabled = gc . isenabled ( ) <EOL> try : <EOL> func = lambda : None <EOL> gc . enable ( ) <EOL> executable_list = "<STR_LIT>" <EOL> for args , exe_list , cwd , env_list in ( <EOL> ( <NUM_LIT> , [ b"<STR_LIT>" ] , None , [ b"<STR_LIT>" ] ) , <EOL> ( [ b"<STR_LIT>" ] , <NUM_LIT> , None , [ b"<STR_LIT>" ] ) , <EOL> ( [ b"<STR_LIT>" ] , [ b"<STR_LIT>" ] , <NUM_LIT> , [ b"<STR_LIT>" ] ) , <EOL> ( [ b"<STR_LIT>" ] , [ b"<STR_LIT>" ] , None , <NUM_LIT> ) , <EOL> ) : <EOL> with self . assertRaises ( TypeError ) : <EOL> _posixsubprocess . fork_exec ( <EOL> args , exe_list , <EOL> True , [ ] , cwd , env_list , <EOL> - <NUM_LIT:1> , - <NUM_LIT:1> , - <NUM_LIT:1> , - <NUM_LIT:1> , <EOL> <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <EOL> True , True , func ) <EOL> finally : <EOL> if not gc_enabled : <EOL> gc . disable ( ) <EOL> @ unittest . skipUnless ( mswindows , "<STR_LIT>" ) <EOL> class Win32ProcessTestCase ( BaseTestCase ) : <EOL> def test_startupinfo ( self ) : <EOL> STARTF_USESHOWWINDOW = <NUM_LIT:1> <EOL> SW_MAXIMIZE = <NUM_LIT:3> <EOL> startupinfo = subprocess . STARTUPINFO ( ) <EOL> startupinfo . dwFlags = STARTF_USESHOWWINDOW <EOL> startupinfo . wShowWindow = SW_MAXIMIZE <EOL> subprocess . call ( [ sys . executable , "<STR_LIT:-c>" , "<STR_LIT>" ] , <EOL> startupinfo = startupinfo ) <EOL> def test_creationflags ( self ) : <EOL> CREATE_NEW_CONSOLE = <NUM_LIT:16> <EOL> sys . stderr . write ( "<STR_LIT>" ) <EOL> subprocess . call ( sys . executable + <EOL> '<STR_LIT>' , <EOL> creationflags = CREATE_NEW_CONSOLE ) <EOL> def test_invalid_args ( self ) : <EOL> self . assertRaises ( ValueError , subprocess . call , <EOL> [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] , <EOL> preexec_fn = lambda : <NUM_LIT:1> ) <EOL> self . assertRaises ( ValueError , subprocess . call , <EOL> [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] , <EOL> stdout = subprocess . PIPE , <EOL> close_fds = True ) <EOL> def test_close_fds ( self ) : <EOL> rc = subprocess . call ( [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] , <EOL> close_fds = True ) <EOL> self . assertEqual ( rc , <NUM_LIT> ) <EOL> def test_shell_sequence ( self ) : <EOL> newenv = os . environ . copy ( ) <EOL> newenv [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> p = subprocess . Popen ( [ "<STR_LIT>" ] , shell = <NUM_LIT:1> , <EOL> stdout = subprocess . PIPE , <EOL> env = newenv ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . assertIn ( b"<STR_LIT>" , p . stdout . read ( ) ) <EOL> def test_shell_string ( self ) : <EOL> newenv = os . environ . copy ( ) <EOL> newenv [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> p = subprocess . Popen ( "<STR_LIT>" , shell = <NUM_LIT:1> , <EOL> stdout = subprocess . PIPE , <EOL> env = newenv ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . assertIn ( b"<STR_LIT>" , p . stdout . read ( ) ) <EOL> def test_call_string ( self ) : <EOL> rc = subprocess . call ( sys . executable + <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( rc , <NUM_LIT> ) <EOL> def _kill_process ( self , method , * args ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , """<STR_LIT>""" ] , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> self . addCleanup ( p . stdin . close ) <EOL> p . stdout . read ( <NUM_LIT:1> ) <EOL> getattr ( p , method ) ( * args ) <EOL> _ , stderr = p . communicate ( ) <EOL> self . assertStderrEqual ( stderr , b'<STR_LIT>' ) <EOL> returncode = p . wait ( ) <EOL> self . assertNotEqual ( returncode , <NUM_LIT:0> ) <EOL> def _kill_dead_process ( self , method , * args ) : <EOL> p = subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , """<STR_LIT>""" ] , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . addCleanup ( p . stderr . close ) <EOL> self . addCleanup ( p . stdin . close ) <EOL> p . stdout . read ( <NUM_LIT:1> ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> getattr ( p , method ) ( * args ) <EOL> _ , stderr = p . communicate ( ) <EOL> self . assertStderrEqual ( stderr , b'<STR_LIT>' ) <EOL> rc = p . wait ( ) <EOL> self . assertEqual ( rc , <NUM_LIT> ) <EOL> def test_send_signal ( self ) : <EOL> self . _kill_process ( '<STR_LIT>' , signal . SIGTERM ) <EOL> def test_kill ( self ) : <EOL> self . _kill_process ( '<STR_LIT>' ) <EOL> def test_terminate ( self ) : <EOL> self . _kill_process ( '<STR_LIT>' ) <EOL> def test_send_signal_dead ( self ) : <EOL> self . _kill_dead_process ( '<STR_LIT>' , signal . SIGTERM ) <EOL> def test_kill_dead ( self ) : <EOL> self . _kill_dead_process ( '<STR_LIT>' ) <EOL> def test_terminate_dead ( self ) : <EOL> self . _kill_dead_process ( '<STR_LIT>' ) <EOL> class CommandTests ( unittest . TestCase ) : <EOL> def test_getoutput ( self ) : <EOL> self . assertEqual ( subprocess . getoutput ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( subprocess . getstatusoutput ( '<STR_LIT>' ) , <EOL> ( <NUM_LIT:0> , '<STR_LIT>' ) ) <EOL> dir = None <EOL> try : <EOL> dir = tempfile . mkdtemp ( ) <EOL> name = os . path . join ( dir , "<STR_LIT:foo>" ) <EOL> status , output = subprocess . getstatusoutput ( <EOL> ( "<STR_LIT>" if mswindows else "<STR_LIT>" ) + name ) <EOL> self . assertNotEqual ( status , <NUM_LIT:0> ) <EOL> finally : <EOL> if dir is not None : <EOL> os . rmdir ( dir ) <EOL> @ unittest . skipUnless ( hasattr ( selectors , '<STR_LIT>' ) , <EOL> "<STR_LIT>" ) <EOL> class ProcessTestCaseNoPoll ( ProcessTestCase ) : <EOL> def setUp ( self ) : <EOL> self . orig_selector = subprocess . _PopenSelector <EOL> subprocess . _PopenSelector = selectors . SelectSelector <EOL> ProcessTestCase . setUp ( self ) <EOL> def tearDown ( self ) : <EOL> subprocess . _PopenSelector = self . orig_selector <EOL> ProcessTestCase . tearDown ( self ) <EOL> class HelperFunctionTests ( unittest . TestCase ) : <EOL> @ unittest . skipIf ( mswindows , "<STR_LIT>" ) <EOL> def test_eintr_retry_call ( self ) : <EOL> record_calls = [ ] <EOL> def fake_os_func ( * args ) : <EOL> record_calls . append ( args ) <EOL> if len ( record_calls ) == <NUM_LIT:2> : <EOL> raise OSError ( errno . EINTR , "<STR_LIT>" ) <EOL> return tuple ( reversed ( args ) ) <EOL> self . assertEqual ( ( <NUM_LIT> , <NUM_LIT> ) , <EOL> subprocess . _eintr_retry_call ( fake_os_func , <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . assertEqual ( [ ( <NUM_LIT> , <NUM_LIT> ) ] , record_calls ) <EOL> self . assertEqual ( ( <NUM_LIT> , ) , <EOL> subprocess . _eintr_retry_call ( fake_os_func , <NUM_LIT> ) ) <EOL> self . assertEqual ( [ ( <NUM_LIT> , <NUM_LIT> ) , ( <NUM_LIT> , ) , ( <NUM_LIT> , ) ] , record_calls ) <EOL> @ unittest . skipUnless ( mswindows , "<STR_LIT>" ) <EOL> class CommandsWithSpaces ( BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ) . setUp ( ) <EOL> f , fname = mkstemp ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . fname = fname . lower ( ) <EOL> os . write ( f , b"<STR_LIT>" <EOL> b"<STR_LIT>" <EOL> ) <EOL> os . close ( f ) <EOL> def tearDown ( self ) : <EOL> os . remove ( self . fname ) <EOL> super ( ) . tearDown ( ) <EOL> def with_spaces ( self , * args , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = subprocess . PIPE <EOL> p = subprocess . Popen ( * args , ** kwargs ) <EOL> self . addCleanup ( p . stdout . close ) <EOL> self . assertEqual ( <EOL> p . stdout . read ( ) . decode ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" % self . fname <EOL> ) <EOL> def test_shell_string_with_spaces ( self ) : <EOL> self . with_spaces ( '<STR_LIT>' % ( sys . executable , self . fname , <EOL> "<STR_LIT>" ) , shell = <NUM_LIT:1> ) <EOL> def test_shell_sequence_with_spaces ( self ) : <EOL> self . with_spaces ( [ sys . executable , self . fname , "<STR_LIT>" ] , shell = <NUM_LIT:1> ) <EOL> def test_noshell_string_with_spaces ( self ) : <EOL> self . with_spaces ( '<STR_LIT>' % ( sys . executable , self . fname , <EOL> "<STR_LIT>" ) ) <EOL> def test_noshell_sequence_with_spaces ( self ) : <EOL> self . with_spaces ( [ sys . executable , self . fname , "<STR_LIT>" ] ) <EOL> class ContextManagerTests ( BaseTestCase ) : <EOL> def test_pipe ( self ) : <EOL> with subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) as proc : <EOL> self . assertEqual ( proc . stdout . read ( ) , b"<STR_LIT>" ) <EOL> self . assertStderrEqual ( proc . stderr . read ( ) , b"<STR_LIT>" ) <EOL> self . assertTrue ( proc . stdout . closed ) <EOL> self . assertTrue ( proc . stderr . closed ) <EOL> def test_returncode ( self ) : <EOL> with subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" ] ) as proc : <EOL> pass <EOL> self . assertEqual ( proc . returncode , <NUM_LIT:100> ) <EOL> def test_communicate_stdin ( self ) : <EOL> with subprocess . Popen ( [ sys . executable , "<STR_LIT:-c>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] , <EOL> stdin = subprocess . PIPE ) as proc : <EOL> proc . communicate ( b"<STR_LIT>" ) <EOL> self . assertEqual ( proc . returncode , <NUM_LIT:1> ) <EOL> def test_invalid_args ( self ) : <EOL> with self . assertRaises ( FileNotFoundError ) as c : <EOL> with subprocess . Popen ( [ '<STR_LIT>' ] , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE ) as proc : <EOL> pass <EOL> def test_broken_pipe_cleanup ( self ) : <EOL> """<STR_LIT>""" <EOL> proc = subprocess . Popen ( [ sys . executable , '<STR_LIT:-c>' , '<STR_LIT>' ] , <EOL> stdin = subprocess . PIPE , <EOL> bufsize = support . PIPE_MAX_SIZE * <NUM_LIT:2> ) <EOL> proc = proc . __enter__ ( ) <EOL> proc . stdin . write ( b'<STR_LIT:x>' * support . PIPE_MAX_SIZE ) <EOL> self . assertIsNone ( proc . returncode ) <EOL> self . assertRaises ( OSError , proc . __exit__ , None , None , None ) <EOL> self . assertEqual ( proc . returncode , <NUM_LIT:0> ) <EOL> self . assertTrue ( proc . stdin . closed ) <EOL> def test_main ( ) : <EOL> unit_tests = ( ProcessTestCase , <EOL> POSIXProcessTestCase , <EOL> Win32ProcessTestCase , <EOL> CommandTests , <EOL> ProcessTestCaseNoPoll , <EOL> HelperFunctionTests , <EOL> CommandsWithSpaces , <EOL> ContextManagerTests , <EOL> ) <EOL> support . run_unittest ( * unit_tests ) <EOL> support . reap_children ( ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
