Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 1 addition & 12 deletions CHANGELOG.rst
Original file line number Diff line number Diff line change
Expand Up @@ -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
------------
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -446,6 +446,7 @@ ignore_decorators = [
"@pytest.fixture",
# Flask
"@*APP.route",
"@*APP.after_request",
"@*APP.before_request",
"@*APP.errorhandler",
]
Expand Down
11 changes: 11 additions & 0 deletions src/mock_vws/_flask_server/vwq.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
"""

import email.utils
import time
from enum import StrEnum, auto
from http import HTTPMethod, HTTPStatus

Expand Down Expand Up @@ -58,6 +59,7 @@ class VWQSettings(BaseSettings):
query_image_matcher: _ImageMatcherChoice = (
_ImageMatcherChoice.STRUCTURAL_SIMILARITY
)
response_delay_seconds: float = 0.0


@beartype
Expand Down Expand Up @@ -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."""
Expand Down
11 changes: 11 additions & 0 deletions src/mock_vws/_flask_server/vws.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -73,6 +74,7 @@ class VWSSettings(BaseSettings):
duplicates_image_matcher: _ImageMatcherChoice = (
_ImageMatcherChoice.STRUCTURAL_SIMILARITY
)
response_delay_seconds: float = 0.0


@beartype
Expand Down Expand Up @@ -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."""
Expand Down
62 changes: 62 additions & 0 deletions tests/mock_vws/test_flask_app_usage.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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