Updated bdb.py + test_bdb.py (#6593)

* Updated bdb.py + test_bdb.py

* Double checked test_bdb.py with the script

Lib/bdb.py

@@ -3,6 +3,7 @@
 import fnmatch
 import sys
 import os
+from contextlib import contextmanager
 from inspect import CO_GENERATOR, CO_COROUTINE, CO_ASYNC_GENERATOR
 
 __all__ = ["BdbQuit", "Bdb", "Breakpoint"]
@@ -32,7 +33,12 @@ class Bdb:
         self.skip = set(skip) if skip else None
         self.breaks = {}
         self.fncache = {}
+        self.frame_trace_lines_opcodes = {}
         self.frame_returning = None
+        self.trace_opcodes = False
+        self.enterframe = None
+        self.cmdframe = None
+        self.cmdlineno = None
 
         self._load_breaks()
 
@@ -60,6 +66,12 @@ class Bdb:
         self.botframe = None
         self._set_stopinfo(None, None)
 
+    @contextmanager
+    def set_enterframe(self, frame):
+        self.enterframe = frame
+        yield
+        self.enterframe = None
+
     def trace_dispatch(self, frame, event, arg):
         """Dispatch a trace function for debugged frames based on the event.
 
@@ -84,24 +96,28 @@ class Bdb:
 
         The arg parameter depends on the previous event.
         """
-        if self.quitting:
-            return # None
-        if event == 'line':
-            return self.dispatch_line(frame)
-        if event == 'call':
-            return self.dispatch_call(frame, arg)
-        if event == 'return':
-            return self.dispatch_return(frame, arg)
-        if event == 'exception':
-            return self.dispatch_exception(frame, arg)
-        if event == 'c_call':
+
+        with self.set_enterframe(frame):
+            if self.quitting:
+                return # None
+            if event == 'line':
+                return self.dispatch_line(frame)
+            if event == 'call':
+                return self.dispatch_call(frame, arg)
+            if event == 'return':
+                return self.dispatch_return(frame, arg)
+            if event == 'exception':
+                return self.dispatch_exception(frame, arg)
+            if event == 'c_call':
+                return self.trace_dispatch
+            if event == 'c_exception':
+                return self.trace_dispatch
+            if event == 'c_return':
+                return self.trace_dispatch
+            if event == 'opcode':
+                return self.dispatch_opcode(frame, arg)
+            print('bdb.Bdb.dispatch: unknown debugging event:', repr(event))
             return self.trace_dispatch
-        if event == 'c_exception':
-            return self.trace_dispatch
-        if event == 'c_return':
-            return self.trace_dispatch
-        print('bdb.Bdb.dispatch: unknown debugging event:', repr(event))
-        return self.trace_dispatch
 
     def dispatch_line(self, frame):
         """Invoke user function and return trace function for line event.
@@ -110,7 +126,12 @@ class Bdb:
         self.user_line(). Raise BdbQuit if self.quitting is set.
         Return self.trace_dispatch to continue tracing in this scope.
         """
-        if self.stop_here(frame) or self.break_here(frame):
+        # GH-136057
+        # For line events, we don't want to stop at the same line where
+        # the latest next/step command was issued.
+        if (self.stop_here(frame) or self.break_here(frame)) and not (
+            self.cmdframe == frame and self.cmdlineno == frame.f_lineno
+        ):
             self.user_line(frame)
             if self.quitting: raise BdbQuit
         return self.trace_dispatch
@@ -157,6 +178,11 @@ class Bdb:
             # The user issued a 'next' or 'until' command.
             if self.stopframe is frame and self.stoplineno != -1:
                 self._set_stopinfo(None, None)
+            # The previous frame might not have f_trace set, unless we are
+            # issuing a command that does not expect to stop, we should set
+            # f_trace
+            if self.stoplineno != -1:
+                self._set_caller_tracefunc(frame)
         return self.trace_dispatch
 
     def dispatch_exception(self, frame, arg):
@@ -186,6 +212,17 @@ class Bdb:
 
         return self.trace_dispatch
 
+    def dispatch_opcode(self, frame, arg):
+        """Invoke user function and return trace function for opcode event.
+        If the debugger stops on the current opcode, invoke
+        self.user_opcode(). Raise BdbQuit if self.quitting is set.
+        Return self.trace_dispatch to continue tracing in this scope.
+        """
+        if self.stop_here(frame) or self.break_here(frame):
+            self.user_opcode(frame)
+            if self.quitting: raise BdbQuit
+        return self.trace_dispatch
+
     # Normally derived classes don't override the following
     # methods, but they may if they want to redefine the
     # definition of stopping and breakpoints.
@@ -272,7 +309,22 @@ class Bdb:
         """Called when we stop on an exception."""
         pass
 
-    def _set_stopinfo(self, stopframe, returnframe, stoplineno=0):
+    def user_opcode(self, frame):
+        """Called when we are about to execute an opcode."""
+        pass
+
+    def _set_trace_opcodes(self, trace_opcodes):
+        if trace_opcodes != self.trace_opcodes:
+            self.trace_opcodes = trace_opcodes
+            frame = self.enterframe
+            while frame is not None:
+                frame.f_trace_opcodes = trace_opcodes
+                if frame is self.botframe:
+                    break
+                frame = frame.f_back
+
+    def _set_stopinfo(self, stopframe, returnframe, stoplineno=0, opcode=False,
+                      cmdframe=None, cmdlineno=None):
         """Set the attributes for stopping.
 
         If stoplineno is greater than or equal to 0, then stop at line
@@ -285,6 +337,21 @@ class Bdb:
         # stoplineno >= 0 means: stop at line >= the stoplineno
         # stoplineno -1 means: don't stop at all
         self.stoplineno = stoplineno
+        # cmdframe/cmdlineno is the frame/line number when the user issued
+        # step/next commands.
+        self.cmdframe = cmdframe
+        self.cmdlineno = cmdlineno
+        self._set_trace_opcodes(opcode)
+
+    def _set_caller_tracefunc(self, current_frame):
+        # Issue #13183: pdb skips frames after hitting a breakpoint and running
+        # step commands.
+        # Restore the trace function in the caller (that may not have been set
+        # for performance reasons) when returning from the current frame, unless
+        # the caller is the botframe.
+        caller_frame = current_frame.f_back
+        if caller_frame and not caller_frame.f_trace and caller_frame is not self.botframe:
+            caller_frame.f_trace = self.trace_dispatch
 
     # Derived classes and clients can call the following methods
     # to affect the stepping state.
@@ -299,19 +366,17 @@ class Bdb:
 
     def set_step(self):
         """Stop after one line of code."""
-        # Issue #13183: pdb skips frames after hitting a breakpoint and running
-        # step commands.
-        # Restore the trace function in the caller (that may not have been set
-        # for performance reasons) when returning from the current frame.
-        if self.frame_returning:
-            caller_frame = self.frame_returning.f_back
-            if caller_frame and not caller_frame.f_trace:
-                caller_frame.f_trace = self.trace_dispatch
-        self._set_stopinfo(None, None)
+        # set_step() could be called from signal handler so enterframe might be None
+        self._set_stopinfo(None, None, cmdframe=self.enterframe,
+                           cmdlineno=getattr(self.enterframe, 'f_lineno', None))
+
+    def set_stepinstr(self):
+        """Stop before the next instruction."""
+        self._set_stopinfo(None, None, opcode=True)
 
     def set_next(self, frame):
         """Stop on the next line in or below the given frame."""
-        self._set_stopinfo(frame, None)
+        self._set_stopinfo(frame, None, cmdframe=frame, cmdlineno=frame.f_lineno)
 
     def set_return(self, frame):
         """Stop when returning from the given frame."""
@@ -328,11 +393,15 @@ class Bdb:
         if frame is None:
             frame = sys._getframe().f_back
         self.reset()
-        while frame:
-            frame.f_trace = self.trace_dispatch
-            self.botframe = frame
-            frame = frame.f_back
-        self.set_step()
+        with self.set_enterframe(frame):
+            while frame:
+                frame.f_trace = self.trace_dispatch
+                self.botframe = frame
+                self.frame_trace_lines_opcodes[frame] = (frame.f_trace_lines, frame.f_trace_opcodes)
+                # We need f_trace_lines == True for the debugger to work
+                frame.f_trace_lines = True
+                frame = frame.f_back
+            self.set_stepinstr()
         sys.settrace(self.trace_dispatch)
 
     def set_continue(self):
@@ -349,6 +418,9 @@ class Bdb:
             while frame and frame is not self.botframe:
                 del frame.f_trace
                 frame = frame.f_back
+            for frame, (trace_lines, trace_opcodes) in self.frame_trace_lines_opcodes.items():
+                frame.f_trace_lines, frame.f_trace_opcodes = trace_lines, trace_opcodes
+            self.frame_trace_lines_opcodes = {}
 
     def set_quit(self):
         """Set quitting attribute to True.
@@ -387,6 +459,14 @@ class Bdb:
             return 'Line %s:%d does not exist' % (filename, lineno)
         self._add_to_breaks(filename, lineno)
         bp = Breakpoint(filename, lineno, temporary, cond, funcname)
+        # After we set a new breakpoint, we need to search through all frames
+        # and set f_trace to trace_dispatch if there could be a breakpoint in
+        # that frame.
+        frame = self.enterframe
+        while frame:
+            if self.break_anywhere(frame):
+                frame.f_trace = self.trace_dispatch
+            frame = frame.f_back
         return None
 
     def _load_breaks(self):

Lib/test/test_bdb.py

@@ -228,6 +228,10 @@ class Tracer(Bdb):
         self.process_event('exception', frame)
         self.next_set_method()
 
+    def user_opcode(self, frame):
+        self.process_event('opcode', frame)
+        self.next_set_method()
+
     def do_clear(self, arg):
         # The temporary breakpoints are deleted in user_line().
         bp_list = [self.currentbp]
@@ -366,7 +370,7 @@ class Tracer(Bdb):
         set_method = getattr(self, 'set_' + set_type)
 
         # The following set methods give back control to the tracer.
-        if set_type in ('step', 'continue', 'quit'):
+        if set_type in ('step', 'stepinstr', 'continue', 'quit'):
             set_method()
             return
         elif set_type in ('next', 'return'):
@@ -586,7 +590,7 @@ class BaseTestCase(unittest.TestCase):
 class StateTestCase(BaseTestCase):
     """Test the step, next, return, until and quit 'set_' methods."""
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_step(self):
         self.expect_set = [
             ('line', 2, 'tfunc_main'),  ('step', ),
@@ -597,7 +601,7 @@ class StateTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             tracer.runcall(tfunc_main)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_step_next_on_last_statement(self):
         for set_type in ('step', 'next'):
             with self.subTest(set_type=set_type):
@@ -612,7 +616,18 @@ class StateTestCase(BaseTestCase):
                 with TracerRun(self) as tracer:
                     tracer.runcall(tfunc_main)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON')
+    # AssertionError: All paired tuples have not been processed, the last one was number 1 [('next',), ('quit',)]
+    def test_stepinstr(self):
+        self.expect_set = [
+            ('line',   2, 'tfunc_main'),  ('stepinstr', ),
+            ('opcode', 2, 'tfunc_main'),  ('next', ),
+            ('line',   3, 'tfunc_main'),  ('quit', ),
+        ]
+        with TracerRun(self) as tracer:
+            tracer.runcall(tfunc_main)
+
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_next(self):
         self.expect_set = [
             ('line', 2, 'tfunc_main'),   ('step', ),
@@ -624,7 +639,7 @@ class StateTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             tracer.runcall(tfunc_main)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_next_over_import(self):
         code = """
             def main():
@@ -639,7 +654,7 @@ class StateTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_next_on_plain_statement(self):
         # Check that set_next() is equivalent to set_step() on a plain
         # statement.
@@ -652,7 +667,7 @@ class StateTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             tracer.runcall(tfunc_main)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_next_in_caller_frame(self):
         # Check that set_next() in the caller frame causes the tracer
         # to stop next in the caller frame.
@@ -666,7 +681,7 @@ class StateTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             tracer.runcall(tfunc_main)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_return(self):
         self.expect_set = [
             ('line', 2, 'tfunc_main'),    ('step', ),
@@ -679,7 +694,7 @@ class StateTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             tracer.runcall(tfunc_main)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_return_in_caller_frame(self):
         self.expect_set = [
             ('line', 2, 'tfunc_main'),   ('step', ),
@@ -691,7 +706,7 @@ class StateTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             tracer.runcall(tfunc_main)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_until(self):
         self.expect_set = [
             ('line', 2, 'tfunc_main'),  ('step', ),
@@ -703,7 +718,7 @@ class StateTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             tracer.runcall(tfunc_main)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_until_with_too_large_count(self):
         self.expect_set = [
             ('line', 2, 'tfunc_main'),               break_in_func('tfunc_first'),
@@ -714,7 +729,7 @@ class StateTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             tracer.runcall(tfunc_main)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_until_in_caller_frame(self):
         self.expect_set = [
             ('line', 2, 'tfunc_main'),  ('step', ),
@@ -726,7 +741,8 @@ class StateTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             tracer.runcall(tfunc_main)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
+    @patch_list(sys.meta_path)
     def test_skip(self):
         # Check that tracing is skipped over the import statement in
         # 'tfunc_import()'.
@@ -759,7 +775,7 @@ class StateTestCase(BaseTestCase):
         bdb = Bdb(skip=['anything*'])
         self.assertIs(bdb.is_skipped_module(None), False)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_down(self):
         # Check that set_down() raises BdbError at the newest frame.
         self.expect_set = [
@@ -768,7 +784,7 @@ class StateTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             self.assertRaises(BdbError, tracer.runcall, tfunc_main)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_up(self):
         self.expect_set = [
             ('line', 2, 'tfunc_main'),  ('step', ),
@@ -782,7 +798,7 @@ class StateTestCase(BaseTestCase):
 class BreakpointTestCase(BaseTestCase):
     """Test the breakpoint set method."""
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_bp_on_non_existent_module(self):
         self.expect_set = [
             ('line', 2, 'tfunc_import'), ('break', ('/non/existent/module.py', 1))
@@ -790,7 +806,7 @@ class BreakpointTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             self.assertRaises(BdbError, tracer.runcall, tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_bp_after_last_statement(self):
         code = """
             def main():
@@ -804,7 +820,7 @@ class BreakpointTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 self.assertRaises(BdbError, tracer.runcall, tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_temporary_bp(self):
         code = """
             def func():
@@ -828,7 +844,7 @@ class BreakpointTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_disabled_temporary_bp(self):
         code = """
             def func():
@@ -857,7 +873,7 @@ class BreakpointTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_bp_condition(self):
         code = """
             def func(a):
@@ -878,7 +894,7 @@ class BreakpointTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_bp_exception_on_condition_evaluation(self):
         code = """
             def func(a):
@@ -898,7 +914,7 @@ class BreakpointTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_bp_ignore_count(self):
         code = """
             def func():
@@ -920,7 +936,7 @@ class BreakpointTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_ignore_count_on_disabled_bp(self):
         code = """
             def func():
@@ -948,7 +964,7 @@ class BreakpointTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_clear_two_bp_on_same_line(self):
         code = """
             def func():
@@ -974,7 +990,7 @@ class BreakpointTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_clear_at_no_bp(self):
         self.expect_set = [
             ('line', 2, 'tfunc_import'), ('clear', (__file__, 1))
@@ -1028,7 +1044,7 @@ class BreakpointTestCase(BaseTestCase):
 class RunTestCase(BaseTestCase):
     """Test run, runeval and set_trace."""
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_run_step(self):
         # Check that the bdb 'run' method stops at the first line event.
         code = """
@@ -1041,7 +1057,7 @@ class RunTestCase(BaseTestCase):
         with TracerRun(self) as tracer:
             tracer.run(compile(textwrap.dedent(code), '<string>', 'exec'))
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_runeval_step(self):
         # Test bdb 'runeval'.
         code = """
@@ -1064,7 +1080,7 @@ class RunTestCase(BaseTestCase):
 class IssuesTestCase(BaseTestCase):
     """Test fixed bdb issues."""
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_step_at_return_with_no_trace_in_caller(self):
         # Issue #13183.
         # Check that the tracer does step into the caller frame when the
@@ -1095,7 +1111,7 @@ class IssuesTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_next_until_return_in_generator(self):
         # Issue #16596.
         # Check that set_next(), set_until() and set_return() do not treat the
@@ -1137,7 +1153,7 @@ class IssuesTestCase(BaseTestCase):
                     with TracerRun(self) as tracer:
                         tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_next_command_in_generator_for_loop(self):
         # Issue #16596.
         code = """
@@ -1169,7 +1185,7 @@ class IssuesTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_next_command_in_generator_with_subiterator(self):
         # Issue #16596.
         code = """
@@ -1201,7 +1217,7 @@ class IssuesTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
-    @unittest.skip("TODO: RUSTPYTHON, Error in atexit._run_exitfuncs")
+    @unittest.skip('TODO: RUSTPYTHON; Error in atexit._run_exitfuncs')
     def test_return_command_in_generator_with_subiterator(self):
         # Issue #16596.
         code = """
@@ -1233,6 +1249,21 @@ class IssuesTestCase(BaseTestCase):
             with TracerRun(self) as tracer:
                 tracer.runcall(tfunc_import)
 
+    @unittest.skip('TODO: RUSTPYTHON')
+    # AssertionError: All paired tuples have not been processed, the last one was number 1 [('next',)]
+    def test_next_to_botframe(self):
+        # gh-125422
+        # Check that next command won't go to the bottom frame.
+        code = """
+            lno = 2
+        """
+        self.expect_set = [
+            ('line', 2, '<module>'),   ('step', ),
+            ('return', 2, '<module>'), ('next', ),
+        ]
+        with TracerRun(self) as tracer:
+            tracer.run(compile(textwrap.dedent(code), '<string>', 'exec'))
+
 
 class TestRegressions(unittest.TestCase):
     def test_format_stack_entry_no_lineno(self):