diff --git a/_pydevd_bundle/pydevd_safe_repr.py b/_pydevd_bundle/pydevd_safe_repr.py index df5e8f4a..df315849 100644 --- a/_pydevd_bundle/pydevd_safe_repr.py +++ b/_pydevd_bundle/pydevd_safe_repr.py @@ -4,6 +4,8 @@ # Gotten from ptvsd for supporting the format expected there. import sys +from collections import defaultdict +from types import FunctionType, BuiltinFunctionType from _pydevd_bundle.pydevd_constants import IS_PY36_OR_GREATER import locale from _pydev_bundle import pydev_log @@ -97,6 +99,18 @@ def has_obj_repr(t): except Exception: return obj_repr is r + # Bound mapping values without invoking a custom factory representation. + # Preserve existing behavior for subclasses and explicit raw/hex formats. + if type(obj) is defaultdict and not self.raw_value and not self.convert_to_hex: + factory = obj.default_factory + if ( + factory is None + or type(factory) is FunctionType + or type(factory) is BuiltinFunctionType + or any(factory is t for t in (list, dict, set, tuple, int, float, str, bool, bytes, object, frozenset)) + ): + return self._repr_defaultdict(obj, level) + for t, prefix, suffix, comma in self.collection_types: if isinstance(obj, t) and has_obj_repr(t): return self._repr_iter(obj, level, prefix, suffix, comma) @@ -114,6 +128,15 @@ def has_obj_repr(t): return self._repr_other(obj, level) + def _repr_defaultdict(self, obj, level): + yield "defaultdict(" + for part in self._repr_other(obj.default_factory, level + 1): + yield part + yield ", " + for part in self._repr_dict(obj, level, "{", "}", "", ": ", ""): + yield part + yield ")" + # Determines whether an iterable exceeds the limits set in # maxlimits, and is therefore unsafe to repr(). def _is_long_iter(self, obj, level=0): diff --git a/tests_python/test_safe_repr_defaultdict.py b/tests_python/test_safe_repr_defaultdict.py new file mode 100644 index 00000000..05f12d0f --- /dev/null +++ b/tests_python/test_safe_repr_defaultdict.py @@ -0,0 +1,228 @@ +"""Regression coverage for bounded defaultdict display and real DAP handling.""" + +import collections +import json +import sys + +import pytest + +from _pydevd_bundle.pydevd_safe_repr import SafeRepr + + +class CountedValue: + calls = 0 + + def __repr__(self): + type(self).calls += 1 + return "leaf" + + +def nested_values(): + value = collections.defaultdict(lambda: collections.defaultdict(list)) + for outer in "abcdef": + for inner in "ghijk": + value[outer][inner] = [CountedValue() for _ in range(100)] + return value + + +@pytest.mark.parametrize("factory", [None, list, dict, set, tuple, int, float, str, bool, bytes, object, frozenset, len, lambda: []]) +def test_defaultdict_small(factory): + value = collections.defaultdict(factory, {"a": [1, 2]}) + assert SafeRepr()(value) == repr(value) + assert value.default_factory is factory + + +def test_defaultdict_does_not_format_hidden_leaves(): + value = nested_values() + CountedValue.calls = 0 + result = SafeRepr()(value) + assert CountedValue.calls == 0 + assert result.startswith("defaultdict(" in result + assert "[...]" in result + assert len(value) == 6 + assert all(len(items) == 100 for inner in value.values() for items in inner.values()) + + +def test_defaultdict_many_keys(): + value = collections.defaultdict(list, {i: CountedValue() for i in range(100)}) + CountedValue.calls = 0 + result = SafeRepr()(value) + assert result.startswith("defaultdict(, {") + assert "..." in result + assert CountedValue.calls < SafeRepr.maxcollection[0] + assert len(value) == 100 + + +@pytest.mark.parametrize("indirect", [False, True]) +def test_defaultdict_recursive(indirect): + value = collections.defaultdict(list) + value["self"] = [value] if indirect else value + result = SafeRepr()(value) + assert result.startswith("defaultdict(, ") + assert "..." in result + assert len(result) < 200 + + +def test_defaultdict_custom_repr_and_factory_are_preserved(): + class Custom(collections.defaultdict): + def __repr__(self): + return "custom-defaultdict" + + class Factory: + def __call__(self): + pytest.fail("Formatting must not invoke the default factory") + + def __repr__(self): + return "custom-factory" + + class Meta(type): + def __repr__(cls): + return "custom-metaclass" + + class CustomType(metaclass=Meta): + pass + + for value in (Custom(list), collections.defaultdict(Factory(), {"a": 1}), collections.defaultdict(CustomType, {"a": 1})): + assert SafeRepr()(value) == repr(value) + + +@pytest.mark.parametrize("attribute", ["raw_value", "convert_to_hex"]) +def test_defaultdict_explicit_formats(attribute): + value = collections.defaultdict(list, {"a": [16]}) + formatter = SafeRepr() + setattr(formatter, attribute, True) + assert formatter(value) == repr(value) + + +def frame_with_value(value): + return sys._getframe() + + +@pytest.fixture +def real_debugger(): + # Instantiate the actual debugger without connecting, tracing, or starting a listener. + from pydevd import PyDB + + debugger = PyDB(set_as_global=False) + commands = [] + + class OutputCollector: + # Only the output transport is replaced. Request dispatch, queue entries, + # resolution, representation, response construction and ID mapping are real. + def add_command(self, command): + commands.append(command) + + debugger.writer = OutputCollector() + yield debugger, commands + debugger.writer = None + + +def request_variables(debugger, commands, reference, seq, fmt=None, dap_id=False): + from _pydevd_bundle._debug_adapter import pydevd_base_schema + from _pydevd_bundle.pydevd_process_net_command_json import PyDevJsonCommandProcessor + + args = {"variablesReference": reference} + if fmt is not None: + args["format"] = fmt + request = pydevd_base_schema.from_json( + json.dumps({"type": "request", "seq": seq, "command": "variables", "arguments": args}), + update_ids_from_dap=dap_id, + ) + processor = PyDevJsonCommandProcessor(pydevd_base_schema.from_json) + assert processor.on_variables_request(debugger, request) is None + queue, _event = debugger.get_internal_queue_and_event("defaultdict-test") + command = queue.get_nowait() + command.do_it(debugger) + assert queue.empty() + response = json.loads(commands.pop()._as_bytes.decode("utf-8")) + assert response["success"] is True + assert response["request_seq"] == seq + assert response["command"] == "variables" + assert not commands + return {entry["name"]: entry for entry in response["body"]["variables"]} + + +def test_defaultdict_variables_request_and_child_expansion(real_debugger): + from _pydevd_bundle.pydevd_frame_utils import create_frames_list_from_frame + + debugger, commands = real_debugger + value = nested_values() + original_keys = tuple(value) + original_factories = (value.default_factory, tuple(v.default_factory for v in value.values())) + original_ids = tuple(id(item) for inner in value.values() for items in inner.values() for item in items) + frame = frame_with_value(value) + manager = debugger.suspended_frames_manager + with manager.track_frames(debugger) as tracker: + tracker.track("defaultdict-test", create_frames_list_from_frame(frame)) + CountedValue.calls = 0 + parent = request_variables(debugger, commands, id(frame), 1)["value"] + assert CountedValue.calls == 0 + assert parent["type"] == "defaultdict" + assert parent["evaluateName"] == "value" + assert parent["value"].startswith("defaultdict( 0 + children = request_variables(debugger, commands, parent["variablesReference"], 2, dap_id=True) + assert CountedValue.calls <= 6 * 5 * (SafeRepr.maxcollection[1] - 1) + child = children["'a'"] + assert child["type"] == "defaultdict" + assert child["evaluateName"] == "value['a']" + assert child["variablesReference"] != parent["variablesReference"] + grandchildren = request_variables(debugger, commands, child["variablesReference"], 3, dap_id=True) + grandchild = grandchildren["'g'"] + assert grandchild["type"] == "list" + assert grandchild["evaluateName"] == "value['a']['g']" + leaves = request_variables(debugger, commands, grandchild["variablesReference"], 4, dap_id=True) + first_leaf = next(entry for entry in leaves.values() if entry.get("evaluateName") == "value['a']['g'][0]") + assert first_leaf["value"] == "leaf" + assert manager.get_thread_id_for_variable_reference(id(frame)) is None + assert tuple(value) == original_keys + assert (value.default_factory, tuple(v.default_factory for v in value.values())) == original_factories + assert tuple(id(item) for inner in value.values() for items in inner.values() for item in items) == original_ids + + +@pytest.mark.parametrize("fmt", [{"hex": True}, {"rawString": True}]) +def test_defaultdict_request_format_propagation(real_debugger, fmt): + from _pydevd_bundle.pydevd_frame_utils import create_frames_list_from_frame + + debugger, commands = real_debugger + value = collections.defaultdict(list, {"a": [16]}) + frame = frame_with_value(value) + with debugger.suspended_frames_manager.track_frames(debugger) as tracker: + tracker.track("defaultdict-test", create_frames_list_from_frame(frame)) + entry = request_variables(debugger, commands, id(frame), 1, fmt=fmt)["value"] + assert entry["value"] == repr(value) + assert entry["type"] == "defaultdict" + assert entry["variablesReference"] > 0 + + +def test_defaultdict_presentation_provider_precedes_safe_repr(real_debugger, monkeypatch): + from _pydevd_bundle import pydevd_xml + from _pydevd_bundle.pydevd_extension_api import StrPresentationProvider + from _pydevd_bundle.pydevd_frame_utils import create_frames_list_from_frame + + class Provider(StrPresentationProvider): + def can_provide(self, type_object, type_name): + return type_object is collections.defaultdict + + def get_str(self, value): + return "provider-defaultdict" + + def get_str_in_context(self, value, context): + return "provider-" + context + + handler = pydevd_xml._TYPE_RESOLVE_HANDLER + if not handler._initialized: + handler._initialize() + monkeypatch.setattr(handler, "_str_providers", [Provider()]) + monkeypatch.setattr(handler, "_type_to_str_provider_cache", {}) + debugger, commands = real_debugger + frame = frame_with_value(nested_values()) + with debugger.suspended_frames_manager.track_frames(debugger) as tracker: + tracker.track("defaultdict-test", create_frames_list_from_frame(frame)) + CountedValue.calls = 0 + entry = request_variables(debugger, commands, id(frame), 1)["value"] + assert entry["value"] == "provider-defaultdict" + variable = debugger.suspended_frames_manager.get_variable(id(frame)).get_child_variable_named("value") + assert variable.get_var_data(context="watch")["value"] == "provider-watch" + assert CountedValue.calls == 0