Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,11 @@ on version bumps.
`intk -x create.package --provider <scheme>`.


### 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
Expand Down
28 changes: 28 additions & 0 deletions src/invoke_toolkit/runners/rich.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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."""

Expand Down Expand Up @@ -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
Expand Down
5 changes: 1 addition & 4 deletions tests/extensions/test_package_template_entrypoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
61 changes: 61 additions & 0 deletions tests/test_pty_stdin.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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)
Loading