Skip to content
Merged
Show file tree
Hide file tree
Changes from 16 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 @@ -340,7 +340,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 @@ -372,7 +379,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

Comment thread
yihong0618 marked this conversation as resolved.
if self.is_apple_terminal:
os.write(self.output_fd, b"\033[?7h")
Expand Down Expand Up @@ -411,6 +422,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
99 changes: 99 additions & 0 deletions Lib/test/test_pyrepl/eio_test_script.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
import os
import sys
import pty
import fcntl
import termios
import signal
import errno
Comment thread
ambv marked this conversation as resolved.
Outdated


def handler(sig, f):
pass


def create_eio_condition():
Comment thread
ambv marked this conversation as resolved.
try:
# gh-135329: try to create a condition that will actually produce EIO
Comment thread
ambv marked this conversation as resolved.
Outdated
master_fd, slave_fd = pty.openpty()
child_pid = os.fork()
if child_pid == 0:
try:
os.setsid()
fcntl.ioctl(slave_fd, termios.TIOCSCTTY, 0)
p2_pgid = os.getpgrp()
Comment thread
ambv marked this conversation as resolved.
Outdated
grandchild_pid = os.fork()
if grandchild_pid == 0:
# Grandchild - set up process group
os.setpgid(0, 0)
# Redirect stdin to slave
os.dup2(slave_fd, 0)
Comment thread
ambv marked this conversation as resolved.
Outdated
if slave_fd > 2:
os.close(slave_fd)
# Fork great-grandchild for terminal control manipulation
ggc_pid = os.fork()
if ggc_pid == 0:
Comment thread
ambv marked this conversation as resolved.
Outdated
# Great-grandchild - just exit quickly
sys.exit(0)
Comment thread
ambv marked this conversation as resolved.
Outdated
else:
# Back to grandchild
try:
os.tcsetpgrp(0, p2_pgid)
except OSError:
pass
sys.exit(0)
Comment thread
ambv marked this conversation as resolved.
Outdated
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, p2_pgid)
# 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
result = input()
print(f"Unexpectedly got input: {repr(result)}", file=sys.stderr)
sys.exit(0)
Comment thread
ambv marked this conversation as resolved.
Outdated
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()
51 changes: 50 additions & 1 deletion Lib/test/test_pyrepl/test_unix_console.py
Original file line number Diff line number Diff line change
@@ -1,12 +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, force_not_colorized_test_class
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

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


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
Comment thread
ambv marked this conversation as resolved.
Outdated
self.assertEqual(proc.returncode, 1, f"Expected EIO error, got return code {proc.returncode}")
self.assertIn("Got EIO:", err, f"Expected EIO error message in stderr: {err}")
Comment thread
ambv marked this conversation as resolved.
Outdated
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