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_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 diff --git a/tests/test_pipeline_job_control.py b/tests/test_pipeline_job_control.py new file mode 100644 index 000000000..bdb5154d9 --- /dev/null +++ b/tests/test_pipeline_job_control.py @@ -0,0 +1,733 @@ +"""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 = "" + # 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()) + + 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)}\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. + + 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)) + resizes.append(f"requested {rows}x80, read back {terminal_size()}") + 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()