Merge pull request #810 from python-cmd2/read_input

cmd2-specific input() function

4 files changed, 245 insertions, 221 deletions

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8c25bc9b..630f6892 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,10 @@
+## 0.9.21 (TBD, 2019)
+* Bug Fixes
+    * Fixed bug where pipe processes were not being stopped by Ctrl-C on Linux/Mac
+* 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.
+
 ## 0.9.20 (November 12, 2019)
 * Bug Fixes
     * Fixed bug where setting `use_ipython` to False removed ipy command from the entire `cmd2.Cmd` class instead of
diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py
index e50d9512..051e6f7c 100644
--- a/cmd2/cmd2.py
+++ b/cmd2/cmd2.py
@@ -416,6 +416,10 @@ class Cmd(cmd.Cmd):
             self.perror('Invalid value: {} (valid values: {}, {}, {})'.format(new_val, ansi.ANSI_TERMINAL,
                                                                               ansi.ANSI_ALWAYS, ansi.ANSI_NEVER))
 
+    def _completion_supported(self) -> bool:
+        """Return whether tab completion is supported"""
+        return self.use_rawinput and self.completekey and rl_type != RlType.NONE
+
     @property
     def visible_prompt(self) -> str:
         """Read-only property to get the visible prompt with any ANSI escape codes stripped.
@@ -1322,7 +1326,7 @@ class Cmd(cmd.Cmd):
         """
         # noinspection PyBroadException
         try:
-            if state == 0 and rl_type != RlType.NONE:
+            if state == 0:
                 self._reset_completion_defaults()
 
                 # Check if we are completing a multiline command
@@ -1649,7 +1653,7 @@ class Cmd(cmd.Cmd):
         """Keep accepting lines of input until the command is complete.
 
         There is some pretty hacky code here to handle some quirks of
-        self._pseudo_raw_input(). It returns a literal 'eof' if the input
+        self._read_command_line(). It returns a literal 'eof' if the input
         pipe runs out. We can't refactor it because we need to retain
         backwards compatibility with the standard library version of cmd.
 
@@ -1683,7 +1687,7 @@ class Cmd(cmd.Cmd):
                 # Save the command line up to this point for tab completion
                 self._multiline_in_progress = line + '\n'
 
-                nextline = self._pseudo_raw_input(self.continuation_prompt)
+                nextline = self._read_command_line(self.continuation_prompt)
                 if nextline == 'eof':
                     # they entered either a blank line, or we hit an EOF
                     # for some other reason. Turn the literal 'eof'
@@ -1989,36 +1993,59 @@ class Cmd(cmd.Cmd):
             # Set apply_style to False so default_error's style is not overridden
             self.perror(err_msg, apply_style=False)
 
-    def _pseudo_raw_input(self, prompt: str) -> str:
-        """Began life as a copy of cmd's cmdloop; like raw_input but
-
-        - accounts for changed stdin, stdout
-        - if input is a pipe (instead of a tty), look at self.echo
-          to decide whether to print the prompt and the input
+    def read_input(self, prompt: str, *, allow_completion: bool = False) -> str:
         """
-        if self.use_rawinput:
-            try:
-                if sys.stdin.isatty():
-                    # Wrap in try since terminal_lock may not be locked when this function is called from unit tests
-                    try:
-                        # A prompt is about to be drawn. Allow asynchronous changes to the terminal.
-                        self.terminal_lock.release()
-                    except RuntimeError:
-                        pass
+        Read input from appropriate stdin value. Also allows you to disable tab completion while input is being read.
+        :param prompt: prompt to display to user
+        :param allow_completion: if True, then tab completion of commands is enabled. This generally should be
+        set to False unless reading the command line. Defaults to False.
+        :return: the line read from stdin with all trailing new lines removed
+        :raises any exceptions raised by input() and stdin.readline()
+        """
+        completion_disabled = False
+        orig_completer = None
+
+        def disable_completion():
+            """Turn off completion while entering input"""
+            nonlocal orig_completer
+            nonlocal completion_disabled
+
+            if self._completion_supported() and not completion_disabled:
+                orig_completer = readline.get_completer()
+                readline.set_completer(lambda *args, **kwargs: None)
+                completion_disabled = True
+
+        def enable_completion():
+            """Restore tab completion when finished entering input"""
+            nonlocal completion_disabled
+
+            if self._completion_supported() and completion_disabled:
+                readline.set_completer(orig_completer)
+                completion_disabled = False
 
