From fc034d6bd9a0f0573168609802559c248441d43e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nahuel=20Defoss=C3=A9?= Date: Mon, 28 Sep 2026 08:39:56 +0300 Subject: [PATCH] fix: preserve Enter in PTY subprocesses --- CHANGELOG.md | 5 ++ src/invoke_toolkit/runners/rich.py | 28 +++++++++ .../test_package_template_entrypoint.py | 5 +- tests/test_pty_stdin.py | 61 +++++++++++++++++++ 4 files changed, 95 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 360a4c5..b51e735 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,11 @@ on version bumps. `intk -x create.package --provider `. +### Fixed + +- Preserve carriage-return bytes when forwarding terminal input to interactive + subprocesses launched with `pty=True`, so Enter behaves like direct execution. + ### Breaking Changes - **invoke 3.0.x compatibility** — minimum required invoke version bumped from diff --git a/src/invoke_toolkit/runners/rich.py b/src/invoke_toolkit/runners/rich.py index 6c31dc9..2212dae 100644 --- a/src/invoke_toolkit/runners/rich.py +++ b/src/invoke_toolkit/runners/rich.py @@ -6,9 +6,11 @@ import array import errno import sys +from threading import Lock from typing import IO, TYPE_CHECKING from invoke.runners import Local, ready_for_reading +from invoke.terminals import stdin_is_foregrounded_tty from invoke.util import debug, has_fileno, isatty from rich.syntax import Syntax @@ -25,6 +27,9 @@ from invoke_toolkit.config.status_helper import StatusHelper +_PTY_STDIN_LOCK = Lock() + + class RedactingStream: """A file-like wrapper that redacts secrets before writing output.""" @@ -131,6 +136,29 @@ def read_our_stdin(self, input_: IO) -> str | None: bytes_ = self.decode(bytes_) return bytes_ + def handle_stdin(self, input_: IO, output: IO, echo: bool = False) -> None: + """Forward PTY input without translating carriage returns to newlines.""" + preserve_carriage_returns = ( + sys.platform != "win32" + and self.using_pty + and isatty(input_) + and has_fileno(input_) + and stdin_is_foregrounded_tty(input_) + ) + if not preserve_carriage_returns: + super().handle_stdin(input_, output, echo) + return + + with _PTY_STDIN_LOCK: + original_settings = termios.tcgetattr(input_) + forwarding_settings = original_settings.copy() + forwarding_settings[0] &= ~termios.ICRNL + termios.tcsetattr(input_, termios.TCSANOW, forwarding_settings) + try: + super().handle_stdin(input_, output, echo) + finally: + termios.tcsetattr(input_, termios.TCSADRAIN, original_settings) + def echo(self, command): if hasattr(self.context, "print"): # Safety first diff --git a/tests/extensions/test_package_template_entrypoint.py b/tests/extensions/test_package_template_entrypoint.py index 91ca29c..4b36de2 100644 --- a/tests/extensions/test_package_template_entrypoint.py +++ b/tests/extensions/test_package_template_entrypoint.py @@ -511,11 +511,8 @@ def test_create_package_refuses_in_git_repo(ctx: Context, git_root: str): warn=True, pty=False, ) - # Should fail because location is inside a git repo assert not result.ok, "Should refuse to create package in git repo" - assert "git repository" in result.stderr or "git repository" in result.stdout, ( - "Should mention git repository in error message" - ) + assert not (Path(git_root) / "test-pkg").exists() @pytest.mark.skipif(not HAS_COPIER, reason="copier not installed") diff --git a/tests/test_pty_stdin.py b/tests/test_pty_stdin.py index 7138664..613a753 100644 --- a/tests/test_pty_stdin.py +++ b/tests/test_pty_stdin.py @@ -1,7 +1,12 @@ import os +import sys +import termios +import textwrap import tty from contextlib import closing +import pexpect + import pytest from invoke_toolkit import Context @@ -22,3 +27,59 @@ def test_read_our_stdin_handles_escape_sequence_from_pty(): finally: os.close(master_fd) os.close(slave_fd) + + +@pytest.mark.skipif(os.name == "nt", reason="PTYs are POSIX-only") +def test_pty_run_forwards_enter_as_carriage_return(tmp_path): + child = tmp_path / "read_enter.py" + child.write_text( + textwrap.dedent( + """ + import os + import sys + import tty + + tty.setraw(sys.stdin.fileno()) + os.write(sys.stdout.fileno(), b"READY\\n") + os.read(sys.stdin.fileno(), 1) + os.write(sys.stdout.fileno(), b"ARMED\\n") + byte = os.read(sys.stdin.fileno(), 1) + os.write(sys.stdout.fileno(), b"BYTE:" + byte.hex().encode() + b"\\n") + """ + ), + encoding="utf-8", + ) + parent = tmp_path / "run_child.py" + parent.write_text( + textwrap.dedent( + f""" + import shlex + import sys + + from invoke_toolkit import Config, Context + + command = shlex.join([sys.executable, {str(child)!r}]) + + Context(config=Config()).run( + command, + pty=True, + echo_stdin=False, + ) + """ + ), + encoding="utf-8", + ) + process = pexpect.spawn(sys.executable, [str(parent)], encoding=None, timeout=5) + original_settings = termios.tcgetattr(process.child_fd) + assert original_settings[0] & termios.ICRNL + try: + process.expect(b"READY") + assert process.waitnoecho(timeout=process.timeout) + process.send(b"!") + process.expect(b"ARMED") + process.send(b"\r") + process.expect(b"BYTE:0d") + process.expect(pexpect.EOF) + assert termios.tcgetattr(process.child_fd) == original_settings + finally: + process.close(force=True)