Add tests that verify timeout is actually enforced

Add test_timeout_raises_on_slow_response and test_longer_timeout_succeeds
tests to both VWS and CloudRecoService. These tests simulate slow server
responses and verify that short timeouts actually raise Timeout exceptions.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: adamtheturtle

tests/test_query.py

@@ -1,9 +1,13 @@
 """Tests for the ``CloudRecoService`` querying functionality."""
 
 import io
+import time
 import uuid
 from typing import BinaryIO
+from unittest.mock import patch
 
+import pytest
+import requests
 from mock_vws import MockVWS
 from mock_vws.database import VuforiaDatabase
 
@@ -89,6 +93,106 @@ def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None:
             matches = cloud_reco_client.query(image=image)
             assert len(matches) == 1
 
+    @staticmethod
+    def test_timeout_raises_on_slow_response(
+        image: io.BytesIO | BinaryIO,
+    ) -> None:
+        """A short timeout raises an error when the server is slow."""
+        with MockVWS() as mock:
+            database = VuforiaDatabase()
+            mock.add_database(database=database)
+            vws_client = VWS(
+                server_access_key=database.server_access_key,
+                server_secret_key=database.server_secret_key,
+            )
+            cloud_reco_client = CloudRecoService(
+                client_access_key=database.client_access_key,
+                client_secret_key=database.client_secret_key,
+                request_timeout_seconds=0.1,
+            )
+
+            target_id = vws_client.add_target(
+                name="x",
+                width=1,
+                image=image,
+                active_flag=True,
+                application_metadata=None,
+            )
+            vws_client.wait_for_target_processed(target_id=target_id)
+
+            simulated_slow_threshold = 0.5
+            original_request = requests.request
+
+            def slow_request(
+                *args: object,
+                **kwargs: float | None,
+            ) -> requests.Response:
+                """Simulate a slow server response."""
+                timeout = kwargs.get("timeout")
+                if timeout is not None and timeout < simulated_slow_threshold:
+                    time.sleep(0.2)
+                    raise requests.exceptions.Timeout
+                return original_request(*args, **kwargs)  # type: ignore[arg-type]
+
+            with (
+                patch.object(
+                    requests,
+                    "request",
+                    side_effect=slow_request,
+                ),
+                pytest.raises(requests.exceptions.Timeout),
+            ):
+                cloud_reco_client.query(image=image)
+
+    @staticmethod
+    def test_longer_timeout_succeeds(image: io.BytesIO | BinaryIO) -> None:
+        """A longer timeout allows slow responses to complete."""
+        simulated_slow_threshold = 0.5
+
+        with MockVWS() as mock:
+            database = VuforiaDatabase()
+            mock.add_database(database=database)
+            vws_client = VWS(
+                server_access_key=database.server_access_key,
+                server_secret_key=database.server_secret_key,
+            )
+            cloud_reco_client = CloudRecoService(
+                client_access_key=database.client_access_key,
+                client_secret_key=database.client_secret_key,
+                request_timeout_seconds=1.0,
+            )
+
+            target_id = vws_client.add_target(
+                name="x",
+                width=1,
+                image=image,
+                active_flag=True,
+                application_metadata=None,
+            )
+            vws_client.wait_for_target_processed(target_id=target_id)
+
+            original_request = requests.request
+
+            def slow_request(
+                *args: object,
+                **kwargs: float | None,
+            ) -> requests.Response:
+                """Simulate a slow server response."""
+                timeout = kwargs.get("timeout")
+                if timeout is not None and timeout < simulated_slow_threshold:
+                    time.sleep(0.2)
+                    raise requests.exceptions.Timeout
+                return original_request(*args, **kwargs)  # type: ignore[arg-type]
+
+            with patch.object(
+                requests,
+                "request",
+                side_effect=slow_request,
+            ):
+                # This should succeed because timeout is 1.0 > 0.5
+                matches = cloud_reco_client.query(image=image)
+                assert len(matches) == 1
+
 
 class TestCustomBaseVWQURL:
     """Tests for using a custom base VWQ URL."""

tests/test_vws.py

@@ -4,10 +4,13 @@
 import datetime
 import io
 import secrets
+import time
 import uuid
 from typing import BinaryIO
+from unittest.mock import patch
 
 import pytest
+import requests
 from freezegun import freeze_time
 from mock_vws import MockVWS
 from mock_vws.database import VuforiaDatabase
@@ -132,6 +135,92 @@ def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None:
                 application_metadata=None,
             )
 
+    @staticmethod
+    def test_timeout_raises_on_slow_response(
+        image: io.BytesIO | BinaryIO,
+    ) -> None:
+        """A short timeout raises an error when the server is slow."""
+        simulated_slow_threshold = 0.5
+
+        with MockVWS() as mock:
+            database = VuforiaDatabase()
+            mock.add_database(database=database)
+            vws_client = VWS(
+                server_access_key=database.server_access_key,
+                server_secret_key=database.server_secret_key,
+                request_timeout_seconds=0.1,
+            )
+
+            original_request = requests.request
+
+            def slow_request(
+                *args: object,
+                **kwargs: float | None,
+            ) -> requests.Response:
+                """Simulate a slow server response."""
+                timeout = kwargs.get("timeout")
+                if timeout is not None and timeout < simulated_slow_threshold:
+                    time.sleep(0.2)
+                    raise requests.exceptions.Timeout
+                return original_request(*args, **kwargs)  # type: ignore[arg-type]
+
+            with (
+                patch.object(
+                    requests,
+                    "request",
+                    side_effect=slow_request,
+                ),
+                pytest.raises(requests.exceptions.Timeout),
+            ):
+                vws_client.add_target(
+                    name="x",
+                    width=1,
+                    image=image,
+                    active_flag=True,
+                    application_metadata=None,
+                )
+
+    @staticmethod
+    def test_longer_timeout_succeeds(image: io.BytesIO | BinaryIO) -> None:
+        """A longer timeout allows slow responses to complete."""
+        simulated_slow_threshold = 0.5
+
+        with MockVWS() as mock:
+            database = VuforiaDatabase()
+            mock.add_database(database=database)
+            vws_client = VWS(
+                server_access_key=database.server_access_key,
+                server_secret_key=database.server_secret_key,
+                request_timeout_seconds=1.0,
+            )
+
+            original_request = requests.request
+
+            def slow_request(
+                *args: object,
+                **kwargs: float | None,
+            ) -> requests.Response:
+                """Simulate a slow server response."""
+                timeout = kwargs.get("timeout")
+                if timeout is not None and timeout < simulated_slow_threshold:
+                    time.sleep(0.2)
+                    raise requests.exceptions.Timeout
+                return original_request(*args, **kwargs)  # type: ignore[arg-type]
+
+            with patch.object(
+                requests,
+                "request",
+                side_effect=slow_request,
+            ):
+                # This should succeed because timeout is 1.0 > 0.5
+                vws_client.add_target(
+                    name="x",
+                    width=1,
+                    image=image,
+                    active_flag=True,
+                    application_metadata=None,
+                )
+
 
 class TestCustomBaseVWSURL:
     """Tests for using a custom base VWS URL."""