https://github.com/python/cpython/commit/6cbeb6ab916899c0740104483144e597363fe27a
commit: 6cbeb6ab916899c0740104483144e597363fe27a
branch: main
author: Pablo Galindo Salgado <[email protected]>
committer: pablogsal <[email protected]>
date: 2025-04-25T14:46:44+01:00
summary:
gh-132912: Account for race in test_keyboard_interrupt in test_remote_pdb
(#132929)
files:
M Lib/test/test_remote_pdb.py
diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index 271d974f39bdc0..6226a2563ba316 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -1,4 +1,5 @@
import io
+import time
import json
import os
import signal
@@ -462,8 +463,6 @@ def test_breakpoints(self):
self.assertIn("Function returned: 42", stdout)
self.assertEqual(process.returncode, 0)
- # gh-132912: The test fails randomly
- @unittest.skipIf(True, "flaky test")
def test_keyboard_interrupt(self):
"""Test that sending keyboard interrupt breaks into pdb."""
synchronizer_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -512,15 +511,22 @@ def bar():
# Wait until execution has continued
synchronizer_sock.accept()[0].close()
+ # Wait a bit so the remote leaves create_connection(). This is not
+ # required but makes the rest of the test faster as we will exit
the main
+ # loop immediately by setting iterations to 0.
+ time.sleep(0.1)
+
# Inject a script to interrupt the running process
self._send_interrupt(process.pid)
messages = self._read_until_prompt(client_file)
- # Verify we got the keyboard interrupt message
- interrupt_msg = next(msg['message'] for msg in messages if
'message' in msg)
- self.assertIn("bar()", interrupt_msg)
+ # Verify we got the keyboard interrupt message. Is possible that
we get interrupted somewhere
+ # in bar() or when leving create_connection()
+ interrupt_msgs = [msg['message'] for msg in messages if 'message'
in msg]
+ expected_msg = [msg for msg in interrupt_msgs if "bar()" in msg or
"create_connection()" in msg]
+ self.assertGreater(len(expected_msg), 0)
- # Continue to end
+ # Continue to end as fast as we can
self._send_command(client_file, "iterations = 0")
self._send_command(client_file, "c")
stdout, _ = process.communicate(timeout=5)
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3/lists/python-checkins.python.org/
Member address: [email protected]
