From 92459c6ee1962cf1783cbe0238aa17ffcad9b4b7 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Tue, 24 Feb 2026 12:41:29 +0000 Subject: [PATCH 1/8] Add switchable HTTP transports (requests and httpx) Introduce a Transport protocol and two concrete implementations (RequestsTransport and HTTPXTransport) to allow users to choose between requests and httpx as the HTTP backend. All three client classes (VWS, CloudRecoService, VuMarkService) now accept an optional transport parameter, defaulting to RequestsTransport for backwards compatibility. The transport abstraction handles the full HTTP lifecycle and returns the library's Response dataclass. Co-Authored-By: Claude Haiku 4.5 --- docs/source/api-reference.rst | 4 + pyproject.toml | 1 + spelling_private_dict.txt | 1 + src/vws/__init__.py | 4 + src/vws/_vws_request.py | 37 ++++----- src/vws/query.py | 27 +++--- src/vws/transports.py | 151 ++++++++++++++++++++++++++++++++++ src/vws/vumark_service.py | 15 +++- src/vws/vws.py | 47 ++++++----- 9 files changed, 226 insertions(+), 61 deletions(-) create mode 100644 src/vws/transports.py diff --git a/docs/source/api-reference.rst b/docs/source/api-reference.rst index c94d810e..3fc7441c 100644 --- a/docs/source/api-reference.rst +++ b/docs/source/api-reference.rst @@ -20,3 +20,7 @@ API Reference .. automodule:: vws.response :undoc-members: :members: + +.. automodule:: vws.transports + :undoc-members: + :members: diff --git a/pyproject.toml b/pyproject.toml index 4d14b8f9..ea0d7cd5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,7 @@ dynamic = [ ] dependencies = [ "beartype>=0.22.9", + "httpx>=0.28.0", "requests>=2.32.3", "urllib3>=2.2.3", "vws-auth-tools>=2024.7.12", diff --git a/spelling_private_dict.txt b/spelling_private_dict.txt index 2329ef7f..8831d3d3 100644 --- a/spelling_private_dict.txt +++ b/spelling_private_dict.txt @@ -54,6 +54,7 @@ hmac html http https +httpx iff io issuecomment diff --git a/src/vws/__init__.py b/src/vws/__init__.py index a091641f..ad3d2de3 100644 --- a/src/vws/__init__.py +++ b/src/vws/__init__.py @@ -1,11 +1,15 @@ """A library for Vuforia Web Services.""" from .query import CloudRecoService +from .transports import HTTPXTransport, RequestsTransport, Transport from .vumark_service import VuMarkService from .vws import VWS __all__ = [ "VWS", "CloudRecoService", + "HTTPXTransport", + "RequestsTransport", + "Transport", "VuMarkService", ] diff --git a/src/vws/_vws_request.py b/src/vws/_vws_request.py index 2fc7bd3b..797860dc 100644 --- a/src/vws/_vws_request.py +++ b/src/vws/_vws_request.py @@ -2,11 +2,11 @@ API. """ -import requests from beartype import BeartypeConf, beartype from vws_auth_tools import authorization_header, rfc_1123_date from vws.response import Response +from vws.transports import Transport @beartype(conf=BeartypeConf(is_pep484_tower=True)) @@ -21,27 +21,30 @@ def target_api_request( base_vws_url: str, request_timeout_seconds: float | tuple[float, float], extra_headers: dict[str, str], + transport: Transport, ) -> Response: """Make a request to the Vuforia Target API. - This uses `requests` to make a request against https://vws.vuforia.com. - Args: content_type: The content type of the request. server_access_key: A VWS server access key. server_secret_key: A VWS server secret key. - method: The HTTP method which will be used in the request. - data: The request body which will be used in the request. - request_path: The path to the endpoint which will be used in the + method: The HTTP method which will be used in the + request. + data: The request body which will be used in the request. + request_path: The path to the endpoint which will be + used in the request. base_vws_url: The base URL for the VWS API. - request_timeout_seconds: The timeout for the request, as used by - ``requests.request``. This can be a float to set both the - connect and read timeouts, or a (connect, read) tuple. - extra_headers: Additional headers to include in the request. + request_timeout_seconds: The timeout for the request. + This can be a float to set both the connect and + read timeouts, or a (connect, read) tuple. + extra_headers: Additional headers to include in the + request. + transport: The HTTP transport to use for the request. Returns: - The response to the request made by `requests`. + The response to the request. """ date_string = rfc_1123_date() @@ -64,20 +67,10 @@ def target_api_request( url = base_vws_url.rstrip("/") + request_path - requests_response = requests.request( + return transport( method=method, url=url, headers=headers, data=data, timeout=request_timeout_seconds, ) - - return Response( - text=requests_response.text, - url=requests_response.url, - status_code=requests_response.status_code, - headers=dict(requests_response.headers), - request_body=requests_response.request.body, - tell_position=requests_response.raw.tell(), - content=bytes(requests_response.content), - ) diff --git a/src/vws/query.py b/src/vws/query.py index f5046bd1..fcca282d 100644 --- a/src/vws/query.py +++ b/src/vws/query.py @@ -6,7 +6,6 @@ from http import HTTPMethod, HTTPStatus from typing import Any, BinaryIO -import requests from beartype import BeartypeConf, beartype from urllib3.filepost import encode_multipart_formdata from vws_auth_tools import authorization_header, rfc_1123_date @@ -24,7 +23,7 @@ ) from vws.include_target_data import CloudRecoIncludeTargetData from vws.reports import QueryResult, TargetData -from vws.response import Response +from vws.transports import RequestsTransport, Transport _ImageType = io.BytesIO | BinaryIO @@ -50,21 +49,26 @@ def __init__( client_secret_key: str, base_vwq_url: str = "https://cloudreco.vuforia.com", request_timeout_seconds: float | tuple[float, float] = 30.0, + transport: Transport | None = None, ) -> None: """ Args: client_access_key: A VWS client access key. client_secret_key: A VWS client secret key. base_vwq_url: The base URL for the VWQ API. - request_timeout_seconds: The timeout for each HTTP request, as - used by ``requests.request``. This can be a float to set - both the connect and read timeouts, or a (connect, read) - tuple. + request_timeout_seconds: The timeout for each + HTTP request. This can be a float to set both + the connect and read timeouts, or a + (connect, read) tuple. + transport: The HTTP transport to use for + requests. Defaults to + ``RequestsTransport()``. """ self._client_access_key = client_access_key self._client_secret_key = client_secret_key self._base_vwq_url = base_vwq_url self._request_timeout_seconds = request_timeout_seconds + self._transport = transport or RequestsTransport() def query( self, @@ -143,22 +147,13 @@ def query( "Content-Type": content_type_header, } - requests_response = requests.request( + response = self._transport( method=method, url=self._base_vwq_url.rstrip("/") + request_path, headers=headers, data=content, timeout=self._request_timeout_seconds, ) - response = Response( - text=requests_response.text, - url=requests_response.url, - status_code=requests_response.status_code, - headers=dict(requests_response.headers), - request_body=requests_response.request.body, - tell_position=requests_response.raw.tell(), - content=bytes(requests_response.content), - ) if response.status_code == HTTPStatus.REQUEST_ENTITY_TOO_LARGE: raise RequestEntityTooLargeError(response=response) diff --git a/src/vws/transports.py b/src/vws/transports.py new file mode 100644 index 00000000..d9107691 --- /dev/null +++ b/src/vws/transports.py @@ -0,0 +1,151 @@ +"""HTTP transport implementations for VWS clients.""" + +from __future__ import annotations + +from typing import Protocol, runtime_checkable + +import httpx +import requests +from beartype import BeartypeConf, beartype + +from vws.response import Response + + +@runtime_checkable +class Transport(Protocol): + """Protocol for HTTP transports used by VWS clients. + + A transport is a callable that makes an HTTP request and + returns a ``Response``. + """ + + def __call__( + self, + *, + method: str, + url: str, + headers: dict[str, str], + data: bytes, + timeout: float | tuple[float, float], + ) -> Response: + """Make an HTTP request. + + Args: + method: The HTTP method (e.g. "GET", "POST"). + url: The full URL to request. + headers: Headers to send with the request. + data: The request body as bytes. + timeout: The timeout for the request. A float + sets both the connect and read timeouts. A + (connect, read) tuple sets them individually. + + Returns: + A Response populated from the HTTP response. + """ + ... + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +class RequestsTransport: + """HTTP transport using the ``requests`` library. + + This is the default transport. + """ + + def __call__( + self, + *, + method: str, + url: str, + headers: dict[str, str], + data: bytes, + timeout: float | tuple[float, float], + ) -> Response: + """Make an HTTP request using ``requests``. + + Args: + method: The HTTP method. + url: The full URL. + headers: Request headers. + data: The request body. + timeout: The request timeout. + + Returns: + A Response populated from the requests response. + """ + requests_response = requests.request( + method=method, + url=url, + headers=headers, + data=data, + timeout=timeout, + ) + + return Response( + text=requests_response.text, + url=requests_response.url, + status_code=requests_response.status_code, + headers=dict(requests_response.headers), + request_body=requests_response.request.body, + tell_position=requests_response.raw.tell(), + content=bytes(requests_response.content), + ) + + +@beartype(conf=BeartypeConf(is_pep484_tower=True)) +class HTTPXTransport: + """HTTP transport using the ``httpx`` library. + + Use this transport for environments where ``httpx`` is + preferred over ``requests``. + """ + + def __call__( + self, + *, + method: str, + url: str, + headers: dict[str, str], + data: bytes, + timeout: float | tuple[float, float], + ) -> Response: + """Make an HTTP request using ``httpx``. + + Args: + method: The HTTP method. + url: The full URL. + headers: Request headers. + data: The request body. + timeout: The request timeout. + + Returns: + A Response populated from the httpx response. + """ + if isinstance(timeout, tuple): + connect_timeout, read_timeout = timeout + httpx_timeout = httpx.Timeout( + connect=connect_timeout, + read=read_timeout, + write=None, + pool=None, + ) + else: + httpx_timeout = httpx.Timeout(timeout=timeout) + + httpx_response = httpx.request( + method=method, + url=url, + headers=headers, + content=data, + timeout=httpx_timeout, + ) + + return Response( + text=httpx_response.text, + url=str(object=httpx_response.url), + status_code=httpx_response.status_code, + headers=dict(httpx_response.headers), + request_body=bytes(httpx_response.request.content), + tell_position=0, + content=bytes(httpx_response.content), + ) diff --git a/src/vws/vumark_service.py b/src/vws/vumark_service.py index a23ac33f..adb89ca7 100644 --- a/src/vws/vumark_service.py +++ b/src/vws/vumark_service.py @@ -20,6 +20,7 @@ TooManyRequestsError, UnknownTargetError, ) +from vws.transports import RequestsTransport, Transport from vws.vumark_accept import VuMarkAccept @@ -33,21 +34,26 @@ def __init__( server_secret_key: str, base_vws_url: str = "https://vws.vuforia.com", request_timeout_seconds: float | tuple[float, float] = 30.0, + transport: Transport | None = None, ) -> None: """ Args: server_access_key: A VWS server access key. server_secret_key: A VWS server secret key. base_vws_url: The base URL for the VWS API. - request_timeout_seconds: The timeout for each HTTP request, as - used by ``requests.request``. This can be a float to set - both the connect and read timeouts, or a (connect, read) - tuple. + request_timeout_seconds: The timeout for each + HTTP request. This can be a float to set both + the connect and read timeouts, or a + (connect, read) tuple. + transport: The HTTP transport to use for + requests. Defaults to + ``RequestsTransport()``. """ self._server_access_key = server_access_key self._server_secret_key = server_secret_key self._base_vws_url = base_vws_url self._request_timeout_seconds = request_timeout_seconds + self._transport = transport or RequestsTransport() def generate_vumark_instance( self, @@ -109,6 +115,7 @@ def generate_vumark_instance( base_vws_url=self._base_vws_url, request_timeout_seconds=self._request_timeout_seconds, extra_headers={"Accept": accept}, + transport=self._transport, ) if ( diff --git a/src/vws/vws.py b/src/vws/vws.py index 50b4f5e7..01315e6a 100644 --- a/src/vws/vws.py +++ b/src/vws/vws.py @@ -43,6 +43,7 @@ TargetSummaryReport, ) from vws.response import Response +from vws.transports import RequestsTransport, Transport _ImageType = io.BytesIO | BinaryIO @@ -68,21 +69,26 @@ def __init__( server_secret_key: str, base_vws_url: str = "https://vws.vuforia.com", request_timeout_seconds: float | tuple[float, float] = 30.0, + transport: Transport | None = None, ) -> None: """ Args: server_access_key: A VWS server access key. server_secret_key: A VWS server secret key. base_vws_url: The base URL for the VWS API. - request_timeout_seconds: The timeout for each HTTP request, as - used by ``requests.request``. This can be a float to set - both the connect and read timeouts, or a (connect, read) - tuple. + request_timeout_seconds: The timeout for each + HTTP request. This can be a float to set both + the connect and read timeouts, or a + (connect, read) tuple. + transport: The HTTP transport to use for + requests. Defaults to + ``RequestsTransport()``. """ self._server_access_key = server_access_key self._server_secret_key = server_secret_key self._base_vws_url = base_vws_url self._request_timeout_seconds = request_timeout_seconds + self._transport = transport or RequestsTransport() def make_request( self, @@ -96,29 +102,31 @@ def make_request( ) -> Response: """Make a request to the Vuforia Target API. - This uses `requests` to make a request against Vuforia. - Args: - method: The HTTP method which will be used in the request. - data: The request body which will be used in the request. - request_path: The path to the endpoint which will be used in the + method: The HTTP method which will be used in + the request. + data: The request body which will be used in the request. - expected_result_code: See "VWS API Result Codes" on + request_path: The path to the endpoint which + will be used in the request. + expected_result_code: See + "VWS API Result Codes" on https://developer.vuforia.com/library/web-api/cloud-targets-web-services-api. content_type: The content type of the request. - extra_headers: Additional headers to include in the request. + extra_headers: Additional headers to include in + the request. Returns: - The response to the request made by `requests`. + The response to the request. Raises: - ~vws.exceptions.custom_exceptions.ServerError: There is an error - with Vuforia's servers. - ~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is - rate limiting access. - json.JSONDecodeError: The server did not respond with valid JSON. - This may happen if the server address is not a valid Vuforia - server. + ~vws.exceptions.custom_exceptions.ServerError: + There is an error with Vuforia's servers. + ~vws.exceptions.vws_exceptions.TooManyRequestsError: + Vuforia is rate limiting access. + json.JSONDecodeError: The server did not respond + with valid JSON. This may happen if the + server address is not a valid Vuforia server. """ response = target_api_request( content_type=content_type, @@ -130,6 +138,7 @@ def make_request( base_vws_url=self._base_vws_url, request_timeout_seconds=self._request_timeout_seconds, extra_headers=extra_headers or {}, + transport=self._transport, ) if ( From e9041ec56912d55ee60731ff8c797aeaacfe37f5 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Tue, 24 Feb 2026 12:44:56 +0000 Subject: [PATCH 2/8] Remove transport re-exports from vws.__init__ Users should import transports from vws.transports directly. Co-Authored-By: Claude Opus 4.6 --- pyproject.toml | 2 ++ src/vws/__init__.py | 4 ---- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ea0d7cd5..4fd78dfa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -358,6 +358,8 @@ ignore_path = [ # Ideally we would limit the paths to the source code where we want to ignore names, # but Vulture does not enable this. ignore_names = [ + # Public API classes imported by users from vws.transports + "HTTPXTransport", # pytest configuration "pytest_collect_file", "pytest_collection_modifyitems", diff --git a/src/vws/__init__.py b/src/vws/__init__.py index ad3d2de3..a091641f 100644 --- a/src/vws/__init__.py +++ b/src/vws/__init__.py @@ -1,15 +1,11 @@ """A library for Vuforia Web Services.""" from .query import CloudRecoService -from .transports import HTTPXTransport, RequestsTransport, Transport from .vumark_service import VuMarkService from .vws import VWS __all__ = [ "VWS", "CloudRecoService", - "HTTPXTransport", - "RequestsTransport", - "Transport", "VuMarkService", ] From c6c313bc84e86466daf6639fb8b3831e6744863d Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Tue, 24 Feb 2026 12:54:42 +0000 Subject: [PATCH 3/8] Fix pylint issues in transports and vumark_service Make VuMarkService.__init__ keyword-only to fix too-many-positional-arguments. Suppress unnecessary-ellipsis on Transport protocol method body. Co-Authored-By: Claude Opus 4.6 --- src/vws/transports.py | 2 +- src/vws/vumark_service.py | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/src/vws/transports.py b/src/vws/transports.py index d9107691..a719f8da 100644 --- a/src/vws/transports.py +++ b/src/vws/transports.py @@ -42,7 +42,7 @@ def __call__( Returns: A Response populated from the HTTP response. """ - ... + ... # pylint: disable=unnecessary-ellipsis @beartype(conf=BeartypeConf(is_pep484_tower=True)) diff --git a/src/vws/vumark_service.py b/src/vws/vumark_service.py index adb89ca7..47d68d94 100644 --- a/src/vws/vumark_service.py +++ b/src/vws/vumark_service.py @@ -30,6 +30,7 @@ class VuMarkService: def __init__( self, + *, server_access_key: str, server_secret_key: str, base_vws_url: str = "https://vws.vuforia.com", From 5b3fd2c0703341d82f1b22017467cf148ca44fcd Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Tue, 24 Feb 2026 13:00:18 +0000 Subject: [PATCH 4/8] Remove future annotations, add HTTPXTransport tests Remove `from __future__ import annotations` from transports.py. Add tests for HTTPXTransport covering both float and tuple timeout branches to achieve 100% coverage. Co-Authored-By: Claude Opus 4.6 --- src/vws/transports.py | 2 -- tests/test_transports.py | 57 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 57 insertions(+), 2 deletions(-) create mode 100644 tests/test_transports.py diff --git a/src/vws/transports.py b/src/vws/transports.py index a719f8da..c4e46665 100644 --- a/src/vws/transports.py +++ b/src/vws/transports.py @@ -1,7 +1,5 @@ """HTTP transport implementations for VWS clients.""" -from __future__ import annotations - from typing import Protocol, runtime_checkable import httpx diff --git a/tests/test_transports.py b/tests/test_transports.py new file mode 100644 index 00000000..70d55186 --- /dev/null +++ b/tests/test_transports.py @@ -0,0 +1,57 @@ +"""Tests for HTTP transport implementations.""" + +from http import HTTPStatus + +import httpx +import respx + +from vws.response import Response +from vws.transports import HTTPXTransport + + +class TestHTTPXTransport: + """Tests for ``HTTPXTransport``.""" + + @respx.mock + def test_float_timeout(self) -> None: + """``HTTPXTransport`` works with a float timeout.""" + respx.post(url="https://example.com/test").mock( + return_value=httpx.Response( + status_code=HTTPStatus.OK, + text="OK", + ), + ) + transport = HTTPXTransport() + response = transport( + method="POST", + url="https://example.com/test", + headers={"Content-Type": "text/plain"}, + data=b"hello", + timeout=30.0, + ) + assert isinstance(response, Response) + assert response.status_code == HTTPStatus.OK + assert response.text == "OK" + assert response.tell_position == 0 + + @respx.mock + def test_tuple_timeout(self) -> None: + """``HTTPXTransport`` works with a (connect, read) timeout + tuple. + """ + respx.post(url="https://example.com/test").mock( + return_value=httpx.Response( + status_code=HTTPStatus.OK, + text="OK", + ), + ) + transport = HTTPXTransport() + response = transport( + method="POST", + url="https://example.com/test", + headers={"Content-Type": "text/plain"}, + data=b"hello", + timeout=(5.0, 30.0), + ) + assert isinstance(response, Response) + assert response.status_code == HTTPStatus.OK From fbff605d2419b88b7ed2b833a05cf05c4f3d6136 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Tue, 24 Feb 2026 13:12:44 +0000 Subject: [PATCH 5/8] Address Bugbot review feedback on HTTPXTransport - Set tell_position to len(content) instead of 0 - Add follow_redirects=True to httpx.request() - Handle request_body None when content is empty Co-Authored-By: Claude Opus 4.6 --- src/vws/transports.py | 10 +++++++--- tests/test_transports.py | 2 +- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/src/vws/transports.py b/src/vws/transports.py index c4e46665..bc0f9688 100644 --- a/src/vws/transports.py +++ b/src/vws/transports.py @@ -136,14 +136,18 @@ def __call__( headers=headers, content=data, timeout=httpx_timeout, + follow_redirects=True, ) + content = bytes(httpx_response.content) + request_content = httpx_response.request.content + return Response( text=httpx_response.text, url=str(object=httpx_response.url), status_code=httpx_response.status_code, headers=dict(httpx_response.headers), - request_body=bytes(httpx_response.request.content), - tell_position=0, - content=bytes(httpx_response.content), + request_body=bytes(request_content) or None, + tell_position=len(content), + content=content, ) diff --git a/tests/test_transports.py b/tests/test_transports.py index 70d55186..bbd6e633 100644 --- a/tests/test_transports.py +++ b/tests/test_transports.py @@ -32,7 +32,7 @@ def test_float_timeout(self) -> None: assert isinstance(response, Response) assert response.status_code == HTTPStatus.OK assert response.text == "OK" - assert response.tell_position == 0 + assert response.tell_position == len(b"OK") @respx.mock def test_tuple_timeout(self) -> None: From c3076542dcede6bb9f54799edbdf4204cb21b91a Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Tue, 24 Feb 2026 13:16:29 +0000 Subject: [PATCH 6/8] Suppress pylint no-self-use for test files Use @staticmethod on test methods that don't reference self. Co-Authored-By: Claude Opus 4.6 --- tests/test_transports.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/test_transports.py b/tests/test_transports.py index bbd6e633..972c83a1 100644 --- a/tests/test_transports.py +++ b/tests/test_transports.py @@ -12,8 +12,9 @@ class TestHTTPXTransport: """Tests for ``HTTPXTransport``.""" + @staticmethod @respx.mock - def test_float_timeout(self) -> None: + def test_float_timeout() -> None: """``HTTPXTransport`` works with a float timeout.""" respx.post(url="https://example.com/test").mock( return_value=httpx.Response( @@ -34,8 +35,9 @@ def test_float_timeout(self) -> None: assert response.text == "OK" assert response.tell_position == len(b"OK") + @staticmethod @respx.mock - def test_tuple_timeout(self) -> None: + def test_tuple_timeout() -> None: """``HTTPXTransport`` works with a (connect, read) timeout tuple. """ From e2bf8ce06eaaf63e8434b8a4509238e91f5dc38c Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Tue, 24 Feb 2026 13:34:18 +0000 Subject: [PATCH 7/8] Make float timeout consistent with tuple timeout in HTTPXTransport Both paths now only set connect and read timeouts, matching requests behavior. Previously the float path also set write and pool timeouts. Co-Authored-By: Claude Opus 4.6 --- src/vws/transports.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/vws/transports.py b/src/vws/transports.py index bc0f9688..9e4c550c 100644 --- a/src/vws/transports.py +++ b/src/vws/transports.py @@ -128,7 +128,12 @@ def __call__( pool=None, ) else: - httpx_timeout = httpx.Timeout(timeout=timeout) + httpx_timeout = httpx.Timeout( + connect=timeout, + read=timeout, + write=None, + pool=None, + ) httpx_response = httpx.request( method=method, From b977b52f5e853a622e26cfcb818e1010ef846551 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Tue, 24 Feb 2026 14:23:36 +0000 Subject: [PATCH 8/8] Assert httpx mock routes are called in transport tests Co-Authored-By: Claude Opus 4.6 --- tests/test_transports.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/test_transports.py b/tests/test_transports.py index 972c83a1..7b77107c 100644 --- a/tests/test_transports.py +++ b/tests/test_transports.py @@ -16,7 +16,7 @@ class TestHTTPXTransport: @respx.mock def test_float_timeout() -> None: """``HTTPXTransport`` works with a float timeout.""" - respx.post(url="https://example.com/test").mock( + route = respx.post(url="https://example.com/test").mock( return_value=httpx.Response( status_code=HTTPStatus.OK, text="OK", @@ -30,6 +30,7 @@ def test_float_timeout() -> None: data=b"hello", timeout=30.0, ) + assert route.called assert isinstance(response, Response) assert response.status_code == HTTPStatus.OK assert response.text == "OK" @@ -41,7 +42,7 @@ def test_tuple_timeout() -> None: """``HTTPXTransport`` works with a (connect, read) timeout tuple. """ - respx.post(url="https://example.com/test").mock( + route = respx.post(url="https://example.com/test").mock( return_value=httpx.Response( status_code=HTTPStatus.OK, text="OK", @@ -55,5 +56,6 @@ def test_tuple_timeout() -> None: data=b"hello", timeout=(5.0, 30.0), ) + assert route.called assert isinstance(response, Response) assert response.status_code == HTTPStatus.OK