Retain explicitly provided falsy custom transports

newsfragments/3092.change.rst

@@ -0,0 +1 @@
+Retain explicitly provided falsy custom transports instead of replacing them with defaults.

spelling_private_dict.txt

@@ -2,6 +2,7 @@ AuthenticationFailure
 BadImage
 ConnectionErrorPossiblyImageTooLarge
 DateRangeError
+Falsy
 ImageTooLarge
 InactiveProject
 JSONDecodeError
@@ -45,6 +46,7 @@ dev
 dict
 docstring
 enum
+falsy
 filename
 foo
 formdata

src/vws/async_query.py

@@ -60,7 +60,9 @@ def __init__(
         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 AsyncHTTPXTransport()
+        self._transport = (
+            transport if transport is not None else AsyncHTTPXTransport()
+        )
 
     async def aclose(self) -> None:
         """Close the underlying transport if it supports closing."""

src/vws/async_vumark_service.py

@@ -46,7 +46,9 @@ def __init__(
         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 AsyncHTTPXTransport()
+        self._transport = (
+            transport if transport is not None else AsyncHTTPXTransport()
+        )
 
     async def aclose(self) -> None:
         """Close the underlying transport if it supports closing."""

src/vws/async_vws.py

@@ -57,7 +57,9 @@ def __init__(
         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 AsyncHTTPXTransport()
+        self._transport = (
+            transport if transport is not None else AsyncHTTPXTransport()
+        )
 
     async def aclose(self) -> None:
         """Close the underlying transport if it supports closing."""

src/vws/query.py

@@ -56,7 +56,9 @@ def __init__(
         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()
+        self._transport = (
+            transport if transport is not None else RequestsTransport()
+        )
 
     def query(
         self,

src/vws/vumark_service.py

@@ -43,7 +43,9 @@ def __init__(
         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()
+        self._transport = (
+            transport if transport is not None else RequestsTransport()
+        )
 
     def generate_vumark_instance(
         self,

src/vws/vws.py

@@ -56,7 +56,9 @@ def __init__(
         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()
+        self._transport = (
+            transport if transport is not None else RequestsTransport()
+        )
 
     def make_request(
         self,

tests/test_transports.py

@@ -1,13 +1,24 @@
 """Tests for HTTP transport implementations."""
 
+import io
+import uuid
 from http import HTTPStatus
 
 import httpx
 import pytest
 import respx
 
+from vws import (
+    VWS,
+    AsyncCloudRecoService,
+    AsyncVuMarkService,
+    AsyncVWS,
+    CloudRecoService,
+    VuMarkService,
+)
 from vws.response import Response
 from vws.transports import AsyncHTTPXTransport, HTTPXTransport
+from vws.vumark_accept import VuMarkAccept
 
 
 class TestHTTPXTransport:
@@ -206,3 +217,191 @@ async def test_context_manager() -> None:
         assert route.called
         assert isinstance(response, Response)
         assert response.status_code == HTTPStatus.OK
+
+
+class _FalsyTransport:
+    """A sync transport that is falsy but protocol-conforming."""
+
+    def __bool__(self) -> bool:
+        """Return ``False`` so truthiness checks would skip this
+        transport.
+        """
+        return False
+
+    def close(self) -> None:
+        """Close the transport."""
+
+    def __call__(
+        self,
+        *,
+        method: str,
+        url: str,
+        headers: dict[str, str],
+        data: bytes,
+        request_timeout: float | tuple[float, float],
+    ) -> Response:
+        """Return a successful API response for the requested URL."""
+        del method, headers, request_timeout
+        if url.endswith("/query"):
+            body = '{"result_code":"Success","results":[]}'
+            return Response(
+                text=body,
+                url=url,
+                status_code=HTTPStatus.OK,
+                headers={},
+                request_body=data,
+                tell_position=0,
+                content=body.encode(),
+            )
+        if "/instances" in url:
+            content = b"vumark-bytes"
+            return Response(
+                text="",
+                url=url,
+                status_code=HTTPStatus.OK,
+                headers={},
+                request_body=data,
+                tell_position=0,
+                content=content,
+            )
+        body = '{"result_code":"Success","results":[]}'
+        return Response(
+            text=body,
+            url=url,
+            status_code=HTTPStatus.OK,
+            headers={},
+            request_body=data,
+            tell_position=0,
+            content=body.encode(),
+        )
+
+
+class _FalsyAsyncTransport:
+    """An async transport that is falsy but protocol-conforming."""
+
+    def __bool__(self) -> bool:
+        """Return ``False`` so truthiness checks would skip this
+        transport.
+        """
+        return False
+
+    async def aclose(self) -> None:
+        """Close the transport."""
+
+    async def __call__(
+        self,
+        *,
+        method: str,
+        url: str,
+        headers: dict[str, str],
+        data: bytes,
+        request_timeout: float | tuple[float, float],
+    ) -> Response:
+        """Return a successful API response for the requested URL."""
+        del method, headers, request_timeout
+        if url.endswith("/query"):
+            body = '{"result_code":"Success","results":[]}'
+            return Response(
+                text=body,
+                url=url,
+                status_code=HTTPStatus.OK,
+                headers={},
+                request_body=data,
+                tell_position=0,
+                content=body.encode(),
+            )
+        if "/instances" in url:
+            content = b"vumark-bytes"
+            return Response(
+                text="",
+                url=url,
+                status_code=HTTPStatus.OK,
+                headers={},
+                request_body=data,
+                tell_position=0,
+                content=content,
+            )
+        body = '{"result_code":"Success","results":[]}'
+        return Response(
+            text=body,
+            url=url,
+            status_code=HTTPStatus.OK,
+            headers={},
+            request_body=data,
+            tell_position=0,
+            content=body.encode(),
+        )
+
+
+def test_falsy_sync_transport_is_retained(
+    high_quality_image: io.BytesIO,
+) -> None:
+    """Falsy custom sync transports are not replaced by the default."""
+    access_key = uuid.uuid4().hex
+    secret_key = uuid.uuid4().hex
+    transport = _FalsyTransport()
+    assert not transport
+
+    targets = VWS(
+        server_access_key=access_key,
+        server_secret_key=secret_key,
+        transport=transport,
+    ).list_targets()
+    assert not targets
+
+    query_results = CloudRecoService(
+        client_access_key=access_key,
+        client_secret_key=secret_key,
+        transport=transport,
+    ).query(image=high_quality_image)
+    assert not query_results
+
+    vumark_bytes = VuMarkService(
+        server_access_key=access_key,
+        server_secret_key=secret_key,
+        transport=transport,
+    ).generate_vumark_instance(
+        target_id="target",
+        instance_id="instance",
+        accept=VuMarkAccept.PNG,
+    )
+    assert vumark_bytes == b"vumark-bytes"
+
+
+@pytest.mark.asyncio
+async def test_falsy_async_transport_is_retained(
+    high_quality_image: io.BytesIO,
+) -> None:
+    """Falsy custom async transports are not replaced by the default."""
+    access_key = uuid.uuid4().hex
+    secret_key = uuid.uuid4().hex
+    transport = _FalsyAsyncTransport()
+    assert not transport
+
+    async with AsyncVWS(
+        server_access_key=access_key,
+        server_secret_key=secret_key,
+        transport=transport,
+    ) as vws_client:
+        assert not await vws_client.list_targets()
+
+    async with AsyncCloudRecoService(
+        client_access_key=access_key,
+        client_secret_key=secret_key,
+        transport=transport,
+    ) as cloud_reco_client:
+        assert not await cloud_reco_client.query(image=high_quality_image)
+
+    async with AsyncVuMarkService(
+        server_access_key=access_key,
+        server_secret_key=secret_key,
+        transport=transport,
+    ) as vumark_client:
+        assert (
+            await vumark_client.generate_vumark_instance(
+                target_id="target",
+                instance_id="instance",
+                accept=VuMarkAccept.PNG,
+            )
+            == b"vumark-bytes"
+        )