Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 20 additions & 18 deletions cmd2/cmd2.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,8 @@
Easy transcript-based testing of applications (see examples/example.py)
Bash-style ``select`` available

Note that redirection with > and | will only work if `self.poutput()`
is used in place of `print`.
Note, if self.stdout is different than sys.stdout, then redirection with > and |
will only work if `self.poutput()` is used in place of `print`.

- Catherine Devlin, Jan 03 2008 - catherinedevlin.blogspot.com

Expand Down Expand Up @@ -200,8 +200,6 @@ def __init__(self) -> None:
self.readline_settings = _SavedReadlineSettings()
self.readline_module: Optional[ModuleType] = None
self.history: list[str] = []
self.sys_stdout: Optional[TextIO] = None
self.sys_stdin: Optional[TextIO] = None


# Contains data about a disabled command which is used to restore its original functions when the command is enabled
Expand Down Expand Up @@ -2854,9 +2852,12 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
"""
import subprocess

# Only redirect sys.stdout if it's the same as self.stdout
stdouts_match = self.stdout == sys.stdout

# Initialize the redirection saved state
redir_saved_state = utils.RedirectionSavedState(
cast(TextIO, self.stdout), sys.stdout, self._cur_pipe_proc_reader, self._redirecting
cast(TextIO, self.stdout), stdouts_match, self._cur_pipe_proc_reader, self._redirecting
)

# The ProcReader for this command
Expand Down Expand Up @@ -2912,7 +2913,10 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
raise RedirectionError(f'Pipe process exited with code {proc.returncode} before command could run')
redir_saved_state.redirecting = True # type: ignore[unreachable]
cmd_pipe_proc_reader = utils.ProcReader(proc, cast(TextIO, self.stdout), sys.stderr)
sys.stdout = self.stdout = new_stdout

self.stdout = new_stdout
if stdouts_match:
sys.stdout = self.stdout

elif statement.output:
if statement.output_to:
Expand All @@ -2926,7 +2930,10 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
raise RedirectionError('Failed to redirect output') from ex

redir_saved_state.redirecting = True
sys.stdout = self.stdout = new_stdout

self.stdout = new_stdout
if stdouts_match:
sys.stdout = self.stdout

else:
# Redirecting to a paste buffer
Expand All @@ -2944,7 +2951,10 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
# create a temporary file to store output
new_stdout = cast(TextIO, tempfile.TemporaryFile(mode="w+")) # noqa: SIM115
redir_saved_state.redirecting = True
sys.stdout = self.stdout = new_stdout

self.stdout = new_stdout
if stdouts_match:
sys.stdout = self.stdout

if statement.output == constants.REDIRECTION_APPEND:
self.stdout.write(current_paste_buffer)
Expand Down Expand Up @@ -2974,7 +2984,8 @@ def _restore_output(self, statement: Statement, saved_redir_state: utils.Redirec

# Restore the stdout values
self.stdout = cast(TextIO, saved_redir_state.saved_self_stdout)
sys.stdout = cast(TextIO, saved_redir_state.saved_sys_stdout)
if saved_redir_state.stdouts_match:
sys.stdout = self.stdout

# Check if we need to wait for the process being piped to
if self._cur_pipe_proc_reader is not None:
Expand Down Expand Up @@ -4449,22 +4460,13 @@ def _set_up_py_shell_env(self, interp: InteractiveConsole) -> _SavedCmd2Env:
# Set up sys module for the Python console
self._reset_py_display()

cmd2_env.sys_stdout = sys.stdout
sys.stdout = self.stdout # type: ignore[assignment]

cmd2_env.sys_stdin = sys.stdin
sys.stdin = self.stdin # type: ignore[assignment]

return cmd2_env

def _restore_cmd2_env(self, cmd2_env: _SavedCmd2Env) -> None:
"""Restore cmd2 environment after exiting an interactive Python shell.

:param cmd2_env: the environment settings to restore
"""
sys.stdout = cmd2_env.sys_stdout # type: ignore[assignment]
sys.stdin = cmd2_env.sys_stdin # type: ignore[assignment]

# Set up readline for cmd2
if rl_type != RlType.NONE:
# Save py's history
Expand Down
18 changes: 12 additions & 6 deletions cmd2/py_bridge.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,7 @@
"""

import sys
from contextlib import (
redirect_stderr,
redirect_stdout,
)
from contextlib import redirect_stderr
from typing import (
IO,
TYPE_CHECKING,
Expand Down Expand Up @@ -113,6 +110,8 @@ def __call__(self, command: str, *, echo: Optional[bool] = None) -> CommandResul
if echo is None:
echo = self.cmd_echo

stdouts_match = self._cmd2_app.stdout == sys.stdout

# This will be used to capture _cmd2_app.stdout and sys.stdout
copy_cmd_stdout = StdSim(cast(Union[TextIO, StdSim], self._cmd2_app.stdout), echo=echo)

Expand All @@ -126,8 +125,12 @@ def __call__(self, command: str, *, echo: Optional[bool] = None) -> CommandResul

stop = False
try:
self._cmd2_app.stdout = cast(TextIO, copy_cmd_stdout)
with redirect_stdout(cast(IO[str], copy_cmd_stdout)), redirect_stderr(cast(IO[str], copy_stderr)):
with self._cmd2_app.sigint_protection:
self._cmd2_app.stdout = cast(TextIO, copy_cmd_stdout)
if stdouts_match:
sys.stdout = self._cmd2_app.stdout

with redirect_stderr(cast(IO[str], copy_stderr)):
stop = self._cmd2_app.onecmd_plus_hooks(
command,
add_to_history=self._add_to_history,
Expand All @@ -136,6 +139,9 @@ def __call__(self, command: str, *, echo: Optional[bool] = None) -> CommandResul
finally:
with self._cmd2_app.sigint_protection:
self._cmd2_app.stdout = cast(IO[str], copy_cmd_stdout.inner_stream)
if stdouts_match:
sys.stdout = self._cmd2_app.stdout

self.stop = stop or self.stop

# Save the result
Expand Down
8 changes: 4 additions & 4 deletions cmd2/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -683,23 +683,23 @@ class RedirectionSavedState:
def __init__(
self,
self_stdout: Union[StdSim, TextIO],
sys_stdout: Union[StdSim, TextIO],
stdouts_match: bool,
pipe_proc_reader: Optional[ProcReader],
saved_redirecting: bool,
) -> None:
"""RedirectionSavedState initializer.

:param self_stdout: saved value of Cmd.stdout
:param sys_stdout: saved value of sys.stdout
:param stdouts_match: True if Cmd.stdout is equal to sys.stdout
:param pipe_proc_reader: saved value of Cmd._cur_pipe_proc_reader
:param saved_redirecting: saved value of Cmd._redirecting.
"""
# Tells if command is redirecting
self.redirecting = False

# Used to restore values after redirection ends
# Used to restore stdout values after redirection ends
self.saved_self_stdout = self_stdout
self.saved_sys_stdout = sys_stdout
self.stdouts_match = stdouts_match

# Used to restore values after command ends regardless of whether the command redirected
self.saved_pipe_proc_reader = pipe_proc_reader
Expand Down
Loading