+        # Check we are reading from sys.stdin
+        if self.use_rawinput:
+            if sys.stdin.isatty():
+                try:
                     # Deal with the vagaries of readline and ANSI escape codes
                     safe_prompt = rl_make_safe_prompt(prompt)
+
+                    with self.sigint_protection:
+                        # Check if tab completion should be disabled
+                        if not allow_completion:
+                            disable_completion()
                     line = input(safe_prompt)
-                else:
-                    line = input()
-                    if self.echo:
-                        sys.stdout.write('{}{}\n'.format(prompt, line))
-            except EOFError:
-                line = 'eof'
-            finally:
-                if sys.stdin.isatty():
-                    # The prompt is gone. Do not allow asynchronous changes to the terminal.
-                    self.terminal_lock.acquire()
+                finally:
+                    with self.sigint_protection:
+                        # Check if we need to re-enable tab completion
+                        if not allow_completion:
+                            enable_completion()
+            else:
+                line = input()
+                if self.echo:
+                    sys.stdout.write('{}{}\n'.format(prompt, line))
+
+        # Otherwise read from self.stdin
         else:
             if self.stdin.isatty():
                 # on a tty, print the prompt first, then read the line
@@ -2041,6 +2068,28 @@ class Cmd(cmd.Cmd):
 
         return line.rstrip('\r\n')
 
+    def _read_command_line(self, prompt: str) -> str:
+        """
+        Read command line from appropriate stdin
+
+        :param prompt: prompt to display to user
+        :return: command line text of 'eof' if an EOFError was caught
+        :raises whatever exceptions are raised by input() except for EOFError
+        """
+        try:
+            # Wrap in try since terminal_lock may not be locked
+            try:
+                # Command line is about to be drawn. Allow asynchronous changes to the terminal.
+                self.terminal_lock.release()
+            except RuntimeError:
+                pass
+            return self.read_input(prompt, allow_completion=True)
+        except EOFError:
+            return 'eof'
+        finally:
+            # Command line is gone. Do not allow asynchronous changes to the terminal.
+            self.terminal_lock.acquire()
+
     def _set_up_cmd2_readline(self) -> _SavedReadlineSettings:
         """
         Set up readline with cmd2-specific settings
@@ -2048,7 +2097,7 @@ class Cmd(cmd.Cmd):
         """
         readline_settings = _SavedReadlineSettings()
 
-        if self.use_rawinput and self.completekey and rl_type != RlType.NONE:
+        if self._completion_supported():
 
             # Set up readline for our tab completion needs
             if rl_type == RlType.GNU:
@@ -2080,7 +2129,7 @@ class Cmd(cmd.Cmd):
         Restore saved readline settings
         :param readline_settings: the readline settings to restore
         """
-        if self.use_rawinput and self.completekey and rl_type != RlType.NONE:
+        if self._completion_supported():
 
             # Restore what we changed in readline
             readline.set_completer(readline_settings.completer)
@@ -2114,7 +2163,7 @@ class Cmd(cmd.Cmd):
             while not stop:
                 # Get commands from user
                 try:
-                    line = self._pseudo_raw_input(self.prompt)
+                    line = self._read_command_line(self.prompt)
                 except KeyboardInterrupt as ex:
                     if self.quit_on_sigint:
                         raise ex
@@ -2693,27 +2742,6 @@ class Cmd(cmd.Cmd):
                                    that the return value can differ from
                                    the text advertised to the user """
 
-        completion_disabled = False
-        orig_completer = None
-
-        def disable_completion():
-            """Turn off completion during the select input line"""
-            nonlocal orig_completer
-            nonlocal completion_disabled
-
-            if rl_type != RlType.NONE and not completion_disabled:
-                orig_completer = readline.get_completer()
-                readline.set_completer(lambda *args, **kwargs: None)
-                completion_disabled = True
-
-        def enable_completion():
-            """Restore tab completion when select is done reading input"""
-            nonlocal completion_disabled
-
-            if rl_type != RlType.NONE and completion_disabled:
-                readline.set_completer(orig_completer)
-                completion_disabled = False
-
         local_opts = opts
         if isinstance(opts, str):
             local_opts = list(zip(opts.split(), opts.split()))
