diff --git a/tests/mock_vws/fixtures/vuforia_backends.py b/tests/mock_vws/fixtures/vuforia_backends.py index f936f4103..fae9bce5c 100644 --- a/tests/mock_vws/fixtures/vuforia_backends.py +++ b/tests/mock_vws/fixtures/vuforia_backends.py @@ -14,6 +14,7 @@ import responses from beartype import beartype from pytest_multi_backend import backend_fixture +from requests.exceptions import Timeout as RequestsTimeout from requests_mock_flask import add_flask_app_to_mock from vws import VWS from vws.exceptions.vws_exceptions import ( @@ -50,6 +51,9 @@ retrying_transient_real_backend_failures, ) from tests.mock_vws.utils.recognition_counts import SeedRecognitionCounts +from tests.mock_vws.utils.request_timeout_policy import ( + handling_request_timeouts, +) from tests.mock_vws.utils.retries import ( RETRY_ON_TARGET_STILL_PROCESSING, RETRY_ON_TRANSIENT_VWS_FAILURE, @@ -166,6 +170,13 @@ def _vumark_database( ) +@beartype +def _handle_real_request_timeout(*, url: str, error: RequestsTimeout) -> None: + """Skip a shared request which timed out against the real service.""" + del url, error + pytest.skip(reason="The real Vuforia service timed out.") + + @beartype def _enable_use_real_vuforia( *, @@ -186,7 +197,8 @@ def _enable_use_real_vuforia( _delete_all_targets( database_keys=working_database, base_vws_url=base_vws_url ) - yield + with handling_request_timeouts(handler=_handle_real_request_timeout): + yield @beartype @@ -396,6 +408,7 @@ def _enable_use_real_model_target_vuforia( assert bool(monkeypatch) with ( retrying_transient_real_backend_failures(), + handling_request_timeouts(handler=_handle_real_request_timeout), handling_unexpected_model_target_responses( handler=_handle_real_model_target_response, ), diff --git a/tests/mock_vws/requests_mock/test_request_timeout_policy.py b/tests/mock_vws/requests_mock/test_request_timeout_policy.py new file mode 100644 index 000000000..2e9211798 --- /dev/null +++ b/tests/mock_vws/requests_mock/test_request_timeout_policy.py @@ -0,0 +1,148 @@ +"""Tests for timeout handling supplied by backend adapters.""" + +from collections.abc import Generator + +import pytest +from requests.exceptions import ConnectTimeout, ReadTimeout +from requests.exceptions import Timeout as RequestsTimeout + +from tests.mock_vws.fixtures.vuforia_backends import VuforiaBackend +from tests.mock_vws.utils.request_timeout_policy import ( + handling_request_timeouts, + propagating_request_timeouts, +) + + +@pytest.fixture(name="_model_target_backend") +def fixture_model_target_backend() -> VuforiaBackend: + """Select the real policy without credential lookup or requests.""" + return VuforiaBackend.REAL + + +@pytest.fixture( + name="base_vws_url", + params=[ + "https://vws.vuforia.com", + "https://cloudreco.vuforia.com", + "http://consumer.example", + ], +) +def fixture_base_vws_url(*, request: pytest.FixtureRequest) -> str: + """Supply canonical and custom URLs for controlled timeouts.""" + url = request.param + assert isinstance(url, str) + return url + + +@pytest.fixture( + name="request_timeout", + params=[RequestsTimeout, ConnectTimeout, ReadTimeout], +) +def fixture_request_timeout( + *, request: pytest.FixtureRequest +) -> RequestsTimeout: + """Supply a controlled base or derived timeout exception.""" + timeout_type = request.param + assert isinstance(timeout_type, type) + assert issubclass(timeout_type, RequestsTimeout) + return timeout_type("Controlled timeout") + + +@pytest.fixture(autouse=True) +def fixture_check_timeout_policy_cleanup( + *, request_timeout: RequestsTimeout +) -> Generator[None]: + """Check the default before setup and after backend fixtures finish. + + Yields: + ``None``, while the test and its backend fixtures run. + """ + with ( + pytest.raises(expected_exception=RequestsTimeout), + propagating_request_timeouts(url="https://vws.vuforia.com/targets"), + ): + raise request_timeout + yield + with ( + pytest.raises(expected_exception=RequestsTimeout), + propagating_request_timeouts(url="https://vws.vuforia.com/targets"), + ): + raise request_timeout + + +def test_default_propagates_timeouts( + *, base_vws_url: str, request_timeout: RequestsTimeout +) -> None: + """A host name alone never changes the strict shared timeout + behavior. + """ + with ( + pytest.raises(expected_exception=RequestsTimeout) as exc, + propagating_request_timeouts(url=f"{base_vws_url}/targets"), + ): + raise request_timeout + assert exc.value is request_timeout + + +@pytest.mark.usefixtures("verify_model_target_mock_vuforia") +def test_real_adapter_skips_timeouts( + *, base_vws_url: str, request_timeout: RequestsTimeout +) -> None: + """The real adapter skips a timeout at any supplied URL.""" + with ( + pytest.raises( + expected_exception=pytest.skip.Exception, + match=r"The real Vuforia service timed out\.", + ), + propagating_request_timeouts(url=f"{base_vws_url}/targets"), + ): + raise request_timeout + + +def test_nested_timeout_handlers(*, request_timeout: RequestsTimeout) -> None: + """Nested policies restore the outer handler after a raised + exception. + """ + calls: list[str] = [] + url = "http://consumer.example/targets" + + def outer_handler(*, url: str, error: RequestsTimeout) -> None: + """Record a timeout handled by the outer context.""" + assert url == "http://consumer.example/targets" + assert error is request_timeout + calls.append("outer") + + def inner_handler(*, url: str, error: RequestsTimeout) -> None: + """Skip a timeout handled by the inner context.""" + assert url == "http://consumer.example/targets" + assert error is request_timeout + calls.append("inner") + pytest.skip(reason="Controlled inner timeout") + + with handling_request_timeouts(handler=outer_handler): + with ( + pytest.raises(expected_exception=RequestsTimeout) as exc, + propagating_request_timeouts(url=url), + ): + raise request_timeout + assert exc.value is request_timeout + with ( + pytest.raises( + expected_exception=pytest.skip.Exception, + match="Controlled inner timeout", + ), + handling_request_timeouts(handler=inner_handler), + propagating_request_timeouts(url=url), + ): + raise request_timeout + with ( + pytest.raises(expected_exception=RequestsTimeout), + propagating_request_timeouts(url=url), + ): + raise request_timeout + with ( + pytest.raises(expected_exception=RequestsTimeout), + propagating_request_timeouts(url=url), + ): + raise request_timeout + assert calls == ["outer", "inner", "outer"] diff --git a/tests/mock_vws/utils/__init__.py b/tests/mock_vws/utils/__init__.py index 8db33fb66..a093f6562 100644 --- a/tests/mock_vws/utils/__init__.py +++ b/tests/mock_vws/utils/__init__.py @@ -7,11 +7,9 @@ from typing import Literal from urllib.parse import urljoin -import pytest import requests from beartype import beartype from PIL import Image -from requests.exceptions import Timeout as RequestsTimeout from requests.structures import CaseInsensitiveDict from vws.response import Response @@ -19,6 +17,9 @@ from tests.mock_vws.utils.model_target_retries import ( send_with_transient_retries, ) +from tests.mock_vws.utils.request_timeout_policy import ( + propagating_request_timeouts, +) _REQUEST_TIMEOUT_SECONDS = 5 @@ -41,18 +42,11 @@ def _send_request( prepared_request = request.prepare() prepared_request.headers = CaseInsensitiveDict(data=headers) session = requests.Session() - try: + with propagating_request_timeouts(url=url): requests_response = session.send( request=prepared_request, timeout=_REQUEST_TIMEOUT_SECONDS, ) - # A live-service timeout is nondeterministic and cannot be forced reliably. - except RequestsTimeout: # pragma: no cover - if url.startswith( - ("https://vws.vuforia.com/", "https://cloudreco.vuforia.com/"), - ): - pytest.skip(reason="The real Vuforia service timed out.") - raise return Response( text=requests_response.text, url=requests_response.url, diff --git a/tests/mock_vws/utils/request_timeout_policy.py b/tests/mock_vws/utils/request_timeout_policy.py new file mode 100644 index 000000000..a7fd6f317 --- /dev/null +++ b/tests/mock_vws/utils/request_timeout_policy.py @@ -0,0 +1,53 @@ +"""Backend-supplied handling of request timeouts.""" + +import contextlib +from collections.abc import Generator +from typing import Protocol, runtime_checkable + +from beartype import beartype +from requests.exceptions import Timeout as RequestsTimeout + + +@runtime_checkable +class RequestTimeoutHandler(Protocol): + """A backend operation for a request which timed out.""" + + def __call__(self, *, url: str, error: RequestsTimeout) -> None: + """Handle a timeout before the shared sender propagates it.""" + ... + + +_TIMEOUT_HANDLERS: list[RequestTimeoutHandler] = [] + + +@contextlib.contextmanager +@beartype +def handling_request_timeouts( + *, handler: RequestTimeoutHandler +) -> Generator[None]: + """Use the supplied timeout handler within this context. + + Yields: + ``None``, with the supplied handler active. + """ + _TIMEOUT_HANDLERS.append(handler) + try: + yield + finally: + _ = _TIMEOUT_HANDLERS.pop() + + +@contextlib.contextmanager +@beartype +def propagating_request_timeouts(*, url: str) -> Generator[None]: + """Apply the active policy to timeouts raised by a request. + + Yields: + ``None``, while the request runs. + """ + try: + yield + except RequestsTimeout as error: + if bool(_TIMEOUT_HANDLERS): + _TIMEOUT_HANDLERS[-1](url=url, error=error) + raise