Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -394,16 +394,27 @@ def track(self, thread_id, frames_list, frame_custom_thread_id=None):

frame_ids_from_thread = self._thread_id_to_frame_ids.setdefault(coroutine_or_main_thread_id, [])

self._thread_id_to_frames_list[coroutine_or_main_thread_id] = frames_list
for frame in frames_list:
def _register_frame(frame):
frame_id = id(frame)
self._frame_id_to_frame[frame_id] = frame
_FrameVariable(self.py_db, frame, self._register_variable) # Instancing is enough to register.
self._suspended_frames_manager._variable_reference_to_frames_tracker[frame_id] = self
frame_ids_from_thread.append(frame_id)

self._frame_id_to_main_thread_id[frame_id] = thread_id

self._thread_id_to_frames_list[coroutine_or_main_thread_id] = frames_list
for frame in frames_list:
_register_frame(frame)

# Also track frames from chained exceptions (e.g. __cause__ / __context__)
# so that variable evaluation works for chained exception frames displayed
# in the call stack.
chained = getattr(frames_list, 'chained_frames_list', None)
while chained is not None and len(chained) > 0:
for frame in chained:
_register_frame(frame)
chained = getattr(chained, 'chained_frames_list', None)

frame = None

def untrack_all(self):
Expand Down
74 changes: 74 additions & 0 deletions src/debugpy/_vendored/pydevd/tests_python/test_debugger_json.py
Original file line number Diff line number Diff line change
Expand Up @@ -926,6 +926,80 @@ def additional_output_checks(writer, stdout, stderr):
writer.finished_ok = True


def test_case_chained_exception_variables(case_setup_dap, pyfile):
"""
When stopped on a chained exception, variable evaluation must work for
frames belonging to the chained (cause) exception, not just the primary one.
"""

@pyfile
def target():
def inner():
cause_var = "from_cause" # noqa
raise RuntimeError("the cause")

def outer():
outer_var = "from_outer" # noqa
try:
inner()
except Exception as e:
raise ValueError("the effect") from e # raise line

outer()

def check_test_suceeded_msg(self, stdout, stderr):
return "the cause" in "".join(stderr)

def additional_output_checks(writer, stdout, stderr):
assert 'raise RuntimeError("the cause")' in stderr
assert 'raise ValueError("the effect") from e' in stderr

with case_setup_dap.test_file(
target,
EXPECTED_RETURNCODE=1,
check_test_suceeded_msg=check_test_suceeded_msg,
additional_output_checks=additional_output_checks,
) as writer:
json_facade = JsonFacade(writer)

json_facade.write_launch(justMyCode=False)
json_facade.write_set_exception_breakpoints(["uncaught"])
json_facade.write_make_initial_run()

json_hit = json_facade.wait_for_thread_stopped(
reason="exception", line=writer.get_line_index_with_content("raise line")
)

stack_frames = json_hit.stack_trace_response.body.stackFrames

# Find the chained exception frames.
chained_frames = [f for f in stack_frames if f["name"].startswith("[Chained Exc:")]
assert len(chained_frames) > 0, "Expected chained exception frames in stack trace"

# Verify variables can be retrieved for chained frames (this is the
# operation that previously failed with "Unable to find thread to
# evaluate variable reference.").
for chained_frame in chained_frames:
variables_response = json_facade.get_variables_response(chained_frame["id"])
assert variables_response.success

# Find the inner() chained frame and verify its local variable.
inner_frames = [f for f in chained_frames if "inner" in f["name"]]
assert len(inner_frames) == 1
variables_response = json_facade.get_variables_response(inner_frames[0]["id"])
var_names = [v["name"] for v in variables_response.body.variables]
assert "cause_var" in var_names, "Expected 'cause_var' in chained frame variables, got: %s" % var_names

# Also verify that primary frame variables still work.
primary_frame_id = json_hit.frame_id
variables_response = json_facade.get_variables_response(primary_frame_id)
assert variables_response.success

json_facade.write_continue()

writer.finished_ok = True


def test_case_throw_exc_reason_shown(case_setup_dap):

def check_test_suceeded_msg(self, stdout, stderr):
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -156,3 +156,73 @@ def test_get_child_variables():
raise AssertionError("Expected to find variable named: %s" % (TOO_LARGE_ATTR,))
if not found_len:
raise AssertionError("Expected to find variable named: len()")


def test_chained_exception_frames_tracked():
"""
When an exception has chained causes (__cause__ / __context__), the chained
frames are shown in the call stack. Variable evaluation must also work for
those frames, which requires them to be registered in the
SuspendedFramesManager. Uses a 3-level chain to verify all levels are walked.
"""
from _pydevd_bundle.pydevd_suspended_frames import SuspendedFramesManager
from _pydevd_bundle.pydevd_constants import EXCEPTION_TYPE_USER_UNHANDLED

def level0():
local0 = "from_level_0" # noqa
raise RuntimeError("level_0")

def level1():
local1 = "from_level_1" # noqa
try:
level0()
except Exception as e:
raise TypeError("level_1") from e

def level2():
local2 = "from_level_2" # noqa
try:
level1()
except Exception as e:
raise ValueError("level_2") from e

try:
level2()
except Exception:
exc_type, exc_desc, trace_obj = sys.exc_info()
frame = sys._getframe()
frames_list = pydevd_frame_utils.create_frames_list_from_traceback(
trace_obj, frame, exc_type, exc_desc,
exception_type=EXCEPTION_TYPE_USER_UNHANDLED,
)

# Collect all chained levels.
chained_levels = []
cur = frames_list
while getattr(cur, "chained_frames_list", None) is not None:
chained_levels.append(cur.chained_frames_list)
cur = cur.chained_frames_list
assert len(chained_levels) == 2

suspended_frames_manager = SuspendedFramesManager()
with suspended_frames_manager.track_frames(_DummyPyDB()) as tracker:
thread_id = "thread1"
tracker.track(thread_id, frames_list)

# Primary and all chained frames must be tracked.
for f in frames_list:
assert suspended_frames_manager.get_thread_id_for_variable_reference(id(f)) == thread_id
for level in chained_levels:
for f in level:
assert suspended_frames_manager.get_thread_id_for_variable_reference(id(f)) == thread_id

# Variable retrieval must work for the deepest chained frames.
for f in chained_levels[-1]:
assert suspended_frames_manager.get_variable(id(f)).get_children_variables() is not None

# After untracking, all references must be gone.
for f in frames_list:
assert suspended_frames_manager.get_thread_id_for_variable_reference(id(f)) is None
for level in chained_levels:
for f in level:
assert suspended_frames_manager.get_thread_id_for_variable_reference(id(f)) is None
Loading