Keep request timeout policy in the backend adapter

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,
         ),

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

tests/mock_vws/utils/__init__.py

@@ -7,18 +7,19 @@
 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
 
 from mock_vws._constants import ResultCodes
 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,

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