added tests for expressions

Author: daniel-sanche

google/cloud/firestore_v1/pipeline_expressions.py

@@ -340,9 +340,23 @@ class Selectable(Expr):
     """Base class for expressions that can be selected or aliased in projection stages."""
 
     @abstractmethod
-    def _to_map(self):
+    def _to_map(self) -> tuple[str, Value]:
+        """
+        Returns a str: Value representation of the Selectable
+        """
         raise NotImplementedError
 
+    @classmethod
+    def _value_from_selectables(cls, *selectables: Selectable) -> Value:
+        """
+        Returns a Value representing a map of Selectables
+        """
+        return Value(
+            map_value={
+                "fields": {m[0]: m[1] for m in [s._to_map() for s in selectables]}
+            }
+        )
+
 
 class Field(Selectable):
     """Represents a reference to a field within a document."""
@@ -384,8 +398,8 @@ class FilterCondition(Function):
     def __init__(
         self,
         *args,
-        use_infix_repr:bool = True,
-        infix_name_override:str | None= None,
+        use_infix_repr: bool = True,
+        infix_name_override: str | None = None,
         **kwargs,
     ):
         self._use_infix_repr = use_infix_repr
@@ -521,7 +535,9 @@ class In(FilterCondition):
     """Represents checking if an expression's value is within a list of values."""
 
     def __init__(self, left: Expr, others: List[Expr]):
-        super().__init__("in", [left, ListOfExprs(others)], infix_name_override="in_any")
+        super().__init__(
+            "in", [left, ListOfExprs(others)], infix_name_override="in_any"
+        )
 
 
 class IsNaN(FilterCondition):

google/cloud/firestore_v1/pipeline_stages.py

@@ -128,15 +128,7 @@ def __init__(self, *selections: str | Selectable):
         self.projections = [Field(s) if isinstance(s, str) else s for s in selections]
 
     def _pb_args(self) -> list[Value]:
-        return [
-            Value(
-                map_value={
-                    "fields": {
-                        m[0]: m[1] for m in [f._to_map() for f in self.projections]
-                    }
-                }
-            )
-        ]
+        return [Selectable._value_from_selectables(*self.projections)]
 
 
 class Sort(Stage):

tests/unit/v1/test_async_pipeline.py

@@ -309,19 +309,22 @@ async def test_async_pipeline_execute_with_transaction():
     assert request.database == "projects/A/databases/B"
     assert request.transaction == b"123"
 
-@pytest.mark.parametrize("method,args,result_cls", [
-    ("select", (), stages.Select),
-    ("where", (mock.Mock(),), stages.Where),
-    ("sort", (), stages.Sort),
-    ("offset", (1,), stages.Offset),
-    ("limit", (1,), stages.Limit),
-
-])
+
+@pytest.mark.parametrize(
+    "method,args,result_cls",
+    [
+        ("select", (), stages.Select),
+        ("where", (mock.Mock(),), stages.Where),
+        ("sort", (), stages.Sort),
+        ("offset", (1,), stages.Offset),
+        ("limit", (1,), stages.Limit),
+    ],
+)
 def test_async_pipeline_methods(method, args, result_cls):
     start_ppl = _make_async_pipeline()
     method_ptr = getattr(start_ppl, method)
     result_ppl = method_ptr(*args)
     assert result_ppl != start_ppl
     assert len(start_ppl.stages) == 0
     assert len(result_ppl.stages) == 1
-    assert isinstance(result_ppl.stages[0], result_cls)
+    assert isinstance(result_ppl.stages[0], result_cls)

tests/unit/v1/test_pipeline.py

@@ -290,19 +290,22 @@ def test_pipeline_execute_with_transaction():
     assert request.database == "projects/A/databases/B"
     assert request.transaction == b"123"
 
