Merge pull request #2845 from VWS-Python/adamtheturtle/test-default-timeout

Add tests for the default request timeout

Author: adamtheturtle

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."""
 

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."""