fix: raise EIO without strace

Signed-off-by: yihong0618 <zouzou0208@gmail.com>
Co-authored-by: graymon <greyschwinger@gmail.com>

Author: yihong0618

Lib/test/test_pyrepl/test_unix_console.py

@@ -6,9 +6,7 @@
 import termios
 from functools import partial
 from test.support import os_helper, force_not_colorized_test_class
-from test.support.strace_helper import requires_strace
 import subprocess
-import shutil
 import signal
 import textwrap
 
@@ -339,38 +337,104 @@ def test_eio_error_handling_in_restore(self, mock_tcgetattr, mock_tcsetattr):
                 self.fail("EIO error should have been handled gracefully in restore()")
             raise
 
-@requires_strace()
-class TestEIOWithStrace(unittest.TestCase):
-    def _attach_strace(self, pid):
-        strace = shutil.which("strace")
-        cmd = [strace, "-qq", "-p", str(pid), "-e", "inject=read:error=EIO:when=1", "-o", "/dev/null"]
-        try:
-            p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
-            try:
-                p.communicate(timeout=0.15)
-            except subprocess.TimeoutExpired:
-                return p
-            except (OSError, subprocess.SubprocessError):
-                pass
-        except (OSError, subprocess.SubprocessError):
-            pass
-
-        # Fallback command
-        cmd = [strace, "-qq", "-p", str(pid), "-e", "fault=read:error=EIO:when=1", "-o", "/dev/null"]
-        p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
-        try:
-            p.communicate(timeout=0.15)
-        except subprocess.TimeoutExpired:
-            pass
-        return p
-
+    @unittest.skipUnless(sys.platform == "linux", "Only valid on Linux")
     def test_repl_eio(self):
+        # Use the pty-based approach to simulate EIO error
         child_code = textwrap.dedent("""
-            import signal, sys
-            signal.signal(signal.SIGUSR1, lambda *a: None)
+            import os, sys, pty, fcntl, termios, signal, time, errno
+
+            def handler(sig, f):
+                pass
+
+            def create_eio_condition():
+                try:
+                    # Try to create a condition that will actually produce EIO
+                    # Method: Use your original script's approach with modifications
+                    master_fd, slave_fd = pty.openpty()
+                    # Fork a child that will manipulate the pty
+                    child_pid = os.fork()
+                    if child_pid == 0:
+                        # Child process
+                        try:
+                            # Set up session and control terminal like your script
+                            os.setsid()
+                            fcntl.ioctl(slave_fd, termios.TIOCSCTTY, 0)
+                            # Get process group
+                            p2_pgid = os.getpgrp()
+                            # Fork grandchild
+                            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)
+                                if slave_fd > 2:
+                                    os.close(slave_fd)
+                                # Fork great-grandchild for terminal control manipulation
+                                ggc_pid = os.fork()
+                                if ggc_pid == 0:
+                                    # Great-grandchild - just exit quickly
+                                    sys.exit(0)
+                                else:
+                                    # Back to grandchild
+                                    try:
+                                        os.tcsetpgrp(0, p2_pgid)
+                                    except:
+                                        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, 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)
+                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)
+            # Set up signal handler for coordination
+            signal.signal(signal.SIGUSR1, lambda *a: create_eio_condition())
             print("READY", flush=True)
             signal.pause()
-            input()
         """).strip()
 
         proc = subprocess.Popen(
@@ -381,22 +445,11 @@ def test_repl_eio(self):
             text=True
         )
 
-        ready = False
-        while not ready:
-            line = proc.stdout.readline().strip()
-            if line == "READY":
-                ready = True
-                break
-            if proc.poll() is not None:
-                self.fail("Child process exited unexpectedly")
+        ready_line = proc.stdout.readline().strip()
+        if ready_line != "READY" or proc.poll() is not None:
+            self.fail("Child process failed to start properly")
 
-        tracer = self._attach_strace(proc.pid)
-
-        try:
-            os.kill(proc.pid, signal.SIGUSR1)
-            _, err = proc.communicate(timeout=5)
-        finally:
-            if tracer and tracer.poll() is None:
-                tracer.terminate()
-        self.assertNotEqual(proc.returncode, 0)
-        self.assertTrue("Errno 5" in err or "Input/output error" in err or "EOFError" in err, err)
+        os.kill(proc.pid, signal.SIGUSR1)
+        _, err = proc.communicate(timeout=5) # sleep for pty to settle
+        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}")