From 77a03bed911a246422842c107176fa146348dbd1 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sat, 21 Feb 2026 01:06:08 +0000 Subject: [PATCH 1/7] Add respx support for mocking httpx requests Implement MockVWSForHttpx context manager using respx to intercept httpx requests to Vuforia APIs. Reuses existing handler logic by converting httpx.Request to requests.PreparedRequest. Includes 13 tests covering real_http parameter, response delays, custom URLs, and database management. Adds httpx and respx as core dependencies. Co-Authored-By: Claude Haiku 4.5 --- pyproject.toml | 2 + spelling_private_dict.txt | 2 + src/mock_vws/__init__.py | 2 + src/mock_vws/_respx_mock_server/__init__.py | 1 + src/mock_vws/_respx_mock_server/decorators.py | 276 ++++++++++++++ tests/mock_vws/test_respx_mock_usage.py | 349 ++++++++++++++++++ 6 files changed, 632 insertions(+) create mode 100644 src/mock_vws/_respx_mock_server/__init__.py create mode 100644 src/mock_vws/_respx_mock_server/decorators.py create mode 100644 tests/mock_vws/test_respx_mock_usage.py 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..957f7e53a --- /dev/null +++ b/src/mock_vws/_respx_mock_server/decorators.py @@ -0,0 +1,276 @@ +"""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 requests import PreparedRequest +from requests.structures import CaseInsensitiveDict + +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_prepared_request(request: httpx.Request) -> PreparedRequest: + """Convert an httpx.Request to a requests.PreparedRequest. + + Args: + request: The httpx request to convert. + + Returns: + A PreparedRequest with headers, body, method, and url set. + The ``path_url`` property is derived automatically from ``url``. + """ + prepared = PreparedRequest() + prepared.method = request.method + prepared.url = str(request.url) + prepared.headers = CaseInsensitiveDict(dict(request.headers)) + prepared.body = request.content + return prepared + + +@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[[PreparedRequest], _ResponseType], + ) -> Callable[[httpx.Request], httpx.Response]: + """Create a respx-compatible callback from a handler. + + Args: + handler: A handler that takes a PreparedRequest 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: + httpx.ReadTimeout: The response delay exceeded the read + timeout. + """ + prepared = _to_prepared_request(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(prepared) + sleep_fn(delay_seconds) + if isinstance(body, str): + body = body.encode() + return httpx.Response( + status_code=status_code, + headers=list(headers.items()), + 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: + httpx.ConnectError: 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..8d88de0b5 --- /dev/null +++ b/tests/mock_vws/test_respx_mock_usage.py @@ -0,0 +1,349 @@ +"""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: + httpx.ConnectError: 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 From 5b302aae8ed74b1a1e7cd530935dfa2e0abc8481 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sat, 21 Feb 2026 01:07:55 +0000 Subject: [PATCH 2/7] Fix mypy type issues in respx implementation --- src/mock_vws/_respx_mock_server/decorators.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/mock_vws/_respx_mock_server/decorators.py b/src/mock_vws/_respx_mock_server/decorators.py index 957f7e53a..b849801b9 100644 --- a/src/mock_vws/_respx_mock_server/decorators.py +++ b/src/mock_vws/_respx_mock_server/decorators.py @@ -49,8 +49,10 @@ def _to_prepared_request(request: httpx.Request) -> PreparedRequest: """ prepared = PreparedRequest() prepared.method = request.method - prepared.url = str(request.url) - prepared.headers = CaseInsensitiveDict(dict(request.headers)) + prepared.url = str(request.url) # type: ignore[arg-type] + prepared.headers = CaseInsensitiveDict( # type: ignore[arg-type] + dict(request.headers) + ) prepared.body = request.content return prepared From 30505101d2bc83b57ed07c87435fba675c4db124 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sat, 21 Feb 2026 01:08:46 +0000 Subject: [PATCH 3/7] Fix mypy error codes for type ignore comments --- src/mock_vws/_respx_mock_server/decorators.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/mock_vws/_respx_mock_server/decorators.py b/src/mock_vws/_respx_mock_server/decorators.py index b849801b9..1536e9ebe 100644 --- a/src/mock_vws/_respx_mock_server/decorators.py +++ b/src/mock_vws/_respx_mock_server/decorators.py @@ -49,8 +49,8 @@ def _to_prepared_request(request: httpx.Request) -> PreparedRequest: """ prepared = PreparedRequest() prepared.method = request.method - prepared.url = str(request.url) # type: ignore[arg-type] - prepared.headers = CaseInsensitiveDict( # type: ignore[arg-type] + prepared.url = str(request.url) # type: ignore[call-overload] + prepared.headers = CaseInsensitiveDict( # type: ignore[misc] dict(request.headers) ) prepared.body = request.content From c0dc2f2419d76418a726dc256dd3e33fe9bfb6b8 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sat, 21 Feb 2026 01:10:46 +0000 Subject: [PATCH 4/7] Fix ruff linting issue with dict comprehension Add noqa comment to suppress C416 ruff error while keeping dict comprehension to satisfy pyrefly type checking requirements. Co-Authored-By: Claude Haiku 4.5 --- src/mock_vws/_respx_mock_server/decorators.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/mock_vws/_respx_mock_server/decorators.py b/src/mock_vws/_respx_mock_server/decorators.py index 1536e9ebe..30dfc04ca 100644 --- a/src/mock_vws/_respx_mock_server/decorators.py +++ b/src/mock_vws/_respx_mock_server/decorators.py @@ -51,7 +51,7 @@ def _to_prepared_request(request: httpx.Request) -> PreparedRequest: prepared.method = request.method prepared.url = str(request.url) # type: ignore[call-overload] prepared.headers = CaseInsensitiveDict( # type: ignore[misc] - dict(request.headers) + {k: v for k, v in request.headers.items()} # noqa: C416 ) prepared.body = request.content return prepared @@ -204,7 +204,7 @@ def callback(request: httpx.Request) -> httpx.Response: body = body.encode() return httpx.Response( status_code=status_code, - headers=list(headers.items()), + headers=[(k, v) for k, v in headers.items()], content=body, ) From 4db463e98ef2da8e8867bd98321dd9792ef0584c Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sat, 21 Feb 2026 08:58:33 +0000 Subject: [PATCH 5/7] Use RequestData instead of PreparedRequest in respx adapter Convert httpx.Request directly to RequestData, removing the PreparedRequest intermediate. This eliminates the requests library dependency from the respx module and removes all type suppression comments (type: ignore, noqa). Co-Authored-By: Claude Opus 4.6 --- src/mock_vws/_respx_mock_server/decorators.py | 30 ++++++++----------- 1 file changed, 13 insertions(+), 17 deletions(-) diff --git a/src/mock_vws/_respx_mock_server/decorators.py b/src/mock_vws/_respx_mock_server/decorators.py index 30dfc04ca..33db1b56d 100644 --- a/src/mock_vws/_respx_mock_server/decorators.py +++ b/src/mock_vws/_respx_mock_server/decorators.py @@ -10,9 +10,8 @@ import httpx import respx from beartype import BeartypeConf, beartype -from requests import PreparedRequest -from requests.structures import CaseInsensitiveDict +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, @@ -37,24 +36,21 @@ _BRISQUE_TRACKING_RATER = BrisqueTargetTrackingRater() -def _to_prepared_request(request: httpx.Request) -> PreparedRequest: - """Convert an httpx.Request to a requests.PreparedRequest. +def _to_request_data(request: httpx.Request) -> RequestData: + """Convert an httpx.Request to a RequestData. Args: request: The httpx request to convert. Returns: - A PreparedRequest with headers, body, method, and url set. - The ``path_url`` property is derived automatically from ``url``. + A RequestData with method, path, headers, and body set. """ - prepared = PreparedRequest() - prepared.method = request.method - prepared.url = str(request.url) # type: ignore[call-overload] - prepared.headers = CaseInsensitiveDict( # type: ignore[misc] - {k: v for k, v in request.headers.items()} # noqa: C416 + return RequestData( + method=request.method, + path=request.url.raw_path.decode(encoding="ascii"), + headers=request.headers, + body=request.content, ) - prepared.body = request.content - return prepared @beartype(conf=BeartypeConf(is_pep484_tower=True)) @@ -158,12 +154,12 @@ def add_vumark_database(self, vumark_database: VuMarkDatabase) -> None: def _make_callback( self, - handler: Callable[[PreparedRequest], _ResponseType], + handler: Callable[[RequestData], _ResponseType], ) -> Callable[[httpx.Request], httpx.Response]: """Create a respx-compatible callback from a handler. Args: - handler: A handler that takes a PreparedRequest and returns a + handler: A handler that takes a RequestData and returns a response tuple. Returns: @@ -187,7 +183,7 @@ def callback(request: httpx.Request) -> httpx.Response: httpx.ReadTimeout: The response delay exceeded the read timeout. """ - prepared = _to_prepared_request(request=request) + request_data = _to_request_data(request=request) timeout_info: dict[str, float | None] = request.extensions.get( "timeout", {} ) @@ -198,7 +194,7 @@ def callback(request: httpx.Request) -> httpx.Response: message="Response delay exceeded read timeout", request=request, ) - status_code, headers, body = handler(prepared) + status_code, headers, body = handler(request_data) sleep_fn(delay_seconds) if isinstance(body, str): body = body.encode() From 5d446ce36b4be72765dee6250e6d14c5ef324019 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sat, 21 Feb 2026 11:16:50 +0000 Subject: [PATCH 6/7] Fix lint and type issues in respx mock docs --- src/mock_vws/_respx_mock_server/decorators.py | 9 +++++---- tests/mock_vws/test_respx_mock_usage.py | 3 ++- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/src/mock_vws/_respx_mock_server/decorators.py b/src/mock_vws/_respx_mock_server/decorators.py index 33db1b56d..3b4b58560 100644 --- a/src/mock_vws/_respx_mock_server/decorators.py +++ b/src/mock_vws/_respx_mock_server/decorators.py @@ -180,8 +180,8 @@ def callback(request: httpx.Request) -> httpx.Response: An httpx.Response built from the handler's return value. Raises: - httpx.ReadTimeout: The response delay exceeded the read - timeout. + 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( @@ -200,7 +200,7 @@ def callback(request: httpx.Request) -> httpx.Response: body = body.encode() return httpx.Response( status_code=status_code, - headers=[(k, v) for k, v in headers.items()], + headers=headers, content=body, ) @@ -214,7 +214,8 @@ def _block_unmatched(request: httpx.Request) -> httpx.Response: request: The unmatched httpx request. Raises: - httpx.ConnectError: Always raised to block unmatched requests. + Exception: A connection error is always raised to block + unmatched requests. """ raise httpx.ConnectError( message="Connection refused by mock", diff --git a/tests/mock_vws/test_respx_mock_usage.py b/tests/mock_vws/test_respx_mock_usage.py index 8d88de0b5..1c408e4fe 100644 --- a/tests/mock_vws/test_respx_mock_usage.py +++ b/tests/mock_vws/test_respx_mock_usage.py @@ -14,7 +14,8 @@ def _request_unmocked_address() -> None: """Make a request using ``httpx`` to an unmocked, free local address. Raises: - httpx.ConnectError: Expected, as there is nothing to connect to. + Exception: A connection error is expected, as there is nothing + to connect to. """ sock = socket.socket() sock.bind(("", 0)) From 503c2d5817d76b996ba71c1ef08b83b690e74478 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 22 Feb 2026 09:47:06 +0000 Subject: [PATCH 7/7] Add documentation for MockVWSForHttpx Document the new httpx/respx mock backend in README, index, getting-started, mock-api-reference, and a new httpx-example.rst file. Co-Authored-By: Claude Sonnet 4.6 --- README.rst | 24 ++++++++++++++++++++++++ docs/source/getting-started.rst | 8 ++++++++ docs/source/httpx-example.rst | 22 ++++++++++++++++++++++ docs/source/index.rst | 5 +++++ docs/source/mock-api-reference.rst | 4 ++++ 5 files changed, 63 insertions(+) create mode 100644 docs/source/httpx-example.rst 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: