Updated bdb.py + test_bdb.py

RustPython · terryluan12 · Dec 30, 2025 · Dec 30, 2025 · Dec 30, 2025 · Dec 30, 2025
commit b0ec8ee0dc77ba2834124b92ab053f0d6afbf179
diff --git a/Lib/bdb.py b/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 @@ def __init__(self, skip=None):
         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 @@ def reset(self):
         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 @@ def trace_dispatch(self, frame, event, arg):
 
         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':
-            return self.trace_dispatch
-        if event == 'c_exception':
-            return self.trace_dispatch
-        if event == 'c_return':
+
+        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
-        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 @@ def dispatch_line(self, frame):
         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 @@ def dispatch_return(self, frame, arg):
             # 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 @@ def dispatch_exception(self, frame, arg):
 
         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 @@ def user_exception(self, frame, exc_info):
         """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 @@ def _set_stopinfo(self, stopframe, returnframe, stoplineno=0):
         # 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 @@ def set_until(self, frame, lineno=None):
 
     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 @@ def set_trace(self, frame=None):
         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 @@ def set_continue(self):
             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 @@ def set_break(self, filename, lineno, temporary=False, cond=None,
             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):

diff --git a/Lib/test/test_bdb.py b/Lib/test/test_bdb.py
@@ -228,6 +228,10 @@ def user_exception(self, frame, exc_info):
         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 @@ def next_set_method(self):
         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'):
@@ -612,6 +616,16 @@ def test_step_next_on_last_statement(self):
                 with TracerRun(self) as tracer:
                     tracer.runcall(tfunc_main)
 
+    @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 = [
@@ -727,6 +741,7 @@ def test_until_in_caller_frame(self):
             tracer.runcall(tfunc_main)
 
     @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()'.
@@ -1233,6 +1248,21 @@ def main():
             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):