Adapt tests

python · ambv · May 2, 2025 · Mar 18, 2025 · Mar 21, 2025 · Mar 21, 2025
commit 9585bd6bb271cfc5362ee162e43be70e1dc42f67
@@ -45,14 +45,20 @@ def run_repl(
         cmdline_args: list[str] | None = None,
         cwd: str | None = None,
         skip: bool = False,
+        timeout: float = SHORT_TIMEOUT,
     ) -> tuple[str, int]:
         temp_dir = None
         if cwd is None:
             temp_dir = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
             cwd = temp_dir.name
         try:
             return self._run_repl(
-                repl_input, env=env, cmdline_args=cmdline_args, cwd=cwd, skip=skip,
+                repl_input,
+                env=env,
+                cmdline_args=cmdline_args,
+                cwd=cwd,
+                skip=skip,
+                timeout=timeout,
             )
         finally:
             if temp_dir is not None:
@@ -66,6 +72,7 @@ def _run_repl(
         cmdline_args: list[str] | None,
         cwd: str,
         skip: bool,
+        timeout: float,
     ) -> tuple[str, int]:
         assert pty
         master_fd, slave_fd = pty.openpty()
@@ -103,7 +110,7 @@ def _run_repl(
         os.write(master_fd, repl_input.encode("utf-8"))
 
         output = []
-        while select.select([master_fd], [], [], SHORT_TIMEOUT)[0]:
+        while select.select([master_fd], [], [], timeout)[0]:
             try:
                 data = os.read(master_fd, 1024).decode("utf-8")
                 if not data:
@@ -114,12 +121,12 @@ def _run_repl(
         else:
             os.close(master_fd)
             process.kill()
-            process.wait(timeout=SHORT_TIMEOUT)
+            process.wait(timeout=timeout)
             self.fail(f"Timeout while waiting for output, got: {''.join(output)}")
 
         os.close(master_fd)
         try:
-            exit_code = process.wait(timeout=SHORT_TIMEOUT)
+            exit_code = process.wait(timeout=timeout)
         except subprocess.TimeoutExpired:
             process.kill()
             exit_code = process.wait()
@@ -1561,25 +1568,29 @@ def test_readline_history_file(self):
 
     def test_history_survive_crash(self):
         env = os.environ.copy()
-        commands = "1\nexit()\n"
-        output, exit_code = self.run_repl(commands, env=env, skip=True)
 
         with tempfile.NamedTemporaryFile() as hfile:
             env["PYTHON_HISTORY"] = hfile.name
-            commands = "spam\nimport time\ntime.sleep(1000)\npreved\n"
+
+            commands = "1\n2\n3\nexit()\n"
+            output, exit_code = self.run_repl(commands, env=env, skip=True)
+
+            commands = "spam\nimport time\ntime.sleep(1000)\nquit\n"
             try:
-                self.run_repl(commands, env=env)
+                self.run_repl(commands, env=env, timeout=3)
             except AssertionError:
                 pass
 
             history = pathlib.Path(hfile.name).read_text()
+            self.assertIn("2", history)
+            self.assertIn("exit()", history)
             self.assertIn("spam", history)
-            self.assertIn("time", history)
+            self.assertIn("import time", history)
             self.assertNotIn("sleep", history)
-            self.assertNotIn("preved", history)
+            self.assertNotIn("quit", history)
 
     def test_keyboard_interrupt_after_isearch(self):
-        output, exit_code = self.run_repl(["\x12", "\x03", "exit"])
+        output, exit_code = self.run_repl("\x12\x03exit\n")
         self.assertEqual(exit_code, 0)
 
     def test_prompt_after_help(self):

@@ -7,7 +7,8 @@
 
 from .support import handle_all_events, handle_events_narrow_console
 from .support import ScreenEqualMixin, code_to_events
-from .support import prepare_reader, prepare_console, reader_force_colors
+from .support import prepare_console, reader_force_colors
+from .support import reader_no_colors as prepare_reader
 from _pyrepl.console import Event
 from _pyrepl.reader import Reader
 from _pyrepl.utils import TAG_TO_ANSI
@@ -131,7 +132,7 @@ def test_control_characters(self):
         events = code_to_events(code)
         reader, _ = handle_all_events(events, prepare_reader=reader_force_colors)
         self.assert_screen_equal(reader, 'flag = "🏳️\\u200d🌈"', clean=True)
-        self.assert_screen_equal(reader, 'flag = {s}"🏳️\\u200d🌈"{z}'.format(**colors))
+        self.assert_screen_equal(reader, 'flag {o}={z} {s}"🏳️\\u200d🌈"{z}'.format(**colors))
 
     def test_setpos_from_xy_multiple_lines(self):
         # fmt: off

@@ -33,10 +33,12 @@ def unix_console(events, **kwargs):
 
 handle_events_unix_console = partial(
     handle_all_events,
-    prepare_console=partial(unix_console),
+    prepare_reader=reader_no_colors,
+    prepare_console=unix_console,
 )
 handle_events_narrow_unix_console = partial(
     handle_all_events,
+    prepare_reader=reader_no_colors,
     prepare_console=partial(unix_console, width=5),
 )
 handle_events_short_unix_console = partial(