test: add deep regressions and internal compatibility coverage

Adds deep stack-safety regressions (depth 12_000), direct merge/decoder deep-path checks, and internal cycle-state compatibility tests for _encode/Utils.merge behavior.

Author: techouse

tests/unit/decode_test.py

@@ -929,6 +929,30 @@ def test_does_not_crash_when_parsing_deep_dicts(self) -> None:
 
         assert actual_depth == depth
 
+    def test_does_not_crash_when_merging_very_deep_keys(self) -> None:
+        # Single high-depth canary for the deep conflicting merge path.
+        depth = 12_000
+
+        path = "a" + ("[p]" * depth)
+        query = f"{path}[left]=1&{path}[right]=2"
+
+        parsed: t.Optional[t.Mapping[str, t.Any]]
+
+        with does_not_raise():
+            parsed = decode(query, DecodeOptions(depth=depth + 2, parameter_limit=10_000_000))
+
+        assert parsed is not None
+        assert "a" in parsed
+
+        current = parsed["a"]
+        for _ in range(depth):
+            assert isinstance(current, dict)
+            current = current["p"]
+
+        assert isinstance(current, dict)
+        assert current["left"] == "1"
+        assert current["right"] == "2"
+
     def test_parses_null_dicts_correctly(self) -> None:
         a: t.Dict[str, t.Any] = {"b": "c"}
         assert decode(a) == {"b": "c"}

tests/unit/encode_test.py

@@ -11,7 +11,7 @@
 import pytest
 
 from qs_codec import Charset, EncodeOptions, Format, ListFormat, dumps, encode
-from qs_codec.encode import _encode, _sentinel
+from qs_codec.encode import _CycleState, _encode, _pop_current_node, _sentinel
 from qs_codec.models.undefined import Undefined
 from qs_codec.models.weak_wrapper import WeakWrapper
 from qs_codec.utils.encode_utils import EncodeUtils
@@ -867,23 +867,33 @@ def test_encode_depth_guard_prevents_recursion_errors(self) -> None:
         with pytest.raises(ValueError, match="Maximum encoding depth exceeded"):
             encode(data, options=EncodeOptions(max_depth=3))
 
