Merge pull request #2915 from VWS-Python/adamtheturtle/flask-response-delay

Add response_delay_seconds to Flask mock

Author: adamtheturtle

CHANGELOG.rst

@@ -4,19 +4,8 @@ Changelog
 Next
 ----
 
-2026.02.15.2
-------------
-
-
-2026.02.15.1
-------------
-
-
-2026.02.15
-----------
-
-
 - Add ``response_delay_seconds`` parameter to ``MockVWS`` for simulating slow server responses and testing timeout handling.
+- Add ``response_delay_seconds`` setting to the Flask mock (``VWSSettings`` and ``VWQSettings``) for simulating slow server responses.
 
 2025.03.10.1
 ------------

pyproject.toml

@@ -446,6 +446,7 @@ ignore_decorators = [
     "@pytest.fixture",
     # Flask
     "@*APP.route",
+    "@*APP.after_request",
     "@*APP.before_request",
     "@*APP.errorhandler",
 ]

src/mock_vws/_flask_server/vwq.py

@@ -5,6 +5,7 @@
 """
 
 import email.utils
+import time
 from enum import StrEnum, auto
 from http import HTTPMethod, HTTPStatus
 
@@ -58,6 +59,7 @@ class VWQSettings(BaseSettings):
     query_image_matcher: _ImageMatcherChoice = (
         _ImageMatcherChoice.STRUCTURAL_SIMILARITY
     )
+    response_delay_seconds: float = 0.0
 
 
 @beartype
@@ -101,6 +103,15 @@ def set_terminate_wsgi_input() -> None:
         request.environ["wsgi.input_terminated"] = True
 
 
+@CLOUDRECO_FLASK_APP.after_request
+@beartype
+def add_response_delay(response: Response) -> Response:
+    """Add a delay to each response."""
+    settings = VWQSettings.model_validate(obj={})
+    time.sleep(settings.response_delay_seconds)
+    return response
+
+
 @CLOUDRECO_FLASK_APP.errorhandler(code_or_exception=ValidatorError)
 def handle_exceptions(exc: ValidatorError) -> Response:
     """Return the error response associated with the given exception."""

src/mock_vws/_flask_server/vws.py

@@ -8,6 +8,7 @@
 import email.utils
 import json
 import logging
+import time
 import uuid
 from enum import StrEnum, auto
 from http import HTTPMethod, HTTPStatus
@@ -73,6 +74,7 @@ class VWSSettings(BaseSettings):
     duplicates_image_matcher: _ImageMatcherChoice = (
         _ImageMatcherChoice.STRUCTURAL_SIMILARITY
     )
+    response_delay_seconds: float = 0.0
 
 
 @beartype
@@ -130,6 +132,15 @@ def validate_request() -> None:
     )
 
 
+@VWS_FLASK_APP.after_request
+@beartype
+def add_response_delay(response: Response) -> Response:
+    """Add a delay to each response."""
+    settings = VWSSettings.model_validate(obj={})
+    time.sleep(settings.response_delay_seconds)
+    return response
+
+
 @VWS_FLASK_APP.errorhandler(code_or_exception=ValidatorError)
 def handle_exceptions(exc: ValidatorError) -> Response:
     """Return the error response associated with the given exception."""

tests/mock_vws/test_flask_app_usage.py

@@ -1,7 +1,9 @@
 """Tests for the usage of the mock Flask application."""
 
+import email.utils
 import io
 import json
+import time
 import uuid
 from collections.abc import Iterator
 from http import HTTPStatus
@@ -605,3 +607,63 @@ def test_random(
         assert lowest_rating >= minimum_rating
         assert highest_rating <= maximum_rating
         assert lowest_rating != highest_rating
+
+
+class TestResponseDelay:
+    """Tests for the response delay feature.
+
+    These tests run through the ``responses`` library, which intercepts
+    requests in-process. Because of this, the client ``timeout`` parameter
+    is not enforced — the delay blocks but never raises
+    ``requests.exceptions.Timeout``. When running the Flask app as a real
+    server (e.g. in Docker), the delay causes a genuinely slow HTTP
+    response and the ``requests`` client will raise ``Timeout`` on its own.
+    """
+
+    DELAY_SECONDS = 0.5
+
+    @staticmethod
+    def _make_request() -> None:
+        """Make a request to the VWS API."""
+        requests.get(
+            url="https://vws.vuforia.com/summary",
+            headers={
+                "Date": email.utils.formatdate(
+                    timeval=None,
+                    localtime=False,
+                    usegmt=True,
+                ),
+                "Authorization": "bad_auth_token",
+            },
+            data=b"",
+            timeout=30,
+        )
+
+    def test_default_no_delay(self) -> None:
+        """By default, there is no response delay."""
+        database = VuforiaDatabase()
+        databases_url = _EXAMPLE_URL_FOR_TARGET_MANAGER + "/databases"
+        requests.post(url=databases_url, json=database.to_dict(), timeout=30)
+
+        start = time.monotonic()
+        self._make_request()
+        elapsed = time.monotonic() - start
+        assert elapsed < self.DELAY_SECONDS
+
+    def test_delay_is_applied(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """When response_delay_seconds is set, the response is delayed."""
+        monkeypatch.setenv(
+            name="RESPONSE_DELAY_SECONDS",
+            value=f"{self.DELAY_SECONDS}",
+        )
+        database = VuforiaDatabase()
+        databases_url = _EXAMPLE_URL_FOR_TARGET_MANAGER + "/databases"
+        requests.post(url=databases_url, json=database.to_dict(), timeout=30)
+
+        start = time.monotonic()
+        self._make_request()
+        elapsed = time.monotonic() - start
+        assert elapsed >= self.DELAY_SECONDS