PYCBC-1894: Correct request-level timeout and scan_wait handling

Changes
--------
* Encode analytics and view request timeouts in milliseconds, the unit
  every marshaller reads, where to_microseconds asked the core for a
  timeout 1000 times too long; to_microseconds is deleted with its
  last two callers
* Read the five duration getters back from the stored milliseconds
  instead of slicing a unit character off a string, which raised
  TypeError on every request that had a timeout or scan_wait set
* Convert in the N1QLQuery.scan_wait setter on every assignment rather
  than only when the key is already present, and pass scan_wait through
  _VALID_OPTS unchanged, which is how timeout already works
* Build TransactionQueryOptions.scan_wait with
  timedelta_as_milliseconds, where dividing an uncalled total_seconds
  by 1000 raised TypeError on any scan_wait at all
* Correct the analytics and views timeout assertions, which encoded the
  wrong unit, and cover the five getters, a direct scan_wait assignment
  and TransactionQueryOptions.scan_wait, none of which had a test

Change-Id: I6962b8ef34ec623dece7368a00728eeea1220e6a
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/252246
Reviewed-by: Sergey Avseyev <sergey.avseyev@gmail.com>
Tested-by: Build Bot <build@couchbase.com>

Author: thejcfactor

couchbase/_utils.py

@@ -69,22 +69,6 @@ def timedelta_as_milliseconds(duration: timedelta) -> int:
     return int(duration.total_seconds() * 1e3 if duration else 0)
 
 
-def to_microseconds(
-    timeout  # type: Union[timedelta, float, int]
-) -> int:
-    if timeout and not isinstance(timeout, (timedelta, float, int)):
-        raise InvalidArgumentException(message=("Expected timeout to be of type "
-                                                f"Union[timedelta, float, int] instead of {timeout}"))
-    if not timeout:
-        total_us = 0
-    elif isinstance(timeout, timedelta):
-        total_us = int(timeout.total_seconds() * 1e6)
-    else:
-        total_us = int(timeout * 1e6)
-
-    return total_us
-
-
 def to_milliseconds(
     timeout  # type: Union[timedelta, float, int]
 ) -> int:

couchbase/logic/analytics.py

@@ -22,7 +22,7 @@
                     Optional,
                     Union)
 
-from couchbase._utils import to_microseconds
+from couchbase._utils import to_milliseconds
 from couchbase.exceptions import ErrorMapper, InvalidArgumentException
 from couchbase.logic.observability import ObservableRequestHandler, SpanProtocol
 from couchbase.logic.options import AnalyticsOptionsBase
@@ -233,20 +233,19 @@ def params(self):
 
     @property
     def timeout(self) -> Optional[float]:
-        value = self._params.get('timeout', None)
-        if not value:
+        total_ms = self._params.get('timeout', None)
+        if not total_ms:
             return None
-        value = value[:-1]
-        return float(value)
+        return total_ms / 1000
 
     @timeout.setter
     def timeout(self, value  # type: Union[timedelta,float,int]
                 ) -> None:
         if not value:
             self._params.pop('timeout', 0)
         else:
-            total_us = to_microseconds(value)
-            self.set_option('timeout', total_us)
+            total_ms = to_milliseconds(value)
+            self.set_option('timeout', total_ms)
 
     @property
     def metrics(self):

couchbase/logic/n1ql.py

@@ -329,7 +329,7 @@ class N1QLQuery:
         "query_context": {"query_context": lambda x: x},
         "raw": {"raw": lambda x: x},
         "scan_cap": {"scan_cap": lambda x: x},
-        "scan_wait": {"scan_wait": timedelta_as_milliseconds},
+        "scan_wait": {"scan_wait": lambda x: x},
         "metrics": {"metrics": lambda x: x},
         "flex_index": {"flex_index": lambda x: x},
         "preserve_expiry": {"preserve_expiry": lambda x: x},
@@ -409,11 +409,10 @@ def statement(self) -> str:
 
     @property
     def timeout(self) -> Optional[float]:
-        value = self._params.get('timeout', None)
-        if not value:
+        total_ms = self._params.get('timeout', None)
+        if not total_ms:
             return None
