Allow snapshot cursors from other collections for collection group queries. (#8882)

Closes #8633.
Supersedes #8810.

Author: tseaver

firestore/google/cloud/firestore_v1/query.py

@@ -390,6 +390,19 @@ def offset(self, num_to_skip):
             all_descendants=self._all_descendants,
         )
 
+    def _check_snapshot(self, document_fields):
+        """Validate local snapshots for non-collection-group queries.
+
+        Raises:
+            ValueError: for non-collection-group queries, if the snapshot
+                is from a different collection.
+        """
+        if self._all_descendants:
+            return
+
+        if document_fields.reference._path[:-1] != self._parent._path:
+            raise ValueError("Cannot use snapshot from another collection as a cursor.")
+
     def _cursor_helper(self, document_fields, before, start):
         """Set values to be used for a ``start_at`` or ``end_at`` cursor.
 
@@ -419,10 +432,7 @@ def _cursor_helper(self, document_fields, before, start):
         if isinstance(document_fields, tuple):
             document_fields = list(document_fields)
         elif isinstance(document_fields, document.DocumentSnapshot):
-            if document_fields.reference._path[:-1] != self._parent._path:
-                raise ValueError(
-                    "Cannot use snapshot from another collection as a cursor."
-                )
+            self._check_snapshot(document_fields)
         else:
             # NOTE: We copy so that the caller can't modify after calling.
             document_fields = copy.deepcopy(document_fields)

firestore/tests/unit/v1/test_query.py

@@ -459,6 +459,28 @@ def test__cursor_helper_w_snapshot_wrong_collection(self):
         with self.assertRaises(ValueError):
             query._cursor_helper(snapshot, False, False)
 
+    def test__cursor_helper_w_snapshot_other_collection_all_descendants(self):
+        values = {"a": 7, "b": "foo"}
+        docref = self._make_docref("there", "doc_id")
+        snapshot = self._make_snapshot(docref, values)
+        collection = self._make_collection("here")
+        query1 = self._make_one(collection, all_descendants=True)
+
+        query2 = query1._cursor_helper(snapshot, False, False)
+
+        self.assertIs(query2._parent, collection)
+        self.assertIsNone(query2._projection)
+        self.assertEqual(query2._field_filters, ())
+        self.assertEqual(query2._orders, ())
+        self.assertIsNone(query2._limit)
+        self.assertIsNone(query2._offset)
+        self.assertIsNone(query2._start_at)
+
+        cursor, before = query2._end_at
+
+        self.assertIs(cursor, snapshot)
+        self.assertFalse(before)
+
     def test__cursor_helper_w_snapshot(self):
         values = {"a": 7, "b": "foo"}
         docref = self._make_docref("here", "doc_id")