@@ -2730,18 +2758,14 @@ class Cmd(cmd.Cmd):
             self.poutput('  %2d. %s' % (idx + 1, text))
 
         while True:
-            safe_prompt = rl_make_safe_prompt(prompt)
-
             try:
-                with self.sigint_protection:
-                    disable_completion()
-                response = input(safe_prompt)
+                response = self.read_input(prompt)
             except EOFError:
                 response = ''
                 self.poutput('\n', end='')
-            finally:
-                with self.sigint_protection:
-                    enable_completion()
+            except KeyboardInterrupt as ex:
+                self.poutput('^C')
+                raise ex
 
             if not response:
                 continue
@@ -2921,7 +2945,7 @@ class Cmd(cmd.Cmd):
             for item in self._py_history:
                 readline.add_history(item)
 
-            if self.use_rawinput and self.completekey:
+            if self._completion_supported():
                 # Set up tab completion for the Python console
                 # rlcompleter relies on the default settings of the Python readline module
                 if rl_type == RlType.GNU:
@@ -2988,7 +3012,7 @@ class Cmd(cmd.Cmd):
             for item in cmd2_env.history:
                 readline.add_history(item)
 
-            if self.use_rawinput and self.completekey:
+            if self._completion_supported():
                 # Restore cmd2's tab completion settings
                 readline.set_completer(cmd2_env.readline_settings.completer)
                 readline.set_completer_delims(cmd2_env.readline_settings.delims)
@@ -3715,7 +3739,7 @@ class Cmd(cmd.Cmd):
 
     def async_alert(self, alert_msg: str, new_prompt: Optional[str] = None) -> None:  # pragma: no cover
         """
-        Display an important message to the user while they are at the prompt in between commands.
+        Display an important message to the user while they are at a command line prompt.
         To the user it appears as if an alert message is printed above the prompt and their current input
         text and cursor location is left alone.
 
@@ -3775,10 +3799,10 @@ class Cmd(cmd.Cmd):
 
     def async_update_prompt(self, new_prompt: str) -> None:  # pragma: no cover
         """
-        Update the prompt while the user is still typing at it. This is good for alerting the user to system
-        changes dynamically in between commands. For instance you could alter the color of the prompt to indicate
-        a system status or increase a counter to report an event. If you do alter the actual text of the prompt,
-        it is best to keep the prompt the same width as what's on screen. Otherwise the user's input text will
+        Update the command line prompt while the user is still typing at it. This is good for alerting the user to
+        system changes dynamically in between commands. For instance you could alter the color of the prompt to
+        indicate a system status or increase a counter to report an event. If you do alter the actual text of the
+        prompt, it is best to keep the prompt the same width as what's on screen. Otherwise the user's input text will
         be shifted and the update will not be seamless.
 
         Raises a `RuntimeError` if called while another thread holds `terminal_lock`.
@@ -3948,7 +3972,7 @@ class Cmd(cmd.Cmd):
         original_sigint_handler = signal.getsignal(signal.SIGINT)
         signal.signal(signal.SIGINT, self.sigint_handler)
 
-        # Grab terminal lock before the prompt has been drawn by readline
+        # Grab terminal lock before the command line prompt has been drawn by readline
         self.terminal_lock.acquire()
 
         # Always run the preloop first
diff --git a/cmd2/utils.py b/cmd2/utils.py
index 88fbc1da..3155c64a 100644
--- a/cmd2/utils.py
+++ b/cmd2/utils.py
@@ -520,10 +520,15 @@ class ProcReader(object):
         """Send a SIGINT to the process similar to if <Ctrl>+C were pressed."""
         import signal
         if sys.platform.startswith('win'):
-            signal_to_send = signal.CTRL_C_EVENT
+            self._proc.send_signal(signal.CTRL_C_EVENT)
         else:
-            signal_to_send = signal.SIGINT
-        self._proc.send_signal(signal_to_send)
+            # 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
+            try:
+                group_id = os.getpgid(self._proc.pid)
+                os.killpg(group_id, signal.SIGINT)
+            except ProcessLookupError:
+                return
 
     def terminate(self) -> None:
         """Terminate the process"""
diff --git a/tests/test_cmd2.py b/tests/test_cmd2.py
index cb66ac9b..f9c3e61d 100755
--- a/tests/test_cmd2.py
+++ b/tests/test_cmd2.py
@@ -847,6 +847,18 @@ def test_cmdloop_without_rawinput():
     out = app.stdout.getvalue()
     assert out == expected
 
