From 13376509e7e766b0953355279b65ddaefdfdf0f8 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Mon, 2 Feb 2026 11:40:32 +0000 Subject: [PATCH 01/17] Add custom request timeout support for VWS and CloudRecoService Allow users to configure request timeouts instead of using a hardcoded 30-second default. Both VWS and CloudRecoService now accept an optional request_timeout_seconds parameter during initialization, with proper defaults and documentation. Co-Authored-By: Claude Haiku 4.5 --- src/vws/query.py | 7 +++++-- src/vws/vws.py | 10 ++++++++-- tests/test_query.py | 48 +++++++++++++++++++++++++++++++++++++++++++++ tests/test_vws.py | 41 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 102 insertions(+), 4 deletions(-) diff --git a/src/vws/query.py b/src/vws/query.py index 6c2dc481e..b9d42d4c8 100644 --- a/src/vws/query.py +++ b/src/vws/query.py @@ -49,16 +49,20 @@ def __init__( client_access_key: str, client_secret_key: str, base_vwq_url: str = "https://cloudreco.vuforia.com", + request_timeout_seconds: float = 30.0, ) -> None: """ Args: client_access_key: A VWS client access key. client_secret_key: A VWS client secret key. base_vwq_url: The base URL for the VWQ API. + request_timeout_seconds: The timeout in seconds for each HTTP + request made to the Cloud Reco API. """ self._client_access_key = client_access_key self._client_secret_key = client_secret_key self._base_vwq_url = base_vwq_url + self.request_timeout_seconds = request_timeout_seconds def query( self, @@ -141,8 +145,7 @@ def query( url=urljoin(base=self._base_vwq_url, url=request_path), headers=headers, data=content, - # We should make the timeout customizable. - timeout=30, + timeout=self.request_timeout_seconds, ) response = Response( text=requests_response.text, diff --git a/src/vws/vws.py b/src/vws/vws.py index 9b3d19b4e..82eb7c5bc 100644 --- a/src/vws/vws.py +++ b/src/vws/vws.py @@ -68,6 +68,7 @@ def _target_api_request( data: bytes, request_path: str, base_vws_url: str, + request_timeout_seconds: float, ) -> Response: """Make a request to the Vuforia Target API. @@ -82,6 +83,7 @@ def _target_api_request( request_path: The path to the endpoint which will be used in the request. base_vws_url: The base URL for the VWS API. + request_timeout_seconds: The timeout in seconds for the request. Returns: The response to the request made by `requests`. @@ -111,8 +113,7 @@ def _target_api_request( url=url, headers=headers, data=data, - # We should make the timeout customizable. - timeout=30, + timeout=request_timeout_seconds, ) return Response( @@ -134,16 +135,20 @@ def __init__( server_access_key: str, server_secret_key: str, base_vws_url: str = "https://vws.vuforia.com", + request_timeout_seconds: float = 30.0, ) -> None: """ Args: server_access_key: A VWS server access key. server_secret_key: A VWS server secret key. base_vws_url: The base URL for the VWS API. + request_timeout_seconds: The timeout in seconds for each HTTP + request made to the VWS API. """ self._server_access_key = server_access_key self._server_secret_key = server_secret_key self._base_vws_url = base_vws_url + self.request_timeout_seconds = request_timeout_seconds def make_request( self, @@ -187,6 +192,7 @@ def make_request( data=data, request_path=request_path, base_vws_url=self._base_vws_url, + request_timeout_seconds=self.request_timeout_seconds, ) if ( diff --git a/tests/test_query.py b/tests/test_query.py index a79a712c0..af9d19799 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -42,6 +42,54 @@ def test_match( assert matching_target.target_id == target_id +class TestCustomRequestTimeout: + """Tests for using a custom request timeout.""" + + @staticmethod + def test_default_timeout() -> None: + """By default, the request timeout is 30 seconds.""" + default_timeout_seconds = 30.0 + with MockVWS() as mock: + database = VuforiaDatabase() + mock.add_database(database=database) + cloud_reco_client = CloudRecoService( + client_access_key=database.client_access_key, + client_secret_key=database.client_secret_key, + ) + expected = default_timeout_seconds + assert cloud_reco_client.request_timeout_seconds == expected + + @staticmethod + def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None: + """It is possible to set a custom request timeout.""" + 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, + ) + custom_timeout = 60.5 + cloud_reco_client = CloudRecoService( + client_access_key=database.client_access_key, + client_secret_key=database.client_secret_key, + request_timeout_seconds=custom_timeout, + ) + assert cloud_reco_client.request_timeout_seconds == custom_timeout + + # Verify requests work with the custom timeout + 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) + matches = cloud_reco_client.query(image=image) + assert len(matches) == 1 + + class TestCustomBaseVWQURL: """Tests for using a custom base VWQ URL.""" diff --git a/tests/test_vws.py b/tests/test_vws.py index 00936bde8..25fe7227e 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -92,6 +92,47 @@ def test_add_two_targets( ) +class TestCustomRequestTimeout: + """Tests for using a custom request timeout.""" + + @staticmethod + def test_default_timeout() -> None: + """By default, the request timeout is 30 seconds.""" + default_timeout_seconds = 30.0 + 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, + ) + expected = default_timeout_seconds + assert vws_client.request_timeout_seconds == expected + + @staticmethod + def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None: + """It is possible to set a custom request timeout.""" + with MockVWS() as mock: + database = VuforiaDatabase() + mock.add_database(database=database) + custom_timeout = 60.5 + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, + request_timeout_seconds=custom_timeout, + ) + assert vws_client.request_timeout_seconds == custom_timeout + + # Verify requests work with the custom timeout + 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.""" From 7f0990fc6cfabefd54e88a485dccf39d9b75ff1c Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Mon, 2 Feb 2026 11:54:49 +0000 Subject: [PATCH 02/17] 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 --- tests/test_query.py | 104 ++++++++++++++++++++++++++++++++++++++++++++ tests/test_vws.py | 89 +++++++++++++++++++++++++++++++++++++ 2 files changed, 193 insertions(+) diff --git a/tests/test_query.py b/tests/test_query.py index af9d19799..a3f4c0331 100644 --- a/tests/test_query.py +++ b/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.""" diff --git a/tests/test_vws.py b/tests/test_vws.py index 25fe7227e..03b5818b3 100644 --- a/tests/test_vws.py +++ b/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.""" From 1a930ea8c1c3ac4e1ef130a26592a75815af28ac Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Mon, 2 Feb 2026 11:56:10 +0000 Subject: [PATCH 03/17] Fix mypy errors in timeout tests Use keyword arguments for patch.object and pytest.raises to satisfy mypy's type checking. Co-Authored-By: Claude Opus 4.5 --- tests/test_query.py | 10 +++++----- tests/test_vws.py | 10 +++++----- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/tests/test_query.py b/tests/test_query.py index a3f4c0331..6e8247ab1 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -136,11 +136,11 @@ def slow_request( with ( patch.object( - requests, - "request", + target=requests, + attribute="request", side_effect=slow_request, ), - pytest.raises(requests.exceptions.Timeout), + pytest.raises(expected_exception=requests.exceptions.Timeout), ): cloud_reco_client.query(image=image) @@ -185,8 +185,8 @@ def slow_request( return original_request(*args, **kwargs) # type: ignore[arg-type] with patch.object( - requests, - "request", + target=requests, + attribute="request", side_effect=slow_request, ): # This should succeed because timeout is 1.0 > 0.5 diff --git a/tests/test_vws.py b/tests/test_vws.py index 03b5818b3..0f03a1ccf 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -166,11 +166,11 @@ def slow_request( with ( patch.object( - requests, - "request", + target=requests, + attribute="request", side_effect=slow_request, ), - pytest.raises(requests.exceptions.Timeout), + pytest.raises(expected_exception=requests.exceptions.Timeout), ): vws_client.add_target( name="x", @@ -208,8 +208,8 @@ def slow_request( return original_request(*args, **kwargs) # type: ignore[arg-type] with patch.object( - requests, - "request", + target=requests, + attribute="request", side_effect=slow_request, ): # This should succeed because timeout is 1.0 > 0.5 From a0b073dc1f3e3c6a77be8425fa65cac5a49b43df Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Mon, 2 Feb 2026 11:56:45 +0000 Subject: [PATCH 04/17] Add pyright ignore comments for type checking Co-Authored-By: Claude Opus 4.5 --- tests/test_query.py | 4 ++-- tests/test_vws.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_query.py b/tests/test_query.py index 6e8247ab1..85b2d03ab 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -132,7 +132,7 @@ def slow_request( 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] + return original_request(*args, **kwargs) # type: ignore[arg-type] # pyright: ignore[reportArgumentType] with ( patch.object( @@ -182,7 +182,7 @@ def slow_request( 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] + return original_request(*args, **kwargs) # type: ignore[arg-type] # pyright: ignore[reportArgumentType] with patch.object( target=requests, diff --git a/tests/test_vws.py b/tests/test_vws.py index 0f03a1ccf..bbf157b4b 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -162,7 +162,7 @@ def slow_request( 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] + return original_request(*args, **kwargs) # type: ignore[arg-type] # pyright: ignore[reportArgumentType] with ( patch.object( @@ -205,7 +205,7 @@ def slow_request( 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] + return original_request(*args, **kwargs) # type: ignore[arg-type] # pyright: ignore[reportArgumentType] with patch.object( target=requests, From 8217323d13e44f8fa65f5a7ffe30288feb47a036 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 07:59:41 +0000 Subject: [PATCH 05/17] Accept tuple for request_timeout_seconds to match requests API The timeout parameter now accepts either a float (applied to both connect and read timeouts) or a (connect, read) tuple, matching the requests library's timeout interface. Updated docstrings to reference requests. Co-Authored-By: Claude Opus 4.6 --- src/vws/query.py | 8 +++++--- src/vws/vws.py | 14 +++++++++----- tests/test_query.py | 31 ++++++++++++++++++++++++++++++- tests/test_vws.py | 24 +++++++++++++++++++++++- 4 files changed, 67 insertions(+), 10 deletions(-) diff --git a/src/vws/query.py b/src/vws/query.py index b9d42d4c8..0e4e33d04 100644 --- a/src/vws/query.py +++ b/src/vws/query.py @@ -49,15 +49,17 @@ def __init__( client_access_key: str, client_secret_key: str, base_vwq_url: str = "https://cloudreco.vuforia.com", - request_timeout_seconds: float = 30.0, + request_timeout_seconds: float | tuple[float, float] = 30.0, ) -> None: """ Args: client_access_key: A VWS client access key. client_secret_key: A VWS client secret key. base_vwq_url: The base URL for the VWQ API. - request_timeout_seconds: The timeout in seconds for each HTTP - request made to the Cloud Reco API. + request_timeout_seconds: The timeout for each HTTP request, as + used by ``requests.request``. This can be a float to set + both the connect and read timeouts, or a (connect, read) + tuple. """ self._client_access_key = client_access_key self._client_secret_key = client_secret_key diff --git a/src/vws/vws.py b/src/vws/vws.py index 82eb7c5bc..dfb868cff 100644 --- a/src/vws/vws.py +++ b/src/vws/vws.py @@ -68,7 +68,7 @@ def _target_api_request( data: bytes, request_path: str, base_vws_url: str, - request_timeout_seconds: float, + request_timeout_seconds: float | tuple[float, float], ) -> Response: """Make a request to the Vuforia Target API. @@ -83,7 +83,9 @@ def _target_api_request( request_path: The path to the endpoint which will be used in the request. base_vws_url: The base URL for the VWS API. - request_timeout_seconds: The timeout in seconds for the request. + request_timeout_seconds: The timeout for the request, as used by + ``requests.request``. This can be a float to set both the + connect and read timeouts, or a (connect, read) tuple. Returns: The response to the request made by `requests`. @@ -135,15 +137,17 @@ def __init__( server_access_key: str, server_secret_key: str, base_vws_url: str = "https://vws.vuforia.com", - request_timeout_seconds: float = 30.0, + request_timeout_seconds: float | tuple[float, float] = 30.0, ) -> None: """ Args: server_access_key: A VWS server access key. server_secret_key: A VWS server secret key. base_vws_url: The base URL for the VWS API. - request_timeout_seconds: The timeout in seconds for each HTTP - request made to the VWS API. + request_timeout_seconds: The timeout for each HTTP request, as + used by ``requests.request``. This can be a float to set + both the connect and read timeouts, or a (connect, read) + tuple. """ self._server_access_key = server_access_key self._server_secret_key = server_secret_key diff --git a/tests/test_query.py b/tests/test_query.py index 85b2d03ab..d1e13f2cf 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -65,7 +65,7 @@ def test_default_timeout() -> None: @staticmethod def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None: - """It is possible to set a custom request timeout.""" + """It is possible to set a custom request timeout as a float.""" with MockVWS() as mock: database = VuforiaDatabase() mock.add_database(database=database) @@ -93,6 +93,35 @@ def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None: matches = cloud_reco_client.query(image=image) assert len(matches) == 1 + @staticmethod + def test_custom_timeout_tuple(image: io.BytesIO | BinaryIO) -> None: + """It is possible to set separate connect and read timeouts.""" + 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, + ) + custom_timeout = (5.0, 30.0) + cloud_reco_client = CloudRecoService( + client_access_key=database.client_access_key, + client_secret_key=database.client_secret_key, + request_timeout_seconds=custom_timeout, + ) + assert cloud_reco_client.request_timeout_seconds == custom_timeout + + 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) + matches = cloud_reco_client.query(image=image) + assert len(matches) == 1 + @staticmethod def test_timeout_raises_on_slow_response( image: io.BytesIO | BinaryIO, diff --git a/tests/test_vws.py b/tests/test_vws.py index bbf157b4b..2350ca365 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -114,7 +114,7 @@ def test_default_timeout() -> None: @staticmethod def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None: - """It is possible to set a custom request timeout.""" + """It is possible to set a custom request timeout as a float.""" with MockVWS() as mock: database = VuforiaDatabase() mock.add_database(database=database) @@ -135,6 +135,28 @@ def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None: application_metadata=None, ) + @staticmethod + def test_custom_timeout_tuple(image: io.BytesIO | BinaryIO) -> None: + """It is possible to set separate connect and read timeouts.""" + with MockVWS() as mock: + database = VuforiaDatabase() + mock.add_database(database=database) + custom_timeout = (5.0, 30.0) + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, + request_timeout_seconds=custom_timeout, + ) + assert vws_client.request_timeout_seconds == custom_timeout + + vws_client.add_target( + name="x", + width=1, + image=image, + active_flag=True, + application_metadata=None, + ) + @staticmethod def test_timeout_raises_on_slow_response( image: io.BytesIO | BinaryIO, From 1b2001ecdf413e236a9ab625df3f4a23324f7eee Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 08:05:28 +0000 Subject: [PATCH 06/17] Fix integer timeouts failing at request time _target_api_request used plain @beartype which rejects int for float, while VWS.__init__ used is_pep484_tower=True which accepts it. This caused integer timeouts to pass construction but fail on first request. Enable is_pep484_tower on _target_api_request and CloudRecoService too. Co-Authored-By: Claude Opus 4.6 --- src/vws/query.py | 4 ++-- src/vws/vws.py | 2 +- tests/test_query.py | 29 +++++++++++++++++++++++++++++ tests/test_vws.py | 22 ++++++++++++++++++++++ 4 files changed, 54 insertions(+), 3 deletions(-) diff --git a/src/vws/query.py b/src/vws/query.py index 0e4e33d04..db1945b11 100644 --- a/src/vws/query.py +++ b/src/vws/query.py @@ -8,7 +8,7 @@ from urllib.parse import urljoin import requests -from beartype import beartype +from beartype import BeartypeConf, beartype from urllib3.filepost import encode_multipart_formdata from vws_auth_tools import authorization_header, rfc_1123_date @@ -40,7 +40,7 @@ def _get_image_data(image: _ImageType) -> bytes: return image_data -@beartype +@beartype(conf=BeartypeConf(is_pep484_tower=True)) class CloudRecoService: """An interface to the Vuforia Cloud Recognition Web APIs.""" diff --git a/src/vws/vws.py b/src/vws/vws.py index dfb868cff..cdff3f906 100644 --- a/src/vws/vws.py +++ b/src/vws/vws.py @@ -58,7 +58,7 @@ def _get_image_data(image: _ImageType) -> bytes: return image_data -@beartype +@beartype(conf=BeartypeConf(is_pep484_tower=True)) def _target_api_request( *, content_type: str, diff --git a/tests/test_query.py b/tests/test_query.py index d1e13f2cf..5516a0ea4 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -93,6 +93,35 @@ def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None: matches = cloud_reco_client.query(image=image) assert len(matches) == 1 + @staticmethod + def test_custom_timeout_int(image: io.BytesIO | BinaryIO) -> None: + """It is possible to set a custom request timeout as an int.""" + 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, + ) + custom_timeout = 60 + cloud_reco_client = CloudRecoService( + client_access_key=database.client_access_key, + client_secret_key=database.client_secret_key, + request_timeout_seconds=custom_timeout, + ) + assert cloud_reco_client.request_timeout_seconds == custom_timeout + + 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) + matches = cloud_reco_client.query(image=image) + assert len(matches) == 1 + @staticmethod def test_custom_timeout_tuple(image: io.BytesIO | BinaryIO) -> None: """It is possible to set separate connect and read timeouts.""" diff --git a/tests/test_vws.py b/tests/test_vws.py index 2350ca365..db3b88b5d 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -135,6 +135,28 @@ def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None: application_metadata=None, ) + @staticmethod + def test_custom_timeout_int(image: io.BytesIO | BinaryIO) -> None: + """It is possible to set a custom request timeout as an int.""" + with MockVWS() as mock: + database = VuforiaDatabase() + mock.add_database(database=database) + custom_timeout = 60 + vws_client = VWS( + server_access_key=database.server_access_key, + server_secret_key=database.server_secret_key, + request_timeout_seconds=custom_timeout, + ) + assert vws_client.request_timeout_seconds == custom_timeout + + vws_client.add_target( + name="x", + width=1, + image=image, + active_flag=True, + application_metadata=None, + ) + @staticmethod def test_custom_timeout_tuple(image: io.BytesIO | BinaryIO) -> None: """It is possible to set separate connect and read timeouts.""" From 7123758816b61dce6b9f132d8625c8a9f31f0266 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 08:18:06 +0000 Subject: [PATCH 07/17] Fix CI failures in timeout tests - Remove branching and time.sleep from slow_request mocks, fixing coverage gaps (partial branches) and Windows CI timeouts - Remove test_longer_timeout_succeeds (redundant with test_custom_timeout) - Assert the timeout value is passed correctly instead of branching - Remove unused time import and unused *args parameter Co-Authored-By: Claude Opus 4.6 --- tests/test_query.py | 66 ++++----------------------------------------- tests/test_vws.py | 59 ++++------------------------------------ 2 files changed, 10 insertions(+), 115 deletions(-) diff --git a/tests/test_query.py b/tests/test_query.py index 5516a0ea4..43af56711 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -1,7 +1,6 @@ """Tests for the ``CloudRecoService`` querying functionality.""" import io -import time import uuid from typing import BinaryIO from unittest.mock import patch @@ -156,6 +155,7 @@ def test_timeout_raises_on_slow_response( image: io.BytesIO | BinaryIO, ) -> None: """A short timeout raises an error when the server is slow.""" + custom_timeout = 0.1 with MockVWS() as mock: database = VuforiaDatabase() mock.add_database(database=database) @@ -166,7 +166,7 @@ def test_timeout_raises_on_slow_response( cloud_reco_client = CloudRecoService( client_access_key=database.client_access_key, client_secret_key=database.client_secret_key, - request_timeout_seconds=0.1, + request_timeout_seconds=custom_timeout, ) target_id = vws_client.add_target( @@ -178,19 +178,12 @@ def test_timeout_raises_on_slow_response( ) 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] # pyright: ignore[reportArgumentType] + """Simulate a server that is too slow to respond.""" + assert kwargs["timeout"] == custom_timeout + raise requests.exceptions.Timeout with ( patch.object( @@ -202,55 +195,6 @@ def slow_request( ): 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] # pyright: ignore[reportArgumentType] - - with patch.object( - target=requests, - attribute="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.""" diff --git a/tests/test_vws.py b/tests/test_vws.py index db3b88b5d..e141b041c 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -4,7 +4,6 @@ import datetime import io import secrets -import time import uuid from typing import BinaryIO from unittest.mock import patch @@ -184,29 +183,22 @@ 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 - + custom_timeout = 0.1 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, + request_timeout_seconds=custom_timeout, ) - 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] # pyright: ignore[reportArgumentType] + """Simulate a server that is too slow to respond.""" + assert kwargs["timeout"] == custom_timeout + raise requests.exceptions.Timeout with ( patch.object( @@ -224,47 +216,6 @@ def slow_request( 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] # pyright: ignore[reportArgumentType] - - with patch.object( - target=requests, - attribute="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.""" From 5d0a72941fdde8e783e56d1ff9e3281ffa785c73 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 08:59:13 +0000 Subject: [PATCH 08/17] Use vws-python-mock response_delay_seconds for timeout tests Upgrade vws-python-mock to 2026.2.15 which adds response_delay_seconds parameter. Replace unittest.mock.patch-based timeout simulation with MockVWS(response_delay_seconds=0.5), testing the real timeout mechanism end-to-end. Co-Authored-By: Claude Opus 4.6 --- pyproject.toml | 2 +- tests/test_query.py | 22 +++------------------- tests/test_vws.py | 22 +++------------------- 3 files changed, 7 insertions(+), 39 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 9f1747298..68f237322 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -77,7 +77,7 @@ optional-dependencies.dev = [ "ty==0.0.14", "types-requests==2.32.4.20260107", "vulture==2.14", - "vws-python-mock==2025.3.10.1", + "vws-python-mock==2026.2.15", "vws-test-fixtures==2023.3.5", "yamlfix==1.19.1", "zizmor==1.22.0", diff --git a/tests/test_query.py b/tests/test_query.py index 43af56711..29a9b3b4e 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -3,7 +3,6 @@ import io import uuid from typing import BinaryIO -from unittest.mock import patch import pytest import requests @@ -155,8 +154,7 @@ def test_timeout_raises_on_slow_response( image: io.BytesIO | BinaryIO, ) -> None: """A short timeout raises an error when the server is slow.""" - custom_timeout = 0.1 - with MockVWS() as mock: + with MockVWS(response_delay_seconds=0.5) as mock: database = VuforiaDatabase() mock.add_database(database=database) vws_client = VWS( @@ -166,7 +164,7 @@ def test_timeout_raises_on_slow_response( cloud_reco_client = CloudRecoService( client_access_key=database.client_access_key, client_secret_key=database.client_secret_key, - request_timeout_seconds=custom_timeout, + request_timeout_seconds=0.1, ) target_id = vws_client.add_target( @@ -178,21 +176,7 @@ def test_timeout_raises_on_slow_response( ) vws_client.wait_for_target_processed(target_id=target_id) - def slow_request( - **kwargs: float | None, - ) -> requests.Response: - """Simulate a server that is too slow to respond.""" - assert kwargs["timeout"] == custom_timeout - raise requests.exceptions.Timeout - - with ( - patch.object( - target=requests, - attribute="request", - side_effect=slow_request, - ), - pytest.raises(expected_exception=requests.exceptions.Timeout), - ): + with pytest.raises(expected_exception=requests.exceptions.Timeout): cloud_reco_client.query(image=image) diff --git a/tests/test_vws.py b/tests/test_vws.py index e141b041c..5d2a58f89 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -6,7 +6,6 @@ import secrets import uuid from typing import BinaryIO -from unittest.mock import patch import pytest import requests @@ -183,31 +182,16 @@ def test_timeout_raises_on_slow_response( image: io.BytesIO | BinaryIO, ) -> None: """A short timeout raises an error when the server is slow.""" - custom_timeout = 0.1 - with MockVWS() as mock: + with MockVWS(response_delay_seconds=0.5) 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=custom_timeout, + request_timeout_seconds=0.1, ) - def slow_request( - **kwargs: float | None, - ) -> requests.Response: - """Simulate a server that is too slow to respond.""" - assert kwargs["timeout"] == custom_timeout - raise requests.exceptions.Timeout - - with ( - patch.object( - target=requests, - attribute="request", - side_effect=slow_request, - ), - pytest.raises(expected_exception=requests.exceptions.Timeout), - ): + with pytest.raises(expected_exception=requests.exceptions.Timeout): vws_client.add_target( name="x", width=1, From 90c8285c54c3f74bc161d2a46437b088fa529970 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 09:01:21 +0000 Subject: [PATCH 09/17] Add changelog entry for custom request timeout support Co-Authored-By: Claude Opus 4.6 --- CHANGELOG.rst | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 5527f9da1..174256880 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -4,6 +4,8 @@ Changelog Next ---- +* Add ``request_timeout_seconds`` parameter to ``VWS`` and ``CloudRecoService``, allowing customization of the request timeout. This accepts a float or a ``(connect, read)`` tuple, matching the ``requests`` library's timeout interface. The default remains 30 seconds. + 2025.03.10.1 ------------ From 88e1d775cf54e3d65280334bfbd4b3565de7b76e Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 09:12:18 +0000 Subject: [PATCH 10/17] Parametrize custom timeout tests Co-Authored-By: Claude Opus 4.6 --- tests/test_query.py | 74 +++++++-------------------------------------- tests/test_vws.py | 60 +++++++----------------------------- 2 files changed, 22 insertions(+), 112 deletions(-) diff --git a/tests/test_query.py b/tests/test_query.py index 29a9b3b4e..dafe5a738 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -62,67 +62,15 @@ def test_default_timeout() -> None: assert cloud_reco_client.request_timeout_seconds == expected @staticmethod - def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None: - """It is possible to set a custom request timeout as a float.""" - 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, - ) - custom_timeout = 60.5 - cloud_reco_client = CloudRecoService( - client_access_key=database.client_access_key, - client_secret_key=database.client_secret_key, - request_timeout_seconds=custom_timeout, - ) - assert cloud_reco_client.request_timeout_seconds == custom_timeout - - # Verify requests work with the custom timeout - 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) - matches = cloud_reco_client.query(image=image) - assert len(matches) == 1 - - @staticmethod - def test_custom_timeout_int(image: io.BytesIO | BinaryIO) -> None: - """It is possible to set a custom request timeout as an int.""" - 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, - ) - custom_timeout = 60 - cloud_reco_client = CloudRecoService( - client_access_key=database.client_access_key, - client_secret_key=database.client_secret_key, - request_timeout_seconds=custom_timeout, - ) - assert cloud_reco_client.request_timeout_seconds == custom_timeout - - 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) - matches = cloud_reco_client.query(image=image) - assert len(matches) == 1 - - @staticmethod - def test_custom_timeout_tuple(image: io.BytesIO | BinaryIO) -> None: - """It is possible to set separate connect and read timeouts.""" + @pytest.mark.parametrize( + argnames="custom_timeout", + argvalues=[60.5, 60, (5.0, 30.0)], + ) + def test_custom_timeout( + image: io.BytesIO | BinaryIO, + custom_timeout: int | float | tuple[float, float], # noqa: PYI041 + ) -> None: + """It is possible to set a custom request timeout.""" with MockVWS() as mock: database = VuforiaDatabase() mock.add_database(database=database) @@ -130,13 +78,13 @@ def test_custom_timeout_tuple(image: io.BytesIO | BinaryIO) -> None: server_access_key=database.server_access_key, server_secret_key=database.server_secret_key, ) - custom_timeout = (5.0, 30.0) cloud_reco_client = CloudRecoService( client_access_key=database.client_access_key, client_secret_key=database.client_secret_key, request_timeout_seconds=custom_timeout, ) - assert cloud_reco_client.request_timeout_seconds == custom_timeout + expected = custom_timeout + assert cloud_reco_client.request_timeout_seconds == expected target_id = vws_client.add_target( name="x", diff --git a/tests/test_vws.py b/tests/test_vws.py index 5d2a58f89..9fe63dc21 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -111,63 +111,25 @@ def test_default_timeout() -> None: assert vws_client.request_timeout_seconds == expected @staticmethod - def test_custom_timeout(image: io.BytesIO | BinaryIO) -> None: - """It is possible to set a custom request timeout as a float.""" - with MockVWS() as mock: - database = VuforiaDatabase() - mock.add_database(database=database) - custom_timeout = 60.5 - vws_client = VWS( - server_access_key=database.server_access_key, - server_secret_key=database.server_secret_key, - request_timeout_seconds=custom_timeout, - ) - assert vws_client.request_timeout_seconds == custom_timeout - - # Verify requests work with the custom timeout - vws_client.add_target( - name="x", - width=1, - image=image, - active_flag=True, - application_metadata=None, - ) - - @staticmethod - def test_custom_timeout_int(image: io.BytesIO | BinaryIO) -> None: - """It is possible to set a custom request timeout as an int.""" - with MockVWS() as mock: - database = VuforiaDatabase() - mock.add_database(database=database) - custom_timeout = 60 - vws_client = VWS( - server_access_key=database.server_access_key, - server_secret_key=database.server_secret_key, - request_timeout_seconds=custom_timeout, - ) - assert vws_client.request_timeout_seconds == custom_timeout - - vws_client.add_target( - name="x", - width=1, - image=image, - active_flag=True, - application_metadata=None, - ) - - @staticmethod - def test_custom_timeout_tuple(image: io.BytesIO | BinaryIO) -> None: - """It is possible to set separate connect and read timeouts.""" + @pytest.mark.parametrize( + argnames="custom_timeout", + argvalues=[60.5, 60, (5.0, 30.0)], + ) + def test_custom_timeout( + image: io.BytesIO | BinaryIO, + custom_timeout: int | float | tuple[float, float], # noqa: PYI041 + ) -> None: + """It is possible to set a custom request timeout.""" with MockVWS() as mock: database = VuforiaDatabase() mock.add_database(database=database) - custom_timeout = (5.0, 30.0) vws_client = VWS( server_access_key=database.server_access_key, server_secret_key=database.server_secret_key, request_timeout_seconds=custom_timeout, ) - assert vws_client.request_timeout_seconds == custom_timeout + expected = custom_timeout + assert vws_client.request_timeout_seconds == expected vws_client.add_target( name="x", From 605d27c4c81ca670e23b25ff2bcdded8cf20db14 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 10:42:29 +0000 Subject: [PATCH 11/17] Update timeout behavior and default-timeout tests --- src/vws/query.py | 4 ++-- src/vws/vws.py | 4 ++-- tests/test_query.py | 33 +++++++++++++++++++++++++++------ tests/test_vws.py | 43 +++++++++++++++++++++++++++++++++++++------ 4 files changed, 68 insertions(+), 16 deletions(-) diff --git a/src/vws/query.py b/src/vws/query.py index db1945b11..dd36e8f21 100644 --- a/src/vws/query.py +++ b/src/vws/query.py @@ -64,7 +64,7 @@ def __init__( self._client_access_key = client_access_key self._client_secret_key = client_secret_key self._base_vwq_url = base_vwq_url - self.request_timeout_seconds = request_timeout_seconds + self._request_timeout_seconds = request_timeout_seconds def query( self, @@ -147,7 +147,7 @@ def query( url=urljoin(base=self._base_vwq_url, url=request_path), headers=headers, data=content, - timeout=self.request_timeout_seconds, + timeout=self._request_timeout_seconds, ) response = Response( text=requests_response.text, diff --git a/src/vws/vws.py b/src/vws/vws.py index cdff3f906..4cb75ca32 100644 --- a/src/vws/vws.py +++ b/src/vws/vws.py @@ -152,7 +152,7 @@ def __init__( self._server_access_key = server_access_key self._server_secret_key = server_secret_key self._base_vws_url = base_vws_url - self.request_timeout_seconds = request_timeout_seconds + self._request_timeout_seconds = request_timeout_seconds def make_request( self, @@ -196,7 +196,7 @@ def make_request( data=data, request_path=request_path, base_vws_url=self._base_vws_url, - request_timeout_seconds=self.request_timeout_seconds, + request_timeout_seconds=self._request_timeout_seconds, ) if ( diff --git a/tests/test_query.py b/tests/test_query.py index dafe5a738..fd014cadf 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -6,6 +6,7 @@ import pytest import requests +from freezegun import freeze_time from mock_vws import MockVWS from mock_vws.database import VuforiaDatabase @@ -48,18 +49,38 @@ class TestCustomRequestTimeout: """Tests for using a custom request timeout.""" @staticmethod - def test_default_timeout() -> None: - """By default, the request timeout is 30 seconds.""" - default_timeout_seconds = 30.0 - with MockVWS() as mock: + @pytest.mark.parametrize( + argnames=("response_delay_seconds", "expect_timeout"), + argvalues=[(29, False), (31, True)], + ) + def test_default_timeout( + image: io.BytesIO | BinaryIO, + *, + response_delay_seconds: int, + expect_timeout: bool, + ) -> None: + """At 29 seconds there is no error; at 31 seconds there is a + timeout. + """ + with ( + freeze_time(auto_tick_seconds=1), + MockVWS(response_delay_seconds=response_delay_seconds) as mock, + ): database = VuforiaDatabase() mock.add_database(database=database) cloud_reco_client = CloudRecoService( client_access_key=database.client_access_key, client_secret_key=database.client_secret_key, ) - expected = default_timeout_seconds - assert cloud_reco_client.request_timeout_seconds == expected + + if expect_timeout: + with pytest.raises( + expected_exception=requests.exceptions.Timeout, + ): + cloud_reco_client.query(image=image) + else: + matches = cloud_reco_client.query(image=image) + assert matches == [] @staticmethod @pytest.mark.parametrize( diff --git a/tests/test_vws.py b/tests/test_vws.py index 9fe63dc21..03dc796c9 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -97,18 +97,49 @@ class TestCustomRequestTimeout: """Tests for using a custom request timeout.""" @staticmethod - def test_default_timeout() -> None: - """By default, the request timeout is 30 seconds.""" - default_timeout_seconds = 30.0 - with MockVWS() as mock: + @pytest.mark.parametrize( + argnames=("response_delay_seconds", "expect_timeout"), + argvalues=[(29, False), (31, True)], + ) + def test_default_timeout( + image: io.BytesIO | BinaryIO, + *, + response_delay_seconds: int, + expect_timeout: bool, + ) -> None: + """At 29 seconds there is no error; at 31 seconds there is a + timeout. + """ + with ( + freeze_time(auto_tick_seconds=1), + MockVWS(response_delay_seconds=response_delay_seconds) 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, ) - expected = default_timeout_seconds - assert vws_client.request_timeout_seconds == expected + + if expect_timeout: + with pytest.raises( + expected_exception=requests.exceptions.Timeout, + ): + vws_client.add_target( + name="x", + width=1, + image=image, + active_flag=True, + application_metadata=None, + ) + else: + vws_client.add_target( + name="x", + width=1, + image=image, + active_flag=True, + application_metadata=None, + ) @staticmethod @pytest.mark.parametrize( From 2af9315334564b8685a13d3180bd67682ed258dd Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 10:43:05 +0000 Subject: [PATCH 12/17] Remove mypy-invalid timeout attribute assertions --- tests/test_query.py | 2 -- tests/test_vws.py | 2 -- 2 files changed, 4 deletions(-) diff --git a/tests/test_query.py b/tests/test_query.py index fd014cadf..32c07989b 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -104,8 +104,6 @@ def test_custom_timeout( client_secret_key=database.client_secret_key, request_timeout_seconds=custom_timeout, ) - expected = custom_timeout - assert cloud_reco_client.request_timeout_seconds == expected target_id = vws_client.add_target( name="x", diff --git a/tests/test_vws.py b/tests/test_vws.py index 03dc796c9..bb71a0236 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -159,8 +159,6 @@ def test_custom_timeout( server_secret_key=database.server_secret_key, request_timeout_seconds=custom_timeout, ) - expected = custom_timeout - assert vws_client.request_timeout_seconds == expected vws_client.add_target( name="x", From 1c987da7c138b886d92fd49a50173338cc3eb422 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 10:48:13 +0000 Subject: [PATCH 13/17] Fix pylint empty-list comparison in timeout test --- tests/test_query.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/test_query.py b/tests/test_query.py index 32c07989b..3c78e7f77 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -80,16 +80,16 @@ def test_default_timeout( cloud_reco_client.query(image=image) else: matches = cloud_reco_client.query(image=image) - assert matches == [] + assert not matches @staticmethod @pytest.mark.parametrize( argnames="custom_timeout", - argvalues=[60.5, 60, (5.0, 30.0)], + argvalues=[60.5, 60.0, (5.0, 30.0)], ) def test_custom_timeout( image: io.BytesIO | BinaryIO, - custom_timeout: int | float | tuple[float, float], # noqa: PYI041 + custom_timeout: float | tuple[float, float], ) -> None: """It is possible to set a custom request timeout.""" with MockVWS() as mock: From 80c214b77f2ebaee417c45b63ef974ce420cdf95 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 10:51:47 +0000 Subject: [PATCH 14/17] Align timeout test parameter types --- tests/test_vws.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_vws.py b/tests/test_vws.py index bb71a0236..be4dd7a7e 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -144,11 +144,11 @@ def test_default_timeout( @staticmethod @pytest.mark.parametrize( argnames="custom_timeout", - argvalues=[60.5, 60, (5.0, 30.0)], + argvalues=[60.5, 60.0, (5.0, 30.0)], ) def test_custom_timeout( image: io.BytesIO | BinaryIO, - custom_timeout: int | float | tuple[float, float], # noqa: PYI041 + custom_timeout: float | tuple[float, float], ) -> None: """It is possible to set a custom request timeout.""" with MockVWS() as mock: From bdaacd891b766e5a5fafd3cef2c9e4b22e4f2ce8 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 11:59:12 +0000 Subject: [PATCH 15/17] Bump mock-vws and use sleep_fn in timeout tests --- pyproject.toml | 2 +- tests/test_query.py | 44 ++++++++------------------------------------ tests/test_vws.py | 36 +++++++----------------------------- 3 files changed, 16 insertions(+), 66 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 2b0fb465f..2f5a29b78 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -77,7 +77,7 @@ optional-dependencies.dev = [ "ty==0.0.16", "types-requests==2.32.4.20260107", "vulture==2.14", - "vws-python-mock==2026.2.15", + "vws-python-mock==2026.2.15.4", "vws-test-fixtures==2023.3.5", "yamlfix==1.19.1", "zizmor==1.22.0", diff --git a/tests/test_query.py b/tests/test_query.py index 3c78e7f77..4e7ccb86d 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -1,5 +1,6 @@ """Tests for the ``CloudRecoService`` querying functionality.""" +import datetime import io import uuid from typing import BinaryIO @@ -63,8 +64,13 @@ def test_default_timeout( timeout. """ with ( - freeze_time(auto_tick_seconds=1), - MockVWS(response_delay_seconds=response_delay_seconds) as mock, + freeze_time() as frozen_datetime, + MockVWS( + response_delay_seconds=response_delay_seconds, + sleep_fn=lambda seconds: frozen_datetime.tick( + delta=datetime.timedelta(seconds=seconds), + ), + ) as mock, ): database = VuforiaDatabase() mock.add_database(database=database) @@ -82,40 +88,6 @@ def test_default_timeout( matches = cloud_reco_client.query(image=image) assert not matches - @staticmethod - @pytest.mark.parametrize( - argnames="custom_timeout", - argvalues=[60.5, 60.0, (5.0, 30.0)], - ) - def test_custom_timeout( - image: io.BytesIO | BinaryIO, - custom_timeout: float | tuple[float, float], - ) -> None: - """It is possible to set a custom request timeout.""" - 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=custom_timeout, - ) - - 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) - matches = cloud_reco_client.query(image=image) - assert len(matches) == 1 - @staticmethod def test_timeout_raises_on_slow_response( image: io.BytesIO | BinaryIO, diff --git a/tests/test_vws.py b/tests/test_vws.py index be4dd7a7e..e93c35063 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -111,8 +111,13 @@ def test_default_timeout( timeout. """ with ( - freeze_time(auto_tick_seconds=1), - MockVWS(response_delay_seconds=response_delay_seconds) as mock, + freeze_time() as frozen_datetime, + MockVWS( + response_delay_seconds=response_delay_seconds, + sleep_fn=lambda seconds: frozen_datetime.tick( + delta=datetime.timedelta(seconds=seconds), + ), + ) as mock, ): database = VuforiaDatabase() mock.add_database(database=database) @@ -141,33 +146,6 @@ def test_default_timeout( application_metadata=None, ) - @staticmethod - @pytest.mark.parametrize( - argnames="custom_timeout", - argvalues=[60.5, 60.0, (5.0, 30.0)], - ) - def test_custom_timeout( - image: io.BytesIO | BinaryIO, - custom_timeout: float | tuple[float, float], - ) -> None: - """It is possible to set a custom request timeout.""" - 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=custom_timeout, - ) - - vws_client.add_target( - name="x", - width=1, - image=image, - active_flag=True, - application_metadata=None, - ) - @staticmethod def test_timeout_raises_on_slow_response( image: io.BytesIO | BinaryIO, From ff977797212068c55040d02ae4326992eba2d1b1 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 11:59:50 +0000 Subject: [PATCH 16/17] Fix sleep_fn return type for mypy --- tests/test_query.py | 9 ++++++--- tests/test_vws.py | 9 ++++++--- 2 files changed, 12 insertions(+), 6 deletions(-) diff --git a/tests/test_query.py b/tests/test_query.py index 4e7ccb86d..8c6a73769 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -67,9 +67,12 @@ def test_default_timeout( freeze_time() as frozen_datetime, MockVWS( response_delay_seconds=response_delay_seconds, - sleep_fn=lambda seconds: frozen_datetime.tick( - delta=datetime.timedelta(seconds=seconds), - ), + sleep_fn=lambda seconds: ( + frozen_datetime.tick( + delta=datetime.timedelta(seconds=seconds), + ), + None, + )[1], ) as mock, ): database = VuforiaDatabase() diff --git a/tests/test_vws.py b/tests/test_vws.py index e93c35063..6bb771f14 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -114,9 +114,12 @@ def test_default_timeout( freeze_time() as frozen_datetime, MockVWS( response_delay_seconds=response_delay_seconds, - sleep_fn=lambda seconds: frozen_datetime.tick( - delta=datetime.timedelta(seconds=seconds), - ), + sleep_fn=lambda seconds: ( + frozen_datetime.tick( + delta=datetime.timedelta(seconds=seconds), + ), + None, + )[1], ) as mock, ): database = VuforiaDatabase() From a69677c294e8fd5ce4db907594cf17cba163b034 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 15 Feb 2026 12:32:59 +0000 Subject: [PATCH 17/17] Add fast tests for custom request timeouts --- tests/test_query.py | 55 +++++++++++++++++++++++++++++++++++++ tests/test_vws.py | 66 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 121 insertions(+) diff --git a/tests/test_query.py b/tests/test_query.py index 31284325d..d7b64aff2 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -92,6 +92,61 @@ def test_default_timeout( assert not matches +class TestCustomRequestTimeout: + """Tests for custom request timeout values.""" + + @staticmethod + @pytest.mark.parametrize( + argnames=( + "custom_timeout", + "response_delay_seconds", + "expect_timeout", + ), + argvalues=[ + (0.1, 0.09, False), + (0.1, 0.11, True), + ((5.0, 0.1), 0.09, False), + ((5.0, 0.1), 0.11, True), + ], + ) + def test_custom_timeout( + image: io.BytesIO | BinaryIO, + *, + custom_timeout: float | tuple[float, float], + response_delay_seconds: float, + expect_timeout: bool, + ) -> None: + """Custom timeouts are honored for both float and tuple forms.""" + with ( + freeze_time() as frozen_datetime, + MockVWS( + response_delay_seconds=response_delay_seconds, + sleep_fn=lambda seconds: ( + frozen_datetime.tick( + delta=datetime.timedelta(seconds=seconds), + ), + None, + )[1], + ) as mock, + ): + database = VuforiaDatabase() + mock.add_database(database=database) + cloud_reco_client = CloudRecoService( + client_access_key=database.client_access_key, + client_secret_key=database.client_secret_key, + request_timeout_seconds=custom_timeout, + ) + + if expect_timeout: + with pytest.raises( + expected_exception=requests.exceptions.Timeout, + ): + cloud_reco_client.query(image=image) + else: + matches = cloud_reco_client.query(image=image) + assert not matches + + class TestCustomBaseVWQURL: """Tests for using a custom base VWQ URL.""" diff --git a/tests/test_vws.py b/tests/test_vws.py index 9a1c8ded9..f09906ee5 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -150,6 +150,72 @@ def test_default_timeout( ) +class TestCustomRequestTimeout: + """Tests for custom request timeout values.""" + + @staticmethod + @pytest.mark.parametrize( + argnames=( + "custom_timeout", + "response_delay_seconds", + "expect_timeout", + ), + argvalues=[ + (0.1, 0.09, False), + (0.1, 0.11, True), + ((5.0, 0.1), 0.09, False), + ((5.0, 0.1), 0.11, True), + ], + ) + def test_custom_timeout( + image: io.BytesIO | BinaryIO, + *, + custom_timeout: float | tuple[float, float], + response_delay_seconds: float, + expect_timeout: bool, + ) -> None: + """Custom timeouts are honored for both float and tuple forms.""" + with ( + freeze_time() as frozen_datetime, + MockVWS( + response_delay_seconds=response_delay_seconds, + sleep_fn=lambda seconds: ( + frozen_datetime.tick( + delta=datetime.timedelta(seconds=seconds), + ), + None, + )[1], + ) 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=custom_timeout, + ) + + if expect_timeout: + with pytest.raises( + expected_exception=requests.exceptions.Timeout, + ): + vws_client.add_target( + name="x", + width=1, + image=image, + active_flag=True, + application_metadata=None, + ) + else: + 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."""