Merge pull request #811 from python-cmd2/win_fix

Fixed bug where pipe processes were not being stopped by Ctrl-C on Windows

Author: kmvanbrunt

CHANGELOG.md

@@ -1,6 +1,6 @@
 ## 0.9.21 (TBD, 2019)
 * Bug Fixes
-    * Fixed bug where pipe processes were not being stopped by Ctrl-C on Linux/Mac
+    * Fixed bug where pipe processes were not being stopped by Ctrl-C
 * Enhancements
     * Added `read_input()` function that is used to read from stdin. Unlike the Python built-in `input()`, it also has
     an argument to disable tab completion while input is being entered.

cmd2/cmd2.py

@@ -56,9 +56,9 @@
 
 # Set up readline
 if rl_type == RlType.NONE:  # pragma: no cover
-    rl_warning = "Readline features including tab completion have been disabled since no \n" \
-                 "supported version of readline was found. To resolve this, install \n" \
-                 "pyreadline on Windows or gnureadline on Mac.\n\n"
+    rl_warning = ("Readline features including tab completion have been disabled since no\n"
+                  "supported version of readline was found. To resolve this, install pyreadline\n"
+                  "on Windows or gnureadline on Mac.\n\n")
     sys.stderr.write(ansi.style_warning(rl_warning))
 else:
     from .rl_utils import rl_force_redisplay, readline
@@ -1824,24 +1824,22 @@ def _redirect_output(self, statement: Statement) -> Tuple[bool, utils.Redirectio
             subproc_stdin = io.open(read_fd, 'r')
             new_stdout = io.open(write_fd, 'w')
 
-            # Set options to not forward signals to the pipe process. If a Ctrl-C event occurs,
-            # our sigint handler will forward it only to the most recent pipe process. This makes
-            # sure pipe processes close in the right order (most recent first).
+            # Create pipe process in a separate group to isolate our signals from it. If a Ctrl-C event occurs,
+            # our sigint handler will forward it only to the most recent pipe process. This makes sure pipe
+            # processes close in the right order (most recent first).
+            kwargs = dict()
             if sys.platform == 'win32':
-                creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
-                start_new_session = False
+                kwargs['creationflags'] = subprocess.CREATE_NEW_PROCESS_GROUP
             else:
-                creationflags = 0
-                start_new_session = True
+                kwargs['start_new_session'] = True
 
             # For any stream that is a StdSim, we will use a pipe so we can capture its output
             proc = subprocess.Popen(statement.pipe_to,
                                     stdin=subproc_stdin,
                                     stdout=subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout,
                                     stderr=subprocess.PIPE if isinstance(sys.stderr, utils.StdSim) else sys.stderr,
-                                    creationflags=creationflags,
-                                    start_new_session=start_new_session,
-                                    shell=True)
+                                    shell=True,
+                                    **kwargs)
 
             # Popen was called with shell=True so the user can chain pipe commands and redirect their output
             # like: !ls -l | grep user | wc -l > out.txt. But this makes it difficult to know if the pipe process

cmd2/utils.py

@@ -517,10 +517,12 @@ def __init__(self, proc: subprocess.Popen, stdout: Union[StdSim, TextIO],
             self._err_thread.start()
 
     def send_sigint(self) -> None:
-        """Send a SIGINT to the process similar to if <Ctrl>+C were pressed."""
+        """Send a SIGINT to the process similar to if <Ctrl>+C were pressed"""
         import signal
         if sys.platform.startswith('win'):
-            self._proc.send_signal(signal.CTRL_C_EVENT)
+            # cmd2 started the Windows process in a new process group. Therefore
+            # a CTRL_C_EVENT can't be sent to it. Send a CTRL_BREAK_EVENT instead.
+            self._proc.send_signal(signal.CTRL_BREAK_EVENT)
         else:
             # Since cmd2 uses shell=True in its Popen calls, we need to send the SIGINT to
             # the whole process group to make sure it propagates further than the shell

tests/test_utils.py

@@ -5,6 +5,7 @@
 """
 import signal
 import sys
+import time
 
 import pytest
 
@@ -228,27 +229,31 @@ def pr_none():
     import subprocess
 
     # Start a long running process so we have time to run tests on it before it finishes
-    # Put the new process into a separate group so signals sent to it won't interfere with this process
+    # Put the new process into a separate group so its signal are isolated from ours
+    kwargs = dict()
     if sys.platform.startswith('win'):
         command = 'timeout -t 5 /nobreak'
-        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
-        start_new_session = False
+        kwargs['creationflags'] = subprocess.CREATE_NEW_PROCESS_GROUP
     else:
         command = 'sleep 5'
-        creationflags = 0
-        start_new_session = True
+        kwargs['start_new_session'] = True
 
-    proc = subprocess.Popen(command,
-                            creationflags=creationflags,
-                            start_new_session=start_new_session,
-                            shell=True)
+    proc = subprocess.Popen(command, shell=True, **kwargs)
     pr = cu.ProcReader(proc, None, None)
     return pr
 
 def test_proc_reader_send_sigint(pr_none):
     assert pr_none._proc.poll() is None
     pr_none.send_sigint()
+
+    wait_start = time.monotonic()
     pr_none.wait()
+    wait_finish = time.monotonic()
+
+    # Make sure the process exited before sleep of 5 seconds finished
+    # 3 seconds accounts for some delay but is long enough for the process to exit
+    assert wait_finish - wait_start < 3
+
     ret_code = pr_none._proc.poll()
     if sys.platform.startswith('win'):
         assert ret_code is not None
@@ -258,7 +263,15 @@ def test_proc_reader_send_sigint(pr_none):
 def test_proc_reader_terminate(pr_none):
     assert pr_none._proc.poll() is None
     pr_none.terminate()
+
+    wait_start = time.monotonic()
     pr_none.wait()
+    wait_finish = time.monotonic()
+
+    # Make sure the process exited before sleep of 5 seconds finished
+    # 3 seconds accounts for some delay but is long enough for the process to exit
+    assert wait_finish - wait_start < 3
+
     ret_code = pr_none._proc.poll()
     if sys.platform.startswith('win'):
         assert ret_code is not None