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:
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")
return self.event_queue.get()
else:
continue
+ elif err.errno == errno.EIO:
+ raise SystemExit(errno.EIO)
else:
raise
else:
--- /dev/null
+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()
+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
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}",
+ )
--- /dev/null
+Prevent infinite traceback loop when sending CTRL^C to Python through ``strace``.
projects / thirdparty / Python / cpython.git / commitdiff
