From 740846a32107c592d48300da2311572c5167a192 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 22 Feb 2026 21:11:55 +0000 Subject: [PATCH 1/4] Make MockVWS intercept both requests and httpx MockVWS now starts both responses (for requests) and respx (for httpx) mocks simultaneously, eliminating the need for a separate MockVWSForHttpx class. Removes MockVWSForHttpx entirely. Updates all tests and docs to reflect this change. Co-Authored-By: Claude Haiku 4.5 --- CHANGELOG.rst | 3 + README.rst | 22 +- docs/source/basic-example.rst | 3 +- docs/source/getting-started.rst | 7 +- docs/source/httpx-example.rst | 12 +- docs/source/index.rst | 7 +- docs/source/mock-api-reference.rst | 4 - src/mock_vws/__init__.py | 8 +- src/mock_vws/_mock_common.py | 23 ++ .../_requests_mock_server/decorators.py | 50 ++- src/mock_vws/_respx_mock_server/decorators.py | 331 ++++++------------ tests/mock_vws/test_requests_mock_usage.py | 48 +++ tests/mock_vws/test_respx_mock_usage.py | 155 +------- 13 files changed, 248 insertions(+), 425 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 8d57f2531..e03655b46 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -4,6 +4,9 @@ Changelog Next ---- +- ``MockVWS`` now intercepts both ``requests`` (via ``responses``) and ``httpx`` (via ``respx``) simultaneously. + ``MockVWSForHttpx`` has been removed — ``MockVWS`` handles both HTTP libraries. + 2026.02.22.2 ------------ diff --git a/README.rst b/README.rst index 0226ac3d5..81e350071 100644 --- a/README.rst +++ b/README.rst @@ -8,10 +8,10 @@ VWS Mock Mock for the Vuforia Web Services (VWS) API and the Vuforia Web Query API. -Mocking calls made to Vuforia with Python ``requests`` ------------------------------------------------------- +Mocking calls made to Vuforia +------------------------------ -Using the mock redirects requests to Vuforia made with `requests`_ to an in-memory implementation. +``MockVWS`` intercepts requests made with `requests`_ or `httpx`_. .. code-block:: shell @@ -34,25 +34,18 @@ This requires Python |minimum-python-version|\+. # This will use the Vuforia mock. requests.get(url="https://vws.vuforia.com/summary", timeout=30) -By default, an exception will be raised if any requests to unmocked addresses are made. - -.. _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. +``MockVWS`` also intercepts `httpx`_ requests: .. code-block:: python - """Make a request to the Vuforia Web Services API mock.""" + """Make a request to the Vuforia Web Services API mock using httpx.""" import httpx - from mock_vws import MockVWSForHttpx + from mock_vws import MockVWS from mock_vws.database import CloudDatabase - with MockVWSForHttpx() as mock: + with MockVWS() as mock: database = CloudDatabase() mock.add_cloud_database(cloud_database=database) # This will use the Vuforia mock. @@ -60,6 +53,7 @@ Using the mock redirects requests to Vuforia made with `httpx`_ to an in-memory By default, an exception will be raised if any requests to unmocked addresses are made. +.. _requests: https://pypi.org/project/requests/ .. _httpx: https://pypi.org/project/httpx/ Using Docker to mock calls to Vuforia from any language diff --git a/docs/source/basic-example.rst b/docs/source/basic-example.rst index 17a0f568c..c6829a3b4 100644 --- a/docs/source/basic-example.rst +++ b/docs/source/basic-example.rst @@ -1,4 +1,4 @@ -Using the mock redirects requests to Vuforia made with `requests`_ to an in-memory implementation. +``MockVWS`` intercepts requests to Vuforia made with `requests`_ or `httpx`_. .. code-block:: python @@ -20,3 +20,4 @@ By default, an exception will be raised if any requests to unmocked addresses ar See :ref:`mock-api-reference` for details of what can be changed and how. .. _requests: https://pypi.org/project/requests/ +.. _httpx: https://pypi.org/project/httpx/ diff --git a/docs/source/getting-started.rst b/docs/source/getting-started.rst index 120f0138b..17c1bce89 100644 --- a/docs/source/getting-started.rst +++ b/docs/source/getting-started.rst @@ -1,12 +1,9 @@ Getting started --------------- -Mocking calls made to Vuforia with Python ``requests`` -~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +Mocking calls made to Vuforia +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ .. 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 index 2d12eb4d7..0a74e9c82 100644 --- a/docs/source/httpx-example.rst +++ b/docs/source/httpx-example.rst @@ -1,22 +1,18 @@ -Using the mock redirects requests to Vuforia made with `httpx`_ to an in-memory implementation. +``MockVWS`` also intercepts requests made with `httpx`_. .. code-block:: python - """Make a request to the Vuforia Web Services API mock.""" + """Make a request to the Vuforia Web Services API mock using httpx.""" import httpx - from mock_vws import MockVWSForHttpx + from mock_vws import MockVWS from mock_vws.database import CloudDatabase - with MockVWSForHttpx() as mock: + with MockVWS() 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 22c386d5d..04e81ebfe 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -1,8 +1,8 @@ |project| ========= -Mocking calls made to Vuforia with Python ``requests`` ------------------------------------------------------- +Mocking calls made to Vuforia +------------------------------ .. code-block:: console @@ -12,9 +12,6 @@ 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 f44d65f1d..1b2ea255a 100644 --- a/docs/source/mock-api-reference.rst +++ b/docs/source/mock-api-reference.rst @@ -7,10 +7,6 @@ API Reference :members: :undoc-members: -.. autoclass:: mock_vws.MockVWSForHttpx - :members: - :undoc-members: - .. autoclass:: mock_vws.MissingSchemeError :members: :undoc-members: diff --git a/src/mock_vws/__init__.py b/src/mock_vws/__init__.py index 42d6d5264..86151570d 100644 --- a/src/mock_vws/__init__.py +++ b/src/mock_vws/__init__.py @@ -1,13 +1,9 @@ """Tools for using a fake implementation of Vuforia.""" -from mock_vws._requests_mock_server.decorators import ( - MissingSchemeError, - MockVWS, -) -from mock_vws._respx_mock_server.decorators import MockVWSForHttpx +from mock_vws._mock_common import MissingSchemeError +from mock_vws._requests_mock_server.decorators import MockVWS __all__ = [ "MissingSchemeError", "MockVWS", - "MockVWSForHttpx", ] diff --git a/src/mock_vws/_mock_common.py b/src/mock_vws/_mock_common.py index 5b0c81a62..15c3776ae 100644 --- a/src/mock_vws/_mock_common.py +++ b/src/mock_vws/_mock_common.py @@ -8,6 +8,29 @@ from beartype import beartype +@beartype +class MissingSchemeError(Exception): + """Raised when a URL is missing a schema.""" + + def __init__(self, url: str) -> None: + """ + Args: + url: The URL which is missing a scheme. + """ + super().__init__() + self.url = url + + def __str__(self) -> str: + """ + Give a string representation of this error with a + suggestion. + """ + return ( + f'Invalid URL "{self.url}": No scheme supplied. ' + f'Perhaps you meant "https://{self.url}".' + ) + + @beartype @dataclass(frozen=True) class RequestData: diff --git a/src/mock_vws/_requests_mock_server/decorators.py b/src/mock_vws/_requests_mock_server/decorators.py index 291a80d6c..f95446fa5 100644 --- a/src/mock_vws/_requests_mock_server/decorators.py +++ b/src/mock_vws/_requests_mock_server/decorators.py @@ -4,7 +4,7 @@ import time from collections.abc import Callable, Mapping from contextlib import ContextDecorator -from typing import Any, Literal, Self +from typing import TYPE_CHECKING, Any, Literal, Self from urllib.parse import urlparse import requests @@ -12,7 +12,11 @@ from requests import PreparedRequest from responses import RequestsMock -from mock_vws._mock_common import RequestData +if TYPE_CHECKING: + import respx + +from mock_vws._mock_common import MissingSchemeError, RequestData +from mock_vws._respx_mock_server.decorators import start_respx_router from mock_vws.database import CloudDatabase, VuMarkDatabase from mock_vws.image_matchers import ( ImageMatcher, @@ -35,32 +39,12 @@ _BRISQUE_TRACKING_RATER = BrisqueTargetTrackingRater() -@beartype -class MissingSchemeError(Exception): - """Raised when a URL is missing a schema.""" - - def __init__(self, url: str) -> None: - """ - Args: - url: The URL which is missing a scheme. - """ - super().__init__() - self.url = url - - def __str__(self) -> str: - """ - Give a string representation of this error with a - suggestion. - """ - return ( - f'Invalid URL "{self.url}": No scheme supplied. ' - f'Perhaps you meant "https://{self.url}".' - ) - - @beartype(conf=BeartypeConf(is_pep484_tower=True)) class MockVWS(ContextDecorator): - """Route requests to Vuforia's Web Service APIs to fakes of those APIs.""" + """Route requests to Vuforia's Web Service APIs to fakes of those APIs. + + Works with both ``requests`` and ``httpx``. + """ def __init__( self, @@ -78,6 +62,8 @@ def __init__( """Route requests to Vuforia's Web Service APIs to fakes of those APIs. + Works with both ``requests`` and ``httpx``. + Args: real_http: Whether or not to forward requests to the real server if they are not handled by the mock. @@ -108,6 +94,7 @@ def __init__( self._response_delay_seconds = response_delay_seconds self._sleep_fn = sleep_fn self._mock: RequestsMock + self._router: respx.MockRouter self._target_manager = TargetManager() self._base_vws_url = base_vws_url @@ -252,6 +239,16 @@ def __enter__(self) -> Self: self._mock = mock self._mock.start() + self._router = start_respx_router( + mock_vws_api=self._mock_vws_api, + mock_vwq_api=self._mock_vwq_api, + base_vws_url=self._base_vws_url, + base_vwq_url=self._base_vwq_url, + response_delay_seconds=self._response_delay_seconds, + sleep_fn=self._sleep_fn, + real_http=self._real_http, + ) + return self def __exit__(self, *exc: object) -> Literal[False]: @@ -265,4 +262,5 @@ def __exit__(self, *exc: object) -> Literal[False]: del exc self._mock.stop() + self._router.stop() return False diff --git a/src/mock_vws/_respx_mock_server/decorators.py b/src/mock_vws/_respx_mock_server/decorators.py index b9c91e078..a232474f0 100644 --- a/src/mock_vws/_respx_mock_server/decorators.py +++ b/src/mock_vws/_respx_mock_server/decorators.py @@ -1,40 +1,22 @@ -"""Decorators for using the mock with httpx via respx.""" +"""Helpers for mocking Vuforia 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 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, @@ -61,227 +43,140 @@ def _to_request_data( ) -@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 _block_unmatched(request: httpx.Request) -> httpx.Response: + """Raise ConnectError for unmatched requests when real_http=False. - 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: + request: The unmatched httpx request. - 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: + Exception: A connection error is always raised to block + unmatched requests. + """ + raise httpx.ConnectError( + message="Connection refused by mock", + request=request, + ) - 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) +def _make_respx_callback( + *, + handler: Callable[[RequestData], _ResponseType], + base_path: str, + delay_seconds: float, + sleep_fn: Callable[[float], None], +) -> Callable[[httpx.Request], httpx.Response]: + """Create a respx-compatible callback from a handler. - 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, - ) + Args: + handler: A handler that takes a RequestData and returns a + response tuple. + base_path: The base path prefix to strip from the request path. + delay_seconds: The number of seconds to delay the response by. + sleep_fn: The function to use for sleeping during delays. - self._mock_vwq_api = MockVuforiaWebQueryAPI( - target_manager=self._target_manager, - query_match_checker=query_match_checker, - ) + Returns: + A callback that takes an httpx.Request and returns an + httpx.Response. + """ - def add_cloud_database(self, cloud_database: CloudDatabase) -> None: - """Add a cloud database. + def callback(request: httpx.Request) -> httpx.Response: + """Handle an httpx request by converting it and calling the + handler. 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. + request: The httpx request to handle. - Args: - vumark_database: The VuMark database to add. + Returns: + An httpx.Response built from the handler's return value. Raises: - ValueError: One of the given database keys matches a key for - an existing database. + Exception: A timeout error is raised when the response + delay exceeds the read timeout. """ - self._target_manager.add_vumark_database( - vumark_database=vumark_database, + request_data = _to_request_data( + request=request, + base_path=base_path, ) - - def _make_callback( - self, - handler: Callable[[RequestData], _ResponseType], - base_path: str, - ) -> 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. - base_path: The base path prefix to strip from the request path. - - 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( + 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, - base_path=base_path, - ) - 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, + 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, ) - def __enter__(self) -> Self: - """Start an instance of a Vuforia mock. + return callback - 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), - ): - base_path = urlparse(url=base_url).path.rstrip("/") - for route in api.routes: - url_pattern = base_url.rstrip("/") + route.path_pattern + "$" - compiled_url_pattern = re.compile(pattern=url_pattern) +def start_respx_router( + *, + mock_vws_api: MockVuforiaWebServicesAPI, + mock_vwq_api: MockVuforiaWebQueryAPI, + base_vws_url: str, + base_vwq_url: str, + response_delay_seconds: float, + sleep_fn: Callable[[float], None], + real_http: bool, +) -> respx.MockRouter: + """Configure and start a respx.MockRouter with Vuforia routes. - 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, - base_path=base_path, - ), - ) + Args: + mock_vws_api: The VWS API handler. + mock_vwq_api: The VWQ API handler. + base_vws_url: The base URL for the VWS API. + base_vwq_url: The base URL for the VWQ API. + response_delay_seconds: The number of seconds to delay responses. + sleep_fn: The function to use for sleeping during delays. + real_http: Whether to pass through unmatched requests. - if self._real_http: - router.route().pass_through() - else: - router.route().mock(side_effect=self._block_unmatched) + Returns: + A started respx.MockRouter. + """ + router = respx.MockRouter( + assert_all_called=False, + assert_all_mocked=False, + ) - router.start() - self._router = router - return self + for api, base_url in ( + (mock_vws_api, base_vws_url), + (mock_vwq_api, base_vwq_url), + ): + base_path = urlparse(url=base_url).path.rstrip("/") + for route in api.routes: + url_pattern = base_url.rstrip("/") + 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=_make_respx_callback( + handler=original_callback, + base_path=base_path, + delay_seconds=response_delay_seconds, + sleep_fn=sleep_fn, + ), + ) - def __exit__(self, *exc: object) -> Literal[False]: - """Stop the Vuforia mock. + if real_http: + router.route().pass_through() + else: + router.route().mock(side_effect=_block_unmatched) - Returns: - False - """ - del exc - self._router.stop() - return False + router.start() + return router diff --git a/tests/mock_vws/test_requests_mock_usage.py b/tests/mock_vws/test_requests_mock_usage.py index 60f07ed3a..eaa82a494 100644 --- a/tests/mock_vws/test_requests_mock_usage.py +++ b/tests/mock_vws/test_requests_mock_usage.py @@ -8,6 +8,7 @@ from http import HTTPStatus from urllib.parse import urlparse +import httpx import pytest import requests from beartype import beartype @@ -1004,3 +1005,50 @@ def test_text(endpoint: Endpoint) -> None: ) response = new_endpoint.send() assert response.status_code == endpoint.successful_headers_status_code + + +class TestHttpxAlsoIntercepted: + """Tests that MockVWS also intercepts httpx requests.""" + + @staticmethod + def test_httpx_vuforia_endpoint_intercepted() -> None: + """``MockVWS`` intercepts ``httpx`` requests to Vuforia + endpoints. + """ + with MockVWS(): + response = httpx.get( + url="https://vws.vuforia.com/summary", + headers={ + "Date": rfc_1123_date(), + "Authorization": "bad_auth_token", + }, + timeout=30, + ) + assert response.status_code is not None + + @staticmethod + def test_httpx_unmocked_address_blocked() -> None: + """``MockVWS`` blocks ``httpx`` requests to non-Vuforia + addresses. + """ + sock = socket.socket() + sock.bind(("", 0)) + port = sock.getsockname()[1] + sock.close() + with MockVWS(), pytest.raises(expected_exception=httpx.ConnectError): + httpx.get(url=f"http://localhost:{port}", timeout=30) + + @staticmethod + def test_httpx_real_http() -> None: + """When ``real_http=True``, ``httpx`` requests to non-Vuforia + addresses are not blocked. + """ + sock = socket.socket() + sock.bind(("", 0)) + port = sock.getsockname()[1] + sock.close() + with ( + MockVWS(real_http=True), + pytest.raises(expected_exception=httpx.ConnectError), + ): + httpx.get(url=f"http://localhost:{port}", timeout=30) diff --git a/tests/mock_vws/test_respx_mock_usage.py b/tests/mock_vws/test_respx_mock_usage.py index 6d66a189c..3a5294225 100644 --- a/tests/mock_vws/test_respx_mock_usage.py +++ b/tests/mock_vws/test_respx_mock_usage.py @@ -1,4 +1,4 @@ -"""Tests for the usage of the mock for ``httpx`` via ``respx``.""" +"""Tests for ``MockVWS`` intercepting ``httpx`` requests.""" import json import socket @@ -9,7 +9,7 @@ import pytest from vws_auth_tools import authorization_header, rfc_1123_date -from mock_vws import MissingSchemeError, MockVWSForHttpx +from mock_vws import MockVWS from mock_vws.database import CloudDatabase, VuMarkDatabase from mock_vws.target import VuMarkTarget @@ -48,7 +48,7 @@ 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 MockVWS(): with pytest.raises(expected_exception=httpx.ConnectError): _request_unmocked_address() @@ -63,11 +63,10 @@ def test_default() -> None: @staticmethod def test_real_http() -> None: """When the ``real_http`` parameter is ``True``, requests to - unmocked - addresses are not stopped. + unmocked addresses are not stopped. """ with ( - MockVWSForHttpx(real_http=True), + MockVWS(real_http=True), pytest.raises(expected_exception=httpx.ConnectError), ): _request_unmocked_address() @@ -79,7 +78,7 @@ class TestResponseDelay: @staticmethod def test_default_no_delay() -> None: """By default, there is no response delay.""" - with MockVWSForHttpx(): + with MockVWS(): response = httpx.get( url="https://vws.vuforia.com/summary", headers={ @@ -96,7 +95,7 @@ def test_delay_causes_timeout() -> None: timeout, a ``ReadTimeout`` exception is raised. """ with ( - MockVWSForHttpx(response_delay_seconds=0.5), + MockVWS(response_delay_seconds=0.5), pytest.raises(expected_exception=httpx.ReadTimeout), ): httpx.get( @@ -113,7 +112,7 @@ 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): + with MockVWS(response_delay_seconds=0.1): response = httpx.get( url="https://vws.vuforia.com/summary", headers={ @@ -130,7 +129,7 @@ def test_custom_sleep_fn_called_on_delay() -> None: ``time.sleep`` for the non-timeout delay path. """ calls: list[float] = [] - with MockVWSForHttpx( + with MockVWS( response_delay_seconds=5.0, sleep_fn=calls.append, ): @@ -151,7 +150,7 @@ def test_custom_sleep_fn_called_on_timeout() -> None: """ calls: list[float] = [] with ( - MockVWSForHttpx( + MockVWS( response_delay_seconds=5.0, sleep_fn=calls.append, ), @@ -174,7 +173,7 @@ class TestCustomBaseURLs: @staticmethod def test_custom_base_vws_url() -> None: """It is possible to use a custom base VWS URL.""" - with MockVWSForHttpx( + with MockVWS( base_vws_url="https://vuforia.vws.example.com", real_http=False, ): @@ -193,7 +192,7 @@ def test_custom_base_vws_url() -> None: @staticmethod def test_custom_base_vwq_url() -> None: """It is possible to use a custom base cloud recognition URL.""" - with MockVWSForHttpx( + with MockVWS( base_vwq_url="https://vuforia.vwq.example.com", real_http=False, ): @@ -217,7 +216,7 @@ def test_custom_base_vws_url_with_path_prefix() -> None: """A custom base VWS URL with a path prefix intercepts at the prefix. """ - with MockVWSForHttpx( + with MockVWS( base_vws_url="https://vuforia.vws.example.com/prefix", real_http=False, ): @@ -237,7 +236,7 @@ def test_custom_base_vwq_url_with_path_prefix() -> None: """A custom base VWQ URL with a path prefix intercepts at the prefix. """ - with MockVWSForHttpx( + with MockVWS( base_vwq_url="https://vuforia.vwq.example.com/prefix", real_http=False, ): @@ -260,7 +259,7 @@ def test_vws_operations_work_with_path_prefix() -> None: database = CloudDatabase() base_vws_url = "https://vuforia.vws.example.com/prefix" - with MockVWSForHttpx(base_vws_url=base_vws_url) as mock: + with MockVWS(base_vws_url=base_vws_url) as mock: mock.add_cloud_database(cloud_database=database) request_path = "/targets" @@ -288,126 +287,6 @@ def test_vws_operations_work_with_path_prefix() -> None: assert response_json["result_code"] == "Success" assert response_json["results"] == [] - @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.""" @@ -416,7 +295,7 @@ class TestVWSEndpoints: def test_database_summary() -> None: """The database summary endpoint is accessible via httpx.""" database = CloudDatabase() - with MockVWSForHttpx() as mock: + with MockVWS() as mock: mock.add_cloud_database(cloud_database=database) response = httpx.get( url="https://vws.vuforia.com/summary", @@ -450,7 +329,7 @@ def test_vumark_bytes_response() -> None: date=date, request_path=request_path, ) - with MockVWSForHttpx() as mock: + with MockVWS() as mock: mock.add_vumark_database(vumark_database=database) response = httpx.post( url="https://vws.vuforia.com" + request_path, From 13c1696cfbaf6d90b9f4fb0005a91f22dbfeb080 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 22 Feb 2026 21:19:16 +0000 Subject: [PATCH 2/4] Fix pylint C0413 and C0402 in respx decorators Move TYPE_CHECKING guard to after all imports to fix wrong-import-position. Replace "MockRouter" in docstrings with "respx router" to fix spelling warning. Co-Authored-By: Claude Sonnet 4.6 --- src/mock_vws/_requests_mock_server/decorators.py | 6 +++--- src/mock_vws/_respx_mock_server/decorators.py | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/mock_vws/_requests_mock_server/decorators.py b/src/mock_vws/_requests_mock_server/decorators.py index f95446fa5..26a0b7967 100644 --- a/src/mock_vws/_requests_mock_server/decorators.py +++ b/src/mock_vws/_requests_mock_server/decorators.py @@ -12,9 +12,6 @@ from requests import PreparedRequest from responses import RequestsMock -if TYPE_CHECKING: - import respx - from mock_vws._mock_common import MissingSchemeError, RequestData from mock_vws._respx_mock_server.decorators import start_respx_router from mock_vws.database import CloudDatabase, VuMarkDatabase @@ -31,6 +28,9 @@ from .mock_web_query_api import MockVuforiaWebQueryAPI from .mock_web_services_api import MockVuforiaWebServicesAPI +if TYPE_CHECKING: + import respx + _ResponseType = tuple[int, Mapping[str, str], str | bytes] _MockCallback = Callable[[RequestData], _ResponseType] _ResponsesCallback = Callable[[PreparedRequest], _ResponseType] diff --git a/src/mock_vws/_respx_mock_server/decorators.py b/src/mock_vws/_respx_mock_server/decorators.py index a232474f0..b7fabe1ed 100644 --- a/src/mock_vws/_respx_mock_server/decorators.py +++ b/src/mock_vws/_respx_mock_server/decorators.py @@ -131,7 +131,7 @@ def start_respx_router( sleep_fn: Callable[[float], None], real_http: bool, ) -> respx.MockRouter: - """Configure and start a respx.MockRouter with Vuforia routes. + """Configure and start a respx router with Vuforia routes. Args: mock_vws_api: The VWS API handler. @@ -143,7 +143,7 @@ def start_respx_router( real_http: Whether to pass through unmatched requests. Returns: - A started respx.MockRouter. + A started respx router. """ router = respx.MockRouter( assert_all_called=False, From cbcd17856b2518ca84181d4b27dd4e9f34d516e2 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 22 Feb 2026 21:58:33 +0000 Subject: [PATCH 3/4] Suppress pyrefly false positive with inline ignore comments pyrefly's dual search path ("." and "src") causes the same class to be seen under two module paths, triggering a spurious bad-argument-type error when passing API objects to start_respx_router. Co-Authored-By: Claude Sonnet 4.6 --- src/mock_vws/_requests_mock_server/decorators.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/mock_vws/_requests_mock_server/decorators.py b/src/mock_vws/_requests_mock_server/decorators.py index 26a0b7967..fa48bb3ea 100644 --- a/src/mock_vws/_requests_mock_server/decorators.py +++ b/src/mock_vws/_requests_mock_server/decorators.py @@ -240,8 +240,8 @@ def __enter__(self) -> Self: self._mock.start() self._router = start_respx_router( - mock_vws_api=self._mock_vws_api, - mock_vwq_api=self._mock_vwq_api, + mock_vws_api=self._mock_vws_api, # pyrefly: ignore + mock_vwq_api=self._mock_vwq_api, # pyrefly: ignore base_vws_url=self._base_vws_url, base_vwq_url=self._base_vwq_url, response_delay_seconds=self._response_delay_seconds, From fdfb20abf8fc93f4c7519da9bf4e07932693e006 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Sun, 22 Feb 2026 22:04:57 +0000 Subject: [PATCH 4/4] Replace concrete type imports with Protocol to fix pyrefly false positive Define _APIHandler Protocol in _respx_mock_server/decorators.py so it no longer imports concrete classes from _requests_mock_server/. This removes the cross-module dependency that caused pyrefly to see the same class under two module paths (mock_vws.* vs src.mock_vws.*) and report a spurious bad-argument-type error. Co-Authored-By: Claude Sonnet 4.6 --- .../_requests_mock_server/decorators.py | 4 ++-- src/mock_vws/_respx_mock_server/decorators.py | 19 ++++++++++--------- 2 files changed, 12 insertions(+), 11 deletions(-) diff --git a/src/mock_vws/_requests_mock_server/decorators.py b/src/mock_vws/_requests_mock_server/decorators.py index fa48bb3ea..26a0b7967 100644 --- a/src/mock_vws/_requests_mock_server/decorators.py +++ b/src/mock_vws/_requests_mock_server/decorators.py @@ -240,8 +240,8 @@ def __enter__(self) -> Self: self._mock.start() self._router = start_respx_router( - mock_vws_api=self._mock_vws_api, # pyrefly: ignore - mock_vwq_api=self._mock_vwq_api, # pyrefly: ignore + mock_vws_api=self._mock_vws_api, + mock_vwq_api=self._mock_vwq_api, base_vws_url=self._base_vws_url, base_vwq_url=self._base_vwq_url, response_delay_seconds=self._response_delay_seconds, diff --git a/src/mock_vws/_respx_mock_server/decorators.py b/src/mock_vws/_respx_mock_server/decorators.py index b7fabe1ed..090695d0a 100644 --- a/src/mock_vws/_respx_mock_server/decorators.py +++ b/src/mock_vws/_respx_mock_server/decorators.py @@ -2,22 +2,23 @@ import re from collections.abc import Callable, Mapping +from typing import Protocol from urllib.parse import urlparse import httpx import respx -from mock_vws._mock_common import RequestData -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._mock_common import RequestData, Route _ResponseType = tuple[int, Mapping[str, str], str | bytes] +class _APIHandler(Protocol): + """An API handler with mock routes.""" + + routes: set[Route] + + def _to_request_data( request: httpx.Request, *, @@ -123,8 +124,8 @@ def callback(request: httpx.Request) -> httpx.Response: def start_respx_router( *, - mock_vws_api: MockVuforiaWebServicesAPI, - mock_vwq_api: MockVuforiaWebQueryAPI, + mock_vws_api: _APIHandler, + mock_vwq_api: _APIHandler, base_vws_url: str, base_vwq_url: str, response_delay_seconds: float,