add unit tests

Author: daniel-sanche

tests/unit/v1/test_base_query.py

@@ -2116,19 +2116,58 @@ def test__query_pipeline_order_sorts():
     assert sort_stage.orders[1].order_dir == expr.Ordering.Direction.DESCENDING
 
 
-def test__query_pipeline_unsupported():
+def test__query_pipeline_cursors():
     client = make_client()
-    query_start = client.collection("my_col").start_at({"field_a": "value"})
-    with pytest.raises(NotImplementedError, match="cursors"):
-        query_start._build_pipeline(client.pipeline())
+    query_start = (
+        client.collection("my_col").order_by("field_a").start_at({"field_a": "value"})
+    )
+    pipeline = query_start._build_pipeline(client.pipeline())
+    assert len(pipeline.stages) >= 2
+
+    query_end = (
+        client.collection("my_col").order_by("field_a").end_at({"field_a": "value"})
+    )
+    pipeline = query_end._build_pipeline(client.pipeline())
+    assert len(pipeline.stages) >= 2
 
-    query_end = client.collection("my_col").end_at({"field_a": "value"})
-    with pytest.raises(NotImplementedError, match="cursors"):
-        query_end._build_pipeline(client.pipeline())
 
-    query_limit_last = client.collection("my_col").limit_to_last(10)
-    with pytest.raises(NotImplementedError, match="limit_to_last"):
-        query_limit_last._build_pipeline(client.pipeline())
+def test__query_pipeline_limit_to_last():
+    from google.cloud.firestore_v1 import pipeline_expressions as expr
+
+    client = make_client()
+    query_limit_last = (
+        client.collection("my_col").order_by("field_a").limit_to_last(10)
+    )
+    pipeline = query_limit_last._build_pipeline(client.pipeline())
+    # stages: collection, exists, sort(desc), limit, sort(asc)
+
+    assert len(pipeline.stages) == 5
+
+    # 0. Collection
+    assert pipeline.stages[0].path == "/my_col"
+
+    # 1. Exists
+    exists_stage = pipeline.stages[1]
+    assert isinstance(exists_stage, stages.Where)
+
+    # 2. Sort DESCENDING (reversed)
+    sort_desc = pipeline.stages[2]
+    assert isinstance(sort_desc, stages.Sort)
+    assert len(sort_desc.orders) == 1
+    assert sort_desc.orders[0].expr.path == "field_a"
+    assert sort_desc.orders[0].order_dir == expr.Ordering.Direction.DESCENDING
+
+    # 3. Limit
+    limit_stage = pipeline.stages[3]
+    assert isinstance(limit_stage, stages.Limit)
+    assert limit_stage.limit == 10
+
+    # 4. Sort ASCENDING (original)
+    sort_asc = pipeline.stages[4]
+    assert isinstance(sort_asc, stages.Sort)
+    assert len(sort_asc.orders) == 1
+    assert sort_asc.orders[0].expr.path == "field_a"
+    assert sort_asc.orders[0].order_dir == expr.Ordering.Direction.ASCENDING
 
 
 def test__query_pipeline_limit():