+@pytest.mark.skipif(sys.platform.startswith('win'),
+                    reason="stty sane only run on Linux/Mac")
+def test_stty_sane(base_app, monkeypatch):
+    """Make sure stty sane is run on Linux/Mac after each command if stdin is a terminal"""
+    with mock.patch('sys.stdin.isatty', mock.MagicMock(name='isatty', return_value=True)):
+        # Mock out the subprocess.Popen call so we don't actually run stty sane
+        m = mock.MagicMock(name='Popen')
+        monkeypatch.setattr("subprocess.Popen", m)
+
+        base_app.onecmd_plus_hooks('help')
+        m.assert_called_once_with(['stty', 'sane'])
+
 class HookFailureApp(cmd2.Cmd):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -1092,10 +1104,10 @@ def select_app():
     app = SelectApp()
     return app
 
-def test_select_options(select_app):
-    # Mock out the input call so we don't actually wait for a user's response on stdin
-    m = mock.MagicMock(name='input', return_value='2')
-    builtins.input = m
+def test_select_options(select_app, monkeypatch):
+    # Mock out the read_input call so we don't actually wait for a user's response on stdin
+    read_input_mock = mock.MagicMock(name='read_input', return_value='2')
+    monkeypatch.setattr("cmd2.Cmd.read_input", read_input_mock)
 
     food = 'bacon'
     out, err = run_cmd(select_app, "eat {}".format(food))
@@ -1106,17 +1118,18 @@ def test_select_options(select_app):
 """.format(food))
 
     # Make sure our mock was called with the expected arguments
-    m.assert_called_once_with('Sauce? ')
+    read_input_mock.assert_called_once_with('Sauce? ')
 
     # And verify the expected output to stdout
     assert out == expected
 
-def test_select_invalid_option_too_big(select_app):
+def test_select_invalid_option_too_big(select_app, monkeypatch):
     # Mock out the input call so we don't actually wait for a user's response on stdin
-    m = mock.MagicMock(name='input')
+    read_input_mock = mock.MagicMock(name='read_input')
+
     # If side_effect is an iterable then each call to the mock will return the next value from the iterable.
-    m.side_effect = ['3', '1']  # First pass an invalid selection, then pass a valid one
-    builtins.input = m
+    read_input_mock.side_effect = ['3', '1']  # First pass an invalid selection, then pass a valid one
+    monkeypatch.setattr("cmd2.Cmd.read_input", read_input_mock)
 
     food = 'fish'
     out, err = run_cmd(select_app, "eat {}".format(food))
@@ -1130,18 +1143,19 @@ def test_select_invalid_option_too_big(select_app):
     # Make sure our mock was called exactly twice with the expected arguments
     arg = 'Sauce? '
     calls = [mock.call(arg), mock.call(arg)]
-    m.assert_has_calls(calls)
-    assert m.call_count == 2
+    read_input_mock.assert_has_calls(calls)
+    assert read_input_mock.call_count == 2
 
     # And verify the expected output to stdout
     assert out == expected
 
-def test_select_invalid_option_too_small(select_app):
+def test_select_invalid_option_too_small(select_app, monkeypatch):
     # Mock out the input call so we don't actually wait for a user's response on stdin
-    m = mock.MagicMock(name='input')
+    read_input_mock = mock.MagicMock(name='read_input')
+
     # If side_effect is an iterable then each call to the mock will return the next value from the iterable.
-    m.side_effect = ['0', '1']  # First pass an invalid selection, then pass a valid one
-    builtins.input = m
+    read_input_mock.side_effect = ['0', '1']  # First pass an invalid selection, then pass a valid one
+    monkeypatch.setattr("cmd2.Cmd.read_input", read_input_mock)
 
     food = 'fish'
     out, err = run_cmd(select_app, "eat {}".format(food))
@@ -1155,16 +1169,16 @@ def test_select_invalid_option_too_small(select_app):
     # Make sure our mock was called exactly twice with the expected arguments
     arg = 'Sauce? '
     calls = [mock.call(arg), mock.call(arg)]
-    m.assert_has_calls(calls)
-    assert m.call_count == 2
+    read_input_mock.assert_has_calls(calls)
+    assert read_input_mock.call_count == 2
 
     # And verify the expected output to stdout
     assert out == expected
 
-def test_select_list_of_strings(select_app):
+def test_select_list_of_strings(select_app, monkeypatch):
     # Mock out the input call so we don't actually wait for a user's response on stdin
-    m = mock.MagicMock(name='input', return_value='2')
-    builtins.input = m
+    read_input_mock = mock.MagicMock(name='read_input', return_value='2')
+    monkeypatch.setattr("cmd2.Cmd.read_input", read_input_mock)
 
     out, err = run_cmd(select_app, "study")
     expected = normalize("""
