Add some test cases

Author: jg-rp

jsonpath/fluent_api.py

@@ -157,10 +157,12 @@ def select(self, *expressions: str) -> Iterable[object]:
                 for match in self._env.finditer(expr, m.obj):  # type: ignore
                     _pointer = match.pointer()
                     _patch_parents(_pointer.parent(), patch, m.obj)  # type: ignore
-                    patch.add(_pointer, match.obj)
+                    patch.addap(_pointer, match.obj)
 
             patch.apply(obj)
-            yield obj
+
+            if obj:
+                yield obj
 
     def first_one(self) -> Optional[JSONPathMatch]:
         """Return the first `JSONPathMatch` or `None` if there were no matches."""

jsonpath/patch.py

@@ -1,4 +1,5 @@
 """JSON Patch, as per RFC 6902."""
+
 from __future__ import annotations
 
 import copy
@@ -85,16 +86,16 @@ def asdict(self) -> Dict[str, object]:
         return {"op": self.name, "path": str(self.path), "value": self.value}
 
 
-class OpAddNe(Op):
-    """A non-standard _add if not exists_ operation."""
+class OpAddNe(OpAdd):
+    """A non-standard _add if not exists_ operation.
 
-    __slots__ = ("path", "value")
+    This is like _OpAdd_, but only adds object/dict keys/values if they key does
+    not already exist.
+    """
 
-    name = "add"
+    __slots__ = ("path", "value")
 
