feat: raise RoutingError for routing failures; simplify merge()

Author: nathanjmcdougall

src/yamltrip/document.py

@@ -26,6 +26,7 @@
     ParseError,
     PatchError,
     QueryError,
+    RoutingError,
 )
 
 if TYPE_CHECKING:
@@ -197,7 +198,13 @@ def add(self, *keys: KeyPart, key: str, value: Any) -> Document:
         route = _make_route(keys)
         op = Op.add(key, value)
         patch = Patch(route=route, operation=op)
-        return self._apply_patches([patch])
+        try:
+            return self._apply_patches([patch])
+        except PatchError as e:
+            if _is_routing_error(_classify_patch_error(e)):
+                msg = f"Route passes through a non-mapping node at {format_path(keys)}"
+                raise RoutingError(msg) from None
+            raise
 
     def upsert(self, *keys: KeyPart, value: Any) -> Document:
         """Replace if exists, create (with intermediate mappings) if not."""
@@ -253,24 +260,30 @@ def _create_at(
         for k in reversed(child_keys[1:]):
             nested_value = {k: nested_value}
         route = _make_route(parent_keys)
-        if isinstance(nested_value, dict) and any(
-            isinstance(v, (dict, list, tuple)) for v in nested_value.values()
-        ):
-            # Op.merge_into is scoped to flat mappings (uniform indent); for
-            # nested values, add a placeholder then replace via complex-replace
-            # which preserves relative indentation.
-            add_op = Op.add(first_key, None)
-            add_patch = Patch(route=route, operation=add_op)
-            replace_route = _make_route((*parent_keys, first_key))
-            replace_op = Op.replace(nested_value)
-            replace_patch = Patch(route=replace_route, operation=replace_op)
-            return self._apply_patches([add_patch, replace_patch])
-        elif isinstance(nested_value, dict):
-            op = Op.merge_into(first_key, nested_value)
-        else:
-            op = Op.add(first_key, nested_value)
-        patch = Patch(route=route, operation=op)
-        return self._apply_patches([patch])
+        try:
+            if isinstance(nested_value, dict) and any(
+                isinstance(v, (dict, list, tuple)) for v in nested_value.values()
+            ):
+                # Op.merge_into is scoped to flat mappings (uniform indent); for
+                # nested values, add a placeholder then replace via complex-replace
+                # which preserves relative indentation.
+                add_op = Op.add(first_key, None)
+                add_patch = Patch(route=route, operation=add_op)
+                replace_route = _make_route((*parent_keys, first_key))
+                replace_op = Op.replace(nested_value)
+                replace_patch = Patch(route=replace_route, operation=replace_op)
+                return self._apply_patches([add_patch, replace_patch])
+            elif isinstance(nested_value, dict):
+                op = Op.merge_into(first_key, nested_value)
+            else:
+                op = Op.add(first_key, nested_value)
+            patch = Patch(route=route, operation=op)
+            return self._apply_patches([patch])
+        except PatchError as e:
+            if _is_routing_error(_classify_patch_error(e)):
+                msg = f"Route passes through a non-mapping node at {format_path(parent_keys)}"
+                raise RoutingError(msg) from None
+            raise
 
     def _is_empty_document(self) -> bool:
         """True if the document has no root data node."""
@@ -500,20 +513,7 @@ def merge(self, *keys: KeyPart, value: Any) -> Document:
         if normalized:
             route = _make_route(normalized)
             if not self._core_doc.query_exists(route):
-                try:
-                    return self.upsert(*normalized, value=value)
-                except PatchError as e:
-                    if _classify_patch_error(e) == _PatchErrorKind.UNEXPECTED_NODE:
-                        # Find deepest existing ancestor to report
-                        failing = normalized
-                        for i in range(len(normalized), 0, -1):
-                            sub = normalized[:i]
-                            if self._core_doc.query_exists(_make_route(sub)):
-                                failing = sub
-                                break
-                        msg = f"Value at {format_path(failing)} is not a mapping"
-                        raise NodeTypeError(msg) from None
-                    raise
+                return self.upsert(*normalized, value=value)
 
         # Get current value and diff
         try:
@@ -640,6 +640,8 @@ class _PatchErrorKind(enum.Enum):
     FLOW_SEQUENCE = "flow sequence"
     NOT_A_SEQUENCE = "only permitted against sequence"
     BLOCK_SEQUENCE_EXPECTED = "expected BlockSequence"
+    NON_MAPPING_ROUTE = "non-mapping route"
+    EXPECTED_MAPPING = "expected mapping containing key"
     UNEXPECTED_NODE = "unexpected node"
     UNKNOWN = ""
 
@@ -651,12 +653,16 @@ def _classify_patch_error(err: PatchError) -> _PatchErrorKind:
     callers can branch on the enum rather than matching raw strings.
     """
     msg = str(err)
-    if _PatchErrorKind.FLOW_SEQUENCE.value in msg:
-        return _PatchErrorKind.FLOW_SEQUENCE
-    if _PatchErrorKind.NOT_A_SEQUENCE.value in msg:
-        return _PatchErrorKind.NOT_A_SEQUENCE
-    if _PatchErrorKind.BLOCK_SEQUENCE_EXPECTED.value in msg:
-        return _PatchErrorKind.BLOCK_SEQUENCE_EXPECTED
-    if _PatchErrorKind.UNEXPECTED_NODE.value in msg:
-        return _PatchErrorKind.UNEXPECTED_NODE
+    for kind in _PatchErrorKind:
+        if kind.value and kind.value in msg:
+            return kind
     return _PatchErrorKind.UNKNOWN
+
+
+def _is_routing_error(kind: _PatchErrorKind) -> bool:
+    """Return True if kind represents a routing failure through a non-mapping node."""
+    return kind in (
+        _PatchErrorKind.NON_MAPPING_ROUTE,
+        _PatchErrorKind.EXPECTED_MAPPING,
+        _PatchErrorKind.UNEXPECTED_NODE,
+    )

tests/test_edge_cases.py

@@ -178,6 +178,32 @@ def test_classify_unexpected_node(self):
         msg = self._raw_error(doc, "name", op=Op.merge_into("nested", {"foo": "bar"}))
         assert _classify_patch_error(PatchError(msg)) == _PatchErrorKind.UNEXPECTED_NODE
 
+    def test_non_mapping_route_substring(self):
+        """Op.add on a scalar node raises an error containing 'non-mapping route'."""
+        doc = Document("name: scalar\n")
+        msg = self._raw_error(doc, "name", op=Op.add("child", "val"))
+        assert _PatchErrorKind.NON_MAPPING_ROUTE.value in msg
+
+    def test_classify_non_mapping_route(self):
+        doc = Document("name: scalar\n")
+        msg = self._raw_error(doc, "name", op=Op.add("child", "val"))
+        assert (
+            _classify_patch_error(PatchError(msg)) == _PatchErrorKind.NON_MAPPING_ROUTE
+        )
+
+    def test_expected_mapping_substring(self):
+        """Op.merge_into on a list node raises an error containing 'expected mapping containing key'."""
+        doc = Document("items:\n  - a\n")
+        msg = self._raw_error(doc, "items", op=Op.merge_into("k", {"a": 1}))
+        assert _PatchErrorKind.EXPECTED_MAPPING.value in msg
+
+    def test_classify_expected_mapping(self):
+        doc = Document("items:\n  - a\n")
+        msg = self._raw_error(doc, "items", op=Op.merge_into("k", {"a": 1}))
+        assert (
+            _classify_patch_error(PatchError(msg)) == _PatchErrorKind.EXPECTED_MAPPING
+        )
+
     def test_classify_unknown(self):
         assert (
             _classify_patch_error(PatchError("some unrelated error"))

tests/test_errors.py

@@ -1,5 +1,6 @@
 import pytest
 
+from yamltrip import Document
 from yamltrip.errors import (
     KeyExistsError,
     KeyMissingError,
@@ -59,3 +60,25 @@ def test_raise_and_catch_as_patch_error(self):
     def test_message(self):
         err = RoutingError("Route passes through a non-mapping node at a")
         assert "a" in str(err)
+
+
+class TestRoutingErrorBehaviour:
+    def test_upsert_through_scalar_raises_routing_error(self):
+        doc = Document("a: scalar\n")
+        with pytest.raises(RoutingError, match="non-mapping node at a"):
+            doc.upsert("a", "b", value="foo")
+
+    def test_upsert_through_list_raises_routing_error(self):
+        doc = Document("items:\n  - x\n")
+        with pytest.raises(RoutingError, match="non-mapping node at items"):
+            doc.upsert("items", "b", value="foo")
+
+    def test_add_through_scalar_raises_routing_error(self):
+        doc = Document("name: scalar\n")
+        with pytest.raises(RoutingError, match="non-mapping node at name"):
+            doc.add("name", key="child", value="foo")
+
+    def test_routing_error_is_catchable_as_patch_error(self):
+        doc = Document("a: scalar\n")
+        with pytest.raises(PatchError):
+            doc.upsert("a", "b", value="foo")

tests/test_merge.py

@@ -2,7 +2,7 @@
 
 from yamltrip import edit
 from yamltrip.document import Document
-from yamltrip.errors import NodeTypeError
+from yamltrip.errors import RoutingError
 
 
 class TestMergeMapping:
@@ -108,7 +108,7 @@ def test_editor_merge(self, tmp_path):
 
 class TestMergeErrors:
     def test_merge_through_scalar_raises(self):
-        """Merging through a scalar path raises NodeTypeError."""
+        """Merging through a scalar path raises RoutingError."""
         doc = Document("a:\n  b: 1\n")
-        with pytest.raises(NodeTypeError, match="Value at a > b"):
+        with pytest.raises(RoutingError, match="non-mapping node at a > b"):
             doc.merge("a", "b", "c", value={"x": 1})