@@ -1174,15 +1188,15 @@ Good luck learning {}!
 """.format('science'))
 
     # Make sure our mock was called with the expected arguments
-    m.assert_called_once_with('Subject? ')
+    read_input_mock.assert_called_once_with('Subject? ')
 
     # And verify the expected output to stdout
     assert out == expected
 
-def test_select_list_of_tuples(select_app):
+def test_select_list_of_tuples(select_app, monkeypatch):
     # Mock out the input call so we don't actually wait for a user's response on stdin
-    m = mock.MagicMock(name='input', return_value='2')
-    builtins.input = m
+    read_input_mock = mock.MagicMock(name='read_input', return_value='2')
+    monkeypatch.setattr("cmd2.Cmd.read_input", read_input_mock)
 
     out, err = run_cmd(select_app, "procrastinate")
     expected = normalize("""
@@ -1192,16 +1206,16 @@ Have fun procrasinating with {}!
 """.format('YouTube'))
 
     # Make sure our mock was called with the expected arguments
-    m.assert_called_once_with('How would you like to procrastinate? ')
+    read_input_mock.assert_called_once_with('How would you like to procrastinate? ')
 
     # And verify the expected output to stdout
     assert out == expected
 
 
-def test_select_uneven_list_of_tuples(select_app):
+def test_select_uneven_list_of_tuples(select_app, monkeypatch):
     # Mock out the input call so we don't actually wait for a user's response on stdin
-    m = mock.MagicMock(name='input', return_value='2')
-    builtins.input = m
+    read_input_mock = mock.MagicMock(name='read_input', return_value='2')
+    monkeypatch.setattr("cmd2.Cmd.read_input", read_input_mock)
 
     out, err = run_cmd(select_app, "play")
     expected = normalize("""