-    def __init__(self, path: JSONPointer, value: object) -> None:
-        self.path = path
-        self.value = value
+    name = "addne"
 
     def apply(
         self, data: Union[MutableSequence[object], MutableMapping[str, object]]
@@ -115,20 +116,45 @@ def apply(
                     raise JSONPatchError("index out of range")
             else:
                 parent.insert(int(target), self.value)
-        elif (
-            isinstance(parent, MutableMapping)
-            and parent.get(target, UNDEFINED) == UNDEFINED
-        ):
+        elif isinstance(parent, MutableMapping) and target not in parent:
             parent[target] = self.value
-        # else:
-        #     raise JSONPatchError(
-        #         f"unexpected operation on {parent.__class__.__name__!r}"
-        #     )
         return data
 
-    def asdict(self) -> Dict[str, object]:
-        """Return a dictionary representation of this operation."""
-        return {"op": self.name, "path": str(self.path), "value": self.value}
+
+class OpAddAp(OpAdd):
+    """A non-standard add operation that appends to arrays/lists .
+
+    This is like _OpAdd_, but assumes an index of "-" if the path can not
+    be resolved.
+    """
+
+    __slots__ = ("path", "value")
+
+    name = "addap"
+
+    def apply(
+        self, data: Union[MutableSequence[object], MutableMapping[str, object]]
+    ) -> Union[MutableSequence[object], MutableMapping[str, object]]:
+        """Apply this patch operation to _data_."""
+        parent, obj = self.path.resolve_parent(data)
+        if parent is None:
+            # Replace the root object.
+            # The following op, if any, will raise a JSONPatchError if needed.
+            return self.value  # type: ignore
+
+        target = self.path.parts[-1]
+        if isinstance(parent, MutableSequence):
+            if obj is UNDEFINED:
+                parent.append(self.value)
+            else:
+                parent.insert(int(target), self.value)
+        elif isinstance(parent, MutableMapping):
+            parent[target] = self.value
+        else:
+            raise JSONPatchError(
+                f"unexpected operation on {parent.__class__.__name__!r}"
+            )
+        return data
 
 
 class OpRemove(Op):
@@ -388,8 +414,13 @@ def _build(self, patch: Iterable[Mapping[str, object]]) -> None:
                 )
             elif op == "addne":
                 self.addne(
-                    path=self._op_pointer(operation, "path", "add", i),
-                    value=self._op_value(operation, "value", "add", i),
+                    path=self._op_pointer(operation, "path", "addne", i),
+                    value=self._op_value(operation, "value", "addne", i),
+                )
+            elif op == "addap":
+                self.addne(
+                    path=self._op_pointer(operation, "path", "addap", i),
+                    value=self._op_value(operation, "value", "addap", i),
                 )
             elif op == "remove":
                 self.remove(path=self._op_pointer(operation, "path", "add", i))
@@ -491,6 +522,22 @@ def addne(self: Self, path: Union[str, JSONPointer], value: object) -> Self:
         self.ops.append(OpAddNe(path=pointer, value=value))
         return self
 
+    def addap(self: Self, path: Union[str, JSONPointer], value: object) -> Self:
+        """Append an _addap_ operation to this patch.
+
+        Arguments:
+            path: A string representation of a JSON Pointer, or one that has
+                already been parsed.
+            value: The object to add.
+
+        Returns:
+            This `JSONPatch` instance, so we can build a JSON Patch by chaining
+                calls to JSON Patch operation methods.
+        """
+        pointer = self._ensure_pointer(path)
+        self.ops.append(OpAddAp(path=pointer, value=value))
+        return self
+
     def remove(self: Self, path: Union[str, JSONPointer]) -> Self:
         """Append a _remove_ operation to this patch.
 

tests/test_query_project.py

@@ -0,0 +1,78 @@
+"""Test cases for the fluent API projection."""
+
+from typing import Any
+from typing import List
+
+import jsonpath
+
+
+def test_top_level_array() -> None:
+    expr = "$.*"
+    data = [{"a": 1, "b": 1}, {"a": 2, "b": 2}, {"b": 3, "a": 3}]
+    projection = ("a",)
+    it = jsonpath.query(expr, data).select(*projection)
+    assert list(it) == [{"a": 1}, {"a": 2}, {"a": 3}]
+
+
+def test_top_level_array_partial_existence() -> None:
+    expr = "$.*"
+    data = [{"a": 1, "b": 1}, {"b": 2}, {"b": 3, "a": 3}]
+    projection = ("a",)
+    it = jsonpath.query(expr, data).select(*projection)
+    assert list(it) == [{"a": 1}, {"a": 3}]
+
+
+def test_top_level_array_projection_does_not_existence() -> None:
+    expr = "$.*"
+    data = [{"a": 1, "b": 1}, {"b": 2}, {"b": 3, "a": 3}]
+    projection = ("x",)
+    it = jsonpath.query(expr, data).select(*projection)
+    assert list(it) == []
+
+
+def test_empty_top_level_array() -> None:
+    expr = "$.*"
+    data: List[Any] = []
+    projection = ("a",)
+    it = jsonpath.query(expr, data).select(*projection)
+    assert list(it) == []
+
+
+def test_top_level_array_select_many() -> None:
+    expr = "$.*"
+    data = [{"a": 1, "b": 1, "c": 1}, {"a": 2, "b": 2, "c": 2}, {"b": 3, "a": 3}]
+    projection = ("a", "c")
+    it = jsonpath.query(expr, data).select(*projection)
+    assert list(it) == [{"a": 1, "c": 1}, {"a": 2, "c": 2}, {"a": 3}]
+
+
+def test_singular_query() -> None:
+    expr = "$.a"
+    data = {"a": {"foo": 42, "bar": 7}, "b": 1}
+    projection = ("foo",)
+    it = jsonpath.query(expr, data).select(*projection)
+    assert list(it) == [{"foo": 42}]
+
+
+def test_select_array_element() -> None:
+    expr = "$.a"
+    data = {"a": {"foo": [42, 7], "bar": 7}, "b": 1}
+    projection = ("foo[0]",)
+    it = jsonpath.query(expr, data).select(*projection)
+    assert list(it) == [{"foo": [42]}]
+
+
+def test_select_array_slice() -> None:
+    expr = "$.a"
+    data = {"a": {"foo": [1, 2, 42, 7, 3], "bar": 7}, "b": 1}
+    projection = ("foo[2:4]",)
+    it = jsonpath.query(expr, data).select(*projection)
+    assert list(it) == [{"foo": [42, 7]}]
+
+
+def test_select_nested_objects() -> None:
+    expr = "$.a"
+    data = {"a": {"foo": {"bar": 42}, "bar": 7}, "b": 1}
+    projection = ("foo.bar",)
+    it = jsonpath.query(expr, data).select(*projection)
+    assert list(it) == [{"foo": {"bar": 42}}]