https://github.com/python/cpython/commit/98d462cf4de82d4ef40b20bd346db1ba7edfa35c
commit: 98d462cf4de82d4ef40b20bd346db1ba7edfa35c
branch: main
author: Sam Gross <colesb...@gmail.com>
committer: colesbury <colesb...@gmail.com>
date: 2025-07-29T14:25:50-04:00
summary:

gh-137179: Fix flaky test_history_survive_crash test (gh-137180)

Kill the REPL subprocess once it prints the output from the command
immediately before the `time.sleep()`.

files:
M Lib/test/test_pyrepl/test_pyrepl.py

diff --git a/Lib/test/test_pyrepl/test_pyrepl.py 
b/Lib/test/test_pyrepl/test_pyrepl.py
index 2588a5ba27d776..8e4450fdf99ecd 100644
--- a/Lib/test/test_pyrepl/test_pyrepl.py
+++ b/Lib/test/test_pyrepl/test_pyrepl.py
@@ -51,6 +51,7 @@ def run_repl(
         cwd: str | None = None,
         skip: bool = False,
         timeout: float = SHORT_TIMEOUT,
+        exit_on_output: str | None = None,
     ) -> tuple[str, int]:
         temp_dir = None
         if cwd is None:
@@ -64,6 +65,7 @@ def run_repl(
                 cwd=cwd,
                 skip=skip,
                 timeout=timeout,
+                exit_on_output=exit_on_output,
             )
         finally:
             if temp_dir is not None:
@@ -78,6 +80,7 @@ def _run_repl(
         cwd: str,
         skip: bool,
         timeout: float,
+        exit_on_output: str | None,
     ) -> tuple[str, int]:
         assert pty
         master_fd, slave_fd = pty.openpty()
@@ -123,6 +126,11 @@ def _run_repl(
             except OSError:
                 break
             output.append(data)
+            if exit_on_output is not None:
+                output = ["".join(output)]
+                if exit_on_output in output[0]:
+                    process.kill()
+                    break
         else:
             os.close(master_fd)
             process.kill()
@@ -1718,18 +1726,24 @@ def test_history_survive_crash(self):
 
             commands = "1\n2\n3\nexit()\n"
             output, exit_code = self.run_repl(commands, env=env, skip=True)
+            self.assertEqual(exit_code, 0)
 
-            commands = "spam\nimport time\ntime.sleep(1000)\nquit\n"
-            try:
-                self.run_repl(commands, env=env, timeout=3)
-            except AssertionError:
-                pass
+            # Run until "0xcafe" is printed (as "51966") and then kill the
+            # process to simulate a crash. Note that the output also includes
+            # the echoed input commands.
+            commands = "spam\nimport time\n0xcafe\ntime.sleep(1000)\nquit\n"
+            output, exit_code = self.run_repl(commands, env=env,
+                                              exit_on_output="51966")
+            self.assertNotEqual(exit_code, 0)
 
             history = pathlib.Path(hfile.name).read_text()
             self.assertIn("2", history)
             self.assertIn("exit()", history)
             self.assertIn("spam", history)
             self.assertIn("import time", history)
+            # History is written after each command's output is printed to the
+            # console, so depending on how quickly the process is killed,
+            # the last command may or may not be written to the history file.
             self.assertNotIn("sleep", history)
             self.assertNotIn("quit", history)
 

_______________________________________________
Python-checkins mailing list -- python-checkins@python.org
To unsubscribe send an email to python-checkins-le...@python.org
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: arch...@mail-archive.com

Reply via email to