-        value = value[:-1]
-        return float(value)
+        return total_ms / 1000
 
     @timeout.setter
     def timeout(self, value  # type: Union[timedelta,float,int]
@@ -606,23 +605,18 @@ def scan_cap(self, value  # type: int
 
     @property
     def scan_wait(self) -> Optional[float]:
-        value = self._params.get('scan_wait', None)
-        if not value:
+        total_ms = self._params.get('scan_wait', None)
+        if not total_ms:
             return None
-        value = value[:-1]
-        return float(value)
+        return total_ms / 1000
 
     @scan_wait.setter
     def scan_wait(self, value  # type: timedelta
                   ) -> None:
         if not value:
             self._params.pop('scan_wait', 0)
         else:
-            # if using the setter, need to validate/transform timedelta, otherwise, just add the value
-            if 'scan_wait' in self._params:
-                value = timedelta_as_milliseconds(value)
-
-            self.set_option('scan_wait', value)
+            self.set_option('scan_wait', timedelta_as_milliseconds(value))
 
     @property
     def flex_index(self) -> bool:

couchbase/logic/search.py

@@ -1027,11 +1027,10 @@ def params(self) -> Dict[str, Any]:
 
     @property
     def timeout(self) -> Optional[float]:
-        value = self._params.get('timeout', None)
-        if not value:
+        total_ms = self._params.get('timeout', None)
+        if not total_ms:
             return None
-        value = value[:-1]
-        return float(value)
+        return total_ms / 1000
 
     @timeout.setter
     def timeout(self, value  # type: Union[timedelta,float,int]

couchbase/logic/views.py

@@ -26,7 +26,7 @@
                     Optional,
                     Union)
 
-from couchbase._utils import to_microseconds
+from couchbase._utils import to_milliseconds
 from couchbase.exceptions import ErrorMapper, InvalidArgumentException
 from couchbase.logic.observability import ObservableRequestHandler, SpanProtocol
 from couchbase.logic.options import ViewOptionsBase
@@ -198,20 +198,19 @@ def as_encodable(self) -> Dict[str, Any]:
 
     @property
     def timeout(self) -> Optional[float]:
-        value = self._params.get('timeout', None)
-        if not value:
+        total_ms = self._params.get('timeout', None)
+        if not total_ms:
             return None
-        value = value[:-1]
-        return float(value)
+        return total_ms / 1000
 
     @timeout.setter
     def timeout(self, value  # type: Union[timedelta,float,int]
                 ) -> None:
         if not value:
             self._params.pop('timeout', 0)
         else:
-            total_us = to_microseconds(value)
-            self.set_option('timeout', total_us)
+            total_ms = to_milliseconds(value)
+            self.set_option('timeout', total_ms)
 
     @property
     def limit(self) -> Optional[int]:

couchbase/options.py

@@ -2009,7 +2009,7 @@ def __init__(self,   # noqa: C901
             kwargs["bucket_name"] = scope.bucket_name
             kwargs["scope_name"] = scope.name
         if kwargs.get("scan_wait", None):
-            kwargs["scan_wait"] = kwargs["scan_wait"].total_seconds/1000
+            kwargs["scan_wait"] = timedelta_as_milliseconds(kwargs["scan_wait"])
         if kwargs.get("scan_consistency", None):
             kwargs["scan_consistency"] = kwargs["scan_consistency"].value
             if kwargs["scan_consistency"] == "at_plus":

couchbase/tests/analytics_params_t.py

@@ -34,6 +34,7 @@ class AnalyticsParamTestSuite:
         'test_params_read_only',
         'test_params_serializer',
         'test_params_timeout',
+        'test_params_timeout_getter',
         'test_status'
     ]
 
@@ -117,23 +118,31 @@ def test_params_timeout(self, base_opts):
         query = AnalyticsQuery.create_query_object(q_str, q_opts)
 
         exp_opts = base_opts.copy()
-        exp_opts['timeout'] = 120000000
+        exp_opts['timeout'] = 120000
         assert query.params == exp_opts
 
         q_opts = AnalyticsOptions(timeout=20)
         query = AnalyticsQuery.create_query_object(q_str, q_opts)
 
         exp_opts = base_opts.copy()
-        exp_opts['timeout'] = 20000000
+        exp_opts['timeout'] = 20000
         assert query.params == exp_opts
 
         q_opts = AnalyticsOptions(timeout=25.5)
         query = AnalyticsQuery.create_query_object(q_str, q_opts)
 
         exp_opts = base_opts.copy()
-        exp_opts['timeout'] = 25500000
+        exp_opts['timeout'] = 25500
         assert query.params == exp_opts
 
+    def test_params_timeout_getter(self):
+        q_str = 'SELECT * FROM default'
+        query = AnalyticsQuery.create_query_object(q_str, AnalyticsOptions(timeout=timedelta(seconds=120)))
+        assert query.timeout == 120
+
+        query = AnalyticsQuery.create_query_object(q_str, AnalyticsOptions())
+        assert query.timeout is None
+
     @pytest.mark.parametrize('value, expected', [(k, v) for k, v in AnalyticsStatus.__members__.items()])
     def test_status(self, value, expected):
         a_status = AnalyticsStatus[value]

couchbase/tests/query_params_t.py

@@ -46,8 +46,10 @@ class QueryParamTestSuite:
         'test_params_scan_cap',
         'test_params_scan_consistency',
         'test_params_scan_wait',
+        'test_params_scan_wait_direct',
         'test_params_serializer',
         'test_params_timeout',
+        'test_params_timeout_getter',
         'test_params_uninterpretable_enum_option',
         'test_params_use_replica',
     ]
@@ -291,6 +293,14 @@ def test_params_scan_wait(self, base_opts):
         exp_opts['scan_wait'] = 30000
         assert query.params == exp_opts
 
+    def test_params_scan_wait_direct(self):
+        # Assigning the property is a public entry point in its own right, and it has to
+        # convert on the first assignment rather than the second.
+        query = N1QLQuery('SELECT * FROM default')
+        query.scan_wait = timedelta(seconds=30)
+        assert query.params['scan_wait'] == 30000
+        assert query.scan_wait == 30
+
     def test_params_scan_consistency(self, base_opts):
         q_str = 'SELECT * FROM default'
         q_opts = QueryOptions(scan_consistency=QueryScanConsistency.REQUEST_PLUS)
@@ -345,6 +355,14 @@ def test_params_timeout(self, base_opts):
         exp_opts['timeout'] = 25500
         assert query.params == exp_opts
 
+    def test_params_timeout_getter(self):
+        q_str = 'SELECT * FROM default'
+        query = N1QLQuery.create_query_object(q_str, QueryOptions(timeout=timedelta(seconds=20)))
+        assert query.timeout == 20
+
+        query = N1QLQuery.create_query_object(q_str, QueryOptions())
+        assert query.timeout is None
+
 
 class ClassicQueryParamTests(QueryParamTestSuite):
     @pytest.fixture(scope='class')

couchbase/tests/search_params_t.py

@@ -73,6 +73,7 @@ class SearchParamTestSuite:
         'test_params_sort_invalid',
         'test_params_sort_mixed',
         'test_params_timeout',
+        'test_params_timeout_getter',
         'test_phrase_query',
         'test_prefix_query',
         'test_query_string_query',
@@ -1064,6 +1065,18 @@ def test_params_timeout(self, cb_env, base_query_opts):
         exp_opts['timeout'] = 25500
         assert search_query.params == exp_opts
 
+    def test_params_timeout_getter(self, cb_env, base_query_opts):
+        q, _ = base_query_opts
+        search_query = search.SearchQueryBuilder.create_search_query_object(
+            cb_env.TEST_INDEX_NAME, q, SearchOptions(timeout=timedelta(seconds=20))
+        )
+        assert search_query.timeout == 20
+
+        search_query = search.SearchQueryBuilder.create_search_query_object(
+            cb_env.TEST_INDEX_NAME, q, SearchOptions()
+        )
+        assert search_query.timeout is None
+
     def test_phrase_query(self, cb_env):
         exp_json = {
             'query': {

couchbase/tests/transactions_t.py

@@ -106,6 +106,7 @@ class TransactionTestSuite:
         'test_rollback',
         'test_rollback_eating_exceptions',
         'test_scan_consistency',
+        'test_scan_wait',
         'test_scope_qualifier',
         'test_timeout',
         'test_transaction_config_durability',
@@ -1022,6 +1023,12 @@ def test_scan_consistency(self, cls, consistency):
             assert cfg_consistency is not None
             assert cfg_consistency == consistency.value
 
+    def test_scan_wait(self):
+        cfg = TransactionQueryOptions(scan_wait=timedelta(seconds=30))
+        cfg_scan_wait = cfg._base.to_dict().get('scan_wait', None)
+        assert cfg_scan_wait is not None
+        assert cfg_scan_wait == 30000
+
     def test_scope_qualifier(self, cb_env):
         pytest.skip('CBD-5091: Pending Transactions changes')
         cfg = TransactionQueryOptions(scope=cb_env.collection._impl._scope)