Added tests.

Author: kmvanbrunt

cmd2/cmd2.py

@@ -444,7 +444,7 @@ def __init__(
         if auto_suggest:
             self.auto_suggest = AutoSuggestFromHistory()
 
-        self.session = self._build_session()
+        self.session = self._init_session()
 
         # Commands to exclude from the history command
         self.exclude_from_history = [constants.EOF, 'history']
@@ -609,8 +609,8 @@ def __init__(
         # the current command being executed
         self.current_command: Statement | None = None
 
-    def _build_session(self) -> PromptSession[str]:
-        """Construct the primary PromptSession for the cmd2 application.
+    def _init_session(self) -> PromptSession[str]:
+        """Initialize and return the core PromptSession for the application.
 
         Builds an interactive session if stdin is a TTY. Otherwise, uses
         dummy drivers to support non-interactive streams like pipes or files.
@@ -3197,21 +3197,33 @@ def _read_raw_input(
         prompt: Callable[[], ANSI | str] | ANSI | str,
         session: PromptSession[str],
         completer: Completer,
-        **prompt_kwargs: Any,  # optional keyword args for session.prompt()
+        **prompt_kwargs: Any,
     ) -> str:
-        """Read input from either an interactive terminal session or a redirected stream."""
-        # _build_session() sets session.input to a DummyInput when not in a TTY.
+        """Execute the low-level input read from either a terminal or a redirected stream.
+
+        If the session is interactive (TTY), it uses `prompt_toolkit` to render a
+        rich UI with completion and `patch_stdout` protection. If non-interactive
+        (Pipe/File), it performs a direct line read from `stdin`.
+
+        :param prompt: the prompt text or a callable that returns the prompt.
+        :param session: the PromptSession instance to use for reading.
+        :param completer: the completer to use for this specific input.
+        :param prompt_kwargs: additional arguments passed directly to session.prompt().
+        :return: the stripped input string.
+        :raises EOFError: if the input stream is closed or the user signals EOF (e.g., Ctrl+D)
+        """
+        # Check if the session is configured for interactive terminal use.
         if not isinstance(session.input, DummyInput):
             with patch_stdout():
-                return session.prompt(prompt, completer=completer, **prompt_kwargs)  # type: ignore[arg-type]
+                return session.prompt(prompt, completer=completer, **prompt_kwargs)
 
         # We're not at a terminal, so we're likely reading from a file or a pipe.
         # We wait for a line of data before we print anything.
         line = self.stdin.readline()
 
         # If the stream is empty, we've reached the end of the input.
         if not line:
-            return constants.EOF
+            raise EOFError
 
         # If echo is on, we want the output to look like a session transcript.
         # Print the prompt and the command before the results.
@@ -3266,6 +3278,10 @@ def read_input(
     ) -> str:
         """Read a line of input with optional completion and history.
 
+        :param prompt: prompt to display to user
+        :param history: optional Sequence of strings to use for up-arrow history. The passed in history
+                        will not be edited. It is the caller's responsibility to add the returned input
+                        to history if desired. Defaults to None.
         :param preserve_quotes: if True, then quoted tokens will keep their quotes when processed by
                                 ArgparseCompleter. This is helpful in cases when you're completing
                                 flag-like tokens (e.g. -o, --option) and you don't want them to be
@@ -3278,7 +3294,8 @@ def read_input(
         :param completer: completion function that provides choices for single argument
         :param parser: an argument parser which supports the completion of multiple arguments
         :return: the line read from stdin with all trailing new lines removed
-        :raises Exception: any exceptions raised by prompt()
+        :raises EOFError: if the input stream is closed or the user signals EOF (e.g., Ctrl+D)
+        :raises Exception: any other exceptions raised by prompt()
         """
         completer_to_use = self._resolve_completer(
             preserve_quotes=preserve_quotes,
@@ -3303,7 +3320,7 @@ def _read_command_line(self, prompt: str) -> str:
 
         :param prompt: prompt to display to user
         :return: command line text or 'eof' if an EOFError was caught
-        :raises Exception: whatever exceptions are raised by input() except for EOFError
+        :raises Exception: any exceptions raised by prompt(), except EOFError
         """
         try:
             # Use dynamic prompt if the prompt matches self.prompt

tests/test_cmd2.py

@@ -1936,8 +1936,8 @@ def test_read_raw_input_pipe_echo(capsys) -> None:
 
 def test_read_raw_input_eof() -> None:
     app = cmd2.Cmd(stdin=io.StringIO(""))
-    result = app._read_raw_input("prompt> ", app.session, DummyCompleter())
-    assert result == constants.EOF
+    with pytest.raises(EOFError):
+        app._read_raw_input("prompt> ", app.session, DummyCompleter())
 
 
 def test_resolve_completer_none(base_app: cmd2.Cmd) -> None:
@@ -3502,7 +3502,7 @@ def test_custom_completekey():
     assert app.completekey == '?'
 
 
-def test_build_session_exception(monkeypatch):
+def test_init_session_exception(monkeypatch):
 
     # Mock PromptSession to raise ValueError on first call, then succeed
     valid_session_mock = mock.MagicMock(spec=PromptSession)
@@ -3590,7 +3590,7 @@ def test_multiline_complete_statement_keyboard_interrupt(multiline_app, monkeypa
     poutput_mock.assert_called_with('^C')
 
 
-def test_build_session_no_console_error(monkeypatch):
+def test_init_session_no_console_error(monkeypatch):
     from cmd2.cmd2 import NoConsoleScreenBufferError
 
     # Mock PromptSession to raise NoConsoleScreenBufferError on first call, then succeed
@@ -3610,6 +3610,40 @@ def test_build_session_no_console_error(monkeypatch):
     assert isinstance(kwargs['output'], DummyOutput)
 
 
+def test_init_session_with_custom_tty() -> None:
+    # Create a mock stdin with says it's a TTY
+    custom_stdin = mock.MagicMock(spec=io.TextIOWrapper)
+    custom_stdin.isatty.return_value = True
+    assert custom_stdin is not sys.stdin
+
+    # Create a mock stdout which is not sys.stdout
+    custom_stdout = mock.MagicMock(spec=io.TextIOWrapper)
+    assert custom_stdout is not sys.stdout
+
+    # Check if the streams were wrapped
+    with (
+        mock.patch('cmd2.cmd2.create_input') as mock_create_input,
+        mock.patch('cmd2.cmd2.create_output') as mock_create_output,
+    ):
+        app = cmd2.Cmd()
+        app.stdin = custom_stdin
+        app.stdout = custom_stdout
+        app._init_session()
+
+        mock_create_input.assert_called_once_with(stdin=custom_stdin)
+        mock_create_output.assert_called_once_with(stdout=custom_stdout)
+
+
+def test_init_session_non_interactive() -> None:
+    # Set up a mock for a non-TTY stream (like a pipe)
+    mock_stdin = mock.MagicMock(spec=io.TextIOWrapper)
+    mock_stdin.isatty.return_value = False
+
+    app = cmd2.Cmd(stdin=mock_stdin)
+    assert isinstance(app.session.input, DummyInput)
+    assert isinstance(app.session.output, DummyOutput)
+
+
 def test_no_console_screen_buffer_error_dummy():
     from cmd2.cmd2 import NoConsoleScreenBufferError