-@pytest.mark.parametrize("method,args,result_cls", [
-    ("select", (), stages.Select),
-    ("where", (mock.Mock(),), stages.Where),
-    ("sort", (), stages.Sort),
-    ("offset", (1,), stages.Offset),
-    ("limit", (1,), stages.Limit),
-
-])
+
+@pytest.mark.parametrize(
+    "method,args,result_cls",
+    [
+        ("select", (), stages.Select),
+        ("where", (mock.Mock(),), stages.Where),
+        ("sort", (), stages.Sort),
+        ("offset", (1,), stages.Offset),
+        ("limit", (1,), stages.Limit),
+    ],
+)
 def test_pipeline_methods(method, args, result_cls):
     start_ppl = _make_pipeline()
     method_ptr = getattr(start_ppl, method)
     result_ppl = method_ptr(*args)
     assert result_ppl != start_ppl
     assert len(start_ppl.stages) == 0
     assert len(result_ppl.stages) == 1
-    assert isinstance(result_ppl.stages[0], result_cls)
+    assert isinstance(result_ppl.stages[0], result_cls)

tests/unit/v1/test_pipeline_expressions.py

@@ -33,6 +33,47 @@ def mock_client():
     return client
 
 
+class TestOrdering:
+    @pytest.mark.parametrize(
+        "direction_arg,expected_direction",
+        [
+            ("ASCENDING", expr.Ordering.Direction.ASCENDING),
+            ("DESCENDING", expr.Ordering.Direction.DESCENDING),
+            ("ascending", expr.Ordering.Direction.ASCENDING),
+            ("descending", expr.Ordering.Direction.DESCENDING),
+            (expr.Ordering.Direction.ASCENDING, expr.Ordering.Direction.ASCENDING),
+            (expr.Ordering.Direction.DESCENDING, expr.Ordering.Direction.DESCENDING),
+        ],
+    )
+    def test_ctor(self, direction_arg, expected_direction):
+        instance = expr.Ordering("field1", direction_arg)
+        assert isinstance(instance.expr, expr.Field)
+        assert instance.expr.path == "field1"
+        assert instance.order_dir == expected_direction
+
+    def test_repr(self):
+        field_expr = expr.Field.of("field1")
+        instance = expr.Ordering(field_expr, "ASCENDING")
+        repr_str = repr(instance)
+        assert repr_str == "Field.of('field1').ascending()"
+
+        instance = expr.Ordering(field_expr, "DESCENDING")
+        repr_str = repr(instance)
+        assert repr_str == "Field.of('field1').descending()"
+
+    def test_to_pb(self):
+        field_expr = expr.Field.of("field1")
+        instance = expr.Ordering(field_expr, "ASCENDING")
+        result = instance._to_pb()
+        assert result.map_value.fields["expression"].field_reference_value == "field1"
+        assert result.map_value.fields["direction"].string_value == "ascending"
+
+        instance = expr.Ordering(field_expr, "DESCENDING")
+        result = instance._to_pb()
+        assert result.map_value.fields["expression"].field_reference_value == "field1"
+        assert result.map_value.fields["direction"].string_value == "descending"
+
+
 class TestExpr:
     def test_ctor(self):
         """
@@ -116,6 +157,65 @@ def test_repr(self, input_val, expected):
         assert repr_string == expected
 
 
+class TestListOfExprs:
+    def test_to_pb(self):
+        instance = expr.ListOfExprs([expr.Constant(1), expr.Constant(2)])
+        result = instance._to_pb()
+        assert len(result.array_value.values) == 2
+        assert result.array_value.values[0].integer_value == 1
+        assert result.array_value.values[1].integer_value == 2
+
+    def test_empty_to_pb(self):
+        instance = expr.ListOfExprs([])
+        result = instance._to_pb()
+        assert len(result.array_value.values) == 0
+
+    def test_repr(self):
+        instance = expr.ListOfExprs([expr.Constant(1), expr.Constant(2)])
+        repr_string = repr(instance)
+        assert repr_string == "ListOfExprs([Constant.of(1), Constant.of(2)])"
+        empty_instance = expr.ListOfExprs([])
+        empty_repr_string = repr(empty_instance)
+        assert empty_repr_string == "ListOfExprs([])"
+
+
+class TestSelectable:
+    def test_ctor(self):
+        """
+        Base class should be abstract
+        """
+        with pytest.raises(TypeError):
+            expr.Selectable()
+
+    def test_value_from_selectables(self):
+        selectable_list = [expr.Field.of("field1"), expr.Field.of("field2")]
+        result = expr.Selectable._value_from_selectables(*selectable_list)
+        assert len(result.map_value.fields) == 2
+        assert result.map_value.fields["field1"].field_reference_value == "field1"
+        assert result.map_value.fields["field2"].field_reference_value == "field2"
+
+    class TestField:
+        def test_repr(self):
+            instance = expr.Field.of("field1")
+            repr_string = repr(instance)
+            assert repr_string == "Field.of('field1')"
+
+        def test_of(self):
+            instance = expr.Field.of("field1")
+            assert instance.path == "field1"
+
+        def test_to_pb(self):
+            instance = expr.Field.of("field1")
+            result = instance._to_pb()
+            assert result.field_reference_value == "field1"
+
+        def test_to_map(self):
+            instance = expr.Field.of("field1")
+            result = instance._to_map()
+            assert result[0] == "field1"
+            assert result[1] == Value(field_reference_value="field1")
+
+
 class TestFilterCondition:
     def test__from_query_filter_pb_composite_filter_or(self, mock_client):
         """