@@ -1211,15 +1225,15 @@ Charm us with the {}...
 """.format('Drums'))
 
     # Make sure our mock was called with the expected arguments
-    m.assert_called_once_with('Instrument? ')
+    read_input_mock.assert_called_once_with('Instrument? ')
 
     # And verify the expected output to stdout
     assert out == expected
 
-def test_select_eof(select_app):
+def test_select_eof(select_app, monkeypatch):
     # Ctrl-D during select causes an EOFError that just reprompts the user
-    m = mock.MagicMock(name='input', side_effect=[EOFError, 2])
-    builtins.input = m
+    read_input_mock = mock.MagicMock(name='read_input', side_effect=[EOFError, 2])
+    monkeypatch.setattr("cmd2.Cmd.read_input", read_input_mock)
 
     food = 'fish'
     out, err = run_cmd(select_app, "eat {}".format(food))
@@ -1227,8 +1241,19 @@ def test_select_eof(select_app):
     # Make sure our mock was called exactly twice with the expected arguments
     arg = 'Sauce? '
     calls = [mock.call(arg), mock.call(arg)]
-    m.assert_has_calls(calls)
-    assert m.call_count == 2
+    read_input_mock.assert_has_calls(calls)
+    assert read_input_mock.call_count == 2
+
+def test_select_ctrl_c(outsim_app, monkeypatch, capsys):
+    # Ctrl-C during select prints ^C and raises a KeyboardInterrupt
+    read_input_mock = mock.MagicMock(name='read_input', side_effect=KeyboardInterrupt)
+    monkeypatch.setattr("cmd2.Cmd.read_input", read_input_mock)
+
+    with pytest.raises(KeyboardInterrupt):
+        outsim_app.select([('Guitar', 'Electric Guitar'), ('Drums',)], 'Instrument? ')
+
+    out = outsim_app.stdout.getvalue()
+    assert out.rstrip().endswith('^C')
 
 class HelpNoDocstringApp(cmd2.Cmd):
     greet_parser = argparse.ArgumentParser()
@@ -1419,131 +1444,94 @@ def test_echo(capsys):
     out, err = capsys.readouterr()
     assert out.startswith('{}{}\n'.format(app.prompt, commands[0]) + HELP_HISTORY.split()[0])
 
-def test_pseudo_raw_input_tty_rawinput_true():
-    # use context managers so original functions get put back when we are done
-    # we dont use decorators because we need m_input for the assertion
-    with mock.patch('sys.stdin.isatty', mock.MagicMock(name='isatty', return_value=True)):
-        with mock.patch('builtins.input', mock.MagicMock(name='input', side_effect=['set', EOFError])) as m_input:
-            # run the cmdloop, which should pull input from our mocks
-            app = cmd2.Cmd(allow_cli_args=False)
-            app.use_rawinput = True
-            app._cmdloop()
-            # because we mocked the input() call, we won't get the prompt
-            # or the name of the command in the output, so we can't check
-            # if its there. We assume that if input got called twice, once
-            # for the 'set' command, and once for the 'quit' command,
-            # that the rest of it worked
-            assert m_input.call_count == 2
-
-def test_pseudo_raw_input_tty_rawinput_false():
-    # gin up some input like it's coming from a tty
-    fakein = io.StringIO(u'{}'.format('set\n'))
-    mtty = mock.MagicMock(name='isatty', return_value=True)
-    fakein.isatty = mtty
-    mreadline = mock.MagicMock(name='readline', wraps=fakein.readline)
-    fakein.readline = mreadline
-
-    # run the cmdloop, telling it where to get input from
-    app = cmd2.Cmd(stdin=fakein, allow_cli_args=False)
-    app.use_rawinput = False
-    app._cmdloop()
-
-    # because we mocked the readline() call, we won't get the prompt
-    # or the name of the command in the output, so we can't check
-    # if its there. We assume that if readline() got called twice, once
-    # for the 'set' command, and once for the 'quit' command,
-    # that the rest of it worked
-    assert mreadline.call_count == 2
-
-# the next helper function and two tests check for piped
-# input when use_rawinput is True.
-def piped_rawinput_true(capsys, echo, command):
-    app = cmd2.Cmd(allow_cli_args=False)
-    app.use_rawinput = True
-    app.echo = echo
-    # run the cmdloop, which should pull input from our mock
-    app._cmdloop()
-    out, err = capsys.readouterr()
-    return app, out
-
-# using the decorator puts the original input function back when this unit test returns
-@mock.patch('builtins.input', mock.MagicMock(name='input', side_effect=['set', EOFError]))
-def test_pseudo_raw_input_piped_rawinput_true_echo_true(capsys):
-    command = 'set'
-    app, out = piped_rawinput_true(capsys, True, command)
-    out = out.splitlines()
-    assert out[0] == '{}{}'.format(app.prompt, command)
-    assert out[1].startswith('allow_ansi:')
-
-# using the decorator puts the original input function back when this unit test returns
-@mock.patch('builtins.input', mock.MagicMock(name='input', side_effect=['set', EOFError]))
-def test_pseudo_raw_input_piped_rawinput_true_echo_false(capsys):
-    command = 'set'
-    app, out = piped_rawinput_true(capsys, False, command)
-    firstline = out.splitlines()[0]
-    assert firstline.startswith('allow_ansi:')
-    assert not '{}{}'.format(app.prompt, command) in out
-
-# the next helper function and two tests check for piped
-# input when use_rawinput=False
-def piped_rawinput_false(capsys, echo, command):
-    fakein = io.StringIO(u'{}'.format(command))
-    app = cmd2.Cmd(stdin=fakein, allow_cli_args=False)
-    app.use_rawinput = False
-    app.echo = echo
-    app._cmdloop()
-    out, err = capsys.readouterr()
-    return app, out
-
-def test_pseudo_raw_input_piped_rawinput_false_echo_true(capsys):
-    command = 'set'
-    app, out = piped_rawinput_false(capsys, True, command)
-    out = out.splitlines()
-    assert out[0] == '{}{}'.format(app.prompt, command)
-    assert out[1].startswith('allow_ansi:')
+def test_read_input_rawinput_true(capsys, monkeypatch):
+    prompt_str = 'the_prompt'
+    input_str = 'some input'
 
-def test_pseudo_raw_input_piped_rawinput_false_echo_false(capsys):
-    command = 'set'
-    app, out = piped_rawinput_false(capsys, False, command)
-    firstline = out.splitlines()[0]
-    assert firstline.startswith('allow_ansi:')
-    assert not '{}{}'.format(app.prompt, command) in out
-
-
-# other input tests
-def test_raw_input(base_app):
-    base_app.use_raw_input = True
-    fake_input = 'quit'
+    app = cmd2.Cmd()
+    app.use_rawinput = True
 
-    # Mock out the input call so we don't actually wait for a user's response on stdin
-    m = mock.Mock(name='input', return_value=fake_input)
-    builtins.input = m
+    # Mock out input() to return input_str
+    monkeypatch.setattr("builtins.input", lambda *args: input_str)
 
-    line = base_app._pseudo_raw_input('(cmd2)')
-    assert line == fake_input
+    # isatty is True
+    with mock.patch('sys.stdin.isatty', mock.MagicMock(name='isatty', return_value=True)):
+        line = app.read_input(prompt_str)
+        assert line == input_str
+
+    # isatty is False
+    with mock.patch('sys.stdin.isatty', mock.MagicMock(name='isatty', return_value=False)):
+        # echo True
+        app.echo = True
+        line = app.read_input(prompt_str)
+        out, err = capsys.readouterr()
+        assert line == input_str
+        assert out == "{}{}\n".format(prompt_str, input_str)
+
+        # echo False
+        app.echo = False
+        line = app.read_input(prompt_str)
+        out, err = capsys.readouterr()
+        assert line == input_str
+        assert not out
+
+def test_read_input_rawinput_false(capsys, monkeypatch):
+    prompt_str = 'the_prompt'
+    input_str = 'some input'
+
+    def make_app(isatty: bool, empty_input: bool = False):
+        """Make a cmd2 app with a custom stdin"""
+        app_input_str = '' if empty_input else input_str
+
+        fakein = io.StringIO('{}'.format(app_input_str))
+        fakein.isatty = mock.MagicMock(name='isatty', return_value=isatty)
+
+        new_app = cmd2.Cmd(stdin=fakein)
+        new_app.use_rawinput = False
+        return new_app
+
+    # isatty True
+    app = make_app(isatty=True)
+    line = app.read_input(prompt_str)
+    out, err = capsys.readouterr()
+    assert line == input_str
+    assert out == prompt_str
 
-def test_stdin_input():
-    app = cmd2.Cmd()
-    app.use_rawinput = False
-    fake_input = 'quit'
+    # isatty True, empty input
+    app = make_app(isatty=True, empty_input=True)
+    line = app.read_input(prompt_str)
+    out, err = capsys.readouterr()
+    assert line == 'eof'
+    assert out == prompt_str
 
-    # Mock out the readline call so we don't actually read from stdin
-    m = mock.Mock(name='readline', return_value=fake_input)
-    app.stdin.readline = m
+    # isatty is False, echo is True
+    app = make_app(isatty=False)
+    app.echo = True
+    line = app.read_input(prompt_str)
+    out, err = capsys.readouterr()
+    assert line == input_str
+    assert out == "{}{}\n".format(prompt_str, input_str)
 
-    line = app._pseudo_raw_input('(cmd2)')
-    assert line == fake_input
+    # isatty is False, echo is False
+    app = make_app(isatty=False)
+    app.echo = False
+    line = app.read_input(prompt_str)
+    out, err = capsys.readouterr()
+    assert line == input_str
+    assert not out
 
-def test_empty_stdin_input():
-    app = cmd2.Cmd()
-    app.use_rawinput = False
-    fake_input = ''
+    # isatty is False, empty input
+    app = make_app(isatty=False, empty_input=True)
+    line = app.read_input(prompt_str)
+    out, err = capsys.readouterr()
+    assert line == 'eof'
+    assert not out
 
-    # Mock out the readline call so we don't actually read from stdin
-    m = mock.Mock(name='readline', return_value=fake_input)
-    app.stdin.readline = m
+def test_read_command_line_eof(base_app, monkeypatch):
+    read_input_mock = mock.MagicMock(name='read_input', side_effect=EOFError)
+    monkeypatch.setattr("cmd2.Cmd.read_input", read_input_mock)
 
-    line = app._pseudo_raw_input('(cmd2)')
+    line = base_app._read_command_line("Prompt> ")
     assert line == 'eof'
 
 def test_poutput_string(outsim_app):