Address feedback

Author: edvilme

Lib/_pyrepl/__init__.py

@@ -17,9 +17,3 @@
 # RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
 # CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
 # CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
-
-from collections.abc import Callable as _Callable
-
-
-# Declared here so external tools can detect support and install a callback.
-pre_execution_hook: _Callable[[str], object] | None = None

Lib/_pyrepl/simple_interact.py

@@ -30,11 +30,25 @@
 import code
 import warnings
 
-import _pyrepl
-
 from .readline import _get_reader, multiline_input, append_history_file
 
 
+# Called after the user submits a Python statement, but before it is executed.
+# The hook is not called for PyREPL commands such as ``clear``. Exceptions raised
+# by the hook and its return value are ignored. External tools such as IDEs can
+# install a hook to augment the behavior of the REPL.
+#
+# For example, VS Code can mark the start of command execution:
+#
+#     from _pyrepl import simple_interact
+#
+#     def vscode_statement_submitted(statement: str) -> None:
+#         print("\x1b]633;C\x07", end="")
+#
+#     simple_interact.statement_submitted_hook = vscode_statement_submitted
+statement_submitted_hook = None
+
+
 _error: tuple[type[Exception], ...] | type[Exception]
 try:
     from .unix_console import _error
@@ -147,10 +161,9 @@ def maybe_run_command(statement: str) -> bool:
             if maybe_run_command(statement):
                 continue
 
-            pre_execution_hook = getattr(_pyrepl, "pre_execution_hook", None)
-            if callable(pre_execution_hook):
+            if statement_submitted_hook is not None:
                 try:
-                    pre_execution_hook(statement)
+                    statement_submitted_hook(statement)
                 except Exception:
                     pass
 

Lib/test/test_pyrepl/test_interact.py

@@ -5,10 +5,9 @@
 from unittest.mock import MagicMock, patch
 from textwrap import dedent
 
-import _pyrepl
-
 from test.support import force_not_colorized
 
+from _pyrepl import simple_interact
 from _pyrepl.console import InteractiveColoredConsole
 from _pyrepl.simple_interact import _more_lines
 
@@ -303,14 +302,9 @@ def f():
         self.assertEqual(count, 1)
 
 
-class TestPreExecutionHook(unittest.TestCase):
-
-    def test_hook_is_unset_by_default(self):
-        self.assertIsNone(_pyrepl.pre_execution_hook)
-
-    def _run_interactive(self, statements, *, pre_execution_hook=None):
-        from _pyrepl.simple_interact import run_multiline_interactive_console
+class TestStatementSubmittedHook(unittest.TestCase):
 
+    def _run_interactive(self, statements, hook):
         console = InteractiveColoredConsole()
         statement_iter = iter(statements)
 
@@ -320,51 +314,37 @@ def fake_multiline_input(more_lines, ps1, ps2):
             except StopIteration:
                 raise EOFError
 
-        patches = [
-            patch(
-                "_pyrepl.simple_interact.multiline_input",
+        output = io.StringIO()
+        with (
+            patch.object(simple_interact, "statement_submitted_hook", hook),
+            patch.object(
+                simple_interact,
+                "multiline_input",
                 side_effect=fake_multiline_input,
             ),
-            patch("_pyrepl.simple_interact._get_reader"),
-            patch("_pyrepl.simple_interact.append_history_file"),
+            patch.object(simple_interact, "_get_reader"),
+            patch.object(simple_interact, "append_history_file"),
             patch("_pyrepl.readline._setup"),
-        ]
-        if pre_execution_hook is not None:
-            patches.append(
-                patch.object(
-                    _pyrepl,
-                    "pre_execution_hook",
-                    pre_execution_hook,
-                )
-            )
-
-        output = io.StringIO()
-        with contextlib.ExitStack() as stack:
-            for context_manager in patches:
-                stack.enter_context(context_manager)
-            stack.enter_context(contextlib.redirect_stdout(output))
-            stack.enter_context(contextlib.redirect_stderr(output))
-            run_multiline_interactive_console(console)
+            contextlib.redirect_stdout(output),
+            contextlib.redirect_stderr(output),
+        ):
+            simple_interact.run_multiline_interactive_console(console)
 
         return output.getvalue(), console.locals
 
-    def test_hook_called_with_statement(self):
-        hook = MagicMock()
-        self._run_interactive(["x = 1"], pre_execution_hook=hook)
-        hook.assert_called_once_with("x = 1")
-
-    def test_hook_exception_does_not_break_repl(self):
-        hook = MagicMock(side_effect=RuntimeError("hook error"))
-        output, namespace = self._run_interactive(
-            ["x = 1", "y = 2"],
-            pre_execution_hook=hook,
+    def test_hook_called_before_statement_execution(self):
+        statement = "print('statement executed')"
+        escape_sequence = "\x1b]633;C\x07"
+        hook = MagicMock(
+            side_effect=lambda statement: print(escape_sequence, end="")
         )
-        self.assertEqual(hook.call_count, 2)
-        self.assertEqual(namespace["x"], 1)
-        self.assertEqual(namespace["y"], 2)
-        self.assertNotIn("hook error", output)
+        output, _ = self._run_interactive([statement], hook)
+
+        hook.assert_called_once_with(statement)
+        self.assertEqual(output, f"{escape_sequence}statement executed\n")
 
     def test_hook_not_called_for_repl_commands(self):
         hook = MagicMock()
-        self._run_interactive(["clear"], pre_execution_hook=hook)
+        self._run_interactive(["clear"], hook)
+
         hook.assert_not_called()