Subprocess: reset blocking mode on proc exit. Fix #58

.editorconfig

@@ -11,7 +11,7 @@ trim_trailing_whitespace = true
 [*.{py,ini}]
 max_line_length = 120
 
-[*.{yml,rst}]
+[*.{yml,yaml,rst}]
 indent_size = 2
 
 [Makefile]

exec_helpers/subprocess_runner.py

@@ -93,6 +93,9 @@ def __prepare__(
 
 def set_nonblocking_pipe(pipe):  # type: (typing.Any) -> None
     """Set PIPE unblocked to allow polling of all pipes in parallel."""
+    if pipe is None:  # pragma: no cover
+        return
+
     descriptor = pipe.fileno()  # pragma: no cover
 
     if _posix:  # pragma: no cover
@@ -106,10 +109,10 @@ def set_nonblocking_pipe(pipe):  # type: (typing.Any) -> None
         # noinspection PyPep8Naming
         SetNamedPipeHandleState = windll.kernel32.SetNamedPipeHandleState
         SetNamedPipeHandleState.argtypes = [
-            wintypes.HANDLE,
-            wintypes.LPDWORD,
-            wintypes.LPDWORD,
-            wintypes.LPDWORD
+            wintypes.HANDLE,  # hNamedPipe
+            wintypes.LPDWORD,  # lpMode
+            wintypes.LPDWORD,  # lpMaxCollectionCount
+            wintypes.LPDWORD,  # lpCollectDataTimeout
         ]
         SetNamedPipeHandleState.restype = wintypes.BOOL
         # noinspection PyPep8Naming
@@ -122,6 +125,43 @@ def set_nonblocking_pipe(pipe):  # type: (typing.Any) -> None
         )
 
 
+def set_blocking_pipe(pipe):  # type: (typing.Any) -> None
+    """Set pipe blocking mode for final read on process close.
+
+    This will allow to read pipe until closed on remote side.
+    """
+    if pipe is None:  # pragma: no cover
+        return
+
+    descriptor = pipe.fileno()  # pragma: no cover
+
+    if _posix:  # pragma: no cover
+        # Get flags
+        flags = fcntl.fcntl(descriptor, fcntl.F_GETFL)
+
+        # Set block mode
+        fcntl.fcntl(descriptor, fcntl.F_SETFL, flags & (flags ^ os.O_NONBLOCK))
+
+    elif _win:  # pragma: no cover
+        # noinspection PyPep8Naming
+        SetNamedPipeHandleState = windll.kernel32.SetNamedPipeHandleState
+        SetNamedPipeHandleState.argtypes = [
+            wintypes.HANDLE,  # hNamedPipe
+            wintypes.LPDWORD,  # lpMode
+            wintypes.LPDWORD,  # lpMaxCollectionCount
+            wintypes.LPDWORD,  # lpCollectDataTimeout
+        ]
+        SetNamedPipeHandleState.restype = wintypes.BOOL
+        # noinspection PyPep8Naming
+        PIPE_WAIT = wintypes.DWORD(0x00000000)
+        handle = msvcrt.get_osfhandle(descriptor)
+
+        windll.kernel32.SetNamedPipeHandleState(
+            handle,
+            ctypes.byref(PIPE_WAIT), None, None
+        )
+
+
 class Subprocess(six.with_metaclass(SingletonMeta, api.ExecHelper)):
     """Subprocess helper with timeouts and lock-free FIFO."""
 
@@ -217,6 +257,8 @@ def poll_pipes(stop, ):    # type: (threading.Event) -> None
                 interface.poll()
 
                 if interface.returncode is not None:
+                    set_blocking_pipe(stdout)
+                    set_blocking_pipe(stderr)
                     result.read_stdout(
                         src=stdout,
                         log=logger,

exec_helpers/subprocess_runner.pyi

@@ -29,6 +29,8 @@ class SingletonMeta(type):
 
 def set_nonblocking_pipe(pipe: typing.Any) -> None: ...
 
+def set_blocking_pipe(pipe: typing.Any) -> None: ...
+
 
 class Subprocess(api.ExecHelper, metaclass=SingletonMeta):
     def __init__(self, log_mask_re: typing.Optional[str] = ...) -> None: ...

setup.cfg

@@ -46,4 +46,4 @@ count = True
 test=pytest
 
 [tool:pytest]
-addopts = -vvv
+addopts = -vvv -s -p no:django -p no:ipdb

test/test_subprocess_runner.py

@@ -23,6 +23,7 @@
 import errno
 import logging
 import subprocess
+import sys
 import unittest
 
 import mock
@@ -31,6 +32,13 @@
 import exec_helpers
 from exec_helpers import subprocess_runner
 
+if 'posix' in sys.builtin_module_names:
+    mock_block = 'fcntl.fcntl'
+elif sys.platform == "win32":
+    mock_block = 'windll.kernel32.SetNamedPipeHandleState'
+else:
+    mock_block = 'logging.captureWarnings'
+
 command = 'ls ~\nline 2\nline 3\nline с кирилицей'
 command_log = u"Executing command:\n{!r}\n".format(command.rstrip())
 stdout_list = [b' \n', b'2\n', b'3\n', b' \n']
@@ -52,7 +60,7 @@ def fileno(self):
 
 @mock.patch('exec_helpers.subprocess_runner.logger', autospec=True)
 @mock.patch('select.select', autospec=True)
-@mock.patch('exec_helpers.subprocess_runner.set_nonblocking_pipe', autospec=True)
+@mock.patch(mock_block, autospec=True)
 @mock.patch('subprocess.Popen', autospec=True, name='subprocess.Popen')
 class TestSubprocessRunner(unittest.TestCase):
     def setUp(self):
@@ -222,8 +230,11 @@ def test_003_context_manager(
     @mock.patch('time.sleep', autospec=True)
     def test_004_execute_timeout_fail(
         self,
-        sleep,
-        popen, _, select, logger
+        sleep,  # type: mock.MagicMock
+        popen,  # type: mock.MagicMock
+        _,  # type: mock.MagicMock
+        select,  # type: mock.MagicMock
+        logger  # type: mock.MagicMock
     ):
         popen_obj, exp_result = self.prepare_close(popen)
         popen_obj.configure_mock(returncode=None)
@@ -255,7 +266,13 @@ def test_004_execute_timeout_fail(
             ),
         ))
 
