Add some tests for handling SIGINT in the PDB client

python · pablogsal · May 5, 2025 · Apr 30, 2025 · Apr 30, 2025 · Apr 30, 2025
commit b26ed5c039cb71422ffca70da91a4e2e051aa7ac
@@ -87,7 +87,9 @@ def do_test(
        *,
        incoming,
        simulate_send_failure=False,
+        simulate_sigint_during_stdout_write=False,
        expected_outgoing=None,
+        expected_outgoing_signals=None,
        expected_completions=None,
        expected_exception=None,
        expected_stdout="",
@@ -96,6 +98,8 @@ def do_test(
    ):
        if expected_outgoing is None:
            expected_outgoing = []
+        if expected_outgoing_signals is None:
+            expected_outgoing_signals = []
        if expected_completions is None:
            expected_completions = []
        if expected_state is None:
@@ -130,7 +134,9 @@ def mock_input(prompt):
            reply = message["input"]
            if isinstance(reply, BaseException):
                raise reply
-            return reply
+            if isinstance(reply, str):
+                return reply
+            return reply()

        with ExitStack() as stack:
            client_sock, server_sock = socket.socketpair()
@@ -155,15 +161,25 @@ def mock_input(prompt):

            stdout = io.StringIO()

+            if simulate_sigint_during_stdout_write:
+                orig_stdout_write = stdout.write
+
+                def sigint_stdout_write(s):
+                    signal.raise_signal(signal.SIGINT)
+                    return orig_stdout_write(s)
+
+                stdout.write = sigint_stdout_write
+
            input_mock = stack.enter_context(
                unittest.mock.patch("pdb.input", side_effect=mock_input)
            )
            stack.enter_context(redirect_stdout(stdout))

+            interrupt_sock = unittest.mock.Mock(spec=socket.socket)
            client = _PdbClient(
                pid=0,
                server_socket=server_sock,
-                interrupt_sock=unittest.mock.Mock(spec=socket.socket),
+                interrupt_sock=interrupt_sock,
            )

            if expected_exception is not None:
@@ -188,6 +204,12 @@ def mock_input(prompt):
        actual_state = {k: getattr(client, k) for k in expected_state}
        self.assertEqual(actual_state, expected_state)

+        outgoing_signals = [
+            signal.Signals(int.from_bytes(call.args[0]))
+            for call in interrupt_sock.sendall.call_args_list
+        ]
+        self.assertEqual(outgoing_signals, expected_outgoing_signals)
+
    def test_remote_immediately_closing_the_connection(self):
        """Test the behavior when the remote closes the connection immediately."""
        incoming = []
@@ -382,11 +404,17 @@ def test_handling_unrecognized_prompt_type(self):
            expected_state={"state": "dumb"},
        )

-    def test_keyboard_interrupt_at_prompt(self):
-        """Test signaling when a prompt gets a KeyboardInterrupt."""
+    def test_sigint_at_prompt(self):
+        """Test signaling when a prompt gets interrupted."""
        incoming = [
            ("server", {"prompt": "(Pdb) ", "state": "pdb"}),
-            ("user", {"prompt": "(Pdb) ", "input": KeyboardInterrupt()}),
+            (
+                "user",
+                {
+                    "prompt": "(Pdb) ",
+                    "input": lambda: signal.raise_signal(signal.SIGINT),
+                },
+            ),
        ]
        self.do_test(
            incoming=incoming,
@@ -396,6 +424,40 @@ def test_keyboard_interrupt_at_prompt(self):
            expected_state={"state": "pdb"},
        )

+    def test_sigint_at_continuation_prompt(self):
+        """Test signaling when a continuation prompt gets interrupted."""
+        incoming = [
+            ("server", {"prompt": "(Pdb) ", "state": "pdb"}),
+            ("user", {"prompt": "(Pdb) ", "input": "if True:"}),
+            (
+                "user",
+                {
+                    "prompt": "...   ",
+                    "input": lambda: signal.raise_signal(signal.SIGINT),
+                },
+            ),
+        ]
+        self.do_test(
+            incoming=incoming,
+            expected_outgoing=[
+                {"signal": "INT"},
+            ],
+            expected_state={"state": "pdb"},
+        )
+
+    def test_sigint_when_writing(self):
+        """Test siginaling when sys.stdout.write() gets interrupted."""
+        incoming = [
+            ("server", {"message": "Some message or other\n", "type": "info"}),
+        ]
+        self.do_test(
+            incoming=incoming,
+            simulate_sigint_during_stdout_write=True,
+            expected_outgoing=[],
+            expected_outgoing_signals=[signal.SIGINT],
+            expected_stdout="Some message or other\n",
+        )
+
    def test_eof_at_prompt(self):
        """Test signaling when a prompt gets an EOFError."""
        incoming = [