| #!/usr/bin/env vpython3 |
| # Copyright 2013 The LUCI Authors. All rights reserved. |
| # Use of this source code is governed under the Apache License, Version 2.0 |
| # that can be found in the LICENSE file. |
| |
| from __future__ import print_function |
| |
| import ctypes |
| import errno |
| import itertools |
| import os |
| import platform |
| import signal |
| import sys |
| import tempfile |
| import textwrap |
| import time |
| import unittest |
| |
| from nose2.tools import params |
| |
| # Mutates sys.path. |
| import test_env |
| |
| from utils import subprocess42 |
| |
| # Disable pre-set unbuffered output to not interfere with the testing being done |
| # here. Otherwise everything would test with unbuffered; which is fine but |
| # that's not what we specifically want to test here. |
| ENV = os.environ.copy() |
| ENV.pop('PYTHONUNBUFFERED', None) |
| |
| SCRIPT_OUT = ('import signal, sys, time;\n' |
| 'l = [];\n' |
| 'def handler(signum, _):\n' |
| ' l.append(signum);\n' |
| ' sys.stdout.write(\'got signal %%d\\n\' %% signum);\n' |
| ' sys.stdout.flush();\n' |
| 'signal.signal(%s, handler);\n' |
| 'sys.stdout.write(\'hi\\n\');\n' |
| 'sys.stdout.flush();\n' |
| 'while not l:\n' |
| ' try:\n' |
| ' time.sleep(0.01);\n' |
| ' except IOError:\n' |
| ' sys.stdout.write(\'ioerror\\n\');\n' |
| ' sys.stdout.flush();\n' |
| 'sys.stdout.write(\'bye\\n\');\n' |
| 'sys.stdout.flush();\n') % ('signal.SIGBREAK' if sys.platform == |
| 'win32' else 'signal.SIGTERM') |
| |
| SCRIPT_ERR = ('import signal, sys, time;\n' |
| 'l = [];\n' |
| 'def handler(signum, _):\n' |
| ' l.append(signum);\n' |
| ' sys.stderr.write(\'got signal %%d\\n\' %% signum);\n' |
| ' sys.stderr.flush();\n' |
| 'signal.signal(%s, handler);\n' |
| 'sys.stderr.write(\'hi\\n\');\n' |
| 'sys.stderr.flush();\n' |
| 'while not l:\n' |
| ' try:\n' |
| ' time.sleep(0.01);\n' |
| ' except IOError:\n' |
| ' sys.stderr.write(\'ioerror\\n\');\n' |
| ' sys.stderr.flush();\n' |
| 'sys.stderr.write(\'bye\\n\');\n' |
| 'sys.stderr.flush();\n') % ('signal.SIGBREAK' if sys.platform == |
| 'win32' else 'signal.SIGTERM') |
| |
| OUTPUT_SCRIPT = br""" |
| import os |
| import re |
| import sys |
| import time |
| |
| def main(): |
| try: |
| for command in sys.argv[1:]: |
| if re.match(r'^[0-9\.]+$', command): |
| time.sleep(float(command)) |
| continue |
| |
| if command.startswith('out_'): |
| pipe, other = sys.stdout, sys.stderr |
| elif command.startswith('err_'): |
| pipe, other = sys.stderr, sys.stdout |
| else: |
| return 1 |
| |
| command = command[4:] |
| if command == 'print': |
| pipe.write('printing') |
| elif command == 'sleeping': |
| pipe.write('Sleeping.\n') |
| elif command == 'slept': |
| pipe.write('Slept.\n') |
| elif command == 'lf': |
| pipe.write('\n') |
| elif command == 'flush': |
| pipe.flush() |
| elif command == 'leak': |
| pid = os.fork() |
| if pid > 0: |
| return 0 |
| |
| other.write("leaked child is %s %s\n" % (os.getpid(), os.getpgid(0))) |
| other.write("sleeping\n") |
| time.sleep(30) |
| other.write("woke up\n") |
| return 1 |
| else: |
| return 1 |
| return 0 |
| except OSError: |
| return 0 |
| |
| |
| if __name__ == '__main__': |
| sys.exit(main()) |
| """ |
| |
| def to_native_eol(string): |
| if string is None: |
| return string |
| if sys.platform == 'win32': |
| return string.replace(b'\n', b'\r\n') |
| return string |
| |
| |
| def get_output_sleep_proc(flush, unbuffered, sleep_duration): |
| """Returns process with universal_newlines=True that prints to stdout before |
| after a sleep. |
| |
| It also optionally sys.stdout.flush() before the sleep and optionally enable |
| unbuffered output in python. |
| """ |
| command = [ |
| 'import sys,time', |
| 'print(\'A\')', |
| ] |
| if flush: |
| # Sadly, this doesn't work otherwise in some combination. |
| command.append('sys.stdout.flush()') |
| command.extend(( |
| 'time.sleep(%s)' % sleep_duration, |
| 'print(\'B\')', |
| )) |
| cmd = [sys.executable, '-c', ';'.join(command)] |
| if unbuffered: |
| cmd.append('-u') |
| return subprocess42.Popen( |
| cmd, env=ENV, stdout=subprocess42.PIPE, universal_newlines=True) |
| |
| |
| def get_output_sleep_proc_err(sleep_duration): |
| """Returns process with universal_newlines=True that prints to stderr before |
| and after a sleep. |
| """ |
| command = [ |
| 'import sys,time', |
| 'sys.stderr.write(\'A\\n\')', |
| ] |
| command.extend(( |
| 'time.sleep(%s)' % sleep_duration, |
| 'sys.stderr.write(\'B\\n\')', |
| )) |
| cmd = [sys.executable, '-c', ';'.join(command)] |
| return subprocess42.Popen( |
| cmd, env=ENV, stderr=subprocess42.PIPE, universal_newlines=True) |
| |
| |
| class Subprocess42Test(unittest.TestCase): |
| |
| def setUp(self): |
| self._output_script = None |
| super(Subprocess42Test, self).setUp() |
| |
| def tearDown(self): |
| try: |
| if self._output_script: |
| os.remove(self._output_script) |
| finally: |
| super(Subprocess42Test, self).tearDown() |
| |
| @property |
| def output_script(self): |
| if not self._output_script: |
| handle, self._output_script = tempfile.mkstemp( |
| prefix='subprocess42', suffix='.py') |
| os.write(handle, OUTPUT_SCRIPT) |
| os.close(handle) |
| return self._output_script |
| |
| def test_communicate_timeout(self): |
| timedout = 1 if sys.platform == 'win32' else -9 |
| # Format is: |
| # ( (cmd, stderr_pipe, timeout), (stdout, stderr, returncode) ), ... |
| # See OUTPUT script for the meaning of the commands. |
| test_data = [ |
| # 0 means no timeout, like None. |
| ( |
| (['out_sleeping', '0.001', 'out_slept', 'err_print'], None, 0), |
| (b'Sleeping.\nSlept.\n', None, 0), |
| ), |
| ( |
| (['err_print'], subprocess42.STDOUT, 0), |
| (b'printing', None, 0), |
| ), |
| ( |
| (['err_print'], subprocess42.PIPE, 0), |
| (b'', b'printing', 0), |
| ), |
| |
| # On a loaded system, this can be tight. |
| ( |
| (['out_sleeping', 'out_flush', '60', 'out_slept'], None, 1), |
| (b'Sleeping.\n', None, timedout), |
| ), |
| ( |
| ( |
| # Note that err_flush is necessary on Windows but not on the |
| # other OSes. This means the likelihood of missing stderr output |
| # from a killed child process on Windows is much higher than on |
| # other OSes. |
| [ |
| 'out_sleeping', |
| 'out_flush', |
| 'err_print', |
| 'err_flush', |
| '60', |
| 'out_slept', |
| ], |
| subprocess42.PIPE, |
| 1), |
| (b'Sleeping.\n', b'printing', timedout), |
| ), |
| ( |
| (['out_sleeping', '0.001', 'out_slept'], None, 60), |
| (b'Sleeping.\nSlept.\n', None, 0), |
| ), |
| ( |
| ([], None, 60), |
| (b'', None, 0), |
| ), |
| ( |
| ([], subprocess42.PIPE, 60), |
| (b'', b'', 0), |
| ), |
| ] |
| for i, ((args, errpipe, timeout), expected) in enumerate(test_data): |
| proc = subprocess42.Popen( |
| [sys.executable, self.output_script] + args, |
| env=ENV, |
| stdout=subprocess42.PIPE, |
| stderr=errpipe) |
| try: |
| stdout, stderr = proc.communicate(timeout=timeout) |
| code = proc.returncode |
| except subprocess42.TimeoutExpired as e: |
| stdout = e.output |
| stderr = e.stderr |
| self.assertTrue(proc.kill()) |
| code = proc.wait() |
| finally: |
| duration = proc.duration() |
| expected_duration = 0.0001 if not timeout or timeout == 60 else timeout |
| self.assertTrue(duration >= expected_duration, (i, expected_duration)) |
| self.assertEqual((i, stdout, stderr, code), (i, to_native_eol( |
| expected[0]), to_native_eol(expected[1]), expected[2])) |
| |
| # Try again with universal_newlines=True. |
| proc = subprocess42.Popen( |
| [sys.executable, self.output_script] + args, |
| env=ENV, |
| stdout=subprocess42.PIPE, |
| stderr=errpipe, |
| universal_newlines=True) |
| try: |
| stdout, stderr = proc.communicate(timeout=timeout) |
| code = proc.returncode |
| except subprocess42.TimeoutExpired as e: |
| # With communicate() in the native subprocess.py, output/stderr becomes |
| # bytes even if universal_newlines = True in Python3. |
| # They are str on Windows because the communicate() in subprocess42.py |
| # is used. |
| stdout = e.output |
| stderr = e.stderr |
| if sys.platform != 'win32': |
| stdout = stdout.decode() if stdout else None |
| stderr = stderr.decode() if stderr else None |
| self.assertTrue(proc.kill()) |
| code = proc.wait() |
| finally: |
| duration = proc.duration() |
| self.assertTrue(duration >= expected_duration, (i, expected_duration)) |
| self.assertEqual((i, None if stdout is None else stdout.encode(), |
| None if stderr is None else stderr.encode(), code), |
| (i,) + expected) |
| |
| def test_communicate_input(self): |
| cmd = [ |
| sys.executable, |
| '-u', |
| '-c', |
| 'import sys; sys.stdout.write(sys.stdin.read(5))', |
| ] |
| proc = subprocess42.Popen( |
| cmd, stdin=subprocess42.PIPE, stdout=subprocess42.PIPE) |
| out, err = proc.communicate(input=b'12345') |
| self.assertEqual(b'12345', out) |
| self.assertEqual(None, err) |
| |
| def test_communicate_input_timeout(self): |
| cmd = [sys.executable, '-u', '-c', 'import time; time.sleep(60)'] |
| proc = subprocess42.Popen(cmd, stdin=subprocess42.PIPE) |
| try: |
| proc.communicate(input=b'12345', timeout=0.5) |
| self.fail() |
| except subprocess42.TimeoutExpired as e: |
| self.assertEqual(None, e.output) |
| self.assertEqual(None, e.stderr) |
| self.assertTrue(proc.kill()) |
| proc.wait() |
| self.assertLessEqual(0.5, proc.duration()) |
| |
| def test_communicate_input_stdout_timeout(self): |
| cmd = [ |
| sys.executable, |
| '-u', |
| '-c', |
| """ |
| import sys, time |
| sys.stdout.write(sys.stdin.read(5)) |
| sys.stdout.flush() |
| time.sleep(60) |
| """, |
| ] |
| proc = subprocess42.Popen( |
| cmd, stdin=subprocess42.PIPE, stdout=subprocess42.PIPE) |
| try: |
| proc.communicate(input=b'12345', timeout=2) |
| self.fail() |
| except subprocess42.TimeoutExpired as e: |
| self.assertEqual(b'12345', e.output) |
| self.assertEqual(None, e.stderr) |
| self.assertTrue(proc.kill()) |
| proc.wait() |
| self.assertLessEqual(0.5, proc.duration()) |
| |
| def test_communicate_timeout_no_pipe(self): |
| # In this case, it's effectively a wait() call. |
| cmd = [sys.executable, '-u', '-c', 'import time; time.sleep(60)'] |
| proc = subprocess42.Popen(cmd) |
| try: |
| proc.communicate(timeout=0.5) |
| self.fail() |
| except subprocess42.TimeoutExpired as e: |
| self.assertEqual(None, e.output) |
| self.assertEqual(None, e.stderr) |
| self.assertTrue(proc.kill()) |
| proc.wait() |
| self.assertLessEqual(0.5, proc.duration()) |
| |
| def _test_lower_priority(self, lower_priority): |
| if sys.platform == 'win32': |
| cmd = [ |
| sys.executable, '-u', '-c', |
| 'import ctypes,sys; v=ctypes.windll.kernel32.GetPriorityClass(-1);' |
| 'sys.stdout.write(hex(v))' |
| ] |
| else: |
| cmd = [ |
| sys.executable, |
| '-u', |
| '-c', |
| 'import os,sys;sys.stdout.write(str(os.nice(0)))', |
| ] |
| proc = subprocess42.Popen( |
| cmd, stdout=subprocess42.PIPE, lower_priority=lower_priority) |
| out, err = proc.communicate() |
| self.assertEqual(None, err) |
| return out |
| |
| @unittest.skipIf(sys.platform == 'win32', 'crbug.com/1148174') |
| def test_lower_priority(self): |
| out = self._test_lower_priority(True) |
| if sys.platform == 'win32': |
| # See |
| # https://docs.microsoft.com/en-us/windows/desktop/api/processthreadsapi/nf-processthreadsapi-getpriorityclass |
| BELOW_NORMAL_PRIORITY_CLASS = 0x4000 |
| self.assertEqual(hex(BELOW_NORMAL_PRIORITY_CLASS), out) |
| else: |
| self.assertEqual(str(os.nice(0) + 1).encode(), out) |
| |
| def test_lower_priority_False(self): |
| out = self._test_lower_priority(False) |
| if sys.platform == 'win32': |
| # Should be NORMAL_PRIORITY_CLASS. |
| p = ctypes.windll.kernel32.GetPriorityClass(-1) |
| self.assertEqual(hex(p).encode(), out) |
| else: |
| self.assertEqual(str(os.nice(0)).encode(), out) |
| |
| @staticmethod |
| def _cmd_print_good(): |
| # Used in test_containment_auto and test_containment_auto_limit_process. |
| return [ |
| sys.executable, |
| '-u', |
| '-c', |
| 'import subprocess,sys; ' |
| 'subprocess.call([sys.executable, "-c", "print(\\"good\\")"])', |
| ] |
| |
| def test_containment_none(self): |
| # Minimal test case. Starts two processes. |
| cmd = self._cmd_print_good() |
| containment = subprocess42.Containment( |
| containment_type=subprocess42.Containment.NONE) |
| self.assertEqual(0, subprocess42.check_call(cmd, containment=containment)) |
| |
| def test_containment_auto(self): |
| # Minimal test case. Starts two processes. |
| cmd = self._cmd_print_good() |
| containment = subprocess42.Containment( |
| containment_type=subprocess42.Containment.AUTO, |
| limit_processes=2, |
| limit_total_committed_memory=1024 * 1024 * 1024) |
| self.assertEqual(0, subprocess42.check_call(cmd, containment=containment)) |
| |
| def test_containment_auto_limit_process(self): |
| # Process creates a children process. It should fail, throwing not enough |
| # quota. |
| cmd = self._cmd_print_good() |
| containment = subprocess42.Containment( |
| containment_type=subprocess42.Containment.JOB_OBJECT, limit_processes=1) |
| start = lambda: subprocess42.Popen( |
| cmd, |
| stdout=subprocess42.PIPE, |
| stderr=subprocess42.PIPE, |
| containment=containment) |
| |
| if sys.platform == 'win32': |
| p = start() |
| out, err = p.communicate() |
| self.assertEqual(1, p.returncode) |
| self.assertEqual(b'', out) |
| self.assertIn(b'WinError', err) |
| # Value for ERROR_NOT_ENOUGH_QUOTA. See |
| # https://docs.microsoft.com/windows/desktop/debug/system-error-codes--1700-3999- |
| self.assertIn(b'1816', err) |
| else: |
| # JOB_OBJECT is not usable on non-Windows. |
| with self.assertRaises(NotImplementedError): |
| start() |
| |
| def test_containment_auto_kill(self): |
| # Test process killing. |
| cmd = [ |
| sys.executable, |
| '-u', |
| '-c', |
| 'import sys,time; print("hi");time.sleep(60)', |
| ] |
| containment = subprocess42.Containment( |
| containment_type=subprocess42.Containment.AUTO, |
| limit_processes=1, |
| limit_total_committed_memory=1024 * 1024 * 1024) |
| p = subprocess42.Popen( |
| cmd, stdout=subprocess42.PIPE, containment=containment) |
| itr = p.yield_any_line() |
| self.assertEqual(('stdout', b'hi'), next(itr)) |
| p.kill() |
| p.wait() |
| if sys.platform != 'win32': |
| # signal.SIGKILL is not defined on Windows. Validate our assumption here. |
| self.assertEqual(9, signal.SIGKILL) |
| if sys.platform == 'win32': |
| # p.returncode is unsigned in python3 on windows |
| self.assertEqual(4294967287, p.returncode) |
| else: |
| self.assertEqual(-9, p.returncode) |
| |
| @unittest.skipIf(sys.platform == 'win32', 'pgid test') |
| def test_kill_background(self): |
| # Test process group killing. |
| |
| # Leaking a pipe through to the grandchild process is the only way to be |
| # sure that we have a way to detect that this grandchild process is still |
| # running or not. When all handles to the `w` end of the pipe have been |
| # dropped, the `r` end will unblock. |
| r, w = os.pipe() |
| |
| # Use fcntl instead of os.set_blocking because of python2 |
| import fcntl |
| fcntl.fcntl(r, fcntl.F_SETFL, fcntl.fcntl(r, fcntl.F_GETFL) | os.O_NONBLOCK) |
| |
| p = subprocess42.Popen( |
| [sys.executable, self.output_script, 'out_leak'], |
| stdout=w, detached=True) |
| os.close(w) # close so we don't think that a child is hanging onto it. |
| |
| self.assertEqual(p.wait(), 0) # our immediate child has exited! |
| |
| with self.assertRaises(OSError): |
| # oops, something still has a handle to this pipe! it's the grandchild! |
| os.read(r, 1) |
| |
| # kill the group! That'll show 'em (unless the child actually daemonized, in |
| # which case we're hosed). |
| p.kill() |
| |
| # sleepy-loop until the pipe is closed, should take O(ms) but we generously |
| # wait up to 5s. The sub-child will wait 30s and should outlive this loop if |
| # somehow it survived the kill. |
| now = time.time() |
| while True: |
| try: |
| self.assertEqual(os.read(r, 1), b'') # i.e. EOF |
| return |
| except OSError as ex: |
| if ex.errno != errno.EWOULDBLOCK: |
| raise |
| time.sleep(0.1) |
| if time.time() - now > 5: |
| raise Exception('pipe not unblocked after 5s, bailing') |
| |
| @staticmethod |
| def _cmd_large_memory(): |
| # Used in test_large_memory and test_containment_auto_limit_memory. |
| return [ |
| sys.executable, |
| '-u', |
| '-c', |
| 'list(range(50*1024*1024)); print("hi")', |
| ] |
| |
| def test_large_memory(self): |
| # Just assert the process works normally. |
| cmd = self._cmd_large_memory() |
| self.assertEqual(b'hi', subprocess42.check_output(cmd).strip()) |
| |
| def test_containment_auto_limit_memory(self): |
| # Process allocates a lot of memory. It should fail due to quota. |
| cmd = self._cmd_large_memory() |
| containment = subprocess42.Containment( |
| containment_type=subprocess42.Containment.JOB_OBJECT, |
| # 20 MiB. |
| limit_total_committed_memory=20 * 1024 * 1024) |
| start = lambda: subprocess42.Popen( |
| cmd, |
| stdout=subprocess42.PIPE, |
| stderr=subprocess42.PIPE, |
| containment=containment) |
| |
| if sys.platform == 'win32': |
| p = start() |
| out, err = p.communicate() |
| self.assertEqual(1, p.returncode) |
| self.assertEqual(b'', out) |
| self.assertIn(b'MemoryError', err) |
| else: |
| # JOB_OBJECT is not usable on non-Windows. |
| with self.assertRaises(NotImplementedError): |
| start() |
| |
| def test_call(self): |
| cmd = [sys.executable, '-u', '-c', 'import sys; sys.exit(0)'] |
| self.assertEqual(0, subprocess42.call(cmd)) |
| |
| cmd = [sys.executable, '-u', '-c', 'import sys; sys.exit(1)'] |
| self.assertEqual(1, subprocess42.call(cmd)) |
| |
| def test_check_call(self): |
| cmd = [sys.executable, '-u', '-c', 'import sys; sys.exit(0)'] |
| self.assertEqual(0, subprocess42.check_call(cmd)) |
| |
| cmd = [sys.executable, '-u', '-c', 'import sys; sys.exit(1)'] |
| try: |
| self.assertEqual(1, subprocess42.check_call(cmd)) |
| self.fail() |
| except subprocess42.CalledProcessError as e: |
| self.assertEqual(None, e.output) |
| |
| def test_check_output(self): |
| cmd = [sys.executable, '-u', '-c', 'print(\'.\')'] |
| self.assertEqual('.\n', |
| subprocess42.check_output(cmd, universal_newlines=True)) |
| |
| cmd = [sys.executable, '-u', '-c', 'import sys; print(\'.\'); sys.exit(1)'] |
| try: |
| subprocess42.check_output(cmd, universal_newlines=True) |
| self.fail() |
| except subprocess42.CalledProcessError as e: |
| self.assertEqual('.\n', e.output) |
| |
| def test_recv_any(self): |
| # Test all pipe direction and output scenarios. |
| combinations = [ |
| { |
| 'cmd': ['out_print', 'err_print'], |
| 'stdout': None, |
| 'stderr': None, |
| 'expected': {}, |
| 'universal_newlines': False, |
| }, |
| { |
| 'cmd': ['out_print', 'err_print'], |
| 'stdout': None, |
| 'stderr': subprocess42.STDOUT, |
| 'universal_newlines': False, |
| 'expected': {}, |
| }, |
| { |
| 'cmd': ['out_print'], |
| 'stdout': subprocess42.PIPE, |
| 'stderr': subprocess42.PIPE, |
| 'universal_newlines': False, |
| 'expected': { |
| 'stdout': b'printing' |
| }, |
| }, |
| { |
| 'cmd': ['out_print'], |
| 'stdout': subprocess42.PIPE, |
| 'stderr': None, |
| 'universal_newlines': False, |
| 'expected': { |
| 'stdout': b'printing' |
| }, |
| }, |
| { |
| 'cmd': ['out_print'], |
| 'stdout': subprocess42.PIPE, |
| 'stderr': subprocess42.STDOUT, |
| 'universal_newlines': False, |
| 'expected': { |
| 'stdout': b'printing' |
| }, |
| }, |
| { |
| 'cmd': ['err_print'], |
| 'stdout': subprocess42.PIPE, |
| 'stderr': subprocess42.PIPE, |
| 'universal_newlines': False, |
| 'expected': { |
| 'stderr': b'printing' |
| }, |
| }, |
| { |
| 'cmd': ['err_print'], |
| 'stdout': None, |
| 'stderr': subprocess42.PIPE, |
| 'universal_newlines': False, |
| 'expected': { |
| 'stderr': b'printing' |
| }, |
| }, |
| { |
| 'cmd': ['err_print'], |
| 'stdout': subprocess42.PIPE, |
| 'stderr': subprocess42.STDOUT, |
| 'universal_newlines': False, |
| 'expected': { |
| 'stdout': b'printing' |
| }, |
| }, |
| { |
| 'cmd': ['out_print', 'err_print'], |
| 'stdout': subprocess42.PIPE, |
| 'stderr': subprocess42.PIPE, |
| 'universal_newlines': False, |
| 'expected': { |
| 'stderr': b'printing', |
| 'stdout': b'printing' |
| }, |
| }, |
| { |
| 'cmd': ['out_print', 'err_print'], |
| 'stdout': subprocess42.PIPE, |
| 'stderr': subprocess42.STDOUT, |
| 'universal_newlines': False, |
| 'expected': { |
| 'stdout': b'printingprinting' |
| }, |
| }, |
| { |
| 'cmd': ['out_print', 'err_print'], |
| 'stdout': subprocess42.PIPE, |
| 'stderr': subprocess42.PIPE, |
| 'universal_newlines': True, |
| 'expected': { |
| 'stderr': 'printing', |
| 'stdout': 'printing' |
| }, |
| }, |
| ] |
| for i, testcase in enumerate(combinations): |
| default = '' if testcase['universal_newlines'] else b'' |
| cmd = [sys.executable, self.output_script] + testcase['cmd'] |
| p = subprocess42.Popen( |
| cmd, |
| env=ENV, |
| stdout=testcase['stdout'], |
| stderr=testcase['stderr'], |
| universal_newlines=testcase['universal_newlines']) |
| actual = {} |
| while p.poll() is None: |
| pipe, data = p.recv_any() |
| if data: |
| actual.setdefault(pipe, default) |
| actual[pipe] += data |
| |
| # The process exited, read any remaining data in the pipes. |
| while True: |
| pipe, data = p.recv_any() |
| if pipe is None: |
| break |
| actual.setdefault(pipe, default) |
| actual[pipe] += data |
| self.assertEqual(testcase['expected'], actual, |
| (i, testcase['cmd'], testcase['expected'], actual)) |
| self.assertEqual((None, None), p.recv_any()) |
| self.assertEqual(0, p.returncode) |
| |
| def test_recv_any_different_buffering(self): |
| # Specifically test all buffering scenarios. |
| for flush, unbuffered in itertools.product([True, False], [True, False]): |
| actual = '' |
| proc = get_output_sleep_proc(flush, unbuffered, 0.5) |
| while True: |
| p, data = proc.recv_any() |
| if not p: |
| break |
| self.assertEqual('stdout', p) |
| self.assertTrue(data, (p, data)) |
| actual += data |
| |
| self.assertEqual('A\nB\n', actual) |
| # Contrary to yield_any() or recv_any(0), wait() needs to be used here. |
| proc.wait() |
| self.assertEqual(0, proc.returncode) |
| |
| def test_recv_any_timeout_0(self): |
| self._test_recv_any_timeout(False, False) |
| self._test_recv_any_timeout(False, True) |
| self._test_recv_any_timeout(True, False) |
| self._test_recv_any_timeout(True, True) |
| |
| def _test_recv_any_timeout(self, flush, unbuffered): |
| # rec_any() is expected to timeout and return None with no data pending at |
| # least once, due to the sleep of 'duration' and the use of timeout=0. |
| for duration in (0.05, 0.1, 0.5, 2): |
| got_none = False |
| actual = '' |
| try: |
| proc = get_output_sleep_proc(flush, unbuffered, duration) |
| try: |
| while True: |
| p, data = proc.recv_any(timeout=0) |
| if p: |
| self.assertEqual('stdout', p) |
| self.assertTrue(data, (p, data)) |
| actual += data |
| continue |
| if proc.poll() is None: |
| got_none = True |
| continue |
| break |
| |
| self.assertEqual('A\nB\n', actual) |
| self.assertEqual(0, proc.returncode) |
| self.assertEqual(True, got_none) |
| break |
| finally: |
| proc.kill() |
| proc.wait() |
| except AssertionError: |
| if duration != 2: |
| print('Sleeping rocks. Trying slower.') |
| continue |
| raise |
| |
| def test_yield_any_no_timeout(self): |
| for duration in (0.05, 0.1, 0.5, 2): |
| try: |
| proc = get_output_sleep_proc(True, True, duration) |
| try: |
| expected = [ |
| 'A\n', |
| 'B\n', |
| ] |
| for p, data in proc.yield_any(): |
| self.assertEqual('stdout', p) |
| self.assertEqual(expected.pop(0), data) |
| self.assertEqual(0, proc.returncode) |
| self.assertEqual([], expected) |
| break |
| finally: |
| proc.kill() |
| proc.wait() |
| except AssertionError: |
| if duration != 2: |
| print('Sleeping rocks. Trying slower.') |
| continue |
| raise |
| |
| def test_yield_any_timeout_0(self): |
| # rec_any() is expected to timeout and return None with no data pending at |
| # least once, due to the sleep of 'duration' and the use of timeout=0. |
| for duration in (0.05, 0.1, 0.5, 2): |
| try: |
| proc = get_output_sleep_proc(True, True, duration) |
| try: |
| expected = [ |
| 'A\n', |
| 'B\n', |
| ] |
| got_none = False |
| for p, data in proc.yield_any(timeout=0): |
| if not p: |
| got_none = True |
| continue |
| self.assertEqual('stdout', p) |
| self.assertEqual(expected.pop(0), data) |
| self.assertEqual(0, proc.returncode) |
| self.assertEqual([], expected) |
| self.assertEqual(True, got_none) |
| break |
| finally: |
| proc.kill() |
| proc.wait() |
| except AssertionError: |
| if duration != 2: |
| print('Sleeping rocks. Trying slower.') |
| continue |
| raise |
| |
| def test_yield_any_timeout_0_called(self): |
| # rec_any() is expected to timeout and return None with no data pending at |
| # least once, due to the sleep of 'duration' and the use of timeout=0. |
| for duration in (0.05, 0.1, 0.5, 2): |
| got_none = False |
| expected = ['A\n', 'B\n'] |
| called = [] |
| |
| def timeout(): |
| # pylint: disable=cell-var-from-loop |
| called.append(0) |
| return 0 |
| |
| try: |
| proc = get_output_sleep_proc(True, True, duration) |
| try: |
| for p, data in proc.yield_any(timeout=timeout): |
| if not p: |
| got_none = True |
| continue |
| self.assertEqual('stdout', p) |
| self.assertEqual(expected.pop(0), data) |
| self.assertEqual(0, proc.returncode) |
| self.assertEqual([], expected) |
| self.assertEqual(True, got_none) |
| self.assertTrue(called) |
| break |
| finally: |
| proc.kill() |
| proc.wait() |
| except AssertionError: |
| if duration != 2: |
| print('Sleeping rocks. Trying slower.') |
| continue |
| raise |
| |
| def test_yield_any_returncode(self): |
| proc = subprocess42.Popen( |
| [sys.executable, '-c', 'import sys;sys.stdout.write("yo");sys.exit(1)'], |
| stdout=subprocess42.PIPE) |
| for p, d in proc.yield_any(): |
| self.assertEqual('stdout', p) |
| self.assertEqual(b'yo', d) |
| # There was a bug where the second call to wait() would overwrite |
| # proc.returncode with 0 when timeout is not None. |
| self.assertEqual(1, proc.wait()) |
| self.assertEqual(1, proc.wait(timeout=0)) |
| self.assertEqual(1, proc.poll()) |
| self.assertEqual(1, proc.returncode) |
| # On Windows, the clock resolution is 15ms so Popen.duration() will likely |
| # be 0. |
| self.assertLessEqual(0, proc.duration()) |
| |
| def _wait_for_hi(self, proc, err): |
| actual = b'' |
| while True: |
| if err: |
| data = proc.recv_err(timeout=5) |
| else: |
| data = proc.recv_out(timeout=5) |
| if not data: |
| self.fail('%r' % actual) |
| self.assertTrue(data) |
| actual += data |
| if actual in (b'hi\n', b'hi\r\n'): |
| break |
| |
| def _proc(self, err, **kwargs): |
| # Do not use the -u flag here, we want to test when it is buffered by |
| # default. See reference above about PYTHONUNBUFFERED. |
| # That's why the two scripts uses .flush(). Sadly, the flush() call is |
| # needed on Windows even for sys.stderr (!) |
| cmd = [sys.executable, '-c', SCRIPT_ERR if err else SCRIPT_OUT] |
| # TODO(maruel): Make universal_newlines=True work and not hang. |
| if err: |
| kwargs['stderr'] = subprocess42.PIPE |
| else: |
| kwargs['stdout'] = subprocess42.PIPE |
| return subprocess42.Popen(cmd, **kwargs) |
| |
| def test_detached(self): |
| self._test_detached(False) |
| self._test_detached(True) |
| |
| def _test_detached(self, err): |
| is_win = (sys.platform == 'win32') |
| key = 'stderr' if err else 'stdout' |
| proc = self._proc(err, detached=True) |
| try: |
| self._wait_for_hi(proc, err) |
| proc.terminate() |
| if is_win: |
| # What happens on Windows is that the process is immediately killed |
| # after handling SIGBREAK. |
| self.assertEqual(0, proc.wait()) |
| # Windows... |
| self.assertIn(proc.recv_any(), ( |
| (key, b'got signal 21\r\nioerror\r\nbye\r\n'), |
| (key, b'got signal 21\nioerror\nbye\n'), |
| (key, b'got signal 21\r\nbye\r\n'), |
| (key, b'got signal 21\nbye\n'), |
| )) |
| else: |
| self.assertEqual(0, proc.wait()) |
| self.assertEqual((key, b'got signal 15\nbye\n'), proc.recv_any()) |
| finally: |
| # In case the test fails. |
| proc.kill() |
| proc.wait() |
| |
| def test_attached(self): |
| self._test_attached(False) |
| self._test_attached(True) |
| |
| def _test_attached(self, err): |
| is_win = (sys.platform == 'win32') |
| key = 'stderr' if err else 'stdout' |
| proc = self._proc(err, detached=False) |
| try: |
| self._wait_for_hi(proc, err) |
| proc.terminate() |
| if is_win: |
| # If attached, it's hard killed. |
| self.assertEqual(1, proc.wait()) |
| self.assertEqual((None, None), proc.recv_any()) |
| else: |
| self.assertEqual(0, proc.wait()) |
| self.assertEqual((key, b'got signal 15\nbye\n'), proc.recv_any()) |
| finally: |
| # In case the test fails. |
| proc.kill() |
| proc.wait() |
| |
| def test_split(self): |
| data = [ |
| ('stdout', b'o1\no2\no3\n'), |
| ('stderr', b'e1\ne2\ne3\n'), |
| ('stdout', b'\n\n'), |
| ('stdout', b'\n'), |
| ('stdout', b'o4\no5'), |
| ('stdout', b'_sameline\npart1 of one line '), |
| ('stderr', b'err inserted between two parts of stdout\n'), |
| ('stdout', b'part2 of one line\n'), |
| ('stdout', b'incomplete last stdout'), |
| ('stderr', b'incomplete last stderr'), |
| ] |
| expected = [ |
| ('stdout', b'o1'), |
| ('stdout', b'o2'), |
| ('stdout', b'o3'), |
| ('stderr', b'e1'), |
| ('stderr', b'e2'), |
| ('stderr', b'e3'), |
| ('stdout', b''), |
| ('stdout', b''), |
| ('stdout', b''), |
| ('stdout', b'o4'), |
| ('stdout', b'o5_sameline'), |
| ('stderr', b'err inserted between two parts of stdout'), |
| ('stdout', b'part1 of one line part2 of one line'), |
| ('stderr', b'incomplete last stderr'), |
| ('stdout', b'incomplete last stdout'), |
| ] |
| if sys.platform == 'win32': |
| data = [(d[0], d[1].replace(b'\n', b'\r\n')) for d in data] |
| |
| # With universal_newlines=False |
| self.assertEqual(list(subprocess42.split(data, False)), expected) |
| |
| # With universal_newlines=True |
| data = [(d[0], d[1].decode()) for d in data] |
| expected = [(e[0], e[1].decode()) for e in expected] |
| self.assertEqual(list(subprocess42.split(data, True)), expected) |
| |
| @params((None,), (10,)) |
| def test_wait_can_be_interrupted(self, timeout): |
| cmd = [ |
| sys.executable, |
| '-c', |
| textwrap.dedent(r""" |
| import signal |
| import sys |
| import textwrap |
| import time |
| |
| from utils import subprocess42 |
| |
| class ExitError(Exception): |
| pass |
| def handler(signum, _frame): |
| raise ExitError |
| |
| sleep_script = textwrap.dedent(''' |
| import time |
| for _ in range(50): |
| time.sleep(0.2) |
| ''') |
| proc = subprocess42.Popen([sys.executable, '-c', sleep_script], |
| detached=True) |
| sig = signal.SIGBREAK if sys.platform =='win32' else signal.SIGTERM |
| with subprocess42.set_signal_handler([sig], handler): |
| try: |
| sys.stdout.write('hi\n') |
| sys.stdout.flush() |
| proc.wait(%s) |
| except ExitError: |
| sys.stdout.write('wait is interrupted') |
| sys.stdout.flush() |
| proc.kill() |
| """ % timeout), |
| ] |
| # Set cwd to CLIENT_DIR so that the script can import subprocess42. |
| proc = subprocess42.Popen(cmd, stdout=subprocess42.PIPE, |
| cwd=test_env.CLIENT_DIR, detached=True) |
| self._wait_for_hi(proc, False) |
| time.sleep(0.5) |
| proc.terminate() |
| # proc is waiting for 'timeout' and SIGTERM/SIGBREAK is sent at 0.5s mark. |
| # Expect proc to write to stdout and exit almost immediately (decided |
| # by the poll interval of wait method). We wait for 8 second to give |
| # some buffer here. |
| start = time.time() |
| got = proc.recv_out(timeout=8) |
| want = b'wait is interrupted' |
| self.assertEqual( |
| got, want, |
| "%s != %s after %s seconds" % (got, want, time.time() - start)) |
| |
| |
| if __name__ == '__main__': |
| test_env.main() |
