diff --git a/README.rst b/README.rst index 483954a59..0226ac3d5 100644 --- a/README.rst +++ b/README.rst @@ -38,6 +38,30 @@ By default, an exception will be raised if any requests to unmocked addresses ar .. _requests: https://pypi.org/project/requests/ +Mocking calls made to Vuforia with Python ``httpx`` +---------------------------------------------------- + +Using the mock redirects requests to Vuforia made with `httpx`_ to an in-memory implementation. + +.. code-block:: python + + """Make a request to the Vuforia Web Services API mock.""" + + import httpx + + from mock_vws import MockVWSForHttpx + from mock_vws.database import CloudDatabase + + with MockVWSForHttpx() as mock: + database = CloudDatabase() + mock.add_cloud_database(cloud_database=database) + # This will use the Vuforia mock. + httpx.get(url="https://vws.vuforia.com/summary", timeout=30) + +By default, an exception will be raised if any requests to unmocked addresses are made. + +.. _httpx: https://pypi.org/project/httpx/ + Using Docker to mock calls to Vuforia from any language ------------------------------------------------------- diff --git a/docs/source/getting-started.rst b/docs/source/getting-started.rst index f55e3d324..120f0138b 100644 --- a/docs/source/getting-started.rst +++ b/docs/source/getting-started.rst @@ -1,4 +1,12 @@ Getting started --------------- +Mocking calls made to Vuforia with Python ``requests`` +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + .. include:: basic-example.rst + +Mocking calls made to Vuforia with Python ``httpx`` +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +.. include:: httpx-example.rst diff --git a/docs/source/httpx-example.rst b/docs/source/httpx-example.rst new file mode 100644 index 000000000..2d12eb4d7 --- /dev/null +++ b/docs/source/httpx-example.rst @@ -0,0 +1,22 @@ +Using the mock redirects requests to Vuforia made with `httpx`_ to an in-memory implementation. + +.. code-block:: python + + """Make a request to the Vuforia Web Services API mock.""" + + import httpx + + from mock_vws import MockVWSForHttpx + from mock_vws.database import CloudDatabase + + with MockVWSForHttpx() as mock: + database = CloudDatabase() + mock.add_cloud_database(cloud_database=database) + # This will use the Vuforia mock. + httpx.get(url="https://vws.vuforia.com/summary", timeout=30) + +By default, an exception will be raised if any requests to unmocked addresses are made. + +See :ref:`mock-api-reference` for details of what can be changed and how. + +.. _httpx: https://pypi.org/project/httpx/ diff --git a/docs/source/index.rst b/docs/source/index.rst index 6f39583a2..22c386d5d 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -12,6 +12,11 @@ This requires Python |minimum-python-version|\+. .. include:: basic-example.rst +Mocking calls made to Vuforia with Python ``httpx`` +---------------------------------------------------- + +.. include:: httpx-example.rst + Using Docker to mock calls to Vuforia from any language ------------------------------------------------------- diff --git a/docs/source/mock-api-reference.rst b/docs/source/mock-api-reference.rst index 1b2ea255a..f44d65f1d 100644 --- a/docs/source/mock-api-reference.rst +++ b/docs/source/mock-api-reference.rst @@ -7,6 +7,10 @@ API Reference :members: :undoc-members: +.. autoclass:: mock_vws.MockVWSForHttpx + :members: + :undoc-members: + .. autoclass:: mock_vws.MissingSchemeError :members: :undoc-members: diff --git a/pyproject.toml b/pyproject.toml index df251f8d2..463151133 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,12 +36,14 @@ dynamic = [ dependencies = [ "beartype>=0.22.9", "flask>=3.0.3", + "httpx>=0.27.0", "numpy>=1.26.4", "pillow>=11.0.0", "piq>=0.8.0", "pydantic-settings>=2.6.1", "requests>=2.32.3", "responses>=0.25.3", + "respx>=0.21.0", "torch>=2.5.1", "torchmetrics>=1.5.1", "torchvision>=0.20.1", diff --git a/spelling_private_dict.txt b/spelling_private_dict.txt index 0b053e3bb..b1ad02e08 100644 --- a/spelling_private_dict.txt +++ b/spelling_private_dict.txt @@ -45,6 +45,7 @@ hmac html http https +httpx iff io issuecomment @@ -98,6 +99,7 @@ reqjsonarr resheader resjson resjsonarr +respx rfc rgb str diff --git a/src/mock_vws/__init__.py b/src/mock_vws/__init__.py index d6d5e053a..42d6d5264 100644 --- a/src/mock_vws/__init__.py +++ b/src/mock_vws/__init__.py @@ -4,8 +4,10 @@ MissingSchemeError, MockVWS, ) +from mock_vws._respx_mock_server.decorators import MockVWSForHttpx __all__ = [ "MissingSchemeError", "MockVWS", + "MockVWSForHttpx", ] diff --git a/src/mock_vws/_respx_mock_server/__init__.py b/src/mock_vws/_respx_mock_server/__init__.py new file mode 100644 index 000000000..a5ffb7cab --- /dev/null +++ b/src/mock_vws/_respx_mock_server/__init__.py @@ -0,0 +1 @@ +"""A fake implementation of Vuforia Web Services for use with respx.""" diff --git a/src/mock_vws/_respx_mock_server/decorators.py b/src/mock_vws/_respx_mock_server/decorators.py new file mode 100644 index 000000000..3b4b58560 --- /dev/null +++ b/src/mock_vws/_respx_mock_server/decorators.py @@ -0,0 +1,275 @@ +"""Decorators for using the mock with httpx via respx.""" + +import re +import time +from collections.abc import Callable, Mapping +from contextlib import ContextDecorator +from typing import Literal, Self +from urllib.parse import urljoin, urlparse + +import httpx +import respx +from beartype import BeartypeConf, beartype + +from mock_vws._mock_common import RequestData +from mock_vws._requests_mock_server.decorators import MissingSchemeError +from mock_vws._requests_mock_server.mock_web_query_api import ( + MockVuforiaWebQueryAPI, +) +from mock_vws._requests_mock_server.mock_web_services_api import ( + MockVuforiaWebServicesAPI, +) +from mock_vws.database import CloudDatabase, VuMarkDatabase +from mock_vws.image_matchers import ( + ImageMatcher, + StructuralSimilarityMatcher, +) +from mock_vws.target_manager import TargetManager +from mock_vws.target_raters import ( + BrisqueTargetTrackingRater, + TargetTrackingRater, +) + +_ResponseType = tuple[int, Mapping[str, str], str | bytes] + +_STRUCTURAL_SIMILARITY_MATCHER = StructuralSimilarityMatcher() +_BRISQUE_TRACKING_RATER = BrisqueTargetTrackingRater() + + +def _to_request_data(request: httpx.Request) -> RequestData: + """Convert an httpx.Request to a RequestData. + + Args: + request: The httpx request to convert. + + Returns: + A RequestData with method, path, headers, and body set. + """ + return RequestData( + method=request.method, + path=request.url.raw_path.decode(encoding="ascii"), + headers=request.headers, + body=request.content, + ) + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +class MockVWSForHttpx(ContextDecorator): + """Route httpx requests to Vuforia's Web Service APIs to fakes of those + APIs. + """ + + def __init__( + self, + *, + base_vws_url: str = "https://vws.vuforia.com", + base_vwq_url: str = "https://cloudreco.vuforia.com", + duplicate_match_checker: ImageMatcher = _STRUCTURAL_SIMILARITY_MATCHER, + query_match_checker: ImageMatcher = _STRUCTURAL_SIMILARITY_MATCHER, + processing_time_seconds: float = 2.0, + target_tracking_rater: TargetTrackingRater = _BRISQUE_TRACKING_RATER, + real_http: bool = False, + response_delay_seconds: float = 0.0, + sleep_fn: Callable[[float], None] = time.sleep, + ) -> None: + """Route httpx requests to Vuforia's Web Service APIs to fakes of + those APIs. + + Args: + real_http: Whether or not to forward requests to the real + server if they are not handled by the mock. + processing_time_seconds: The number of seconds to process each + image for. + In the real Vuforia Web Services, this is not deterministic. + base_vwq_url: The base URL for the VWQ API. + base_vws_url: The base URL for the VWS API. + query_match_checker: A callable which takes two image values and + returns whether they will match in a query request. + duplicate_match_checker: A callable which takes two image values + and returns whether they are duplicates. + target_tracking_rater: A callable for rating targets for tracking. + response_delay_seconds: The number of seconds to delay each + response by. This can be used to test timeout handling. + sleep_fn: The function to use for sleeping during response + delays. Defaults to ``time.sleep``. Inject a custom + function to control virtual time in tests without + monkey-patching. + + Raises: + MissingSchemeError: There is no scheme in a given URL. + """ + super().__init__() + self._real_http = real_http + self._response_delay_seconds = response_delay_seconds + self._sleep_fn = sleep_fn + self._router: respx.MockRouter + self._target_manager = TargetManager() + + self._base_vws_url = base_vws_url + self._base_vwq_url = base_vwq_url + for url in (base_vwq_url, base_vws_url): + parse_result = urlparse(url=url) + if not parse_result.scheme: + raise MissingSchemeError(url=url) + + self._mock_vws_api = MockVuforiaWebServicesAPI( + target_manager=self._target_manager, + processing_time_seconds=float(processing_time_seconds), + duplicate_match_checker=duplicate_match_checker, + target_tracking_rater=target_tracking_rater, + ) + + self._mock_vwq_api = MockVuforiaWebQueryAPI( + target_manager=self._target_manager, + query_match_checker=query_match_checker, + ) + + def add_cloud_database(self, cloud_database: CloudDatabase) -> None: + """Add a cloud database. + + Args: + cloud_database: The cloud database to add. + + Raises: + ValueError: One of the given cloud database keys matches a key for + an existing cloud database. + """ + self._target_manager.add_cloud_database( + cloud_database=cloud_database, + ) + + def add_vumark_database(self, vumark_database: VuMarkDatabase) -> None: + """Add a VuMark database. + + Args: + vumark_database: The VuMark database to add. + + Raises: + ValueError: One of the given database keys matches a key for + an existing database. + """ + self._target_manager.add_vumark_database( + vumark_database=vumark_database, + ) + + def _make_callback( + self, + handler: Callable[[RequestData], _ResponseType], + ) -> Callable[[httpx.Request], httpx.Response]: + """Create a respx-compatible callback from a handler. + + Args: + handler: A handler that takes a RequestData and returns a + response tuple. + + Returns: + A callback that takes an httpx.Request and returns an + httpx.Response. + """ + delay_seconds = self._response_delay_seconds + sleep_fn = self._sleep_fn + + def callback(request: httpx.Request) -> httpx.Response: + """Handle an httpx request by converting it and calling the + handler. + + Args: + request: The httpx request to handle. + + Returns: + An httpx.Response built from the handler's return value. + + Raises: + Exception: A timeout error is raised when the response + delay exceeds the read timeout. + """ + request_data = _to_request_data(request=request) + timeout_info: dict[str, float | None] = request.extensions.get( + "timeout", {} + ) + read_timeout = timeout_info.get("read") + if read_timeout is not None and delay_seconds > read_timeout: + sleep_fn(read_timeout) + raise httpx.ReadTimeout( + message="Response delay exceeded read timeout", + request=request, + ) + status_code, headers, body = handler(request_data) + sleep_fn(delay_seconds) + if isinstance(body, str): + body = body.encode() + return httpx.Response( + status_code=status_code, + headers=headers, + content=body, + ) + + return callback + + @staticmethod + def _block_unmatched(request: httpx.Request) -> httpx.Response: + """Raise ConnectError for unmatched requests when real_http=False. + + Args: + request: The unmatched httpx request. + + Raises: + Exception: A connection error is always raised to block + unmatched requests. + """ + raise httpx.ConnectError( + message="Connection refused by mock", + request=request, + ) + + def __enter__(self) -> Self: + """Start an instance of a Vuforia mock. + + Returns: + ``self``. + """ + router = respx.MockRouter( + assert_all_called=False, + assert_all_mocked=False, + ) + + for api, base_url in ( + (self._mock_vws_api, self._base_vws_url), + (self._mock_vwq_api, self._base_vwq_url), + ): + for route in api.routes: + url_pattern = urljoin( + base=base_url, + url=f"{route.path_pattern}$", + ) + compiled_url_pattern = re.compile(pattern=url_pattern) + + for http_method in route.http_methods: + original_callback = getattr(api, route.route_name) + router.route( + method=http_method, + url=compiled_url_pattern, + ).mock( + side_effect=self._make_callback( + handler=original_callback, + ), + ) + + if self._real_http: + router.route().pass_through() + else: + router.route().mock(side_effect=self._block_unmatched) + + router.start() + self._router = router + return self + + def __exit__(self, *exc: object) -> Literal[False]: + """Stop the Vuforia mock. + + Returns: + False + """ + del exc + self._router.stop() + return False diff --git a/tests/mock_vws/test_respx_mock_usage.py b/tests/mock_vws/test_respx_mock_usage.py new file mode 100644 index 000000000..1c408e4fe --- /dev/null +++ b/tests/mock_vws/test_respx_mock_usage.py @@ -0,0 +1,350 @@ +"""Tests for the usage of the mock for ``httpx`` via ``respx``.""" + +import socket + +import httpx +import pytest +from vws_auth_tools import rfc_1123_date + +from mock_vws import MissingSchemeError, MockVWSForHttpx +from mock_vws.database import CloudDatabase, VuMarkDatabase + + +def _request_unmocked_address() -> None: + """Make a request using ``httpx`` to an unmocked, free local address. + + Raises: + Exception: A connection error is expected, as there is nothing + to connect to. + """ + sock = socket.socket() + sock.bind(("", 0)) + port = sock.getsockname()[1] + sock.close() + httpx.get(url=f"http://localhost:{port}", timeout=30) + + +def _request_mocked_address() -> None: + """Make a request using ``httpx`` to a mocked Vuforia endpoint.""" + httpx.get( + url="https://vws.vuforia.com/summary", + headers={ + "Date": rfc_1123_date(), + "Authorization": "bad_auth_token", + }, + timeout=30, + ) + + +class TestRealHTTP: + """Tests for making requests to mocked and unmocked addresses.""" + + @staticmethod + def test_default() -> None: + """By default, the mock stops any requests made with ``httpx`` to + non-Vuforia addresses, but not to mocked Vuforia endpoints. + """ + with MockVWSForHttpx(): + with pytest.raises(expected_exception=httpx.ConnectError): + _request_unmocked_address() + + # No exception is raised when making a request to a mocked + # endpoint. + _request_mocked_address() + + # The mocking stops when the context manager stops. + with pytest.raises(expected_exception=httpx.ConnectError): + _request_unmocked_address() + + @staticmethod + def test_real_http() -> None: + """When the ``real_http`` parameter is ``True``, requests to + unmocked + addresses are not stopped. + """ + with ( + MockVWSForHttpx(real_http=True), + pytest.raises(expected_exception=httpx.ConnectError), + ): + _request_unmocked_address() + + +class TestResponseDelay: + """Tests for the response delay feature.""" + + @staticmethod + def test_default_no_delay() -> None: + """By default, there is no response delay.""" + with MockVWSForHttpx(): + response = httpx.get( + url="https://vws.vuforia.com/summary", + headers={ + "Date": rfc_1123_date(), + "Authorization": "bad_auth_token", + }, + timeout=0.5, + ) + assert response.status_code is not None + + @staticmethod + def test_delay_causes_timeout() -> None: + """When ``response_delay_seconds`` is set higher than the client + timeout, a ``ReadTimeout`` exception is raised. + """ + with ( + MockVWSForHttpx(response_delay_seconds=0.5), + pytest.raises(expected_exception=httpx.ReadTimeout), + ): + httpx.get( + url="https://vws.vuforia.com/summary", + headers={ + "Date": rfc_1123_date(), + "Authorization": "bad_auth_token", + }, + timeout=0.1, + ) + + @staticmethod + def test_delay_allows_completion() -> None: + """When ``response_delay_seconds`` is set lower than the client + timeout, the request completes successfully. + """ + with MockVWSForHttpx(response_delay_seconds=0.1): + response = httpx.get( + url="https://vws.vuforia.com/summary", + headers={ + "Date": rfc_1123_date(), + "Authorization": "bad_auth_token", + }, + timeout=2.0, + ) + assert response.status_code is not None + + @staticmethod + def test_custom_sleep_fn_called_on_delay() -> None: + """When a custom ``sleep_fn`` is provided, it is called instead of + ``time.sleep`` for the non-timeout delay path. + """ + calls: list[float] = [] + with MockVWSForHttpx( + response_delay_seconds=5.0, + sleep_fn=calls.append, + ): + httpx.get( + url="https://vws.vuforia.com/summary", + headers={ + "Date": rfc_1123_date(), + "Authorization": "bad_auth_token", + }, + timeout=30, + ) + assert calls == [5.0] + + @staticmethod + def test_custom_sleep_fn_called_on_timeout() -> None: + """When a custom ``sleep_fn`` is provided, it is called with the + effective timeout when the delay exceeds it. + """ + calls: list[float] = [] + with ( + MockVWSForHttpx( + response_delay_seconds=5.0, + sleep_fn=calls.append, + ), + pytest.raises(expected_exception=httpx.ReadTimeout), + ): + httpx.get( + url="https://vws.vuforia.com/summary", + headers={ + "Date": rfc_1123_date(), + "Authorization": "bad_auth_token", + }, + timeout=1.0, + ) + assert calls == [1.0] + + +class TestCustomBaseURLs: + """Tests for using custom base URLs.""" + + @staticmethod + def test_custom_base_vws_url() -> None: + """It is possible to use a custom base VWS URL.""" + with MockVWSForHttpx( + base_vws_url="https://vuforia.vws.example.com", + real_http=False, + ): + with pytest.raises(expected_exception=httpx.ConnectError): + httpx.get(url="https://vws.vuforia.com/summary", timeout=30) + + httpx.get( + url="https://vuforia.vws.example.com/summary", + timeout=30, + ) + httpx.post( + url="https://cloudreco.vuforia.com/v1/query", + timeout=30, + ) + + @staticmethod + def test_custom_base_vwq_url() -> None: + """It is possible to use a custom base cloud recognition URL.""" + with MockVWSForHttpx( + base_vwq_url="https://vuforia.vwq.example.com", + real_http=False, + ): + with pytest.raises(expected_exception=httpx.ConnectError): + httpx.post( + url="https://cloudreco.vuforia.com/v1/query", + timeout=30, + ) + + httpx.post( + url="https://vuforia.vwq.example.com/v1/query", + timeout=30, + ) + httpx.get( + url="https://vws.vuforia.com/summary", + timeout=30, + ) + + @staticmethod + def test_no_scheme() -> None: + """An error is raised if a URL is given with no scheme.""" + with pytest.raises(expected_exception=MissingSchemeError) as vws_exc: + MockVWSForHttpx(base_vws_url="vuforia.vws.example.com") + + expected = ( + 'Invalid URL "vuforia.vws.example.com": No scheme supplied. ' + 'Perhaps you meant "https://vuforia.vws.example.com".' + ) + assert str(object=vws_exc.value) == expected + with pytest.raises(expected_exception=MissingSchemeError) as vwq_exc: + MockVWSForHttpx(base_vwq_url="vuforia.vwq.example.com") + expected = ( + 'Invalid URL "vuforia.vwq.example.com": No scheme supplied. ' + 'Perhaps you meant "https://vuforia.vwq.example.com".' + ) + assert str(object=vwq_exc.value) == expected + + +class TestAddDatabase: + """Tests for adding databases to the mock.""" + + @staticmethod + def test_duplicate_keys() -> None: + """It is not possible to have multiple databases with matching + keys. + """ + database = CloudDatabase( + server_access_key="1", + server_secret_key="2", + client_access_key="3", + client_secret_key="4", + database_name="5", + ) + + bad_server_access_key_db = CloudDatabase(server_access_key="1") + bad_server_secret_key_db = CloudDatabase(server_secret_key="2") + bad_client_access_key_db = CloudDatabase(client_access_key="3") + bad_client_secret_key_db = CloudDatabase(client_secret_key="4") + bad_database_name_db = CloudDatabase(database_name="5") + + server_access_key_conflict_error = ( + "All server access keys must be unique. " + 'There is already a database with the server access key "1".' + ) + server_secret_key_conflict_error = ( + "All server secret keys must be unique. " + 'There is already a database with the server secret key "2".' + ) + client_access_key_conflict_error = ( + "All client access keys must be unique. " + 'There is already a database with the client access key "3".' + ) + client_secret_key_conflict_error = ( + "All client secret keys must be unique. " + 'There is already a database with the client secret key "4".' + ) + database_name_conflict_error = ( + "All names must be unique. " + 'There is already a database with the name "5".' + ) + + with MockVWSForHttpx() as mock: + mock.add_cloud_database(cloud_database=database) + for bad_database, expected_message in ( + (bad_server_access_key_db, server_access_key_conflict_error), + (bad_server_secret_key_db, server_secret_key_conflict_error), + (bad_client_access_key_db, client_access_key_conflict_error), + (bad_client_secret_key_db, client_secret_key_conflict_error), + (bad_database_name_db, database_name_conflict_error), + ): + with pytest.raises( + expected_exception=ValueError, + match=expected_message + "$", + ): + mock.add_cloud_database(cloud_database=bad_database) + + @staticmethod + def test_duplicate_vumark_keys() -> None: + """It is not possible to have multiple databases with matching + keys, + including VuMark databases. + """ + database = VuMarkDatabase( + server_access_key="1", + server_secret_key="2", + database_name="3", + ) + + bad_server_access_key_db = VuMarkDatabase(server_access_key="1") + bad_server_secret_key_db = VuMarkDatabase(server_secret_key="2") + bad_database_name_db = VuMarkDatabase(database_name="3") + + server_access_key_conflict_error = ( + "All server access keys must be unique. " + 'There is already a database with the server access key "1".' + ) + server_secret_key_conflict_error = ( + "All server secret keys must be unique. " + 'There is already a database with the server secret key "2".' + ) + database_name_conflict_error = ( + "All names must be unique. " + 'There is already a database with the name "3".' + ) + + with MockVWSForHttpx() as mock: + mock.add_vumark_database(vumark_database=database) + for bad_database, expected_message in ( + (bad_server_access_key_db, server_access_key_conflict_error), + (bad_server_secret_key_db, server_secret_key_conflict_error), + (bad_database_name_db, database_name_conflict_error), + ): + with pytest.raises( + expected_exception=ValueError, + match=expected_message + "$", + ): + mock.add_vumark_database(vumark_database=bad_database) + + +class TestVWSEndpoints: + """Tests that VWS endpoints are accessible via httpx.""" + + @staticmethod + def test_database_summary() -> None: + """The database summary endpoint is accessible via httpx.""" + database = CloudDatabase() + with MockVWSForHttpx() as mock: + mock.add_cloud_database(cloud_database=database) + response = httpx.get( + url="https://vws.vuforia.com/summary", + headers={ + "Date": rfc_1123_date(), + "Authorization": "bad_auth_token", + }, + timeout=30, + ) + # We just verify we get a response (auth will fail but endpoint works) + assert response.status_code is not None