Skip to content
Open
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
10 changes: 8 additions & 2 deletions invoke/terminals.py
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,12 @@ def bytes_to_read(input_: IO) -> int:
# it's not a tty but has a fileno, or vice versa; neither is typically
# going to work re: ioctl().
if not WINDOWS and isatty(input_) and has_fileno(input_):
fionread = fcntl.ioctl(input_, termios.FIONREAD, b" ")
return int(struct.unpack("h", fionread)[0])
# FIONREAD writes a C int (typically 4 bytes). Pass a 4-byte buffer
# and unpack as a signed int. The previous 2-byte buffer ("h")
# happened to work on <=3.13 thanks to CPython's internal 1024-byte
# static copy buffer, but CPython 3.14 tightened buffer-size checks
# (cpython#144206) so undersized mutable buffers now raise
# SystemError: buffer overflow. See pyinvoke/invoke#1070.
fionread = fcntl.ioctl(input_, termios.FIONREAD, b"\x00\x00\x00\x00")
return int(struct.unpack("i", fionread)[0])
return 1
2 changes: 1 addition & 1 deletion tests/runners.py
Original file line number Diff line number Diff line change
Expand Up @@ -1226,7 +1226,7 @@ def fake_ioctl(fd, cmd, buf):
# This works since each mocked attr will still be its own mock
# object with a distinct 'is' identity.
if cmd is termios.FIONREAD:
return struct.pack("h", len(stdin_data))
return struct.pack("i", len(stdin_data))

ioctl.side_effect = fake_ioctl
# Set up our runner as one w/ mocked stdin writing (simplest way to
Expand Down
17 changes: 17 additions & 0 deletions tests/terminals.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import fcntl
import struct
import termios

from unittest.mock import Mock, patch
Expand Down Expand Up @@ -71,6 +72,22 @@ def returns_1_when_stream_has_fileno_but_is_not_a_tty(self, fcntl):
assert bytes_to_read(stream) == 1
assert not fcntl.ioctl.called

@patch("invoke.terminals.fcntl")
def uses_4_byte_buffer_for_FIONREAD(self, fcntl):
# FIONREAD writes a C int (4 bytes); the buffer must match to
# avoid SystemError on Python 3.14+ (#1070)
fcntl.ioctl.return_value = struct.pack("i", 42)
stream = Mock(isatty=lambda: True, fileno=lambda: 7)
assert bytes_to_read(stream) == 42
buf = fcntl.ioctl.call_args[0][2]
assert len(buf) == 4

@patch("invoke.terminals.fcntl")
def returns_0_when_FIONREAD_is_zero(self, fcntl):
fcntl.ioctl.return_value = struct.pack("i", 0)
stream = Mock(isatty=lambda: True, fileno=lambda: 7)
assert bytes_to_read(stream) == 0

def returns_FIONREAD_result_when_stream_is_a_tty(self):
skip()

Expand Down