1  """Tests for OSError EIO suppression during interrupt shutdown (#13710).
  2  
  3  When the user interrupts a running task, prompt_toolkit tries to flush
  4  stdout during emergency shutdown.  If stdout is already in a broken state
  5  (redirected to /dev/null, pipe closed, etc.), the flush raises
  6  ``OSError: [Errno 5] Input/output error``.
  7  
  8  The ``_suppress_closed_loop_errors`` asyncio exception handler and the
  9  outer ``except (KeyError, OSError)`` block must both suppress this error
 10  to prevent a hard crash.
 11  """
 12  
 13  from __future__ import annotations
 14  
 15  import errno
 16  import os
 17  from unittest.mock import MagicMock
 18  
 19  import pytest
 20  
 21  
 22  # ---------------------------------------------------------------------------
 23  # _suppress_closed_loop_errors – asyncio exception handler
 24  # ---------------------------------------------------------------------------
 25  
 26  def _make_suppress_fn():
 27      """Build a standalone copy of ``_suppress_closed_loop_errors``.
 28  
 29      The real function is defined as a closure inside
 30      ``CLI._run_interactive``; we reconstruct an equivalent here so the
 31      unit tests don't need a full CLI instance.
 32      """
 33      def _suppress_closed_loop_errors(loop, context):
 34          exc = context.get("exception")
 35          if isinstance(exc, RuntimeError) and "Event loop is closed" in str(exc):
 36              return
 37          if isinstance(exc, KeyError) and "is not registered" in str(exc):
 38              return
 39          if isinstance(exc, OSError) and getattr(exc, "errno", None) == errno.EIO:
 40              return
 41          loop.default_exception_handler(context)
 42      return _suppress_closed_loop_errors
 43  
 44  
 45  class TestSuppressClosedLoopErrors:
 46      """Verify the asyncio exception handler suppresses expected errors."""
 47  
 48      def test_suppresses_event_loop_closed(self):
 49          handler = _make_suppress_fn()
 50          loop = MagicMock()
 51          handler(loop, {"exception": RuntimeError("Event loop is closed")})
 52          loop.default_exception_handler.assert_not_called()
 53  
 54      def test_suppresses_key_not_registered(self):
 55          handler = _make_suppress_fn()
 56          loop = MagicMock()
 57          handler(loop, {"exception": KeyError("0 is not registered")})
 58          loop.default_exception_handler.assert_not_called()
 59  
 60      def test_suppresses_oserror_eio(self):
 61          """OSError with errno.EIO must be suppressed (#13710)."""
 62          handler = _make_suppress_fn()
 63          loop = MagicMock()
 64          exc = OSError(errno.EIO, "Input/output error")
 65          handler(loop, {"exception": exc})
 66          loop.default_exception_handler.assert_not_called()
 67  
 68      def test_does_not_suppress_oserror_other_errno(self):
 69          """OSError with a different errno must still propagate."""
 70          handler = _make_suppress_fn()
 71          loop = MagicMock()
 72          exc = OSError(errno.EACCES, "Permission denied")
 73          handler(loop, {"exception": exc})
 74          loop.default_exception_handler.assert_called_once()
 75  
 76      def test_does_not_suppress_unrelated_exception(self):
 77          """Unrelated exceptions must still propagate."""
 78          handler = _make_suppress_fn()
 79          loop = MagicMock()
 80          handler(loop, {"exception": ValueError("something else")})
 81          loop.default_exception_handler.assert_called_once()
 82  
 83      def test_no_exception_key(self):
 84          """Context without 'exception' must propagate to default handler."""
 85          handler = _make_suppress_fn()
 86          loop = MagicMock()
 87          handler(loop, {"message": "some log"})
 88          loop.default_exception_handler.assert_called_once()
 89  
 90  
 91  # ---------------------------------------------------------------------------
 92  # Outer except block – EIO handling
 93  # ---------------------------------------------------------------------------
 94  
 95  class TestOuterExceptEIO:
 96      """Verify the outer ``except (KeyError, OSError)`` block logic."""
 97  
 98      def test_eio_does_not_reraise(self):
 99          """OSError with errno.EIO should be silently suppressed."""
100          exc = OSError(errno.EIO, "Input/output error")
101          # Simulate the condition check from the outer except block:
102          assert isinstance(exc, OSError)
103          assert getattr(exc, "errno", None) == errno.EIO
104  
105      def test_bad_file_descriptor_matches(self):
106          """'Bad file descriptor' string should be caught."""
107          exc = OSError(errno.EBADF, "Bad file descriptor")
108          assert "Bad file descriptor" in str(exc)
109  
110      def test_other_oserror_reraises(self):
111          """Other OSError variants must not match the EIO guard."""
112          exc = OSError(errno.EACCES, "Permission denied")
113          assert not (getattr(exc, "errno", None) == errno.EIO)
114          assert "is not registered" not in str(exc)
115          assert "Bad file descriptor" not in str(exc)