PYCO-55: Add GHA linting and tests

thejcfactor · thejcfactor · commit e4a00c80b6bf · 2025-08-01T13:06:01.000-06:00
Changes
=======
* Add test.yml workflow
* Update error handling to take into account timeout scenarios after
  `send_request` is called again after retry
* Update error handling to take into account httpx ReadTimeout,
  WriteError and WriteTimeout
* Update JSON streaming/parsing to handle when ijson uses Python as the
  backend and thus can raise different errors
* Update ijson from 3.3.0 to 3.4.0 to get Python 3.13 wheel
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
diff --git a/acouchbase_analytics/protocol/_core/async_json_stream.py b/acouchbase_analytics/protocol/_core/async_json_stream.py
@@ -154,6 +154,20 @@ async def _process_token_stream(self) -> None:
                 await self._send_to_stream(ParsedResult(ex_str.encode('utf-8'), ParsedResultType.ERROR), close=True)
                 self._handle_notification(ParsedResultType.ERROR)
                 return
+            except ijson.common.JSONError as ex:
+                ex_str = str(ex)
+                self._log_message(f'JSON error encountered: {ex_str}', LogLevel.ERROR)
+                self._token_stream_exhausted = True
+                await self._send_to_stream(ParsedResult(ex_str.encode('utf-8'), ParsedResultType.ERROR), close=True)
+                self._handle_notification(ParsedResultType.ERROR)
+                return
+            except ijson.backends.python.UnexpectedSymbol as ex:
+                ex_str = str(ex)
+                self._log_message(f'Unexpected symbol encountered: {ex_str}', LogLevel.ERROR)
+                self._token_stream_exhausted = True
+                await self._send_to_stream(ParsedResult(ex_str.encode('utf-8'), ParsedResultType.ERROR), close=True)
+                self._handle_notification(ParsedResultType.ERROR)
+                return
 
         if self._token_stream_exhausted:
             result_type = ParsedResultType.ERROR if self._json_token_parser.has_errors else ParsedResultType.END
diff --git a/acouchbase_analytics/protocol/_core/retries.py b/acouchbase_analytics/protocol/_core/retries.py
@@ -19,7 +19,7 @@
 from functools import wraps
 from typing import TYPE_CHECKING, Any, Callable, Coroutine, Optional, Union
 
-from httpx import ConnectError, ConnectTimeout
+from httpx import ConnectError, ConnectTimeout, ReadTimeout, WriteError, WriteTimeout
 
 from acouchbase_analytics.protocol._core.anyio_utils import sleep
 from couchbase_analytics.common.errors import AnalyticsError, InternalSDKError, TimeoutError
@@ -39,7 +39,7 @@ class AsyncRetryHandler:
 
     @staticmethod
     async def handle_httpx_retry(
-        ex: Union[ConnectError, ConnectTimeout], ctx: AsyncRequestContext
+        ex: Union[ConnectError, ConnectTimeout, WriteError, WriteTimeout], ctx: AsyncRequestContext
     ) -> Optional[Exception]:
         err_str = str(ex)
         if 'SSL:' in err_str:
@@ -107,17 +107,30 @@ async def wrapped_fn(self: AsyncHttpStreamingResponse) -> None:  # noqa: C901
                         continue
                     await self._request_context.shutdown(type(ex), ex, ex.__traceback__)
                     raise err from None
-                except (ConnectError, ConnectTimeout) as ex:
+                except (ConnectError, ConnectTimeout, WriteError, WriteTimeout) as ex:
                     err = await AsyncRetryHandler.handle_httpx_retry(ex, self._request_context)
                     if err is None:
                         continue
                     await self._request_context.shutdown(type(ex), ex, ex.__traceback__)
                     raise err from None
+                except ReadTimeout as ex:
+                    # we set the read timeout to the query timeout, so if we get a ReadTimeout,
+                    # it means the request timed out from the httpx client
+                    await self._request_context.shutdown(type(ex), ex, ex.__traceback__)
+                    raise TimeoutError(
+                        message='Request timed out.', context=str(self._request_context.error_context)
+                    ) from None
                 except AnalyticsError:
                     # if an AnalyticsError is raised, we have already shut down the request context
                     raise
                 except RuntimeError as ex:
                     await self._request_context.shutdown(type(ex), ex, ex.__traceback__)
