diff --git a/tests/test_query.py b/tests/test_query.py index a79a712c0..31284325d 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -1,9 +1,13 @@ """Tests for the ``CloudRecoService`` querying functionality.""" +import datetime import io import uuid from typing import BinaryIO +import pytest +import requests +from freezegun import freeze_time from mock_vws import MockVWS from mock_vws.database import VuforiaDatabase @@ -42,6 +46,52 @@ def test_match( assert matching_target.target_id == target_id +class TestDefaultRequestTimeout: + """Tests for the default request timeout.""" + + @staticmethod + @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() 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, + ) + + 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 00936bde8..9a1c8ded9 100644 --- a/tests/test_vws.py +++ b/tests/test_vws.py @@ -8,6 +8,7 @@ from typing import BinaryIO import pytest +import requests from freezegun import freeze_time from mock_vws import MockVWS from mock_vws.database import VuforiaDatabase @@ -92,6 +93,63 @@ def test_add_two_targets( ) +class TestDefaultRequestTimeout: + """Tests for the default request timeout.""" + + @staticmethod + @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() 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, + ) + + 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."""