PYCBC-1778: Fix KV range scan consistent_with and ScanResult.expiry_time

Changes
-------
* Fix ScanResult.expiry_time reading the 'cas' field instead of 'expiry',
* Fix consistent_with serialization: wrap tokens as {'tokens': [...]} so the
  C++ binding's mutation_state extraction receives them; a bare list was
  silently dropped, leaving scans with no consistency
* Guard expiry_time access on content results in range scan tests (sync + async)
* Add cluster-free RangeScanRequestBuilderUnitTests asserting the
  consistent_with serialization shape

Change-Id: Ibe6a165ebbc4466bc1ebbf499e3d2e66837f76f7
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/246632
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Dimitris Christodoulou <dimitris.christodoulou@couchbase.com>

Author: thejcfactor

acouchbase/tests/kv_range_scan_t.py

@@ -13,7 +13,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from datetime import timedelta
+from datetime import datetime, timedelta
 from uuid import uuid4
 
 import pytest
@@ -120,6 +120,11 @@ async def _validate_result(self, result, expected_count=0, ids_only=False, retur
                 content = r.content_as[dict]
                 assert content is not None
                 assert content == {'id': r.id}
+                # Regression guard (PYCBC-1778): ScanResult.expiry_time previously read the 'cas' field instead
+                # of 'expiry', passing a huge HLC value to datetime.fromtimestamp() raising an exception.
+                # Accessing it on a content result must not raise. The value is either None (no expiry) or a datetime.
+                expiry_time = r.expiry_time
+                assert expiry_time is None or isinstance(expiry_time, datetime)
             rows.append(r)
 
         if from_sample is True:

couchbase/logic/collection_req_builder.py

@@ -621,7 +621,10 @@ def _process_scan_orchestrator_ops(self, orchestrator_opts: Dict[str, Any]) -> N
             if not (isinstance(consistent_with, MutationState) and len(consistent_with._sv) > 0):
                 raise InvalidArgumentException('Passed empty or invalid mutation state')
             else:
-                orchestrator_opts['consistent_with'] = list(token.as_dict() for token in consistent_with._sv)
+                # The C++ binding expects a mutation_state dict ({'tokens': [...]}), not a bare list.
+                orchestrator_opts['consistent_with'] = {
+                    'tokens': list(token.as_dict() for token in consistent_with._sv)
+                }
 
     def _get_scan_config(self, scan_type: ScanType) -> Dict[str, Any]:  # noqa: C901
         scan_config = {}

couchbase/result.py

@@ -1086,7 +1086,7 @@ def expiry_time(self) -> Optional[datetime]:
         if self.ids_only:
             raise InvalidArgumentException(("No expiry_time available when scan is requested with "
                                             "`ScanOptions` ids_only set to True."))
-        time_ms = self._orig.raw_result.get('scan_item', {}).get('body', {}).get('cas', None)
+        time_ms = self._orig.raw_result.get('scan_item', {}).get('body', {}).get('expiry', None)
         if time_ms:
             return datetime.fromtimestamp(time_ms)
         return None

couchbase/tests/kv_range_scan_t.py

@@ -13,7 +13,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from datetime import timedelta
+from datetime import datetime, timedelta
 from uuid import uuid4
 
 import pytest
@@ -26,9 +26,13 @@
                                      RangeScan,
                                      SamplingScan,
                                      ScanTerm)
+from couchbase.logic.collection_req_builder import CollectionRequestBuilder
+from couchbase.logic.collection_types import CollectionDetails
 from couchbase.mutation_state import MutationState
 from couchbase.options import ScanOptions
-from couchbase.result import ScanResult, ScanResultIterable
+from couchbase.result import (MutationToken,
+                              ScanResult,
+                              ScanResultIterable)
 from tests.environments import CollectionType
 from tests.test_features import EnvironmentFeatures
 
@@ -116,6 +120,11 @@ def _validate_result(self, result, expected_count=0, ids_only=False, return_rows
                 assert content is not None
                 if not skip_content_validation:
                     assert content == {'id': r.id}
+                # Regression guard (PYCBC-1778): ScanResult.expiry_time previously read the 'cas' field instead
+                # of 'expiry', passing a huge HLC value to datetime.fromtimestamp() raising an exception.
+                # Accessing it on a content result must not raise. The value is either None (no expiry) or a datetime.
+                expiry_time = r.expiry_time
+                assert expiry_time is None or isinstance(expiry_time, datetime)
             rows.append(r)
 
         if from_sample is True:
@@ -364,3 +373,52 @@ def couchbase_test_environment(self, cb_base_env, test_manifest_validated, reque
         yield cb_base_env
         self._purge_temp_docs(cb_base_env, test_ids)
         cb_base_env.teardown(request.param)
+
+
+class RangeScanRequestBuilderUnitTests:
+    """Cluster-free unit tests for the range-scan request builder.
+
+    Regression coverage for the consistent_with serialization bug (PYCBC-1778)
+    """
+
+    @staticmethod
+    def _builder():
+        details = CollectionDetails('default', '_default', '_default', None)
+        return CollectionRequestBuilder(details)
+
+    @staticmethod
+    def _mutation_state(count):
+        state = MutationState()
+        for i in range(count):
+            state.add_mutation_token(MutationToken({'partition_id': i,
+                                                    'partition_uuid': 1000 + i,
+                                                    'sequence_number': 1,
+                                                    'bucket_name': 'default'}))
+        return state
+
+    def test_consistent_with_serialized_as_mutation_state(self):
+        opts = {'consistent_with': self._mutation_state(3), 'concurrency': 1}
+        self._builder()._process_scan_orchestrator_ops(opts)
+
+        # Must be a mutation_state dict ({'tokens': [...]}), not a bare list, or the C++
+        # binding drops every token.
+        consistent_with = opts['consistent_with']
+        assert isinstance(consistent_with, dict)
+        assert list(consistent_with.keys()) == ['tokens']
+
+        tokens = consistent_with['tokens']
+        assert isinstance(tokens, list)
+        assert len(tokens) == 3
+        for token in tokens:
+            assert set(token.keys()) == {'partition_id', 'partition_uuid',
+                                         'sequence_number', 'bucket_name'}
+
+    def test_empty_consistent_with_raises(self):
+        opts = {'consistent_with': MutationState()}
+        with pytest.raises(InvalidArgumentException):
+            self._builder()._process_scan_orchestrator_ops(opts)
+
+    def test_zero_concurrency_raises(self):
+        opts = {'concurrency': 0}
+        with pytest.raises(InvalidArgumentException):
+            self._builder()._process_scan_orchestrator_ops(opts)