-    def test_encode_depth_guard_caps_to_recursion_limit(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        import importlib
-
-        encode_module = importlib.import_module("qs_codec.encode")
-
-        limit = encode_module._DEPTH_MARGIN + 3
-        monkeypatch.setattr(encode_module.sys, "getrecursionlimit", lambda: limit)
-
+    def test_encode_depth_guard_does_not_cap_to_recursion_limit(self) -> None:
+        # `_get_max_encode_depth` now uses `sys.maxsize` for None and explicit values directly,
+        # so monkeypatching `sys.getrecursionlimit` is intentionally unnecessary here.
         data: t.Dict[str, t.Any] = {}
         current = data
         for _ in range(5):
             nxt: t.Dict[str, t.Any] = {}
             current["a"] = nxt
             current = nxt
+        current["leaf"] = "x"
 
-        with pytest.raises(ValueError, match="Maximum encoding depth exceeded"):
-            encode(data, options=EncodeOptions(max_depth=10_000))
+        with does_not_raise():
+            result = encode(data, options=EncodeOptions(max_depth=10_000, encode=False))
+
+        assert result.endswith("=x")
+
+    def test_encode_deep_nesting_iterative_stack_safety(self) -> None:
+        # Keep this above common recursion limits so recursion regressions still fail quickly.
+        depth = 12_000
+        data: t.Dict[str, t.Any] = {"leaf": "x"}
+        for _ in range(depth):
+            data = {"a": data}
+
+        with does_not_raise():
+            result = encode(data, options=EncodeOptions(encode=False))
+
+        assert result.endswith("=x")
 
     @pytest.mark.parametrize(
         "data, options, expected",
@@ -1853,6 +1863,93 @@ def test_encode_cycle_detection_marks_prior_visit_without_raising(self) -> None:
 
         assert tokens == ["root%5Bchild%5D=value"]
 
+    def test_encode_cycle_state_prefers_nearest_ancestor_mapping(self) -> None:
+        value: t.Dict[str, t.Any] = {"child": "value"}
+        wrapper = WeakWrapper(value)
+
+        top_channel: WeakKeyDictionary = WeakKeyDictionary()
+        top_channel[wrapper] = 2
+
+        mid_channel: WeakKeyDictionary = WeakKeyDictionary()
+        mid_channel[_sentinel] = top_channel
+        mid_channel[wrapper] = 99
+
+        side_channel: WeakKeyDictionary = WeakKeyDictionary()
+        side_channel[_sentinel] = mid_channel
+
+        tokens = _encode(
+            value=value,
+            is_undefined=False,
+            side_channel=side_channel,
+            prefix="root",
+            comma_round_trip=False,
+            comma_compact_nulls=False,
+            encoder=EncodeUtils.encode,
+            serialize_date=EncodeUtils.serialize_date,
+            sort=None,
+            filter_=None,
+            formatter=Format.RFC3986.formatter,
+            format=Format.RFC3986,
+            generate_array_prefix=ListFormat.INDICES.generator,
+            allow_empty_lists=False,
+            strict_null_handling=False,
+            skip_nulls=False,
+            encode_dot_in_keys=False,
+            allow_dots=False,
+            encode_values_only=False,
+            charset=Charset.UTF8,
+        )
+
+        assert tokens == ["root%5Bchild%5D=value"]
+
+    def test_encode_cycle_state_bootstrap_matches_legacy_side_channel_behavior(self) -> None:
+        value: t.Dict[str, t.Any] = {"child": "value"}
+        wrapper = WeakWrapper(value)
+
+        top_channel: WeakKeyDictionary = WeakKeyDictionary()
+        top_channel[wrapper] = 99
+
+        parent_channel: WeakKeyDictionary = WeakKeyDictionary()
+        parent_channel[_sentinel] = top_channel
+
+        side_channel: WeakKeyDictionary = WeakKeyDictionary()
+        side_channel[_sentinel] = parent_channel
+
+        tokens = _encode(
+            value=value,
+            is_undefined=False,
+            side_channel=side_channel,
+            prefix="root",
+            comma_round_trip=False,
+            comma_compact_nulls=False,
+            encoder=EncodeUtils.encode,
+            serialize_date=EncodeUtils.serialize_date,
+            sort=None,
+            filter_=None,
+            formatter=Format.RFC3986.formatter,
+            format=Format.RFC3986,
+            generate_array_prefix=ListFormat.INDICES.generator,
+            allow_empty_lists=False,
+            strict_null_handling=False,
+            skip_nulls=False,
+            encode_dot_in_keys=False,
+            allow_dots=False,
+            encode_values_only=False,
+            charset=Charset.UTF8,
+        )
+
+        assert tokens == ["root%5Bchild%5D=value"]
+
+    def test_pop_current_node_noop_when_wrapper_not_present(self) -> None:
+        value: t.Dict[str, t.Any] = {"child": "value"}
+        wrapper = WeakWrapper(value)
+        state = _CycleState()
+
+        with does_not_raise():
+            _pop_current_node(state, wrapper)
+
+        assert state.entries == {}
+
     def test_encode_handles_iterable_filter_for_indexable_object(self, monkeypatch: pytest.MonkeyPatch) -> None:
         class Indexable:
             def __getitem__(self, key: str) -> str:

tests/unit/utils_test.py

@@ -594,6 +594,52 @@ def test_merges_array_into_object(self) -> None:
             {"foo": ["xyzzy"]},
         ) == {"foo": {"bar": "baz", "0": "xyzzy"}}
 
+    def test_merge_mapping_target_with_scalar_source_returns_target_unchanged(self) -> None:
+        target = {"a": "b"}
+        source = "scalar"
+
+        result = Utils.merge(target, source)  # type: ignore[arg-type]
+
+        assert result == {"a": "b"}
+        assert result is target
+
+    def test_merge_structured_lists_prefers_source_when_target_slot_is_undefined(self) -> None:
+        options = DecodeOptions()
+        target = [Undefined()]
+        source = [{"from_source": 1}]
+
+        result = Utils.merge(target, source, options)
+
+        assert result == [{"from_source": 1}]
+
+    def test_merge_deep_maps_without_stack_overflow(self) -> None:
+        # Keep this above common recursion limits so recursion regressions still fail quickly.
+        depth = 12_000
+
+        left: t.Dict[str, t.Any] = {}
+        cursor = left
+        for _ in range(depth):
+            nxt: t.Dict[str, t.Any] = {}
+            cursor["a"] = nxt
+            cursor = nxt
+
+        right: t.Dict[str, t.Any] = {}
+        right_cursor = right
+        for _ in range(depth):
+            nxt = {}
+            right_cursor["a"] = nxt
+            right_cursor = nxt
+        right_cursor["leaf"] = "x"
+
+        merged = Utils.merge(left, right)
+        walk = merged
+        for _ in range(depth):
+            assert isinstance(walk, dict)
+            walk = walk["a"]
+
+        assert isinstance(walk, dict)
+        assert walk["leaf"] == "x"
+
     def test_combine_both_arrays(self) -> None:
         a: t.List[int] = [1]
         b: t.List[int] = [2]
@@ -950,6 +996,24 @@ def test_merge_prefers_exact_key_match_before_string_normalization(self) -> None
         assert result == {"1": {"a": "x", "b": "y"}}
         assert 1 not in result
 
+    def test_merge_does_not_match_source_keys_inserted_earlier_in_same_map_pass(self) -> None:
+        target = {"3": ""}
+        source = {"1": {"a": 1}, 1: ("x",)}
+
+        result = Utils.merge(target, source)  # type: ignore[arg-type]
+
+        assert result == {"3": "", "1": {"a": 1}, 1: ("x",)}
+        assert "1" in result
+        assert 1 in result
+
+    def test_merge_normalized_collision_with_existing_key_uses_existing_target_slot(self) -> None:
+        target = {"1": {"z": 0}}
+        source = {"1": {"a": 1}, 1: ("x",)}
+
+        result = Utils.merge(target, source)  # type: ignore[arg-type]
+
+        assert result == {"1": {"z": 0, "a": 1, "0": "x"}}
+
     def test_overflow_dict_copy_preserves_type(self) -> None:
         target = OverflowDict({"0": "a"})
         result = target.copy()