+                    if self._request_context.timed_out:
+                        raise TimeoutError(
+                            message='Request timeout.', context=str(self._request_context.error_context)
+                        ) from None
+                    if self._request_context.cancelled:
+                        raise CancelledError('Request was cancelled.') from None
                     raise ex
                 except BaseException as ex:
                     await self._request_context.shutdown(type(ex), ex, ex.__traceback__)
diff --git a/acouchbase_analytics/tests/json_parsing_t.py b/acouchbase_analytics/tests/json_parsing_t.py
@@ -312,7 +312,8 @@ async def test_invalid_empty(self) -> None:
         res = await parser.get_result()
         assert res.result_type == ParsedResultType.ERROR
         assert res.value is not None
-        assert 'parse error' in str(res.value.decode('utf-8'))
+        decoded_value = res.value.decode('utf-8')
+        assert ('parse error' in decoded_value or 'Incomplete JSON content' in decoded_value) is True
 
     @pytest.mark.anyio
     async def test_invalid_garbage_between_objects(self) -> None:
@@ -324,7 +325,8 @@ async def test_invalid_garbage_between_objects(self) -> None:
         res = await parser.get_result()
         assert res.result_type == ParsedResultType.ERROR
         assert res.value is not None
-        assert 'lexical error' in str(res.value.decode('utf-8'))
+        decoded_value = res.value.decode('utf-8')
+        assert ('lexical error' in decoded_value or 'Unexpected symbol' in decoded_value) is True
 
     @pytest.mark.anyio
     async def test_invalid_leading_garbage(self) -> None:
@@ -336,7 +338,8 @@ async def test_invalid_leading_garbage(self) -> None:
         res = await parser.get_result()
         assert res.result_type == ParsedResultType.ERROR
         assert res.value is not None
-        assert 'lexical error' in str(res.value.decode('utf-8'))
+        decoded_value = res.value.decode('utf-8')
+        assert ('lexical error' in decoded_value or 'Unexpected symbol' in decoded_value) is True
 
     @pytest.mark.anyio
     async def test_invalid_trailing_garbage(self) -> None:
@@ -348,7 +351,8 @@ async def test_invalid_trailing_garbage(self) -> None:
         res = await parser.get_result()
         assert res.result_type == ParsedResultType.ERROR
         assert res.value is not None
-        assert 'parse error' in str(res.value.decode('utf-8'))
+        decoded_value = res.value.decode('utf-8')
+        assert ('parse error' in decoded_value or 'Additional data found' in decoded_value) is True
 
     @pytest.mark.anyio
     async def test_invalid_whitespace_only(self) -> None:
@@ -360,7 +364,8 @@ async def test_invalid_whitespace_only(self) -> None:
         res = await parser.get_result()
         assert res.result_type == ParsedResultType.ERROR
         assert res.value is not None
-        assert 'parse error' in str(res.value.decode('utf-8'))
+        decoded_value = res.value.decode('utf-8')
+        assert ('parse error' in decoded_value or 'Incomplete JSON content' in decoded_value) is True
 
     @pytest.mark.anyio
     async def test_value_bool(self) -> None:
diff --git a/acouchbase_analytics/tests/test_server_t.py b/acouchbase_analytics/tests/test_server_t.py
@@ -148,7 +148,7 @@ async def test_error_timeout(self, test_env: AsyncTestEnvironment, server_side:
         if server_side:
             req_json = {'error_type': ErrorType.Timeout.value, 'timeout': 1, 'server_side': True}
         else:
-            req_json = {'error_type': ErrorType.Timeout.value, 'timeout': 2}
+            req_json = {'error_type': ErrorType.Timeout.value, 'timeout': 3}
 
         test_env.update_request_json(req_json)
         statement = 'SELECT "Hello, data!" AS greeting'
diff --git a/couchbase_analytics/protocol/_core/json_stream.py b/couchbase_analytics/protocol/_core/json_stream.py
@@ -151,6 +151,20 @@ def _process_token_stream(self, request_context: Optional[RequestContext] = None
                 self._put(ParsedResult(ex_str.encode('utf-8'), ParsedResultType.ERROR))
                 self._handle_notification(ParsedResultType.ERROR)
                 return
+            except ijson.common.JSONError as ex:
+                ex_str = str(ex)
+                self._log_message(f'JSON error encountered: {ex_str}', LogLevel.ERROR)
+                self._token_stream_exhausted = True
+                self._put(ParsedResult(ex_str.encode('utf-8'), ParsedResultType.ERROR))
+                self._handle_notification(ParsedResultType.ERROR)
+                return
+            except ijson.backends.python.UnexpectedSymbol as ex:
+                ex_str = str(ex)
+                self._log_message(f'Unexpected symbol encountered: {ex_str}', LogLevel.ERROR)
+                self._token_stream_exhausted = True
+                self._put(ParsedResult(ex_str.encode('utf-8'), ParsedResultType.ERROR))
+                self._handle_notification(ParsedResultType.ERROR)
+                return
 
         if self._token_stream_exhausted:
             result_type = ParsedResultType.ERROR if self._json_token_parser.has_errors else ParsedResultType.END
diff --git a/couchbase_analytics/protocol/_core/request_context.py b/couchbase_analytics/protocol/_core/request_context.py
@@ -212,6 +212,7 @@ def _reset_stream(self) -> None:
             del self._json_stream
         self._request_state = RequestState.ResetAndNotStarted
         self._stage_notification_ft = None
+        self.log_message('Request state has been reset', LogLevel.DEBUG)
 
     def _start_next_stage(
         self,
@@ -454,5 +455,8 @@ def wait_for_stage_notification(self) -> None:
             raise TimeoutError(message='Request timed out waiting for stage notification', context=str(self._error_ctx))
         result_type = self._stage_notification_ft.result(timeout=deadline)
         if result_type == ParsedResultType.ROW:
+            self.log_message('Received row, setting status to streaming', LogLevel.DEBUG)
             # we move to iterating rows
             self._request_state = RequestState.StreamingResults
+        else:
+            self.log_message(f'Received result type {result_type.name}', LogLevel.DEBUG)
diff --git a/couchbase_analytics/protocol/_core/retries.py b/couchbase_analytics/protocol/_core/retries.py
@@ -20,7 +20,7 @@
 from time import sleep
 from typing import TYPE_CHECKING, Callable, Optional, Union
 
-from httpx import ConnectError, ConnectTimeout
+from httpx import ConnectError, ConnectTimeout, ReadTimeout, WriteError, WriteTimeout
 
 from couchbase_analytics.common.errors import AnalyticsError, InternalSDKError, TimeoutError
 from couchbase_analytics.common.logging import LogLevel
@@ -38,7 +38,9 @@ class RetryHandler:
     """
 
     @staticmethod
-    def handle_httpx_retry(ex: Union[ConnectError, ConnectTimeout], ctx: RequestContext) -> Optional[Exception]:
+    def handle_httpx_retry(
+        ex: Union[ConnectError, ConnectTimeout, WriteError, WriteTimeout], ctx: RequestContext
+    ) -> Optional[Exception]:
         err_str = str(ex)
         if 'SSL:' in err_str:
             message = 'TLS connection error occurred.'
@@ -103,17 +105,30 @@ def wrapped_fn(self: HttpStreamingResponse) -> None:  # noqa: C901
                         continue
                     self._request_context.shutdown(ex)
                     raise err from None
-                except (ConnectError, ConnectTimeout) as ex:
+                except (ConnectError, ConnectTimeout, WriteError, WriteTimeout) as ex:
                     err = RetryHandler.handle_httpx_retry(ex, self._request_context)
                     if err is None:
                         continue
                     self._request_context.shutdown(ex)
                     raise err from None
+                except ReadTimeout as ex:
+                    # we set the read timeout to the query timeout, so if we get a ReadTimeout,
+                    # it means the request timed out from the httpx client
+                    self._request_context.shutdown(ex)
+                    raise TimeoutError(
+                        message='Request timed out.', context=str(self._request_context.error_context)
+                    ) from None
                 except AnalyticsError:
                     # if an AnalyticsError is raised, we have already shut down the request context
                     raise
                 except RuntimeError as ex:
                     self._request_context.shutdown(ex)
+                    if self._request_context.timed_out:
+                        raise TimeoutError(
+                            message='Request timeout.', context=str(self._request_context.error_context)
+                        ) from None
+                    if self._request_context.cancelled:
+                        raise CancelledError('Request was cancelled.') from None
                     raise ex
                 except BaseException as ex:
                     self._request_context.shutdown(ex)
diff --git a/couchbase_analytics/tests/json_parsing_t.py b/couchbase_analytics/tests/json_parsing_t.py
@@ -256,7 +256,8 @@ def test_invalid_empty(self) -> None:
         assert isinstance(res, ParsedResult)
         assert res.result_type == ParsedResultType.ERROR
         assert res.value is not None
-        assert 'parse error' in str(res.value.decode('utf-8'))
+        decoded_value = res.value.decode('utf-8')
+        assert ('parse error' in decoded_value or 'Incomplete JSON content' in decoded_value) is True
 
     def test_invalid_garbage_between_objects(self) -> None:
         data = '[{"id":1,"name":"Alice"},garbage,{"id":2,"name":"Bob"}]'
@@ -268,7 +269,8 @@ def test_invalid_garbage_between_objects(self) -> None:
         assert isinstance(res, ParsedResult)
         assert res.result_type == ParsedResultType.ERROR
         assert res.value is not None
-        assert 'lexical error' in str(res.value.decode('utf-8'))
+        decoded_value = res.value.decode('utf-8')
+        assert ('lexical error' in decoded_value or 'Unexpected symbol' in decoded_value) is True
 
     def test_invalid_leading_garbage(self) -> None:
         data = 'garbage{"key":"value"}'
@@ -280,7 +282,8 @@ def test_invalid_leading_garbage(self) -> None:
         assert isinstance(res, ParsedResult)
         assert res.result_type == ParsedResultType.ERROR
         assert res.value is not None
-        assert 'lexical error' in str(res.value.decode('utf-8'))
+        decoded_value = res.value.decode('utf-8')
+        assert ('lexical error' in decoded_value or 'Unexpected symbol' in decoded_value) is True
 
     def test_invalid_trailing_garbage(self) -> None:
         data = '{"key":"value"}garbage'
@@ -292,7 +295,8 @@ def test_invalid_trailing_garbage(self) -> None:
         assert isinstance(res, ParsedResult)
         assert res.result_type == ParsedResultType.ERROR
         assert res.value is not None
-        assert 'parse error' in str(res.value.decode('utf-8'))
+        decoded_value = res.value.decode('utf-8')
+        assert ('parse error' in decoded_value or 'Additional data found' in decoded_value) is True
 
     def test_invalid_whitespace_only(self) -> None:
         data = '   \n\t  '
@@ -304,7 +308,8 @@ def test_invalid_whitespace_only(self) -> None:
         assert isinstance(res, ParsedResult)
         assert res.result_type == ParsedResultType.ERROR
         assert res.value is not None
-        assert 'parse error' in str(res.value.decode('utf-8'))
+        decoded_value = res.value.decode('utf-8')
+        assert ('parse error' in decoded_value or 'Incomplete JSON content' in decoded_value) is True
 
     def test_object(self) -> None:
         data = '{"name":"John","age":30,"city":"New York"}'
diff --git a/couchbase_analytics/tests/test_server_t.py b/couchbase_analytics/tests/test_server_t.py
@@ -150,7 +150,7 @@ def test_error_timeout(self, test_env: BlockingTestEnvironment, server_side: boo
         if server_side:
             req_json = {'error_type': ErrorType.Timeout.value, 'timeout': 1, 'server_side': True}
         else:
-            req_json = {'error_type': ErrorType.Timeout.value, 'timeout': 2}
+            req_json = {'error_type': ErrorType.Timeout.value, 'timeout': 3}
 
         test_env.update_request_json(req_json)
         statement = 'SELECT "Hello, data!" AS greeting'
diff --git a/mypy.ini b/mypy.ini
@@ -1,6 +1,7 @@
 [mypy]
 exclude = (?x)(
     setup\.py$
+    | ci_scripts/
     | docs/
     | tests/utils/
     )
diff --git a/pyproject.toml b/pyproject.toml
@@ -11,7 +11,7 @@ version = "1.0.0.dev1"
 dependencies = [
     "anyio~=4.9.0",
     "httpx~=0.28.1",
-    "ijson~=3.3.0",
+    "ijson~=3.4.0",
     "sniffio~=1.3.1",
     "typing-extensions~=4.11; python_version<'3.11'",
 ]
diff --git a/requirements.in b/requirements.in
@@ -1,6 +1,6 @@
 anyio~=4.9.0
 sniffio~=1.3.1
 httpx~=0.28.1
-ijson~=3.3.0
+ijson~=3.4.0
 # Typing support
 typing-extensions~=4.11; python_version<"3.11.0"
diff --git a/requirements.txt b/requirements.txt
@@ -20,7 +20,7 @@ idna==3.10
     # via
     #   anyio
     #   httpx
-ijson==3.3.0
+ijson==3.4.0
     # via -r requirements.in
 sniffio==1.3.1
     # via
diff --git a/tests/environments/base_environment.py b/tests/environments/base_environment.py
@@ -38,6 +38,7 @@
 from couchbase_analytics.result import BlockingQueryResult
 from couchbase_analytics.scope import Scope
 from tests import AnalyticsTestEnvironmentError
+from tests.test_server import ResultType
 from tests.utils._run_web_server import WebServerHandler
 
 if TYPE_CHECKING:
@@ -208,6 +209,7 @@ def enable_test_server(self) -> BlockingTestEnvironment:
         url = self._cluster._impl.client_adapter.connection_details.url.get_formatted_url()
         print(f'Connecting to test server at {url}')
         self._server_handler.start_server()
+        self.warmup_test_server()
         return self
 
     def setup(self) -> None:
@@ -289,6 +291,24 @@ def update_request_json(self, json: Dict[str, object]) -> None:
             raise AnalyticsTestEnvironmentError('No cluster available, cannot enable test server.')
         self._cluster._impl._client_adapter.update_request_json(json)
 
+    def warmup_test_server(self) -> None:
+        row_count = 5
+        self.set_url_path('/test_results')
+        self.update_request_json({'result_type': ResultType.Object.value, 'row_count': 5, 'stream': True})
+        statement = 'SELECT "Hello, data!" AS greeting'
+        exc = None
+        for _ in range(3):
+            exc = None
+            try:
+                res = self.cluster.execute_query(statement)
+                rows = list(res.rows())
+                if len(rows) == row_count:
+                    break
+            except Exception as ex:
+                exc = AnalyticsTestEnvironmentError(f'Unable to execute statement={statement}. Error: {ex}')
+        if exc is not None:
+            raise exc
+
     @classmethod
     def get_environment(
         cls, config: AnalyticsConfig, server_handler: Optional[WebServerHandler] = None
@@ -407,6 +427,7 @@ async def enable_test_server(self) -> AsyncTestEnvironment:
         url = self._async_cluster._impl.client_adapter.connection_details.url.get_formatted_url()
         print(f'Connecting to test server at {url}')
         self._server_handler.start_server()
+        await self.warmup_test_server()
         return self
 
     async def setup(self) -> None:
@@ -491,6 +512,24 @@ def update_request_json(self, json: Dict[str, object]) -> None:
             raise AnalyticsTestEnvironmentError('No cluster available, cannot enable test server.')
         self._async_cluster._impl._client_adapter.update_request_json(json)
 
+    async def warmup_test_server(self) -> None:
+        row_count = 5
+        self.set_url_path('/test_results')
+        self.update_request_json({'result_type': ResultType.Object.value, 'row_count': 5, 'stream': True})
+        statement = 'SELECT "Hello, data!" AS greeting'
+        exc = None
+        for _ in range(3):
+            exc = None
+            try:
+                res = await self.cluster.execute_query(statement)
+                rows = [r async for r in res.rows()]
+                if len(rows) == row_count:
+                    break
+            except Exception as ex:
+                exc = AnalyticsTestEnvironmentError(f'Unable to execute statement={statement}. Error: {ex}')
+        if exc is not None:
+            raise exc
+
     @classmethod
     def get_environment(
         cls, config: AnalyticsConfig, server_handler: Optional[WebServerHandler] = None, backend: Optional[str] = None
diff --git a/uv.lock b/uv.lock