-    def test_005_execute_no_stdout(self, popen, _, select, logger):
+    def test_005_execute_no_stdout(
+        self,
+        popen,  # type: mock.MagicMock
+        _,  # type: mock.MagicMock
+        select,  # type: mock.MagicMock
+        logger  # type: mock.MagicMock
+    ):
         popen_obj, exp_result = self.prepare_close(popen, open_stdout=False)
         select.return_value = [popen_obj.stdout, popen_obj.stderr], [], []
 
@@ -293,7 +310,13 @@ def test_005_execute_no_stdout(self, popen, _, select, logger):
             mock.call.poll(), popen_obj.mock_calls
         )
 
-    def test_006_execute_no_stderr(self, popen, _, select, logger):
+    def test_006_execute_no_stderr(
+        self,
+        popen,  # type: mock.MagicMock
+        _,  # type: mock.MagicMock
+        select,  # type: mock.MagicMock
+        logger  # type: mock.MagicMock
+    ):
         popen_obj, exp_result = self.prepare_close(popen, open_stderr=False)
         select.return_value = [popen_obj.stdout, popen_obj.stderr], [], []
 
@@ -332,7 +355,13 @@ def test_006_execute_no_stderr(self, popen, _, select, logger):
             mock.call.poll(), popen_obj.mock_calls
         )
 
-    def test_007_execute_no_stdout_stderr(self, popen, _, select, logger):
+    def test_007_execute_no_stdout_stderr(
+        self,
+        popen,  # type: mock.MagicMock
+        _,  # type: mock.MagicMock
+        select,  # type: mock.MagicMock
+        logger  # type: mock.MagicMock
+    ):
         popen_obj, exp_result = self.prepare_close(
             popen,
             open_stdout=False,
@@ -369,7 +398,13 @@ def test_007_execute_no_stdout_stderr(self, popen, _, select, logger):
             mock.call.poll(), popen_obj.mock_calls
         )
 
-    def test_008_execute_mask_global(self, popen, _, select, logger):
+    def test_008_execute_mask_global(
+        self,
+        popen,  # type: mock.MagicMock
+        _,  # type: mock.MagicMock
+        select,  # type: mock.MagicMock
+        logger  # type: mock.MagicMock
+    ):
         cmd = "USE='secret=secret_pass' do task"
         log_mask_re = r"secret\s*=\s*([A-Z-a-z0-9_\-]+)"
         masked_cmd = "USE='secret=<*masked*>' do task"
@@ -424,7 +459,13 @@ def test_008_execute_mask_global(self, popen, _, select, logger):
             mock.call.poll(), popen_obj.mock_calls
         )
 
-    def test_009_execute_mask_local(self, popen, _, select, logger):
+    def test_009_execute_mask_local(
+        self,
+        popen,  # type: mock.MagicMock
+        _,  # type: mock.MagicMock
+        select,  # type: mock.MagicMock
+        logger  # type: mock.MagicMock
+    ):
         cmd = "USE='secret=secret_pass' do task"
         log_mask_re = r"secret\s*=\s*([A-Z-a-z0-9_\-]+)"
         masked_cmd = "USE='secret=<*masked*>' do task"
@@ -823,8 +864,12 @@ def test_012_check_stdin_fail_close(
     @mock.patch('time.sleep', autospec=True)
     def test_013_execute_timeout_done(
         self,
-        sleep,
-        popen, _, select, logger
+        sleep,  # type: mock.MagicMock
+        popen,  # type: mock.MagicMock
+        _,  # type: mock.MagicMock
+        select,  # type: mock.MagicMock
+        logger  # type: mock.MagicMock
+
     ):
         popen_obj, exp_result = self.prepare_close(popen, ec=exec_helpers.ExitCodes.EX_INVALID)
         popen_obj.configure_mock(returncode=None)

tox.ini

@@ -25,7 +25,7 @@ deps =
     py{27,py}: mock
 
 commands =
-    py.test -vv --junitxml=unit_result.xml --html=report.html --cov-config .coveragerc --cov-report html --cov=exec_helpers {posargs:test}
+    py.test --junitxml=unit_result.xml --cov-report html --self-contained-html --html=report.html --cov-config .coveragerc --cov=exec_helpers {posargs:test}
     coverage report --fail-under 97
 
 [testenv:py34-nocov]