From aad3d01939cca7ffbaf93ac4094cdfb09f69ff94 Mon Sep 17 00:00:00 2001 From: Todd Leonhardt Date: Sat, 26 Sep 2026 11:09:22 -0400 Subject: [PATCH 1/3] Run POSIX pipes to interactive programs as the terminal's foreground job A pipe process always ran in its own session, so it never received the terminal. Piped to an interactive program such as less, Ctrl-Z and fg did not suspend and resume it together with cmd2, and the program could not take the keyboard as a shell's foreground job would. When a pipe's output goes to cmd2's controlling terminal and the pipe is started on the main thread, run the pipe process in a new process group and lend it the terminal as it starts, while cmd2 writes to it, and while cmd2 waits for it. Between writes the terminal returns to cmd2, so command code can still read the keyboard. ProcReader watches the group on a thread, relays its job-control stops to cmd2's job, and forwards Ctrl-C without signalling cmd2's own group again. A shell command piped to such a program joins the pipeline's job for as long as it runs. Pipes started off the main thread, or whose output cmd2 captures, still run in their own session. The terminal tests drive cmd2 under an interactive bash on a pseudo-terminal, using pyte to read the screen, and coverage now follows the applications those tests start. Extracted from the reserved-row toolbar work (PR #1761) without the toolbar. --- CHANGELOG.md | 13 + cmd2/cmd2.py | 223 ++++++--- cmd2/utils.py | 347 +++++++++++++- docs/features/redirection.md | 4 + pyproject.toml | 6 + tests/test_cmd2.py | 119 ++++- tests/test_pipeline_job_control.py | 722 +++++++++++++++++++++++++++++ tests/test_utils.py | 408 ++++++++++++++++ 8 files changed, 1772 insertions(+), 70 deletions(-) create mode 100644 tests/test_pipeline_job_control.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 552217630..6d6af0573 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,16 @@ +## 4.2.5 (TBD) + +- Bug Fixes + - On POSIX, piping a command's output to an interactive program such as `less` + (`help -v | less`) now runs that program as the terminal's foreground job, as a shell pipeline + does. It had run in a separate session that never received the terminal, so Ctrl-Z and `fg` + did not suspend and resume it together with cmd2. The program now owns the terminal as it + starts, so a pager can set its terminal modes, and Ctrl-C and Ctrl-Z reach the whole pipeline. + Pipes started from a worker thread, or whose output cmd2 captures, still run in their own + session + - A `shell` command piped to an interactive program, such as `shell git log | less`, now joins + the pipeline's job, so both processes receive Ctrl-C and Ctrl-Z + ## 4.2.4 (September 8, 2026) - Bug Fixes diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index fcd26b2e4..84aae5185 100644 --- a/cmd2/cmd2.py +++ b/cmd2/cmd2.py @@ -3330,46 +3330,102 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState: subproc_stdin = open(read_fd, encoding="utf-8") # noqa: SIM115 new_stdout: TextIO = cast(TextIO, open(write_fd, "w", encoding="utf-8")) # noqa: SIM115 - # 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). + # Isolate pipeline signals from cmd2. Terminal pipelines receive the + # foreground terminal; ProcReader relays their job-control stops. kwargs: dict[str, Any] = {} if sys.platform == "win32": kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP else: - kwargs["start_new_session"] = True - # Attempt to run the pipe process in the user's preferred shell instead of the default behavior of using sh. shell = os.environ.get("SHELL") if shell: kwargs["executable"] = shell # For any stream that is a StdSim, we will use a pipe so we can capture its output - proc = subprocess.Popen( # noqa: S602 - statement.redirect_to, - stdin=subproc_stdin, - stdout=subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout, # type: ignore[unreachable] - stderr=subprocess.PIPE if isinstance(sys.stderr, utils.StdSim) else sys.stderr, - shell=True, - **kwargs, - ) + pipe_stdout = None if isinstance(self.stdout, utils.StdSim) else self.stdout # type: ignore[unreachable] + pipe_stderr = None if isinstance(sys.stderr, utils.StdSim) else sys.stderr + + terminal_fd = None + if sys.platform != "win32": + # Job control installs signal handlers, which only the main thread may do. + # Elsewhere, keep the pipeline in its own session as before. + if threading.current_thread() is threading.main_thread(): + for stream in (pipe_stdout, pipe_stderr): + if stream is not None and stream.isatty(): + with contextlib.suppress(OSError, ValueError): + if os.tcgetpgrp(stream.fileno()) == os.getpgrp(): + terminal_fd = stream.fileno() + break + if terminal_fd is None: + kwargs["start_new_session"] = True + else: + kwargs["process_group"] = 0 - # 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 - # started OK, since the shell itself always starts. Therefore, we will wait a short time and check - # if the pipe process is still running. - with contextlib.suppress(subprocess.TimeoutExpired): - proc.wait(0.2) + with contextlib.ExitStack() as terminal_stack: + with contextlib.ExitStack() as spawn_stack: + if terminal_fd is not None and os.getpgrp() == os.getsid(0): + import signal - # Check if the pipe process already exited - if proc.returncode is not None: + # A session leader's job has no outer shell to resume it. + # Its pipeline must inherit the same Ctrl-Z behavior: the + # new group would otherwise make SIGTSTP actionable again. + previous_tstp = signal.signal(signal.SIGTSTP, signal.SIG_IGN) + spawn_stack.callback(signal.signal, signal.SIGTSTP, previous_tstp) + proc = subprocess.Popen( # noqa: S602 + statement.redirect_to, + stdin=subproc_stdin, + stdout=subprocess.PIPE if pipe_stdout is None else pipe_stdout, + stderr=subprocess.PIPE if pipe_stderr is None else pipe_stderr, + shell=True, + **kwargs, + ) + # Only the child should own a read end. In particular, a consumer + # exit must unblock a producer writing to a full pipe immediately. subproc_stdin.close() - new_stdout.close() - raise RedirectionError(f"Pipe process exited with code {proc.returncode} before command could run") - redir_saved_state.redirecting = True - cmd_pipe_proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr) + if terminal_fd is not None: + cmd_pipe_proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr, terminal_fd=terminal_fd) + terminal_stack.enter_context(cmd_pipe_proc_reader.manage_terminal()) + + # 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 + # started OK, since the shell itself always starts. Therefore, we will wait a short time and check + # if the pipe process is still running. + with contextlib.suppress(subprocess.TimeoutExpired): + if cmd_pipe_proc_reader is None: + proc.wait(0.2) + else: + # A pager such as less sets its terminal modes as it starts, before it + # reads the pipe. It must own the terminal by then: a background + # tcsetattr() stops it with SIGTTOU, and on macOS that call fails with + # EINTR when the process is continued instead of being restarted. less + # ignores the failure and runs on a cooked terminal. + with cmd_pipe_proc_reader.lend_terminal(): + cmd_pipe_proc_reader.wait_for_exit(0.2) + + # Check if the pipe process already exited + if proc.returncode is not None: + if cmd_pipe_proc_reader is not None: + cmd_pipe_proc_reader.wait() + subproc_stdin.close() + new_stdout.close() + raise RedirectionError(f"Pipe process exited with code {proc.returncode} before command could run") + redir_saved_state.redirecting = True + if cmd_pipe_proc_reader is None: + cmd_pipe_proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr) - self.stdout = new_stdout + if terminal_fd is not None: + import io + + pipe_fd = os.dup(new_stdout.fileno()) + new_stdout.close() + new_stdout = io.TextIOWrapper( + io.BufferedWriter(utils.PipelineWriter(pipe_fd, cmd_pipe_proc_reader)), encoding="utf-8" + ) + + self.stdout = new_stdout + + # Keep the pipeline's job control until _restore_output() reaps the pipe process. + redir_saved_state.pipeline_job = terminal_stack.pop_all() elif statement.redirector in (constants.REDIRECTION_OVERWRITE, constants.REDIRECTION_APPEND): if statement.redirect_to: @@ -3428,29 +3484,41 @@ def _restore_output(self, statement: Statement, saved_redir_state: utils.Redirec :param statement: Statement object which contains the parsed input from the user :param saved_redir_state: contains information needed to restore state data """ - if saved_redir_state.redirecting: - # If we redirected output to the clipboard - if ( - statement.redirector in (constants.REDIRECTION_OVERWRITE, constants.REDIRECTION_APPEND) - and not statement.redirect_to - ): - self.stdout.seek(0) - write_to_paste_buffer(self.stdout.read()) + # The pipeline's job control ends once its pipe process has been reaped. + with contextlib.ExitStack() as terminal_stack: + if saved_redir_state.pipeline_job is not None: + terminal_stack.callback(saved_redir_state.pipeline_job.close) + saved_redir_state.pipeline_job = None - with contextlib.suppress(BrokenPipeError): - # Close the file or pipe that stdout was redirected to - self.stdout.close() - - # Restore self.stdout - self.stdout = cast(TextIO, saved_redir_state.saved_self_stdout) - - # Check if we need to wait for the process being piped to - if self._cur_pipe_proc_reader is not None: - self._cur_pipe_proc_reader.wait() - - # These are restored regardless of whether the command redirected - self._cur_pipe_proc_reader = saved_redir_state.saved_pipe_proc_reader - self._redirecting = saved_redir_state.saved_redirecting + try: + if saved_redir_state.redirecting: + # If we redirected output to the clipboard + if ( + statement.redirector in (constants.REDIRECTION_OVERWRITE, constants.REDIRECTION_APPEND) + and not statement.redirect_to + ): + self.stdout.seek(0) + write_to_paste_buffer(self.stdout.read()) + + with contextlib.suppress(BrokenPipeError): + # Close the file or pipe that stdout was redirected to + if self._cur_pipe_proc_reader is not None: + self._cur_pipe_proc_reader.finish_producer() + self.stdout.close() + + # Restore self.stdout + self.stdout = cast(TextIO, saved_redir_state.saved_self_stdout) + + # Check if we need to wait for the process being piped to. Handing the + # terminal back as it finishes can fail, for example after a hangup. + if self._cur_pipe_proc_reader is not None: + self._cur_pipe_proc_reader.wait() + finally: + # These are restored regardless of whether the command redirected, or whether + # restoring it failed: a pipeline left current would keep ppaged() from paging + # and send Ctrl-C to a process group that is gone. + self._cur_pipe_proc_reader = saved_redir_state.saved_pipe_proc_reader + self._redirecting = saved_redir_state.saved_redirecting def get_command_func(self, command: str) -> BoundCommandFunc[...] | None: """Get the bound command function for a command. @@ -4918,19 +4986,52 @@ def do_shell(self, args: argparse.Namespace) -> None: utils.expand_user_in_tokens(tokens) expanded_command = " ".join(tokens) - # Prevent KeyboardInterrupts while in the shell process. The shell process will - # still receive the SIGINT since it is in the same process group as us. - with self.sigint_protection: - # For any stream that is a StdSim, we will use a pipe so we can capture its output - proc = subprocess.Popen( # noqa: S602 - expanded_command, - stdout=subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout, # type: ignore[unreachable] - stderr=subprocess.PIPE if isinstance(sys.stderr, utils.StdSim) else sys.stderr, - shell=True, - **kwargs, - ) - - proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr) + # A terminal pipeline's consumer needs the terminal to drain the pipe, but a shell + # command writes into that pipe itself rather than through self.stdout, which lends + # the terminal per write. Run the command inside the pipeline's job instead, for as + # long as it runs: the consumer keeps the terminal, and Ctrl-C and Ctrl-Z reach both + # processes, as they would in a shell pipeline. + pipeline = self._cur_pipe_proc_reader + pipeline_group = None + if pipeline is not None and not isinstance(self.stdout, utils.StdSim): # type: ignore[unreachable] + pipeline_group = pipeline.terminal_group + + # Prevent KeyboardInterrupts while in the shell process. The shell process still + # receives the SIGINT: it is in our process group or in the foreground pipeline's. + with self.sigint_protection, contextlib.ExitStack() as terminal_stack: + if pipeline is not None and pipeline_group is not None: + kwargs["process_group"] = pipeline_group + terminal_stack.enter_context(pipeline.lend_terminal()) + while True: + try: + # For any stream that is a StdSim, we will use a pipe so we can capture its output. + # A command joining the pipeline is spawned inside the lend, which blocks SIGTTOU. + with utils.unblocked_sigttou() if "process_group" in kwargs else contextlib.nullcontext(): + proc = subprocess.Popen( # noqa: S602 + expanded_command, + stdout=subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout, # type: ignore[unreachable] + stderr=subprocess.PIPE if isinstance(sys.stderr, utils.StdSim) else sys.stderr, + shell=True, + **kwargs, + ) + break + except PermissionError: + # The pipeline exited before the command could join its group. + if kwargs.pop("process_group", None) is None: + raise + # The retry runs in our own group, so take the terminal back from the dead + # pipeline first. Its watcher left it lent, and the command would otherwise + # stop with SIGTTIN on its first terminal read, with nothing to resume it. + terminal_stack.close() + + # A command that joined the pipeline's job is waited for in short polls. Only the + # main thread runs Python signal handlers, and the job-control stop the pipeline's + # watcher relays may wake another thread. Once the consumer and its watcher are + # gone, the same wait relays the command's own stops, such as Ctrl-Z. + joined_pipeline = pipeline if "process_group" in kwargs else None + proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr, pipeline=joined_pipeline) + if joined_pipeline is not None: + proc_reader.wait_for_exit() proc_reader.wait() # Save the return code of the application for use in a pyscript diff --git a/cmd2/utils.py b/cmd2/utils.py index f88a46f20..23fda0c83 100644 --- a/cmd2/utils.py +++ b/cmd2/utils.py @@ -1,9 +1,11 @@ """Shared utility functions.""" import contextlib +import errno import functools import glob import inspect +import io import itertools import os import re @@ -13,6 +15,7 @@ from collections.abc import ( Callable, Iterable, + Iterator, MutableSequence, ) from difflib import SequenceMatcher @@ -533,22 +536,59 @@ def write(self, b: bytes) -> None: self.std_sim_instance.flush() +@contextlib.contextmanager +def unblocked_sigttou() -> Iterator[None]: + """Let a child started inside :meth:`ProcReader.lend_terminal` keep normal job control. + + The lend blocks SIGTTOU for its thread, and a child inherits that mask for life. Spawning + touches no terminal, so unblocking it for the spawn alone cannot stop this thread. + """ + import signal + + previous_mask = signal.pthread_sigmask(signal.SIG_UNBLOCK, {signal.SIGTTOU}) + try: + yield + finally: + signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask) + + class ProcReader: """Used to capture stdout and stderr from a Popen process if any of those were set to subprocess.PIPE. If neither are pipes, then the process will run normally and no output will be captured. """ - def __init__(self, proc: PopenTextIO, stdout: StdSim | TextIO, stderr: StdSim | TextIO) -> None: + def __init__( + self, + proc: PopenTextIO, + stdout: StdSim | TextIO, + stderr: StdSim | TextIO, + *, + terminal_fd: int | None = None, + pipeline: "ProcReader | None" = None, + ) -> None: """ProcReader initializer. :param proc: the Popen process being read from :param stdout: the stream to write captured stdout :param stderr: the stream to write captured stderr. + :param terminal_fd: controlling terminal to lend to a POSIX process in its own group + :param pipeline: terminal pipeline whose process group proc joined as a producer """ self._proc = proc self._stdout = stdout self._stderr = stderr + self._terminal_fd = terminal_fd + self._pipeline = pipeline + self._process_done = threading.Event() + self._producer_finished = False + self._terminal_available = threading.Event() + # Lends in progress, guarded by _terminal_lock. The terminal is available while any is. + self._lends = 0 + self._terminal_lock = threading.RLock() + self._job_resumed = threading.Event() + if terminal_fd is not None: + self._original_group = os.tcgetpgrp(terminal_fd) self._out_thread = threading.Thread(name="out_thread", target=self._reader_thread_func, kwargs={"read_stdout": True}) @@ -573,16 +613,261 @@ def send_sigint(self) -> None: # 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 + # Pipelines lead their own group. A shell command that joined it, such + # as `shell sleep 100 | head -1`, can outlive the reaped consumer. + group_id = self._proc.pid + # Never re-signal our own group: other ProcReader callers may share it + # and already received Ctrl-C. + if group_id != os.getpgrp(): + with contextlib.suppress(ProcessLookupError): + os.killpg(group_id, signal.SIGINT) def terminate(self) -> None: """Terminate the process.""" - self._proc.terminate() + if self._terminal_fd is None: + self._proc.terminate() + else: + import signal + + # Popen.terminate() polls first, which would compete with our waitpid thread. + with contextlib.suppress(ProcessLookupError): + os.kill(self._proc.pid, signal.SIGTERM) + + @property + def terminal_group(self) -> int | None: + """Process group of a running terminal pipeline, which a producer may join, or None.""" + if self._terminal_fd is None or self._proc.returncode is not None: + return None + return self._proc.pid + + @staticmethod + def _set_foreground_group(terminal_fd: int, group_id: int) -> None: + """Transfer the terminal without stopping this background thread with SIGTTOU.""" + import signal + + previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGTTOU}) + try: + os.tcsetpgrp(terminal_fd, group_id) + finally: + signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask) + + @contextlib.contextmanager + def manage_terminal(self) -> Iterator[None]: + """Watch the pipeline and suspend the shell's whole job on the main thread.""" + import signal + + terminal_fd = self._terminal_fd + if terminal_fd is None: + yield + return + previous_handler = signal.getsignal(signal.SIGTSTP) + + def suspend_job(signum: int, frame: Any) -> None: + try: + if previous_handler != signal.SIG_DFL: + if callable(previous_handler): + previous_handler(signum, frame) + return + if os.tcgetpgrp(terminal_fd) == self._proc.pid: + self._set_foreground_group(terminal_fd, self._original_group) + # Ignore our group-directed copy, then stop this thread synchronously. + # Wrappers in our job must stop too. Unlike SIGSTOP, SIGTSTP is + # discarded for orphaned groups, which have no shell to resume them. + signal.signal(signal.SIGTSTP, signal.SIG_IGN) + try: + os.killpg(self._original_group, signal.SIGTSTP) + signal.signal(signal.SIGTSTP, signal.SIG_DFL) + signal.raise_signal(signal.SIGTSTP) + finally: + signal.signal(signal.SIGTSTP, suspend_job) + finally: + try: + if self._terminal_available.is_set() and os.tcgetpgrp(terminal_fd) == self._original_group: + self._set_foreground_group(terminal_fd, self._proc.pid) + finally: + self._job_resumed.set() + + signal.signal(signal.SIGTSTP, suspend_job) + try: + threading.Thread(name="pipe_job", target=self._wait_for_job, args=(terminal_fd,), daemon=True).start() + yield + finally: + signal.signal(signal.SIGTSTP, previous_handler) + + @contextlib.contextmanager + def lend_terminal(self) -> Iterator[None]: + """Lend the terminal only while writing to or waiting for the consumer. + + Command code retains foreground access between writes, including arbitrary + reads through input(), getpass(), or third-party libraries. Lending during + writes lets an interactive consumer drain a full pipe without deadlocking. + """ + import signal + + terminal_fd = self._terminal_fd + if terminal_fd is None or self._proc.returncode is not None: + yield + return + # While the consumer owns the terminal, a signal handler run on this thread may still + # write diagnostics to it. Block SIGTTOU for the lend only: a signal mask survives fork + # and exec, so blocking it for the whole pipeline would leak into every child the + # command starts. A child started during a lend must unblock it; see unblocked_sigttou(). + previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGTTOU}) + try: + with self._terminal_lock: + try: + self._set_foreground_group(terminal_fd, self._proc.pid) + except OSError as error: + # The group can disappear before the watcher has reaped its leader. + if error.errno not in (errno.ESRCH, errno.EINVAL): + raise + self._lends += 1 + self._terminal_available.set() + try: + yield + finally: + with self._terminal_lock: + self._lends -= 1 + # Lends overlap: do_shell() lends for as long as a shell producer runs, + # while a pipe write from another thread lends and returns. Only the last + # to end takes the terminal back, or the producer would stop with SIGTTIN. + if not self._lends: + self._terminal_available.clear() + if os.tcgetpgrp(terminal_fd) == self._proc.pid: + self._set_foreground_group(terminal_fd, self._original_group) + finally: + signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask) + + def _wait_for_job(self, terminal_fd: int) -> None: + """Reap a foreground pipeline and relay its stops to the outer shell's job. + + This is the only waitpid caller for a terminal pipeline. Watching on a separate + thread also catches Ctrl-Z while the command is blocked writing to its pipe. + """ + import signal + + try: + while True: + _, status = os.waitpid(self._proc.pid, os.WUNTRACED) + if not os.WIFSTOPPED(status): + self._proc.returncode = os.waitstatus_to_exitcode(status) + return + if os.WSTOPSIG(status) in (signal.SIGTTIN, signal.SIGTTOU): + # Command code owns the terminal between pipe writes. Defer + # consumer terminal access until the next write or final wait. + while True: + self._terminal_available.wait(0.1) + with self._terminal_lock: + # A stopped consumer can be killed before another write. + # Keep reaping even while command code owns the terminal. + pid, pending_status = os.waitpid(self._proc.pid, os.WNOHANG | os.WUNTRACED) + if pid and not os.WIFSTOPPED(pending_status): + self._proc.returncode = os.waitstatus_to_exitcode(pending_status) + return + # A short write may already have returned the terminal. + # Do not turn that ordinary handoff into a job suspension. + if not self._terminal_available.is_set(): + continue + foreground = os.tcgetpgrp(terminal_fd) + if foreground == self._proc.pid: + os.killpg(self._proc.pid, signal.SIGCONT) + break + if foreground == self._proc.pid: + continue + + if os.tcgetpgrp(terminal_fd) == self._proc.pid: + self._set_foreground_group(terminal_fd, self._original_group) + # Stop every terminal reader before returning control to the outer shell. + os.killpg(self._proc.pid, signal.SIGSTOP) + self._job_resumed.clear() + # Signal the main thread itself. Only it runs Python signal handlers, and a + # process-directed signal may be taken by another thread while the main + # thread sleeps in a system call, which then never returns to run the handler. + signal.pthread_kill(threading.main_thread().ident or 0, signal.SIGTSTP) + self._job_resumed.wait() + os.killpg(self._proc.pid, signal.SIGCONT) + finally: + try: + with self._terminal_lock: + # A shell producer in this group may outlive the consumer and still read + # the terminal. While a lend is active, its holder returns the terminal. + if not self._terminal_available.is_set() and os.tcgetpgrp(terminal_fd) == self._proc.pid: + self._set_foreground_group(terminal_fd, self._original_group) + finally: + self._process_done.set() + + def _relay_producer_stop(self) -> None: + """Suspend the shell's whole job for a stopped producer that outlived the consumer. + + Ctrl-Z reaches only the foreground group, and the producer may be all that is + left of it. The watcher ended with the consumer, so nothing else relays the stop. + """ + import signal + + terminal_fd = self._terminal_fd + if terminal_fd is None or not self._process_done.is_set(): + # A live watcher relays the consumer's stop and continues the whole group. + return + with self._terminal_lock: + if os.tcgetpgrp(terminal_fd) == self._proc.pid: + self._set_foreground_group(terminal_fd, self._original_group) + with contextlib.suppress(ProcessLookupError): + os.killpg(self._proc.pid, signal.SIGSTOP) + self._job_resumed.clear() + signal.pthread_kill(threading.main_thread().ident or 0, signal.SIGTSTP) + self._job_resumed.wait() + with contextlib.suppress(ProcessLookupError): + os.killpg(self._proc.pid, signal.SIGCONT) + + def _wait_for_producer(self, pipeline: "ProcReader", timeout: float | None) -> None: + """Wait for a producer in a terminal pipeline's job, relaying its job-control stops. + + This is the only waitpid caller for such a producer. It polls so that the main + thread keeps returning to Python code, where signal handlers run. + """ + import time + + deadline = None if timeout is None else time.monotonic() + timeout + while self._proc.returncode is None: + pid, status = os.waitpid(self._proc.pid, os.WNOHANG | os.WUNTRACED) + if not pid: + if deadline is not None and time.monotonic() >= deadline: + raise subprocess.TimeoutExpired(self._proc.args, timeout or 0) + time.sleep(0.05) + elif os.WIFSTOPPED(status): + pipeline._relay_producer_stop() + else: + self._proc.returncode = os.waitstatus_to_exitcode(status) + + def finish_producer(self) -> None: + """Disable producer cancellation before flushing and closing its pipe.""" + self._producer_finished = True + + def wait_for_exit(self, timeout: float | None = None) -> None: + """Wait for process exit without competing with the terminal job's waitpid thread. + + :param timeout: maximum seconds to wait, or None to wait indefinitely + :raises subprocess.TimeoutExpired: if the process is still running after timeout + """ + if self._pipeline is not None: + self._wait_for_producer(self._pipeline, timeout) + elif self._terminal_fd is None: + self._proc.wait(timeout) + elif timeout is None: + # A process-directed signal may reach a worker thread. Python still runs + # its handler on the main thread, so periodically return from the wait + # to dispatch it even when the main thread's system call was not interrupted. + while not self._process_done.wait(0.1): + pass + elif not self._process_done.wait(timeout): + raise subprocess.TimeoutExpired(self._proc.args, timeout) def wait(self) -> None: """Wait for the process to finish.""" + if self._terminal_fd is not None: + with self.lend_terminal(): + self.wait_for_exit() if self._out_thread.is_alive(): self._out_thread.join() if self._err_thread.is_alive(): @@ -614,7 +899,8 @@ def _reader_thread_func(self, read_stdout: bool) -> None: raise ValueError("read_stream is None") # Run until process completes - while self._proc.poll() is None: + polled = self._terminal_fd is None and self._pipeline is None + while (self._proc.poll() if polled else self._proc.returncode) is None: available = read_stream.peek() # type: ignore[attr-defined, ty:unresolved-attribute] if available: read_stream.read(len(available)) @@ -635,6 +921,54 @@ def _write_bytes(stream: StdSim | TextIO, to_write: bytes | str) -> None: stream.buffer.write(to_write) +class PipelineWriter(io.FileIO): + """A pipe whose blocking writes temporarily give the consumer terminal access.""" + + def __init__(self, fd: int, reader: ProcReader) -> None: + """Take ownership of a pipe descriptor managed by reader.""" + super().__init__(fd, "w") + self._reader = reader + + def write(self, b: Any) -> int: + """Write all of b while the consumer can interact with the terminal. + + The whole buffer goes out under one lend. Returning the terminal between two + writes, even for an instant, would stop a consumer that had just resumed a + terminal read with SIGTTIN. + + A full pipe is awaited in short polls rather than in one blocking write. Only + the main thread runs Python signal handlers, and the job-control stop ProcReader + relays may wake another thread, so the main thread has to return to Python code + on its own for the handler to run. The descriptor itself stays blocking: a shell + command inherits it, and a producer that found it non-blocking would fail with + EAGAIN once the pipe filled. + """ + import select + import signal + + view = memoryview(b).cast("B") + fd = self.fileno() + poller = select.poll() + poller.register(fd, select.POLLOUT) + try: + with self._reader.lend_terminal(): + written = 0 + while written < len(view): + # Once there is room, a write of at most PIPE_BUF bytes does not block. + if poller.poll(100): + written += os.write(fd, view[written : written + select.PIPE_BUF]) + return written + except BrokenPipeError: + # Ctrl-C during a blocking write must cancel the command, even if it + # normally catches BrokenPipeError. Raise here rather than signaling + # asynchronously: a late signal could interrupt redirection cleanup. + with contextlib.suppress(subprocess.TimeoutExpired): + self._reader.wait_for_exit(0.2) + if not self._reader._producer_finished and self._reader._proc.returncode in (-signal.SIGINT, 128 + signal.SIGINT): + raise KeyboardInterrupt from None + raise + + class ContextFlag: """A context manager which is also used as a boolean flag value within the default sigint handler. @@ -690,6 +1024,9 @@ def __init__( self.saved_pipe_proc_reader = pipe_proc_reader self.saved_redirecting = saved_redirecting + # Holds a terminal pipeline's job control until its pipe process has been reaped + self.pipeline_job: contextlib.ExitStack | None = None + def categorize(func: Callable[..., Any] | Iterable[Callable[..., Any]], category: str) -> None: """Categorize a function. diff --git a/docs/features/redirection.md b/docs/features/redirection.md index 27a238caa..27c8c79c7 100644 --- a/docs/features/redirection.md +++ b/docs/features/redirection.md @@ -30,6 +30,10 @@ Piping the output of a `cmd2` command to a shell command works just like in POSI - pipe as input to a shell command with `|`, as in `mycommand args | wc` +On POSIX systems, a pipe to an interactive program such as `less` runs as the terminal's foreground +job, as it would in a shell: the program can read the keyboard, and Ctrl-C and Ctrl-Z reach it. A +`shell` command whose output is piped this way, as in `shell git log | less`, joins the same job. + ## Multiple Pipes and Redirection Multiple pipes, optionally followed by a redirect, are supported. Thus, it is possible to do diff --git a/pyproject.toml b/pyproject.toml index ec98c9ef6..cd68a4b1f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,6 +43,7 @@ dev = [ "mkdocstrings[python]>=1", "mypy>=2.3.1", "prek>=0.3.5", + "pyte>=0.8.2", "pytest>=8.1.1", "pytest-cov>=5", "pytest-mock>=3.14.1", @@ -62,6 +63,7 @@ quality = ["prek>=0.3.5"] test = [ "codecov>=2.1", "coverage>=7.11.3", + "pyte>=0.8.2", "pytest>=8.1.1", "pytest-cov>=5", "pytest-mock>=3.14.1", @@ -104,6 +106,8 @@ warn_unused_ignores = false testpaths = ["tests"] addopts = [ "-n=auto", + # Spread slow terminal cases across workers instead of queuing them in one batch. + "--maxschedchunk=1", "--cov=cmd2", "--cov-config=pyproject.toml", "--cov-report=xml", @@ -112,6 +116,8 @@ addopts = [ ] [tool.coverage.run] +# Include the cmd2 applications launched by the terminal integration tests. +patch = ["subprocess"] # Use sys.monitoring on Python 3.12+; coverage falls back with a warning on 3.11. core = "sysmon" source = ["cmd2"] diff --git a/tests/test_cmd2.py b/tests/test_cmd2.py index 24009acfd..f4cb4793e 100644 --- a/tests/test_cmd2.py +++ b/tests/test_cmd2.py @@ -428,6 +428,55 @@ def test_shell_manual_call(base_app) -> None: base_app.do_shell(cmd) +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups") +def test_shell_falls_back_to_own_group_when_pipeline_exited(base_app, tmp_path) -> None: + import contextlib + import subprocess + from unittest import mock + + # A group whose only member has exited cannot be joined. The consumer of a terminal + # pipeline can exit between the check and the spawn, like `shell sleep 1 | true`. + leader = subprocess.Popen([sys.executable, "-c", "pass"], process_group=0) + leader.wait() + lent = [] + + @contextlib.contextmanager + def lend_terminal(): + lent.append(True) + try: + yield + finally: + lent.pop() + + # The retry runs in our own group, so the terminal has to come back from the dead + # pipeline first. Otherwise the command stops with SIGTTIN on its first terminal read. + spawned_while_lent = [] + real_popen = subprocess.Popen + + def popen(*args, **kwargs): + spawned_while_lent.append(bool(lent)) + return real_popen(*args, **kwargs) + + base_app._cur_pipe_proc_reader = mock.Mock(terminal_group=leader.pid, lend_terminal=lend_terminal) + with (tmp_path / "output").open("w+") as output, mock.patch("subprocess.Popen", popen): + base_app.stdout = output + base_app.do_shell("echo joined") + output.seek(0) + assert output.read() == "joined\n" + assert base_app.last_result == 0 + assert spawned_while_lent == [True, False] + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell executable") +def test_shell_permission_error_unrelated_to_pipeline(base_app, tmp_path, monkeypatch) -> None: + unusable_shell = tmp_path / "shell" + unusable_shell.write_text("#!/bin/sh\n") + unusable_shell.chmod(0o644) + monkeypatch.setenv("SHELL", str(unusable_shell)) + with pytest.raises(PermissionError): + base_app.do_shell("echo hi") + + def test_base_error(base_app) -> None: _out, err = run_cmd(base_app, "meow") assert "is not a recognized command" in err[0] @@ -865,7 +914,11 @@ def test_pipe_to_shell_and_redirect(redirection_app, running_pipe_process) -> No os.remove(filename) -def test_pipe_to_shell_error(redirection_app, mocker, capsys) -> None: +@pytest.mark.parametrize( + "terminal", + [False, pytest.param(True, marks=pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control"))], +) +def test_pipe_to_shell_error(redirection_app, mocker, capsys, terminal) -> None: """An already-exited pipe process must be reported before the command runs. A real nonexistent command may take longer than the startup probe under load. @@ -876,15 +929,73 @@ def test_pipe_to_shell_error(redirection_app, mocker, capsys) -> None: process = popen.return_value process.returncode = 127 process.wait.return_value = 127 - - out, err = run_cmd(redirection_app, "print_output | foobarbaz.this_does_not_exist") + if terminal: + terminal_stream = mocker.Mock() + terminal_stream.isatty.return_value = True + terminal_stream.fileno.return_value = 10 + redirection_app.stdout = terminal_stream + mocker.patch("os.tcgetpgrp", return_value=os.getpgrp()) + mocker.patch("os.getsid", return_value=os.getpgrp()) + sigmask = mocker.patch("signal.pthread_sigmask", return_value=set()) + reader = mocker.patch("cmd2.utils.ProcReader").return_value + previous_tstp = signal.getsignal(signal.SIGTSTP) + + def start_pipe(*args, **kwargs): + # Session-led pipelines inherit ignored Ctrl-Z, but the caller's + # handler must be restored even when startup reports an early exit. + assert signal.getsignal(signal.SIGTSTP) == signal.SIG_IGN + return process + + popen.side_effect = start_pipe + + if terminal: + # run_cmd captures stderr in a StdSim, which deliberately disables terminal handoff. + redirection_app.onecmd_plus_hooks("print_output | foobarbaz.this_does_not_exist") + out, error_text = capsys.readouterr() + err = error_text.splitlines() + else: + out, err = run_cmd(redirection_app, "print_output | foobarbaz.this_does_not_exist") assert not out assert "Pipe process exited with code 127 before command could run" in " ".join(err) assert capsys.readouterr().out == "" - process.wait.assert_called_once() + if terminal: + assert signal.getsignal(signal.SIGTSTP) == previous_tstp + reader.wait_for_exit.assert_called_once_with(0.2) + reader.wait.assert_called_once_with() + process.wait.assert_not_called() + # SIGTTOU is blocked only inside ProcReader's lends. Blocking it for the whole + # pipeline would leak the mask into every child the command starts. + sigmask.assert_not_called() + else: + process.wait.assert_called_once() assert popen.call_args.kwargs["stdin"].closed +def test_restore_output_resets_pipe_state_when_the_wait_fails(base_app) -> None: + """A failed handback while waiting for the pipe process must not leave it current. + + Otherwise ppaged() would never page again, and Ctrl-C would keep going to a dead group. + """ + import errno + + statement = base_app.statement_parser.parse("help | less") + saved_stdout = base_app.stdout + saved = cmd2.utils.RedirectionSavedState(saved_stdout, None, False) + saved.redirecting = True + reader = mock.Mock() + reader.wait.side_effect = OSError(errno.EIO, "terminal hung up") + base_app._cur_pipe_proc_reader = reader + base_app._redirecting = True + base_app.stdout = io.StringIO() + + with pytest.raises(OSError, match="terminal hung up"): + base_app._restore_output(statement, saved) + + assert base_app.stdout is saved_stdout + assert base_app._cur_pipe_proc_reader is None + assert base_app._redirecting is False + + def test_send_to_paste_buffer(redirection_app: RedirectionApp, capsys: pytest.CaptureFixture[str], mocker) -> None: # Exercise cmd2's real clipboard redirection against a private backend, not the # shared OS clipboard (which another test run or desktop application can alter). diff --git a/tests/test_pipeline_job_control.py b/tests/test_pipeline_job_control.py new file mode 100644 index 000000000..1ad79e343 --- /dev/null +++ b/tests/test_pipeline_job_control.py @@ -0,0 +1,722 @@ +"""Exercise pipeline job control through a real controlling terminal and outer shell.""" + +import codecs +import contextlib +import os +import re +import select +import shlex +import shutil +import signal +import subprocess +import sys +import time +from pathlib import Path + +import pyte +import pytest + +pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="POSIX job control") + + +def describe_processes(root: int, master: int) -> str: + """Report the terminal's foreground group and every descendant of root, for a timeout. + + A silent transcript says only that nothing happened. Process states (T for stopped) + and wait channels say which process was waiting for whom. + """ + try: + foreground: object = os.tcgetpgrp(master) + except OSError as error: + foreground = error + try: + listing = subprocess.run( + ["ps", "-e", "-o", "pid,ppid,pgid,stat,wchan,command"], capture_output=True, text=True, check=False + ) + except OSError as error: + return f"foreground process group: {foreground}\nno process listing: {error}" + rows = listing.stdout.splitlines() + parents = {} + for row in rows[1:]: + fields = row.split(maxsplit=2) + if len(fields) >= 2 and fields[0].isdigit() and fields[1].isdigit(): + parents[int(fields[0])] = int(fields[1]) + family = {root} + while True: + grown = family | {pid for pid, parent in parents.items() if parent in family} + if grown == family: + break + family = grown + described = [row for row in rows[1:] if row.split(maxsplit=1)[0].isdigit() and int(row.split(maxsplit=1)[0]) in family] + return "\n".join([f"foreground process group: {foreground}", rows[0] if rows else "", *described]) + + +@pytest.mark.parametrize( + ("finish", "stop_job", "shell_child", "launcher", "producer", "relay"), + [ + pytest.param("interrupts", True, False, "direct", "command", "main", id="direct-signals-and-job-control"), + pytest.param("interrupts", True, True, "sh", "command", "main", id="wrapper-signals-and-job-control"), + pytest.param("exit_sigint", False, False, "direct", "command", "main", id="interrupt-busy-producer"), + pytest.param("exit_sigint", True, True, "uv", "command", "worker", id="uv-stop-and-interrupt-busy-producer"), + pytest.param("exit_sigint", False, False, "direct", "shell", "main", id="interrupt-busy-shell-producer"), + pytest.param("exit_sigint", True, True, "sh", "shell", "worker", id="wrapper-stop-and-interrupt-busy-shell-producer"), + pytest.param("read_input", False, False, "direct", "command", "main", id="nested-prompt"), + pytest.param("shell_input", False, True, "direct", "command", "main", id="shell-input"), + pytest.param("direct_input", False, False, "sh", "command", "main", id="wrapper-direct-input"), + pytest.param("direct_input", False, False, "exec", "command", "main", id="direct-input-in-orphaned-session"), + pytest.param("interrupts", False, False, "exec", "command", "main", id="orphaned-job-control"), + ], +) +def test_pipeline_stops_with_cmd2_and_returns_terminal( + tmp_path, finish, stop_job, shell_child, launcher, producer, relay +) -> None: + import fcntl + import pty + import struct + import termios + + shell = shutil.which("bash") + if shell is None: + pytest.skip("requires an interactive bash shell") + pager = tmp_path / "pager.py" + pager_pid = tmp_path / "pager.pid" + interrupts = tmp_path / "interrupts" + pager.write_text( + "import errno, os, pathlib, signal, sys, termios, tty\n" + f"if {launcher == 'exec'!r}: assert signal.getsignal(signal.SIGTSTP) == signal.SIG_IGN\n" + # Interactive bash leaves TTIN/TTOU ignored when exec replaces it. Give + # the simulated pager normal terminal-access stops: EOF can arrive before + # cmd2 lends it the terminal, and an ignored TTIN makes that read fail with + # EIO instead of waiting for the handoff. Preserve the inherited TSTP policy. + "signal.signal(signal.SIGTTIN, signal.SIG_DFL)\n" + "signal.signal(signal.SIGTTOU, signal.SIG_DFL)\n" + f"pathlib.Path({str(pager_pid)!r}).write_text(str(os.getpid()))\n" + f"if {finish != 'exit_sigint'!r}: sys.stdin.read()\n" + # Like less, use an inherited terminal descriptor for keyboard input when + # stdin is a pipe. This also works in the broken detached-session case. + "with os.fdopen(os.dup(sys.stderr.fileno()), 'rb', buffering=0) as terminal:\n" + " saved = termios.tcgetattr(terminal)\n" + " def setcbreak():\n" + " while True:\n" + " try:\n" + " tty.setcbreak(terminal)\n" + " return\n" + " except termios.error as error:\n" + " if error.args[0] != errno.EINTR: raise\n" + # os.write rather than print: a signal handler that uses buffered stdout raises + # "reentrant call inside <_io.BufferedWriter>" when the signal lands mid-write, + # which happens when the job is stopped while still reporting readiness. + " def resume(*args):\n" + " setcbreak()\n" + " os.write(1, b'PAGER_RESUMED\\n')\n" + " signal.signal(signal.SIGCONT, resume)\n" + " def interrupt(*args):\n" + f" fd = os.open({str(interrupts)!r}, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)\n" + " os.write(fd, b'I')\n" + " os.close(fd)\n" + " os.write(1, b'PAGER_INTERRUPT\\n')\n" + f" signal.signal(signal.SIGINT, {'signal.SIG_DFL' if finish == 'exit_sigint' else 'interrupt'})\n" + " try:\n" + " setcbreak()\n" + " os.write(1, b'PAGER_READY\\n')\n" + " while True:\n" + " key = os.read(terminal.fileno(), 1)\n" + " if key == b'q': break\n" + " if key == b'p': os.write(1, b'PAGER_ALIVE\\n')\n" + " finally:\n" + " termios.tcsetattr(terminal, termios.TCSANOW, saved)\n", + encoding="utf-8", + ) + application = tmp_path / "application.py" + application_pid = tmp_path / "application.pid" + interrupt_request = tmp_path / "interrupt.request" + last_result = tmp_path / "last_result" + application.write_text( + "from cmd2 import Cmd\n" + "from cmd2.plugin import CommandFinalizationData\n" + "import getpass, os, pathlib, signal, threading, time\n" + "signal.signal(signal.SIGTSTP, signal.SIG_DFL)\n" + f"if {relay == 'worker'!r}:\n" + # The kernel may hand a signal to a thread other than the one it was aimed at. + # Deliver the job-control relay to the watcher that sends it, so the main thread, + # blocked in a pipe write or a wait, only learns of it if it returns on its own. + " _pthread_kill = signal.pthread_kill\n" + " signal.pthread_kill = lambda thread_id, signum: _pthread_kill(threading.get_ident(), signum)\n" + f"pathlib.Path({str(application_pid)!r}).write_text(str(os.getpid()))\n" + "class App(Cmd):\n" + " def do_busy(self, statement):\n" + " os.write(2, b'BUSY_READY\\n')\n" + " self.stdout.write('x' * 262144)\n" + " self.stdout.flush()\n" + " time.sleep(30)\n" + " def do_ask(self, statement):\n" + " self.poutput(self.read_input('INPUT> '))\n" + " def do_direct(self, statement):\n" + " self.stdout.buffer.write(b'x' * 262144)\n" + " self.stdout.flush()\n" + " assert self.select('first second', 'SELECT> ') == 'first'\n" + " assert input('PLAIN> ') == 'answer'\n" + " assert getpass.getpass('SECRET> ') == 'secret'\n" + " os.write(2, b'RAW> ')\n" + " assert os.read(0, 7) == b'direct\\n'\n" + " self.poutput('INPUT_COMPLETE')\n" + " def record_result(self, data: CommandFinalizationData) -> CommandFinalizationData:\n" + f" pathlib.Path({str(last_result)!r}).write_text(repr(self.last_result))\n" + " return data\n" + "app = App()\n" + "app.register_cmdfinalization_hook(app.record_result)\n" + "app.prompt = 'TEST> '\n" + "app.debug = True\n" + f"if {finish == 'interrupts'!r}:\n" + " def interrupt_from_worker():\n" + f" request = pathlib.Path({str(interrupt_request)!r})\n" + " for _ in range(2):\n" + " while not request.exists():\n" + " time.sleep(0.01)\n" + " request.unlink()\n" + # A process-directed signal can be delivered to any unblocked thread. + # Force that case so an indefinite main-thread wait cannot pass by luck. + " signal.pthread_kill(threading.get_ident(), signal.SIGINT)\n" + " threading.Thread(target=interrupt_from_worker, daemon=True).start()\n" + "app.cmdloop()\n", + encoding="utf-8", + ) + master, slave = pty.openpty() + fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0)) + if finish == "exit_sigint": + settings = termios.tcgetattr(slave) + settings[3] |= termios.TOSTOP + termios.tcsetattr(slave, termios.TCSANOW, settings) + # Establish a controlling terminal in a fresh interpreter, avoiding preexec_fn + # (unsafe when pytest or its plugins have started threads). + bootstrap = ( + "import os, fcntl, termios; os.setsid(); " + "fcntl.ioctl(0, termios.TIOCSCTTY, 0); " + "os.execv(os.environ['TEST_SHELL'], ['bash', '--noprofile', '--norc', '-i'])" + ) + # Exercise the same pipeline shell on developer machines and in CI. An + # inherited zsh can exec the pager directly, hiding bash's stop/wait behavior. + env = dict(os.environ, TERM="xterm-256color", PS1="OUTER> ", TEST_SHELL=shell, SHELL=shell) + env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1]) + process = subprocess.Popen([sys.executable, "-c", bootstrap], stdin=slave, stdout=slave, stderr=slave, env=env) + os.close(slave) + screen = pyte.Screen(80, 24) + screen.write_process_input = lambda data: os.write(master, data.encode()) + stream = pyte.Stream(screen) + decoder = codecs.getincrementaldecoder("utf-8")("replace") + transcript = "" + + def send(data): + os.write(master, data.encode()) + + def wait_until(predicate): + nonlocal transcript + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if select.select([master], [], [], 0.05)[0]: + data = decoder.decode(os.read(master, 65536)) + transcript += data + stream.feed(data) + if predicate(): + return + pytest.fail(f"terminal condition timed out:\n{transcript}\n{describe_processes(process.pid, master)}") + + def stopped(*pids: int) -> bool: + """Whether every process is stopped, not merely deprived of the terminal. + + The shell takes the terminal back as soon as its child stops, but a grandchild still + blocked in a one-byte terminal read is woken by the stop signal and, if a keystroke + has arrived by then, consumes it before it stops. Typing has to wait for the whole job. + """ + pids = tuple(set(pids)) + listing = subprocess.run( + ["ps", "-o", "stat=", "-p", ",".join(map(str, pids))], capture_output=True, text=True, check=False + ) + states = listing.stdout.split() + return len(states) == len(pids) and all(state.startswith("T") for state in states) + + job_group = None + pipeline_group = None + try: + wait_until(lambda: "OUTER> " in transcript) + launch = f"{shlex.quote(sys.executable)} {shlex.quote(str(application))}" + if launcher == "sh": + launch = f"{shlex.quote(shell)} -c {shlex.quote(launch + '; :')}" + elif launcher == "uv": + uv = shutil.which("uv") + if uv is None: + pytest.skip("requires uv") + launch = f"{shlex.quote(uv)} run --no-project -- {launch}" + elif launcher == "exec": + launch = "exec " + launch + send(launch + "\n") + wait_until(lambda: "TEST>" in "\n".join(screen.display)) + # A foreground-group query is an observation, not the child's identity. + # It can change during startup and handoffs. Never use an unverified + # foreground query as a kill()/killpg() destination. + app_pid = int(application_pid.read_text()) + job_group = os.getpgid(app_pid) + assert job_group > 1 + wait_until(lambda: os.tcgetpgrp(master) == job_group) + command = "busy" if finish == "exit_sigint" else "help -v" + if producer == "shell": + # A shell command writes into the pipe itself rather than through cmd2's + # stdout, so cmd2 cannot lend the terminal write by write. Like seq or git + # log, this producer dies from SIGINT rather than handling it. + busy_script = tmp_path / "busy.py" + busy_script.write_text( + "import os, signal, sys, time\n" + "signal.signal(signal.SIGINT, signal.SIG_DFL)\n" + "os.write(2, b'BUSY_READY\\n')\n" + "sys.stdout.write('x' * 262144)\n" + "sys.stdout.flush()\n" + "time.sleep(30)\n", + encoding="utf-8", + ) + command = f"shell {shlex.quote(sys.executable)} {shlex.quote(str(busy_script))}" + if finish == "read_input": + command = "ask" + elif finish == "direct_input": + command = "direct" + elif finish == "shell_input": + input_script = tmp_path / "input.py" + input_script.write_text("import os\nos.write(2, b'INPUT> ')\ninput()\n", encoding="utf-8") + command = f"shell {shlex.quote(sys.executable)} {shlex.quote(str(input_script))}" + pipe_command = f"{shlex.quote(sys.executable)} {shlex.quote(str(pager))}" + if shell_child: + # Keep a shell between Popen and the terminal reader, rather than allowing + # the final command to replace it with exec. + pipe_command = f"{shlex.quote(shell)} -c {shlex.quote(pipe_command + '; :')}" + send(f"{command} | {pipe_command}\n") + if finish == "direct_input": + for prompt, response in (("SELECT>", "\r"), ("PLAIN>", "answer\n"), ("SECRET>", "secret\n"), ("RAW>", "direct\n")): + wait_until(lambda prompt=prompt: prompt in "\n".join(screen.display)) + assert os.tcgetpgrp(master) == job_group + send(response) + if finish in ("read_input", "shell_input"): + wait_until(lambda: any(line.startswith("INPUT>") for line in screen.display)) + send("answer\n") + # Whole lines only: a traceback naming the marker must not satisfy the wait. + wait_until(lambda: "PAGER_READY\r\n" in transcript) + if finish == "exit_sigint": + wait_until(lambda: "BUSY_READY\r\n" in transcript) + pager_process = int(pager_pid.read_text()) + pipeline_group = os.getpgid(pager_process) + assert pipeline_group > 1 + assert pipeline_group != job_group + # Readiness output can precede the foreground handoff. Send terminal + # signals and keystrokes only once the pipeline can receive them. + wait_until(lambda: os.tcgetpgrp(master) == pipeline_group) + if launcher == "exec": + # There is no outer shell to run fg: Ctrl-Z must leave the pager + # usable. Require a fresh read acknowledgement, not a SIGCONT. + start = len(transcript) + send("\x1ap") + wait_until(lambda: "PAGER_ALIVE\r\n" in transcript[start:]) + assert os.tcgetpgrp(master) == pipeline_group + for rows in (12, 24) if stop_job else (): + send("\x1a") + wait_until(lambda: os.tcgetpgrp(master) == process.pid) + wait_until(lambda: stopped(job_group, app_pid, pager_process)) + start = len(transcript) + # A child left running can steal these keystrokes from the shell. + send("printf 'SHELL_%s\\n' OWNS_INPUT\n") + wait_until(lambda start=start: "SHELL_OWNS_INPUT" in transcript[start:]) + fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack("HHHH", rows, 80, 0, 0)) + screen.resize(lines=rows, columns=80) + start = len(transcript) + send("stty size\n") + # Bash 5.1+ turns bracketed paste off with "\x1b[?2004l\r" before running the + # command, so the reply may follow a bare "\r" rather than "\r\n". + wait_until(lambda start=start, rows=rows: re.search(rf"[\r\n]{rows} 80\r\n", transcript[start:]) is not None) + start = len(transcript) + send("fg\n") + wait_until(lambda start=start: "PAGER_RESUMED\r\n" in transcript[start:]) + assert os.tcgetpgrp(master) == pipeline_group + if finish == "exit_sigint": + send("\x03") + elif finish == "interrupts": + # Exercise every signal route on this live pipeline, avoiding a fresh + # interpreter and terminal setup for each overlapping matrix combination. + sources = ("terminal", "terminal", "process", "process", "thread", "thread", "group", "group") + for expected_count, source in enumerate(sources, start=1): + if source == "process": + # Signal cmd2 alone, as with `kill -INT `. + os.kill(app_pid, signal.SIGINT) + elif source == "thread": + interrupt_request.touch() + elif source == "group": + os.killpg(pipeline_group, signal.SIGINT) + else: + send("\x03") + wait_until(lambda count=expected_count: transcript.count("PAGER_INTERRUPT\r\n") >= count) + # Keep the handler alive long enough to observe a duplicate delivery, + # then also check that a second real interrupt is not suppressed. + deadline = time.monotonic() + 0.1 + wait_until(lambda deadline=deadline: time.monotonic() >= deadline) + assert interrupts.read_text() == "I" * expected_count + if finish != "exit_sigint": + send("q") + wait_until(lambda: os.tcgetpgrp(master) == job_group and "TEST>" in "\n".join(screen.display)) + if producer == "shell": + # Ctrl-C reached the producer directly, as in a shell pipeline. It did not + # merely die of a broken pipe once the pager was gone. + wait_until(last_result.exists) + assert last_result.read_text() == repr(-signal.SIGINT) + start = len(transcript) + send("help quit\n") + wait_until(lambda: "Exit this application" in transcript[start:]) + send("quit\n") + if launcher != "exec": + wait_until(lambda: os.tcgetpgrp(master) == process.pid) + finally: + # Kill only this test's job, including stopped descendants, on assertion failure. + if pager_pid.exists(): + with contextlib.suppress(ProcessLookupError): + os.kill(int(pager_pid.read_text()), signal.SIGKILL) + if job_group is not None and job_group != process.pid: + with contextlib.suppress(ProcessLookupError): + os.killpg(job_group, signal.SIGKILL) + if pipeline_group is not None: + with contextlib.suppress(ProcessLookupError): + os.killpg(pipeline_group, signal.SIGKILL) + # Release the PTY before reaping its session leader. On macOS, waiting + # while the master is still open can leave terminal teardown blocked. + os.close(master) + process.kill() + process.wait(timeout=5) + + +def test_pipeline_from_worker_thread_stays_isolated(tmp_path) -> None: + """A pipe started off the main thread cannot install job-control handlers. + + It must fall back to running the pipeline in its own session, as before, rather + than failing after Popen and leaving the child unreaped. + """ + import pty + + shell = shutil.which("bash") + if shell is None: + pytest.skip("requires an interactive bash shell") + application = tmp_path / "application.py" + application.write_text( + "from cmd2 import Cmd\n" + "import os, threading\n" + "app = Cmd()\n" + "outcome = []\n" + "worker = threading.Thread(target=lambda: outcome.append(app.onecmd_plus_hooks('help quit | cat')))\n" + "worker.start()\n" + "worker.join()\n" + "try:\n" + " reaped = os.waitpid(-1, os.WNOHANG)\n" + "except ChildProcessError:\n" + " reaped = None\n" + "os.write(1, f'WORKER_DONE {outcome} {reaped}\\n'.encode())\n", + encoding="utf-8", + ) + master, slave = pty.openpty() + bootstrap = ( + "import os, fcntl, termios; os.setsid(); " + "fcntl.ioctl(0, termios.TIOCSCTTY, 0); " + "os.execv(os.environ['TEST_SHELL'], ['bash', '--noprofile', '--norc', '-i'])" + ) + env = dict(os.environ, TERM="xterm-256color", PS1="OUTER> ", TEST_SHELL=shell, SHELL=shell) + env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1]) + process = subprocess.Popen([sys.executable, "-c", bootstrap], stdin=slave, stdout=slave, stderr=slave, env=env) + os.close(slave) + decoder = codecs.getincrementaldecoder("utf-8")("replace") + transcript = "" + + def wait_until(predicate): + nonlocal transcript + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if select.select([master], [], [], 0.05)[0]: + transcript += decoder.decode(os.read(master, 65536)) + if predicate(): + return + pytest.fail(f"terminal condition timed out:\n{transcript}\n{describe_processes(process.pid, master)}") + + try: + wait_until(lambda: "OUTER> " in transcript) + os.write(master, f"{shlex.quote(sys.executable)} {shlex.quote(str(application))}\n".encode()) + # The whole line: a partial read must not satisfy the wait before the reap result arrives. + wait_until(lambda: re.search(r"WORKER_DONE .*\r\n", transcript) is not None) + assert "Exit this application" in transcript + assert "WORKER_DONE [False] None" in transcript + finally: + os.close(master) + process.kill() + process.wait(timeout=5) + + +def test_pipeline_pager_can_set_terminal_modes_at_startup(tmp_path) -> None: + """A pager such as less puts the terminal in raw mode as it starts, before reading its pipe. + + It has to own the terminal by then. A background tcsetattr() stops it with SIGTTOU, and + on macOS the call then fails with EINTR once it is continued rather than being restarted. + less ignores that failure, leaving a cooked terminal: q needs Enter and keys are echoed. + """ + import pty + import termios + + shell = shutil.which("bash") + if shell is None: + pytest.skip("requires an interactive bash shell") + pager = tmp_path / "pager.py" + outcome = tmp_path / "outcome" + pager.write_text( + "import os, pathlib, sys, termios, tty\n" + "with os.fdopen(os.dup(sys.stderr.fileno()), 'rb', buffering=0) as terminal:\n" + " saved = termios.tcgetattr(terminal)\n" + " try:\n" + # Whether cmd2 had lent the terminal yet, should the attempt fail. + " foreground = os.tcgetpgrp(terminal.fileno()) == os.getpgrp()\n" + # Like less, make a single attempt and carry on whatever comes of it. + " try:\n" + " tty.setcbreak(terminal)\n" + " result = 'ok'\n" + " except termios.error as error:\n" + " result = f'{error!r}, foreground before the attempt: {foreground}'\n" + f" pathlib.Path({str(outcome)!r}).write_text(result)\n" + " while os.read(terminal.fileno(), 1) != b'q': pass\n" + " finally:\n" + " termios.tcsetattr(terminal, termios.TCSANOW, saved)\n", + encoding="utf-8", + ) + application = tmp_path / "application.py" + application.write_text( + "import pathlib, time\n" + "from cmd2 import Cmd, utils\n" + # The pipeline's first wait is cmd2's 0.2s startup check. Hold it open until the pager + # has reported, so the test does not race that timer on a busy CI runner: what it checks + # is that the pager owns the terminal throughout the check, however slowly it starts. + "startup_wait = utils.ProcReader.wait_for_exit\n" + "def held_startup_wait(reader, timeout=None):\n" + " utils.ProcReader.wait_for_exit = startup_wait\n" + f" outcome = pathlib.Path({str(outcome)!r})\n" + " deadline = time.monotonic() + 5\n" + " while time.monotonic() < deadline and not (outcome.exists() and outcome.read_text()):\n" + " time.sleep(0.01)\n" + " return startup_wait(reader, timeout)\n" + "utils.ProcReader.wait_for_exit = held_startup_wait\n" + "app = Cmd()\n" + "app.prompt = 'TEST> '\n" + "app.cmdloop()\n", + encoding="utf-8", + ) + master, slave = pty.openpty() + bootstrap = ( + "import os, fcntl, termios; os.setsid(); " + "fcntl.ioctl(0, termios.TIOCSCTTY, 0); " + "os.execv(os.environ['TEST_SHELL'], ['bash', '--noprofile', '--norc', '-i'])" + ) + env = dict(os.environ, TERM="xterm-256color", PS1="OUTER> ", TEST_SHELL=shell, SHELL=shell) + env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1]) + process = subprocess.Popen([sys.executable, "-c", bootstrap], stdin=slave, stdout=slave, stderr=slave, env=env) + os.close(slave) + decoder = codecs.getincrementaldecoder("utf-8")("replace") + transcript = "" + + def wait_until(predicate): + nonlocal transcript + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if select.select([master], [], [], 0.05)[0]: + data = decoder.decode(os.read(master, 65536)) + transcript += data + if "\x1b[6n" in data: + # Answer prompt-toolkit's cursor-position request as a terminal would. + os.write(master, b"\x1b[1;1R") + if predicate(): + return + pytest.fail(f"terminal condition timed out:\n{transcript}\n{describe_processes(process.pid, master)}") + + try: + wait_until(lambda: "OUTER> " in transcript) + os.write(master, f"{shlex.quote(sys.executable)} {shlex.quote(str(application))}\n".encode()) + wait_until(lambda: "TEST>" in transcript) + os.write(master, f"help -v | {shlex.quote(sys.executable)} {shlex.quote(str(pager))}\n".encode()) + wait_until(outcome.exists) + wait_until(lambda: outcome.read_text() != "") + assert outcome.read_text() == "ok" + assert not termios.tcgetattr(master)[3] & termios.ICANON + # A cooked terminal would hold the key back until Enter. + start = len(transcript) + os.write(master, b"q") + wait_until(lambda: "TEST>" in transcript[start:]) + os.write(master, b"quit\n") + wait_until(lambda: os.tcgetpgrp(master) == process.pid) + finally: + os.close(master) + process.kill() + process.wait(timeout=5) + + +@pytest.mark.parametrize("producer", ["command", "shell"]) +def test_pipeline_children_inherit_an_ordinary_signal_mask(tmp_path, producer) -> None: + """Processes started during a terminal pipeline must not inherit a blocked SIGTTOU. + + cmd2 blocks SIGTTOU for itself while it lends the terminal. A signal mask survives fork + and exec, so a child spawned with it blocked -- a shell producer, or a subprocess run by + command code -- would keep it for life, and change terminal modes from the background + where it should be stopped. + """ + import pty + + shell = shutil.which("bash") + if shell is None: + pytest.skip("requires an interactive bash shell") + probe = tmp_path / "probe.py" + outcome = tmp_path / "outcome" + probe.write_text( + "import pathlib, signal\n" + "blocked = signal.SIGTTOU in signal.pthread_sigmask(signal.SIG_BLOCK, [])\n" + f"pathlib.Path({str(outcome)!r}).write_text(repr(blocked))\n", + encoding="utf-8", + ) + application = tmp_path / "application.py" + application.write_text( + "import subprocess, sys\n" + "from cmd2 import Cmd\n" + "class App(Cmd):\n" + " def do_probe(self, _):\n" + " self.poutput('probing')\n" + f" subprocess.run([sys.executable, {str(probe)!r}], check=True)\n" + "app = App()\n" + "app.prompt = 'TEST> '\n" + "app.cmdloop()\n", + encoding="utf-8", + ) + master, slave = pty.openpty() + bootstrap = ( + "import os, fcntl, termios; os.setsid(); " + "fcntl.ioctl(0, termios.TIOCSCTTY, 0); " + "os.execv(os.environ['TEST_SHELL'], ['bash', '--noprofile', '--norc', '-i'])" + ) + env = dict(os.environ, TERM="xterm-256color", PS1="OUTER> ", TEST_SHELL=shell, SHELL=shell) + env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1]) + process = subprocess.Popen([sys.executable, "-c", bootstrap], stdin=slave, stdout=slave, stderr=slave, env=env) + os.close(slave) + decoder = codecs.getincrementaldecoder("utf-8")("replace") + transcript = "" + + def wait_until(predicate): + nonlocal transcript + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if select.select([master], [], [], 0.05)[0]: + data = decoder.decode(os.read(master, 65536)) + transcript += data + if "\x1b[6n" in data: + # Answer prompt-toolkit's cursor-position request as a terminal would. + os.write(master, b"\x1b[1;1R") + if predicate(): + return + pytest.fail(f"terminal condition timed out:\n{transcript}\n{describe_processes(process.pid, master)}") + + command = "probe" if producer == "command" else f"shell {shlex.quote(sys.executable)} {shlex.quote(str(probe))}" + try: + wait_until(lambda: "OUTER> " in transcript) + os.write(master, f"{shlex.quote(sys.executable)} {shlex.quote(str(application))}\n".encode()) + wait_until(lambda: "TEST>" in transcript) + start = len(transcript) + os.write(master, f"{command} | cat\n".encode()) + wait_until(lambda: outcome.exists() and outcome.read_text() != "" and "TEST>" in transcript[start:]) + assert outcome.read_text() == "False" + os.write(master, b"quit\n") + wait_until(lambda: os.tcgetpgrp(master) == process.pid) + finally: + os.close(master) + process.kill() + process.wait(timeout=5) + + +@pytest.mark.parametrize("suspend", [False, True]) +def test_shell_producer_keeps_the_terminal_after_its_consumer_exits(tmp_path, suspend) -> None: + """A shell producer that outlives its consumer still reads the terminal. + + do_shell() lends the terminal to the pipeline's group for as long as the producer runs. + The consumer's exit must not take it back early: the producer would stop with SIGTTIN on + its next terminal read, and nothing watches an ordinary shell command for stops. + + Ctrl-Z then reaches the producer alone, since it is all that is left of the foreground + group. With the consumer's watcher gone, do_shell() has to relay that stop to the + whole job itself, or it waits forever on a stopped child. + """ + import pty + + shell = shutil.which("bash") + if shell is None: + pytest.skip("requires an interactive bash shell") + consumer = tmp_path / "consumer.py" + consumer.write_text("import os, time\ntime.sleep(0.5)\nos.write(2, b'CONSUMER_DONE\\n')\n", encoding="utf-8") + producer = tmp_path / "producer.py" + producer.write_text( + "import os, signal, time\n" + # Interactive bash leaves TTIN ignored in what it execs, which turns a background read into EIO. + "signal.signal(signal.SIGTTIN, signal.SIG_DFL)\n" + "time.sleep(1.5)\n" + "os.write(2, b'PRODUCER> ')\n" + "os.write(2, b'GOT ' + os.read(0, 7))\n", + encoding="utf-8", + ) + application = tmp_path / "application.py" + application.write_text( + "from cmd2 import Cmd\napp = Cmd()\napp.prompt = 'TEST> '\napp.cmdloop()\n", + encoding="utf-8", + ) + master, slave = pty.openpty() + bootstrap = ( + "import os, fcntl, termios; os.setsid(); " + "fcntl.ioctl(0, termios.TIOCSCTTY, 0); " + "os.execv(os.environ['TEST_SHELL'], ['bash', '--noprofile', '--norc', '-i'])" + ) + env = dict(os.environ, TERM="xterm-256color", PS1="OUTER> ", TEST_SHELL=shell, SHELL=shell) + env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1]) + process = subprocess.Popen([sys.executable, "-c", bootstrap], stdin=slave, stdout=slave, stderr=slave, env=env) + os.close(slave) + decoder = codecs.getincrementaldecoder("utf-8")("replace") + transcript = "" + + def wait_until(predicate): + nonlocal transcript + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if select.select([master], [], [], 0.05)[0]: + data = decoder.decode(os.read(master, 65536)) + transcript += data + if "\x1b[6n" in data: + # Answer prompt-toolkit's cursor-position request as a terminal would. + os.write(master, b"\x1b[1;1R") + if predicate(): + return + pytest.fail(f"terminal condition timed out:\n{transcript}\n{describe_processes(process.pid, master)}") + + python = shlex.quote(sys.executable) + try: + wait_until(lambda: "OUTER> " in transcript) + os.write(master, f"{python} {shlex.quote(str(application))}\n".encode()) + wait_until(lambda: "TEST>" in transcript) + os.write(master, f"shell {python} {shlex.quote(str(producer))} | {python} {shlex.quote(str(consumer))}\n".encode()) + wait_until(lambda: "CONSUMER_DONE\r\n" in transcript) + wait_until(lambda: "PRODUCER> " in transcript) + if suspend: + start = len(transcript) + os.write(master, b"\x1a") + wait_until(lambda: os.tcgetpgrp(master) == process.pid and "OUTER> " in transcript[start:]) + os.write(master, b"fg\n") + wait_until(lambda: os.tcgetpgrp(master) not in (process.pid, os.getpgid(process.pid))) + os.write(master, b"answer\n") + wait_until(lambda: "GOT answer" in transcript) + # cmd2 owns the terminal again once the producer is done. + start = len(transcript) + wait_until(lambda: "TEST>" in transcript[start:]) + os.write(master, b"help quit\n") + wait_until(lambda: "Exit this application" in transcript[start:]) + os.write(master, b"quit\n") + wait_until(lambda: os.tcgetpgrp(master) == process.pid) + finally: + os.close(master) + process.kill() + process.wait(timeout=5) diff --git a/tests/test_utils.py b/tests/test_utils.py index 9c737d800..d5aa5a806 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,9 +1,12 @@ """Unit testing for cmd2/utils.py module.""" +import contextlib +import errno import math import os import signal import sys +import threading import time from unittest import ( mock, @@ -219,6 +222,108 @@ def test_proc_reader_send_sigint(pr_none) -> None: assert ret_code == -signal.SIGINT +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups") +def test_proc_reader_does_not_resignal_its_own_group(pr_none) -> None: + try: + with mock.patch("os.getpgrp", return_value=pr_none._proc.pid), mock.patch("os.killpg") as killpg: + pr_none.send_sigint() + killpg.assert_not_called() + finally: + pr_none.terminate() + pr_none.wait() + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups") +def test_proc_reader_sigint_after_pipeline_exit() -> None: + reader = cu.ProcReader(mock.Mock(pid=os.getpid() + 1, stdout=None, stderr=None), sys.stdout, sys.stderr) + with ( + mock.patch("os.getpgid", side_effect=ProcessLookupError), + mock.patch("os.killpg", side_effect=ProcessLookupError) as killpg, + ): + reader.send_sigint() + killpg.assert_called_once_with(reader._proc.pid, signal.SIGINT) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups") +def test_proc_reader_sigint_reaches_group_after_leader_exit() -> None: + """A shell producer joins the pipeline's group and can outlive the consumer that led it.""" + import subprocess + + # A terminal pipeline leads its own group within our session, so a producer may join it. + leader = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"], process_group=0) + reader = cu.ProcReader(leader, sys.stdout, sys.stderr) + member_code = ( + "import signal, time; signal.signal(signal.SIGINT, signal.SIG_DFL); print('ready', flush=True); time.sleep(30)" + ) + member = subprocess.Popen([sys.executable, "-c", member_code], stdout=subprocess.PIPE, process_group=leader.pid) + try: + assert member.stdout is not None + assert member.stdout.readline().strip() == b"ready" + reader.terminate() + reader.wait() + assert leader.returncode == -signal.SIGTERM + + reader.send_sigint() + assert member.wait(timeout=5) == -signal.SIGINT + finally: + member.kill() + member.wait() + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX process groups") +def test_proc_reader_terminal_group() -> None: + proc = mock.Mock(pid=4242, returncode=None, stdout=None, stderr=None) + assert cu.ProcReader(proc, sys.stdout, sys.stderr).terminal_group is None + + with mock.patch("os.tcgetpgrp", return_value=os.getpgrp()): + reader = cu.ProcReader(proc, sys.stdout, sys.stderr, terminal_fd=0) + assert reader.terminal_group == proc.pid + proc.returncode = 0 + assert reader.terminal_group is None + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX pipes") +@pytest.mark.parametrize("producer", ["writer", "child"]) +def test_pipeline_writer_delivers_more_than_the_pipe_holds(producer) -> None: + """Both cmd2's writes and a child inheriting the descriptor must wait for a slow consumer. + + A shell producer gets the descriptor itself, so it must stay blocking: a child that + inherits O_NONBLOCK fails with EAGAIN once the pipe is full. + """ + import subprocess + import threading + + payload = b"x" * 4 * 1024 * 1024 + read_fd, write_fd = os.pipe() + received = bytearray() + + def drain() -> None: + while chunk := os.read(read_fd, 65536): + received.extend(chunk) + time.sleep(0.001) + + reader = mock.Mock(lend_terminal=contextlib.nullcontext) + writer = cu.PipelineWriter(write_fd, reader) + consumer = threading.Thread(target=drain) + consumer.start() + try: + if producer == "writer": + assert writer.write(payload) == len(payload) + else: + child = subprocess.run( + [sys.executable, "-c", f"import sys; sys.stdout.buffer.write(b'x' * {len(payload)})"], + stdout=writer.fileno(), + stderr=subprocess.PIPE, + check=False, + ) + assert child.returncode == 0, child.stderr.decode() + finally: + writer.close() + consumer.join() + os.close(read_fd) + assert bytes(received) == payload + + def test_proc_reader_terminate(pr_none) -> None: assert pr_none._proc.poll() is None pr_none.terminate() @@ -238,6 +343,254 @@ def test_proc_reader_terminate(pr_none) -> None: assert ret_code == -signal.SIGTERM +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control") +@pytest.mark.parametrize("already_exited", [False, True]) +def test_proc_reader_terminate_terminal_job(already_exited) -> None: + proc = mock.Mock(stdout=None, stderr=None) + reader = cu.ProcReader(proc, sys.stdout, sys.stderr) + reader._terminal_fd = 10 + with mock.patch("os.kill", side_effect=ProcessLookupError if already_exited else None) as kill: + reader.terminate() + kill.assert_called_once_with(proc.pid, signal.SIGTERM) + # Only the job watcher may reap this process; Popen.terminate() would poll it. + proc.terminate.assert_not_called() + proc.poll.assert_not_called() + proc.wait.assert_not_called() + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control") +@pytest.mark.parametrize("stop_signal", ["SIGTTIN", "SIGTTOU"]) +@pytest.mark.parametrize("expired_handoff", [False, True]) +def test_proc_reader_resumes_terminal_access_after_handoff(stop_signal, expired_handoff) -> None: + proc = mock.Mock(pid=123, stdout=None, stderr=None, returncode=None) + reader = cu.ProcReader(proc, sys.stdout, sys.stderr) + reader._terminal_fd = 10 + reader._original_group = 456 + reader._terminal_available.set() + stopped_status = (getattr(signal, stop_signal) << 8) | 0x7F + handoffs = iter([False, True] if expired_handoff else [True]) + + def handoff(timeout): + assert timeout == 0.1 + if next(handoffs): + reader._terminal_available.set() + else: + reader._terminal_available.clear() + return True + + with ( + mock.patch( + "os.waitpid", + side_effect=[(proc.pid, stopped_status), *([(0, 0)] * (2 if expired_handoff else 1)), (proc.pid, 0)], + ), + mock.patch("os.tcgetpgrp", return_value=proc.pid), + mock.patch.object(reader, "_set_foreground_group") as foreground, + mock.patch.object(reader._terminal_available, "wait", side_effect=handoff) as available, + mock.patch("os.killpg") as killpg, + mock.patch("signal.raise_signal") as stop, + ): + reader._wait_for_job(10) + assert available.call_count == (2 if expired_handoff else 1) + killpg.assert_called_once_with(proc.pid, signal.SIGCONT) + stop.assert_not_called() + # The lend is still active: its holder returns the terminal, not the watcher. + foreground.assert_not_called() + assert proc.returncode == 0 + assert reader._process_done.is_set() + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control") +@pytest.mark.parametrize("lent", [False, True]) +def test_proc_reader_exit_returns_terminal_unless_lent(lent) -> None: + """A shell producer in a lent pipeline group may outlive the consumer and still need the terminal.""" + proc = mock.Mock(pid=123, stdout=None, stderr=None, returncode=None) + reader = cu.ProcReader(proc, sys.stdout, sys.stderr) + reader._terminal_fd = 10 + reader._original_group = 456 + if lent: + reader._terminal_available.set() + with ( + mock.patch("os.waitpid", return_value=(proc.pid, 0)), + mock.patch("os.tcgetpgrp", return_value=proc.pid), + mock.patch.object(reader, "_set_foreground_group") as foreground, + ): + reader._wait_for_job(10) + if lent: + foreground.assert_not_called() + else: + foreground.assert_called_once_with(10, reader._original_group) + assert reader._process_done.is_set() + + +def test_proc_reader_wait_for_exit_without_terminal() -> None: + proc = mock.Mock(stdout=None, stderr=None) + reader = cu.ProcReader(proc, sys.stdout, sys.stderr) + reader.wait_for_exit(timeout=0.2) + proc.wait.assert_called_once_with(0.2) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control") +@pytest.mark.parametrize("handler_kind", ["default", "ignored", "custom"]) +def test_proc_reader_suspend_restores_signal_handler(handler_kind) -> None: + proc = mock.Mock(pid=123, stdout=None, stderr=None, returncode=None) + reader = cu.ProcReader(proc, sys.stdout, sys.stderr) + reader._terminal_fd = 10 + reader._original_group = 456 + reader._terminal_available.set() + previous = {"default": signal.SIG_DFL, "ignored": signal.SIG_IGN, "custom": mock.Mock()}[handler_kind] + groups = [proc.pid, reader._original_group] if handler_kind == "default" else [reader._original_group] + with ( + mock.patch("signal.getsignal", return_value=previous), + mock.patch("signal.signal") as set_handler, + mock.patch("signal.raise_signal") as stop, + mock.patch("os.killpg") as killpg, + mock.patch("os.tcgetpgrp", side_effect=groups), + mock.patch("threading.Thread"), + mock.patch.object(reader, "_set_foreground_group") as foreground, + reader.manage_terminal(), + ): + handler = set_handler.call_args.args[1] + handler(signal.SIGTSTP, None) + assert reader._job_resumed.is_set() + set_handler.assert_called_with(signal.SIGTSTP, previous) + foreground.assert_called_with(10, proc.pid) + if handler_kind == "default": + killpg.assert_called_once_with(reader._original_group, signal.SIGTSTP) + stop.assert_called_once_with(signal.SIGTSTP) + assert set_handler.call_args_list == [ + mock.call(signal.SIGTSTP, handler), + mock.call(signal.SIGTSTP, signal.SIG_IGN), + mock.call(signal.SIGTSTP, signal.SIG_DFL), + mock.call(signal.SIGTSTP, handler), + mock.call(signal.SIGTSTP, previous), + ] + else: + killpg.assert_not_called() + stop.assert_not_called() + if handler_kind == "custom": + previous.assert_called_once_with(signal.SIGTSTP, None) + + +def test_proc_reader_captured_pipeline_needs_no_terminal() -> None: + reader = cu.ProcReader(mock.Mock(stdout=None, stderr=None), sys.stdout, sys.stderr) + with reader.manage_terminal(), reader.lend_terminal(): + assert not reader._terminal_available.is_set() + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control") +def test_proc_reader_lending_restores_terminal_on_write_error() -> None: + reader = cu.ProcReader(mock.Mock(pid=123, stdout=None, stderr=None, returncode=None), sys.stdout, sys.stderr) + reader._terminal_fd = 10 + reader._original_group = 456 + + def failing_write(): + with reader.lend_terminal(): + assert reader._terminal_available.is_set() + raise BrokenPipeError + + with ( + mock.patch("os.tcgetpgrp", return_value=123), + mock.patch.object(reader, "_set_foreground_group") as foreground, + pytest.raises(BrokenPipeError), + ): + failing_write() + assert not reader._terminal_available.is_set() + assert foreground.call_args_list == [mock.call(10, 123), mock.call(10, 456)] + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control") +@pytest.mark.parametrize("inner", ["nested", "thread"]) +def test_proc_reader_keeps_the_terminal_lent_until_the_last_lend_ends(inner) -> None: + """A shorter lend inside a longer one must not take the terminal back early. + + do_shell() lends for as long as a shell producer runs. A pipe write from another thread + meanwhile lends and returns. If its return took the terminal back, the producer would stop + with SIGTTIN on its next terminal read, and nothing would resume it. + """ + import threading + + reader = cu.ProcReader(mock.Mock(pid=123, stdout=None, stderr=None, returncode=None), sys.stdout, sys.stderr) + reader._terminal_fd = 10 + reader._original_group = 456 + + def short_lend() -> None: + with reader.lend_terminal(): + pass + + with ( + mock.patch("os.tcgetpgrp", return_value=123), + mock.patch.object(reader, "_set_foreground_group") as foreground, + ): + with reader.lend_terminal(): + if inner == "nested": + short_lend() + else: + worker = threading.Thread(target=short_lend) + worker.start() + worker.join() + assert reader._terminal_available.is_set() + assert mock.call(10, 456) not in foreground.call_args_list + assert not reader._terminal_available.is_set() + assert foreground.call_args_list[-1] == mock.call(10, 456) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control") +@pytest.mark.parametrize("error_number", [errno.ESRCH, errno.EINVAL, errno.EBADF]) +def test_proc_reader_handoff_to_disappearing_group(error_number) -> None: + reader = cu.ProcReader(mock.Mock(pid=123, stdout=None, stderr=None, returncode=None), sys.stdout, sys.stderr) + reader._terminal_fd = 10 + reader._original_group = 456 + + def write(): + with reader.lend_terminal(): + assert reader._terminal_available.is_set() + + with ( + mock.patch("os.tcgetpgrp", return_value=456), + mock.patch.object(reader, "_set_foreground_group", side_effect=OSError(error_number, "handoff failed")), + ): + if error_number == errno.EBADF: + with pytest.raises(OSError, match="handoff failed"): + write() + else: + write() + assert not reader._terminal_available.is_set() + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control") +def test_proc_reader_reaps_killed_consumer_without_another_handoff() -> None: + proc = mock.Mock(pid=123, stdout=None, stderr=None, returncode=None) + reader = cu.ProcReader(proc, sys.stdout, sys.stderr) + reader._terminal_fd = 10 + reader._original_group = 456 + stopped_status = (signal.SIGTTIN << 8) | 0x7F + with ( + mock.patch("os.waitpid", side_effect=[(123, stopped_status), (123, signal.SIGKILL)]), + mock.patch("os.tcgetpgrp", return_value=456), + mock.patch.object(reader._terminal_available, "wait", return_value=False), + mock.patch("os.killpg") as killpg, + ): + reader._wait_for_job(10) + assert proc.returncode == -signal.SIGKILL + assert reader._process_done.is_set() + killpg.assert_not_called() + proc.wait.assert_not_called() + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal pipeline writer") +@pytest.mark.parametrize("returncode", [-signal.SIGINT, 128 + signal.SIGINT, 0]) +@pytest.mark.parametrize("finished", [False, True]) +def test_pipeline_writer_cancels_interrupted_producer_but_not_cleanup(returncode, finished) -> None: + reader = cu.ProcReader(mock.Mock(stdout=None, stderr=None, returncode=returncode), sys.stdout, sys.stderr) + if finished: + reader.finish_producer() + read_fd, write_fd = os.pipe() + os.close(read_fd) + expected = KeyboardInterrupt if returncode != 0 and not finished else BrokenPipeError + with cu.PipelineWriter(write_fd, reader) as writer, pytest.raises(expected): + writer.write(b"output") + + @pytest.fixture def context_flag(): return cu.ContextFlag() @@ -438,3 +791,58 @@ def bar_method(self) -> None: cu.categorize([func2, b.bar_method], category) assert getattr(func2, attr_name) == category assert getattr(Bar.bar_method, attr_name) == category + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control") +def test_proc_reader_producer_wait_times_out() -> None: + import subprocess + + pipeline = mock.Mock() + proc = mock.Mock(pid=321, stdout=None, stderr=None, returncode=None) + reader = cu.ProcReader(proc, sys.stdout, sys.stderr, pipeline=pipeline) + with mock.patch("os.waitpid", return_value=(0, 0)), pytest.raises(subprocess.TimeoutExpired): + reader.wait_for_exit(0) + pipeline._relay_producer_stop.assert_not_called() + proc.wait.assert_not_called() + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control") +@pytest.mark.parametrize("watcher_done", [False, True]) +@pytest.mark.parametrize("foreground_group", [123, 456]) +def test_proc_reader_relays_producer_stop_once_the_watcher_is_gone(watcher_done, foreground_group) -> None: + consumer = mock.Mock(pid=123, stdout=None, stderr=None, returncode=0) + pipeline = cu.ProcReader(consumer, sys.stdout, sys.stderr) + pipeline._terminal_fd = 10 + pipeline._original_group = 456 + if watcher_done: + pipeline._process_done.set() + proc = mock.Mock(pid=321, stdout=None, stderr=None, returncode=None) + reader = cu.ProcReader(proc, sys.stdout, sys.stderr, pipeline=pipeline) + stopped_status = (signal.SIGTSTP << 8) | 0x7F + + def resume(thread_id, signum): + assert thread_id == threading.main_thread().ident + assert signum == signal.SIGTSTP + pipeline._job_resumed.set() + + with ( + mock.patch("os.waitpid", side_effect=[(proc.pid, stopped_status), (proc.pid, 0)]), + mock.patch("os.tcgetpgrp", return_value=foreground_group), + mock.patch.object(pipeline, "_set_foreground_group") as foreground, + mock.patch("os.killpg") as killpg, + mock.patch("signal.pthread_kill", side_effect=resume) as relay, + ): + reader.wait_for_exit() + assert proc.returncode == 0 + if not watcher_done: + # The consumer's watcher sees the same Ctrl-Z and suspends the job itself. + relay.assert_not_called() + killpg.assert_not_called() + foreground.assert_not_called() + return + relay.assert_called_once() + assert killpg.call_args_list == [mock.call(consumer.pid, signal.SIGSTOP), mock.call(consumer.pid, signal.SIGCONT)] + if foreground_group == consumer.pid: + foreground.assert_called_once_with(10, pipeline._original_group) + else: + foreground.assert_not_called() From 0380c1c0a65522aba60e89f687a6854d27c89ea1 Mon Sep 17 00:00:00 2001 From: Todd Leonhardt Date: Fri, 11 Sep 2026 20:45:18 -0400 Subject: [PATCH 2/3] Give the corrupt-history tests their own temp files test_history_file_bad_compression and test_history_file_bad_json wrote to a fixed /tmp/doesntmatter, which on the Windows runner is D:\tmp and does not exist on a fresh machine. They passed only when the preceding permission-error test had already run: mocking builtins.open there does not stop the history setup from creating the file's parent directory, so that test created D:\tmp as a side effect. Under xdist the order is not fixed, and adding tests elsewhere shifted the schedule so both writers reached a worker first and failed on every Windows job. All three tests now use tmp_path, so none depends on a directory existing or on another test having run. Each passes alone on a single worker. Validation: 2587 passed, 6 skipped with coverage; make check, make test, make docs-test and git diff --check passed. --- tests/test_history.py | 25 +++++++++++++------------ 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/tests/test_history.py b/tests/test_history.py index 8fff6b7e5..a503a95e2 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -910,38 +910,39 @@ def test_history_cannot_create_directory(mocker, capsys) -> None: assert "Error creating persistent history file directory" in err -def test_history_file_permission_error(mocker, capsys) -> None: +def test_history_file_permission_error(mocker, capsys, tmp_path) -> None: mock_open = mocker.patch("builtins.open") mock_open.side_effect = PermissionError - cmd2.Cmd(persistent_history_file="/tmp/doesntmatter") + # A path under tmp_path rather than a fixed one: mocking open() does not stop the + # history setup from creating the file's parent directory, and a fixed path would leave + # that directory behind as a side effect other tests could come to depend on. + cmd2.Cmd(persistent_history_file=str(tmp_path / "doesntmatter")) out, err = capsys.readouterr() assert not out assert "Cannot read persistent history file" in err -def test_history_file_bad_compression(mocker, capsys) -> None: - history_file = "/tmp/doesntmatter" - with open(history_file, "wb") as f: - f.write(b"THIS IS NOT COMPRESSED DATA") +def test_history_file_bad_compression(capsys, tmp_path) -> None: + history_file = tmp_path / "doesntmatter" + history_file.write_bytes(b"THIS IS NOT COMPRESSED DATA") - cmd2.Cmd(persistent_history_file=history_file) + cmd2.Cmd(persistent_history_file=str(history_file)) out, err = capsys.readouterr() assert not out assert "Error decompressing persistent history data" in err -def test_history_file_bad_json(mocker, capsys) -> None: +def test_history_file_bad_json(capsys, tmp_path) -> None: import lzma data = b"THIS IS NOT JSON" compressed_data = lzma.compress(data) - history_file = "/tmp/doesntmatter" - with open(history_file, "wb") as f: - f.write(compressed_data) + history_file = tmp_path / "doesntmatter" + history_file.write_bytes(compressed_data) - cmd2.Cmd(persistent_history_file=history_file) + cmd2.Cmd(persistent_history_file=str(history_file)) out, err = capsys.readouterr() assert not out assert "Error processing persistent history data" in err From f4c1d676cd5aed53a97f6b8e0c34678b66f3a33c Mon Sep 17 00:00:00 2001 From: Todd Leonhardt Date: Sat, 26 Sep 2026 11:50:45 -0400 Subject: [PATCH 3/3] Report terminal sizes when the job-control test's resize step times out On Ubuntu 3.11 CI the outer shell once printed the old size after the test resized the pseudo-terminal while the job was stopped. Record each resize with the size read back straight after it, and include the size at the timeout, to tell a resize that never took effect from one undone later. --- tests/test_pipeline_job_control.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/tests/test_pipeline_job_control.py b/tests/test_pipeline_job_control.py index 1ad79e343..bdb5154d9 100644 --- a/tests/test_pipeline_job_control.py +++ b/tests/test_pipeline_job_control.py @@ -205,6 +205,13 @@ def test_pipeline_stops_with_cmd2_and_returns_terminal( stream = pyte.Stream(screen) decoder = codecs.getincrementaldecoder("utf-8")("replace") transcript = "" + # Each resize with the size read back straight after it, to tell a resize that never + # took effect from one undone later. + resizes: list[str] = [] + + def terminal_size() -> tuple[int, int]: + rows, columns, _, _ = struct.unpack("HHHH", fcntl.ioctl(master, termios.TIOCGWINSZ, b"\0" * 8)) + return rows, columns def send(data): os.write(master, data.encode()) @@ -219,7 +226,10 @@ def wait_until(predicate): stream.feed(data) if predicate(): return - pytest.fail(f"terminal condition timed out:\n{transcript}\n{describe_processes(process.pid, master)}") + pytest.fail( + f"terminal condition timed out:\n{transcript}\n{describe_processes(process.pid, master)}\n" + f"resizes: {resizes}\nterminal size at timeout: {terminal_size()}" + ) def stopped(*pids: int) -> bool: """Whether every process is stopped, not merely deprived of the terminal. @@ -323,6 +333,7 @@ def stopped(*pids: int) -> bool: send("printf 'SHELL_%s\\n' OWNS_INPUT\n") wait_until(lambda start=start: "SHELL_OWNS_INPUT" in transcript[start:]) fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack("HHHH", rows, 80, 0, 0)) + resizes.append(f"requested {rows}x80, read back {terminal_size()}") screen.resize(lines=rows, columns=80) start = len(transcript) send("stty size\n")