@@ -420,21 +520,23 @@ def test__from_query_filter_pb_unknown_filter_type(self, mock_client):
         with pytest.raises(TypeError, match="Unexpected filter type"):
             FilterCondition._from_query_filter_pb(document_pb.Value(), mock_client)
 
-
-    @pytest.mark.parametrize("method,args,result_cls", [
-        ("eq", (2,), expr.Eq),
-        ("neq", (2,), expr.Neq),
-        ("lt", (2,), expr.Lt),
-        ("lte", (2,), expr.Lte),
-        ("gt", (2,), expr.Gt),
-        ("gte", (2,), expr.Gte),
-        ("in_any", ([None],), expr.In),
-        ("not_in_any", ([None],), expr.Not),
-        ("array_contains", (None,), expr.ArrayContains),
-        ("array_contains_any", ([None],), expr.ArrayContainsAny),
-        ("is_nan", (), expr.IsNaN),
-        ("exists", (), expr.Exists),
-    ])
+    @pytest.mark.parametrize(
+        "method,args,result_cls",
+        [
+            ("eq", (2,), expr.Eq),
+            ("neq", (2,), expr.Neq),
+            ("lt", (2,), expr.Lt),
+            ("lte", (2,), expr.Lte),
+            ("gt", (2,), expr.Gt),
+            ("gte", (2,), expr.Gte),
+            ("in_any", ([None],), expr.In),
+            ("not_in_any", ([None],), expr.Not),
+            ("array_contains", (None,), expr.ArrayContains),
+            ("array_contains_any", ([None],), expr.ArrayContainsAny),
+            ("is_nan", (), expr.IsNaN),
+            ("exists", (), expr.Exists),
+        ],
+    )
     def test_infix_call(self, method, args, result_cls):
         """
         most FilterExpressions should support infix execution
@@ -483,7 +585,10 @@ def test_array_contains_any(self):
         assert isinstance(instance.params[1], ListOfExprs)
         assert instance.params[0] == arg1
         assert instance.params[1].exprs == [arg2, arg3]
-        assert repr(instance) == "ArrayField.array_contains_any(ListOfExprs([Element1, Element2]))"
+        assert (
+            repr(instance)
+            == "ArrayField.array_contains_any(ListOfExprs([Element1, Element2]))"
+        )
 
     def test_exists(self):
         arg1 = self._make_arg("Field")

tests/unit/v1/test_pipeline_source.py

@@ -53,6 +53,7 @@ def test_collection_group(self):
         assert isinstance(first_stage, stages.CollectionGroup)
         assert first_stage.collection_id == "id"
 
+
 class TestPipelineSourceWithAsyncClient(TestPipelineSource):
     """
     When an async client is used, it should produce async pipelines

tests/unit/v1/test_pipeline_stages.py

@@ -168,7 +168,9 @@ def _make_one(self, *args, **kwargs):
     def test_repr(self):
         instance = self._make_one("field1", Field.of("field2"))
         repr_str = repr(instance)
-        assert repr_str == "Select(projections=[Field.of('field1'), Field.of('field2')])"
+        assert (
+            repr_str == "Select(projections=[Field.of('field1'), Field.of('field2')])"
+        )
 
     def test_to_pb(self):
         instance = self._make_one("field1", "field2.subfield", Field.of("field3"))