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
17 changes: 15 additions & 2 deletions Lib/_pyrepl/unix_console.py
Original file line number Diff line number Diff line change
Expand Up @@ -349,7 +349,14 @@ def prepare(self):
raw.lflag |= termios.ISIG
raw.cc[termios.VMIN] = 1
raw.cc[termios.VTIME] = 0
tcsetattr(self.input_fd, termios.TCSADRAIN, raw)
try:
tcsetattr(self.input_fd, termios.TCSADRAIN, raw)
except termios.error as e:
if e.args[0] != errno.EIO:
# gh-135329: when running under external programs (like strace),
# tcsetattr may fail with EIO. We can safely ignore this
# and continue with default terminal settings.
raise

# In macOS terminal we need to deactivate line wrap via ANSI escape code
if self.is_apple_terminal:
Expand Down Expand Up @@ -381,7 +388,11 @@ def restore(self):
self.__disable_bracketed_paste()
self.__maybe_write_code(self._rmkx)
self.flushoutput()
tcsetattr(self.input_fd, termios.TCSADRAIN, self.__svtermstate)
try:
tcsetattr(self.input_fd, termios.TCSADRAIN, self.__svtermstate)
except termios.error as e:
if e.args[0] != errno.EIO:
raise

if self.is_apple_terminal:
os.write(self.output_fd, b"\033[?7h")
Expand Down Expand Up @@ -420,6 +431,8 @@ def get_event(self, block: bool = True) -> Event | None:
return self.event_queue.get()
else:
continue
elif err.errno == errno.EIO:
raise SystemExit(errno.EIO)
else:
raise
else:
Expand Down
94 changes: 94 additions & 0 deletions Lib/test/test_pyrepl/eio_test_script.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
import errno
import fcntl
import os
import pty
import signal
import sys
import termios


def handler(sig, f):
pass


def create_eio_condition():
# SIGINT handler used to produce an EIO.
# See https://github.com/python/cpython/issues/135329.
try:
master_fd, slave_fd = pty.openpty()
child_pid = os.fork()
if child_pid == 0:
try:
os.setsid()
fcntl.ioctl(slave_fd, termios.TIOCSCTTY, 0)
child_process_group_id = os.getpgrp()
grandchild_pid = os.fork()
if grandchild_pid == 0:
os.setpgid(0, 0) # set process group for grandchild
os.dup2(slave_fd, 0) # redirect stdin
if slave_fd > 2:
os.close(slave_fd)
# Fork grandchild for terminal control manipulation
if os.fork() == 0:
sys.exit(0) # exit the child process that was just obtained
else:
try:
os.tcsetpgrp(0, child_process_group_id)
except OSError:
pass
sys.exit(0)
else:
# Back to child
try:
os.setpgid(grandchild_pid, grandchild_pid)
except ProcessLookupError:
pass
os.tcsetpgrp(slave_fd, grandchild_pid)
if slave_fd > 2:
os.close(slave_fd)
os.waitpid(grandchild_pid, 0)
# Manipulate terminal control to create EIO condition
os.tcsetpgrp(master_fd, child_process_group_id)
# Now try to read from master - this might cause EIO
try:
os.read(master_fd, 1)
except OSError as e:
if e.errno == errno.EIO:
print(f"Setup created EIO condition: {e}", file=sys.stderr)
sys.exit(0)
except Exception as setup_e:
print(f"Setup error: {setup_e}", file=sys.stderr)
sys.exit(1)
else:
# Parent process
os.close(slave_fd)
os.waitpid(child_pid, 0)
# Now replace stdin with master_fd and try to read
os.dup2(master_fd, 0)
os.close(master_fd)
# This should now trigger EIO
print(f"Unexpectedly got input: {input()!r}", file=sys.stderr)
sys.exit(0)
except OSError as e:
if e.errno == errno.EIO:
print(f"Got EIO: {e}", file=sys.stderr)
sys.exit(1)
elif e.errno == errno.ENXIO:
print(f"Got ENXIO (no such device): {e}", file=sys.stderr)
sys.exit(1) # Treat ENXIO as success too
else:
print(f"Got other OSError: errno={e.errno} {e}", file=sys.stderr)
sys.exit(2)
except EOFError as e:
print(f"Got EOFError: {e}", file=sys.stderr)
sys.exit(3)
except Exception as e:
print(f"Got unexpected error: {type(e).__name__}: {e}", file=sys.stderr)
sys.exit(4)


if __name__ == "__main__":
# Set up signal handler for coordination
signal.signal(signal.SIGUSR1, lambda *a: create_eio_condition())
print("READY", flush=True)
signal.pause()
63 changes: 62 additions & 1 deletion Lib/test/test_pyrepl/test_unix_console.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,16 @@
import errno
import itertools
import os
import signal
import subprocess
import sys
import unittest
from functools import partial
from test.support import os_helper
from test.support import script_helper

from unittest import TestCase
from unittest.mock import MagicMock, call, patch, ANY
from unittest.mock import MagicMock, call, patch, ANY, Mock

from .support import handle_all_events, code_to_events, reader_no_colors

Expand Down Expand Up @@ -336,3 +341,59 @@ def test_restore_with_invalid_environ_on_macos(self, _os_write):
os.environ = []
console.prepare() # needed to call restore()
console.restore() # this should succeed


@unittest.skipIf(sys.platform == "win32", "No Unix console on Windows")
class TestUnixConsoleEIOHandling(TestCase):

@patch('_pyrepl.unix_console.tcsetattr')
@patch('_pyrepl.unix_console.tcgetattr')
def test_eio_error_handling_in_restore(self, mock_tcgetattr, mock_tcsetattr):

import termios
mock_termios = Mock()
mock_termios.iflag = 0
mock_termios.oflag = 0
mock_termios.cflag = 0
mock_termios.lflag = 0
mock_termios.cc = [0] * 32
mock_termios.copy.return_value = mock_termios
mock_tcgetattr.return_value = mock_termios

console = UnixConsole(term="xterm")
console.prepare()

mock_tcsetattr.side_effect = termios.error(errno.EIO, "Input/output error")

# EIO error should be handled gracefully in restore()
console.restore()

@unittest.skipUnless(sys.platform == "linux", "Only valid on Linux")
def test_repl_eio(self):
# Use the pty-based approach to simulate EIO error
script_path = os.path.join(os.path.dirname(__file__), "eio_test_script.py")

proc = script_helper.spawn_python(
"-S", script_path,
stderr=subprocess.PIPE,
text=True
)

ready_line = proc.stdout.readline().strip()
if ready_line != "READY" or proc.poll() is not None:
self.fail("Child process failed to start properly")

os.kill(proc.pid, signal.SIGUSR1)
_, err = proc.communicate(timeout=5) # sleep for pty to settle
self.assertEqual(
proc.returncode,
1,
f"Expected EIO/ENXIO error, got return code {proc.returncode}",
)
self.assertTrue(
(
"Got EIO:" in err
or "Got ENXIO:" in err
),
f"Expected EIO/ENXIO error message in stderr: {err}",
)
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Prevent infinite traceback loop when sending CTRL^C to Python through ``strace``.
Loading