Cover cloud recognition server errors

newsfragments/3214.change.rst

@@ -0,0 +1 @@
+Cover server errors from synchronous and asynchronous cloud recognition queries.

src/vws/async_query.py

@@ -187,9 +187,7 @@ async def query(
                 response=response,
             )
 
-        if (
-            response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR
-        ):  # pragma: no cover
+        if response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR:
             raise ServerError(response=response)
 
         content_type = {

src/vws/query.py

@@ -157,9 +157,7 @@ def query(
         if "Integer out of range" in response.text:
             raise MaxNumResultsOutOfRangeError(response=response)
 
-        if (
-            response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR
-        ):  # pragma: no cover
+        if response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR:
             raise ServerError(response=response)
 
         content_type = {

tests/test_async_query.py

@@ -2,15 +2,51 @@
 
 from __future__ import annotations
 
+import json
+import secrets
 import uuid
+from http import HTTPStatus
 from typing import TYPE_CHECKING, BinaryIO
 
 import pytest
 from mock_vws import MockVWS
 from mock_vws.database import CloudDatabase
 
 from vws import AsyncCloudRecoService, AsyncVWS
+from vws.exceptions.custom_exceptions import ServerError
 from vws.include_target_data import CloudRecoIncludeTargetData
+from vws.response import Response
+
+
+class _ServerErrorTransport:
+    """An async transport which returns a server error."""
+
+    async def aclose(self) -> None:
+        """Close the transport."""
+
+    async def __call__(
+        self,
+        *,
+        method: str,
+        url: str,
+        headers: dict[str, str],
+        data: bytes,
+        request_timeout: float | tuple[float, float],
+    ) -> Response:
+        """Return a server-error response."""
+        del method, headers, data, request_timeout
+        text = json.dumps(obj={})
+        content = text.encode()
+        return Response(
+            text=text,
+            url=url,
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+            headers={"Content-Type": "application/json"},
+            request_body=None,
+            tell_position=len(content),
+            content=content,
+        )
+
 
 if TYPE_CHECKING:
     import io
@@ -56,6 +92,19 @@ async def test_match(
         )
         assert matching_target.target_id == target_id
 
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_server_error(*, image: io.BytesIO | BinaryIO) -> None:
+        """Server errors are exposed through the public async client."""
+        client = AsyncCloudRecoService(
+            client_access_key="access-key",
+            client_secret_key=secrets.token_hex(),
+            transport=_ServerErrorTransport(),
+        )
+
+        with pytest.raises(expected_exception=ServerError):
+            _ = await client.query(image=image)
+
 
 class TestCustomBaseVWQURL:
     """Tests for using a custom base VWQ URL."""

tests/test_query.py

@@ -16,6 +16,7 @@
 from mock_vws.database import CloudDatabase
 
 from vws import VWS, CloudRecoService
+from vws.exceptions.custom_exceptions import ServerError
 from vws.include_target_data import CloudRecoIncludeTargetData
 from vws.response import Response
 
@@ -26,9 +27,15 @@
 class _JSONResponseTransport:
     """A transport which returns one JSON response body."""
 
-    def __init__(self, *, body: object) -> None:
+    def __init__(
+        self,
+        *,
+        body: object,
+        status_code: HTTPStatus,
+    ) -> None:
         """Create a transport for the given JSON body."""
         self._text = json.dumps(obj=body)
+        self._status_code = status_code
 
     def close(self) -> None:
         """Close the transport."""
@@ -48,7 +55,7 @@ def __call__(
         return Response(
             text=self._text,
             url=url,
-            status_code=HTTPStatus.OK,
+            status_code=self._status_code,
             headers={"Content-Type": "application/json"},
             request_body=None,
             tell_position=len(content),
@@ -92,7 +99,8 @@ def test_match(
     def test_invalid_results(*, image: io.BytesIO | BinaryIO) -> None:
         """Query results in responses must be a list of objects."""
         transport = _JSONResponseTransport(
-            body={"result_code": "Success", "results": 1}
+            body={"result_code": "Success", "results": 1},
+            status_code=HTTPStatus.OK,
         )
         client = CloudRecoService(
             client_access_key="access-key",
@@ -103,6 +111,22 @@ def test_invalid_results(*, image: io.BytesIO | BinaryIO) -> None:
         with pytest.raises(expected_exception=TypeError):
             _ = client.query(image=image)
 
+    @staticmethod
+    def test_server_error(*, image: io.BytesIO | BinaryIO) -> None:
+        """Server errors are exposed through the public query client."""
+        transport = _JSONResponseTransport(
+            body={},
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+        )
+        client = CloudRecoService(
+            client_access_key="access-key",
+            client_secret_key=secrets.token_hex(),
+            transport=transport,
+        )
+
+        with pytest.raises(expected_exception=ServerError):
+            _ = client.query(image=image)
+
 
 class TestDefaultRequestTimeout:
     """Tests for the default request timeout."""