diff --git a/CHANGELOG.md b/CHANGELOG.md index 685741ef..8209847c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ * (Major): `initialize_client()` no longer raises when registration fails. A failed registration is logged and the client keeps running, the same way for both clients. * (Major): Upgraded to yggdrasil-engine 3, new engine built with PyO3 for better performance. * (Minor): Feature fetches are retried the same way by both clients: on a 500, 502 or 504, or when no response arrives, with a short randomised backoff between attempts (up to about 0.5s, 1s and 2s). Before, retries were immediate. -* (Minor): `UnleashClient` takes `transport=` to choose how its requests are sent: a `RequestsTransport` with your own `requests.Session`, for a proxy, TLS settings or connection reuse, or any object with an `execute(Request) -> Response` method, for another HTTP library. See [custom HTTP transport](./README.md#custom-http-transport). +* (Minor): `UnleashClient` and `AsyncUnleashClient` take `transport=` to choose how their requests are sent: a `RequestsTransport` with custom `requests.Session`, an `AiohttpTransport` with custom `aiohttp.ClientSession`, for a proxy, TLS settings or connection reuse, or any object with an `execute(Request) -> Response` method, for another HTTP library. See [custom HTTP transport](./README.md#custom-http-transport). * (Minor): New `FETCH_COMPLETED` event, emitted after every feature fetch with how it ended: `CHANGED`, `NOT_CHANGED`, `UNAUTHORIZED`, `RATE_LIMITED`, `UNAVAILABLE` or `INVALID_RESPONSE`. * (Minor): `feature_definitions()` reports whether each feature is enabled in the environment. * (Minor): New `evaluate_all_toggles()` evaluates every feature for one context in a single call, without recording usage metrics. diff --git a/README.md b/README.md index a623829b..643489ba 100644 --- a/README.md +++ b/README.md @@ -149,7 +149,7 @@ Settings can't be changed after the client is created. | scheduler_executor | Name of the executor to use on your custom `scheduler`. Required with `scheduler`. Sync client only. | None | | multiple_instance_mode | How to handle multiple client instances (`BLOCK`, `WARN`, `SILENTLY_ALLOW`). | `WARN` | | event_callback | Function that receives impression and lifecycle events. See [events and impression data](#events-and-impression-data). | None | -| transport | How requests are sent. Can't be combined with `http` (which will be removed). Sync client only yet. See [custom HTTP transport](#custom-http-transport). | `RequestsTransport` | +| transport | How requests are sent. Can't be combined with `http` (which will be removed). See [custom HTTP transport](#custom-http-transport). | `RequestsTransport` or `AiohttpTransport` | #### `IdentityOptions` @@ -471,12 +471,12 @@ client = UnleashClient( ### Custom HTTP transport -By default, `UnleashClient` sends its requests with `requests`, on a new session for every request. Pass `transport=` to choose how requests are sent: +By default, `UnleashClient` sends its requests with `requests`, on a new session for every request, and `AsyncUnleashClient` with one `aiohttp` session. Pass `transport=` to choose how requests are sent: -- `RequestsTransport(session)` to keep `requests` but configure the session yourself; -- any object with an `execute(request) -> Response` method, to use another HTTP libraryl. +- `RequestsTransport(session)` or `AiohttpTransport(session)` to keep the default library but configure the session; +- any object with an `execute(request) -> Response` method, to use another HTTP library. For `AsyncUnleashClient`, `execute` is `async`. -#### Your own `requests` session +#### Custom `requests` session The session is used for every request and never closed by the client: close it yourself after `client.destroy()`. @@ -515,6 +515,39 @@ A session also keeps connections open between polls, instead of opening a new on `RequestsTransport` takes `headers` and `timeout` with or without a session. The SDK's own headers take precedence over custom ones, and an `Authorization` header raises `ValueError`: pass the API token as `api_key`. +#### Custom `aiohttp` session + +The same for `AsyncUnleashClient`, with an `aiohttp.ClientSession`. Create it inside a coroutine, since aiohttp needs a running event loop, and close it after the client: + +```python +import ssl +import aiohttp +from UnleashClient import AiohttpTransport +from UnleashClient.clients.async_unleash_client import AsyncUnleashClient + +# examples +tls = ssl.create_default_context(cafile="/etc/ssl/certs/internal-ca.pem") +tls.load_cert_chain("/etc/unleash/client.crt", "/etc/unleash/client.key") + +async def main(): + async with aiohttp.ClientSession( + trust_env=True, + connector=aiohttp.TCPConnector(ssl=tls), + trace_configs=[my_trace_config], # e.g. aiohttp tracing for OpenTelemetry + ) as session: + async with AsyncUnleashClient( + url="https://YOUR-API-URL", + app_name="my-python-app", + api_key="", + transport=AiohttpTransport(session, headers={"X-Team": "checkout"}, timeout=10), + ) as client: + ... +``` + +`AiohttpTransport`'s `timeout` applies to every request, in place of the session's own. Every status reaches the client as a response, even on a session created with `raise_for_status=True`. + +An `AiohttpTransport()` without a session opens its own; if you create one and pass it in, close it with `await transport.aclose()` after the client. + #### Another HTTP library A transport receives a `Request` (`method`, `url`, `headers`, `params`, `body`) and returns a `Response` (`status`, `headers`, `body` as bytes). Every status is a response: raise only when no response arrived, such as on a connection error or a timeout. The client logs the error and reports a failed request. For example, with [httpx](https://www.python-httpx.org/): @@ -546,6 +579,38 @@ with httpx.Client(timeout=30) as http_client: ) ``` +For `AsyncUnleashClient`, `execute` is `async`. The same with `httpx.AsyncClient`: + +```python +import httpx +from UnleashClient import Request, Response +from UnleashClient.clients.async_unleash_client import AsyncUnleashClient + +class HttpxAsyncTransport: + def __init__(self, client: httpx.AsyncClient) -> None: + self._client = client + + async def execute(self, request: Request) -> Response: + resp = await self._client.request( + request.method, + request.url, + headers=dict(request.headers), + params=dict(request.params), + content=request.body, + ) + return Response(status=resp.status_code, headers=resp.headers, body=resp.content) + +async def main(): + async with httpx.AsyncClient(timeout=30) as http_client: + async with AsyncUnleashClient( + url="https://YOUR-API-URL", + app_name="my-python-app", + api_key="", + transport=HttpxAsyncTransport(http_client), + ) as client: + ... +``` + The response headers can be a plain `dict`: the client looks them up case-insensitively. #### Tests without a server @@ -580,11 +645,33 @@ def test_new_checkout_is_on(tmp_path): assert client.is_enabled("new-checkout") ``` -Both examples are in [`examples/`](examples/) and run as part of the SDK's tests. +For `AsyncUnleashClient`, make `execute` `async` and the test a coroutine, here with [pytest-asyncio](https://pytest-asyncio.readthedocs.io/): + +```python +import pytest +from UnleashClient.clients.async_unleash_client import AsyncUnleashClient + + +class FakeAsyncTransport(FakeTransport): + async def execute(self, request: Request) -> Response: + return super().execute(request) + +@pytest.mark.asyncio +async def test_new_checkout_is_on(tmp_path): + transport = FakeAsyncTransport(flag("new-checkout", True)) + + async with AsyncUnleashClient( + "http://unleash.test/api", "my-app-tests", + transport=transport, cache_directory=str(tmp_path), + ) as client: + assert client.is_enabled("new-checkout") +``` + +All four examples are in [`examples/`](examples/) and run as part of the SDK's tests. #### Things to know -- **Don't retry in your transport.** The client already retries failed feature fetches, with a backoff. A session with a retrying adapter mounted logs a warning, because failed fetches would be retried twice over. +- **Don't retry in your transport.** The client already retries failed feature fetches, with a backoff. A `requests` session with a retrying adapter mounted logs a warning, because failed fetches would be retried twice over. aiohttp has no retry setting, but resends a GET once by itself when the server drops a reused connection; that's expected, and not a retry to turn off. - **Don't log request headers.** They include your API token. - **Don't add an HTTP cache.** The client already sends `If-None-Match` and handles `304 Not Modified`. A caching session (such as CacheControl or requests-cache) turns each 304 into its cached 200, so the client reloads unchanged flags on every poll. The flags stay correct. - **Create the session in the process that uses it.** With a forking server, such as gunicorn with `--preload` or uWSGI without `lazy-apps`, build the session together with the client in each worker. Forked processes must not share open connections. See [running in multi-process setups](#running-in-multi-process-setups). diff --git a/UnleashClient/__init__.py b/UnleashClient/__init__.py index 495a7fe5..c3618d3d 100644 --- a/UnleashClient/__init__.py +++ b/UnleashClient/__init__.py @@ -13,9 +13,10 @@ PollingOptions, StreamingOptions, ) -from .transport import Request, RequestsTransport, Response, Transport +from .transport import AsyncTransport, Request, RequestsTransport, Response, Transport if TYPE_CHECKING: + from ._aiohttp_transport import AiohttpTransport from .clients.async_unleash_client import AsyncUnleashClient @@ -24,4 +25,8 @@ def __getattr__(name: str): from .clients.async_unleash_client import AsyncUnleashClient # noqa: PLC0415 return AsyncUnleashClient + if name == "AiohttpTransport": + from ._aiohttp_transport import AiohttpTransport # noqa: PLC0415 + + return AiohttpTransport raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/UnleashClient/_aiohttp_transport.py b/UnleashClient/_aiohttp_transport.py index e34aa0bb..64dd637c 100644 --- a/UnleashClient/_aiohttp_transport.py +++ b/UnleashClient/_aiohttp_transport.py @@ -3,14 +3,15 @@ and returns the response. """ -from typing import Optional +from typing import Mapping, Optional from UnleashClient.constants import REQUEST_TIMEOUT from UnleashClient.errors import AlreadyClosedError -from UnleashClient.transport import Request, Response +from UnleashClient.transport import Request, Response, _reject_authorization try: import aiohttp + from multidict import CIMultiDict except ImportError as exc: # pragma: no cover raise ImportError( "AsyncUnleashClient requires aiohttp, which is not installed. " @@ -22,21 +23,47 @@ class AiohttpTransport: """ Sends requests with ``aiohttp``, on one session reused across requests. - The session is opened on the first request, since aiohttp needs a running - event loop to create one, and closed by :meth:`aclose`. A closed transport - stays closed. - :param timeout: Seconds to wait for the connection and for each read. + Example:: + + async with aiohttp.ClientSession(trust_env=True) as session: + async with AsyncUnleashClient( + url, app_name, transport=AiohttpTransport(session) + ) as client: + ... + + :param session: Session to send every request on. It is never closed: its + creator closes it, after the client. Without one, the + transport opens its own on the first request, and + :meth:`aclose` closes it. + :param headers: Extra headers sent with every request. The SDK's own headers + take precedence over them. + :param timeout: Seconds to wait for the connection and for each read. Applies + to every request, in place of the session's own timeout. + :raises ValueError: If ``headers`` contains an ``Authorization`` header: pass + the API key as ``api_key`` to the client. """ - def __init__(self, *, timeout: float = REQUEST_TIMEOUT) -> None: + def __init__( + self, + session: Optional["aiohttp.ClientSession"] = None, + *, + headers: Optional[Mapping[str, str]] = None, + timeout: float = REQUEST_TIMEOUT, + ) -> None: + self._user_session = session + self._headers = dict(headers or {}) self._timeout = timeout self._session: Optional["aiohttp.ClientSession"] = None self._is_closed: bool = False + _reject_authorization(self._headers, "AiohttpTransport") + def _get_session(self) -> "aiohttp.ClientSession": if self._is_closed: raise AlreadyClosedError() + if self._user_session is not None: + return self._user_session if self._session is None: self._session = aiohttp.ClientSession() return self._session @@ -49,21 +76,25 @@ def _client_timeout(self) -> "aiohttp.ClientTimeout": ) async def execute(self, request: Request) -> Response: + headers = CIMultiDict(self._headers) + headers.update(request.headers) async with self._get_session().request( request.method, request.url, - headers=dict(request.headers), + headers=headers, params=dict(request.params) or None, data=request.body, timeout=self._client_timeout(), + raise_for_status=False, ) as resp: body = await resp.read() return Response(status=resp.status, headers=resp.headers, body=body) async def aclose(self) -> None: """ - Close the session. Every later request raises - :class:`AlreadyClosedError`. Closing a closed transport does nothing. + Put the transport out of service, and close its session if the transport + opened it. Every later request raises :class:`AlreadyClosedError`. + Closing a closed transport does nothing. """ self._is_closed = True session, self._session = self._session, None diff --git a/UnleashClient/_async_unleash_api_client.py b/UnleashClient/_async_unleash_api_client.py index 6e71226b..67cbc042 100644 --- a/UnleashClient/_async_unleash_api_client.py +++ b/UnleashClient/_async_unleash_api_client.py @@ -14,6 +14,7 @@ _emit_fetch_completed, _failure_status, _fetched_status, + _header, _log_resp_info, _normalized_url, _text, @@ -141,7 +142,7 @@ async def _fetch_features( "Unleash Client feature fetch failed!" ) - fetched_etag = resp.headers.get("etag", "") + fetched_etag = _header(resp, "etag") if resp.status == 304: return ( diff --git a/UnleashClient/clients/async_unleash_client.py b/UnleashClient/clients/async_unleash_client.py index a3e3073f..01bf60b4 100644 --- a/UnleashClient/clients/async_unleash_client.py +++ b/UnleashClient/clients/async_unleash_client.py @@ -10,6 +10,7 @@ from yggdrasil_engine.engine import UnleashEngine +from UnleashClient._aiohttp_transport import aiohttp from UnleashClient._async_metrics import _AsyncMetricsReporter from UnleashClient._async_scheduler import _AsyncScheduler from UnleashClient._async_unleash_api_client import _AsyncUnleashApiClient @@ -37,6 +38,7 @@ from UnleashClient.constants import ETAG, METRIC_LAST_SENT_TIME from UnleashClient.events import BaseEvent from UnleashClient.impact_metrics import ImpactMetrics +from UnleashClient.transport import AsyncTransport, _check_transport from UnleashClient.utils import LOGGER, InstanceAllowType @@ -78,6 +80,7 @@ class AsyncUnleashClient: :param cache_directory: Location of the cache directory. :param multiple_instance_mode: Determines how multiple instances being instantiated is handled by the SDK. Defaults to InstanceAllowType.WARN. :param event_callback: Function to call if impression events are enabled. Called on a dedicated background thread. + :param transport: Sends the requests to the server, see :class:`~UnleashClient.transport.AsyncTransport`. Pass an :class:`AiohttpTransport` with your own ``aiohttp.ClientSession`` for a proxy, TLS settings or tracing. The client never closes it. Can't be combined with ``http``. """ def __init__( @@ -96,7 +99,16 @@ def __init__( cache_directory: Optional[str] = None, multiple_instance_mode: InstanceAllowType = InstanceAllowType.WARN, event_callback: Optional[Callable[[BaseEvent], None]] = None, + transport: Optional[AsyncTransport] = None, ) -> None: + if transport is not None: + _check_transport( + transport, + http_given=http is not None, + is_async=True, + session_type=aiohttp.ClientSession, + wrapper_name="AiohttpTransport", + ) if isinstance(mode, StreamingOptions): raise ValueError( "AsyncUnleashClient does not support streaming yet. " @@ -143,7 +155,7 @@ def __init__( events=self._event_dispatcher, ) self._api: _AsyncUnleashApiClient = _AsyncUnleashApiClient( - self._config, self._headers, self._event_dispatcher + self._config, self._headers, self._event_dispatcher, transport=transport ) self._metrics_scheduler: _AsyncScheduler = _AsyncScheduler() self._metrics: _AsyncMetricsReporter = _AsyncMetricsReporter( diff --git a/UnleashClient/clients/unleash_client.py b/UnleashClient/clients/unleash_client.py index 69b99ace..70374899 100644 --- a/UnleashClient/clients/unleash_client.py +++ b/UnleashClient/clients/unleash_client.py @@ -1,5 +1,4 @@ # pylint: disable=invalid-name -import inspect import threading import uuid import warnings @@ -48,39 +47,12 @@ UnleashReadyEvent, ) from UnleashClient.impact_metrics import ImpactMetrics -from UnleashClient.transport import Transport +from UnleashClient.transport import Transport, _check_transport from UnleashClient.utils import LOGGER, InstanceAllowType INSTANCES = _get_instance_registry() -# temp check until http optiins is gone -def _check_transport( - transport: Optional[Transport], http: Optional[HttpOptions] -) -> None: - if transport is None: - return - if http is not None: - raise ValueError( - "Pass either http or transport, not both. Set headers and the " - "timeout on the transport: RequestsTransport(headers=..., timeout=...)." - ) - if isinstance(transport, requests.Session): - raise TypeError( - "Wrap the session in a transport: transport=RequestsTransport(session)." - ) - execute = getattr(transport, "execute", None) - if not callable(execute): - raise TypeError( - "transport must have an execute(request) method, see " - "UnleashClient.transport.Transport." - ) - if inspect.iscoroutinefunction(execute): - raise TypeError( - "transport.execute is async: UnleashClient needs a sync transport." - ) - - class _RunState(IntEnum): UNINITIALIZED = 0 INITIALIZED = 1 @@ -175,7 +147,14 @@ def __init__( event_callback: Optional[Callable[[BaseEvent], None]] = None, transport: Optional[Transport] = None, ) -> None: - _check_transport(transport, http) + if transport is not None: + _check_transport( + transport, + http_given=http is not None, + is_async=False, + session_type=requests.Session, + wrapper_name="RequestsTransport", + ) self._config = UnleashConfig( url=url, app_name=app_name, diff --git a/UnleashClient/transport.py b/UnleashClient/transport.py index b03d95b2..1f9d5e09 100644 --- a/UnleashClient/transport.py +++ b/UnleashClient/transport.py @@ -3,10 +3,12 @@ response. Pass a transport to the client to choose how requests are sent: -:class:`RequestsTransport` with your own ``requests.Session``, or any object -with an ``execute(Request) -> Response`` method. +:class:`RequestsTransport` with your own ``requests.Session``, +``AiohttpTransport`` with your own ``aiohttp.ClientSession`` for the async +client, or any object with an ``execute(Request) -> Response`` method. """ +import inspect from dataclasses import dataclass, field from typing import Any, Mapping, Optional, Protocol, runtime_checkable @@ -73,17 +75,70 @@ def execute(self, request: Request) -> Response: @runtime_checkable class AsyncTransport(Protocol): - """Sends requests for the async client.""" + """ + Sends requests for the async client. + + ``request.headers`` include the API key: don't log them. + """ async def execute(self, request: Request) -> Response: """ Send ``request`` once and return the response, whatever its status. + Don't retry: the SDK retries feature fetches itself. :raises Exception: only when no response arrived, e.g. a connection error or a timeout. """ +def _reject_authorization(headers: Mapping[str, str], transport_name: str) -> None: + if any(name.lower() == "authorization" for name in headers): + raise ValueError( + "Pass the API key as api_key to the client, " + f"not as an Authorization header to {transport_name}." + ) + + +def _check_transport( + transport: object, + *, + http_given: bool, + is_async: bool, + session_type: type, + wrapper_name: str, +) -> None: + """ + Raises for a transport the client can't use, so the mistake shows up in the + constructor instead of as a failed request on every poll. + """ + # temp check until HttpOptions is gone + if http_given: + raise ValueError( + "Pass either http or transport, not both. Set headers and the " + f"timeout on the transport: {wrapper_name}(headers=..., timeout=...)." + ) + if isinstance(transport, session_type): + raise TypeError( + f"Wrap the session in a transport: transport={wrapper_name}(session)." + ) + execute = getattr(transport, "execute", None) + if not callable(execute): + raise TypeError( + "transport must have an execute(request) method, see " + "UnleashClient.transport." + ) + is_coroutine = inspect.iscoroutinefunction(execute) + if is_async and not is_coroutine: + is_coroutine = inspect.iscoroutinefunction(inspect.unwrap(execute)) + if is_coroutine != is_async: + raise TypeError( + "transport.execute is not async: AsyncUnleashClient needs an async " + "transport." + if is_async + else "transport.execute is async: UnleashClient needs a sync transport." + ) + + def _session_retries(session: requests.Session) -> bool: """True if a mounted adapter retries; a plain Session's adapters don't.""" for adapter in session.adapters.values(): @@ -132,11 +187,7 @@ def __init__( self._timeout = timeout self._request_kwargs = dict(_request_kwargs or {}) - if any(name.lower() == "authorization" for name in self._headers): - raise ValueError( - "Pass the API key as api_key to the client, " - "not as an Authorization header to RequestsTransport." - ) + _reject_authorization(self._headers, "RequestsTransport") if session is not None and _session_retries(session): LOGGER.warning( "The session passed to RequestsTransport retries requests itself. " diff --git a/examples/fake_async_transport.py b/examples/fake_async_transport.py new file mode 100644 index 00000000..6c2146ac --- /dev/null +++ b/examples/fake_async_transport.py @@ -0,0 +1,31 @@ +""" +Test code that uses the async client without a server or HTTP mocking. + +Builds on ``fake_transport.py``: only ``execute`` is ``async``. +""" + +import pytest + +from examples.fake_transport import FakeTransport, flag +from UnleashClient import Request, Response +from UnleashClient.clients.async_unleash_client import AsyncUnleashClient + + +class FakeAsyncTransport(FakeTransport): + async def execute(self, request: Request) -> Response: # type: ignore[override] + return super().execute(request) + + +# A pytest test with pytest-asyncio, as it would look in your own test suite. +@pytest.mark.asyncio +async def test_new_checkout_is_on(tmp_path): + transport = FakeAsyncTransport(flag("new-checkout", True)) + + async with AsyncUnleashClient( + "http://unleash.test/api", + "my-app-tests", + transport=transport, + cache_directory=str(tmp_path), + ) as client: + assert client.is_enabled("new-checkout") + assert not client.is_enabled("unknown-flag") diff --git a/examples/httpx_async_transport.py b/examples/httpx_async_transport.py new file mode 100644 index 00000000..8035795a --- /dev/null +++ b/examples/httpx_async_transport.py @@ -0,0 +1,45 @@ +""" +Send the async client's requests with httpx instead of aiohttp. + +An async transport only needs an ``async execute(Request) -> Response`` +method. It sends each request once: the SDK retries feature fetches itself. +""" + +import asyncio + +import httpx + +from UnleashClient import Request, Response +from UnleashClient.clients.async_unleash_client import AsyncUnleashClient + + +class HttpxAsyncTransport: + def __init__(self, client: httpx.AsyncClient) -> None: + self._client = client + + async def execute(self, request: Request) -> Response: + resp = await self._client.request( + request.method, + request.url, + headers=dict(request.headers), + params=dict(request.params), + content=request.body, + ) + return Response( + status=resp.status_code, headers=resp.headers, body=resp.content + ) + + +async def main() -> None: + async with httpx.AsyncClient(timeout=30) as http_client: + async with AsyncUnleashClient( + "https://unleash.example.com/api", + "my-app", + api_key="", + transport=HttpxAsyncTransport(http_client), + ) as client: + print(client.is_enabled("new-checkout")) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/unit_tests/clients/test_async_unleash_client_transport.py b/tests/unit_tests/clients/test_async_unleash_client_transport.py new file mode 100644 index 00000000..92729eb6 --- /dev/null +++ b/tests/unit_tests/clients/test_async_unleash_client_transport.py @@ -0,0 +1,287 @@ +import functools +import json + +import aiohttp +import pytest +import pytest_asyncio +from pytest import mark + +from tests.utilities.fake_unleash_server import FakeUnleash +from tests.utilities.testing_constants import APP_NAME, ETAG_VALUE, URL +from UnleashClient import INSTANCES, AiohttpTransport, Request, Response +from UnleashClient.clients.async_unleash_client import AsyncUnleashClient +from UnleashClient.config import HttpOptions, MetricsOptions, PollingOptions +from UnleashClient.constants import FEATURES_URL, METRICS_URL, REGISTER_URL +from UnleashClient.utils import InstanceAllowType + +API_PREFIX = "/api" +FEATURES = { + "version": 1, + "features": [ + { + "name": "testFlag", + "enabled": True, + "strategies": [{"name": "default", "parameters": {}}], + } + ], +} + + +@pytest.fixture(autouse=True) +def before_each(): + INSTANCES._reset() + + +def build_client(tmp_path, transport, url=URL, **kwargs) -> AsyncUnleashClient: + kwargs.setdefault("mode", PollingOptions(interval=3600)) + return AsyncUnleashClient( + url, + APP_NAME, + api_key="project:environment.hash", + cache_directory=str(tmp_path), + transport=transport, + **kwargs, + ) + + +class FakeAsyncTransport: + def __init__(self, headers=None) -> None: + self._headers = headers or {} + self.requests = [] + + async def execute(self, request: Request) -> Response: + self.requests.append(request) + if request.method == "GET": + body = json.dumps(FEATURES).encode() + return Response(status=200, headers=self._headers, body=body) + return Response(status=202, headers={}, body=b"") + + +@mark.asyncio +async def test_the_client_sends_every_request_through_the_transport(tmp_path): + transport = FakeAsyncTransport() + + async with build_client(tmp_path, transport) as client: + assert client.is_enabled("testFlag") + + sent = [(request.method, request.url) for request in transport.requests] + assert ("POST", URL + REGISTER_URL) in sent + assert ("GET", URL + FEATURES_URL) in sent + # the final metrics send on destroy() + assert ("POST", URL + METRICS_URL) in sent + assert all( + request.headers["Authorization"] == "project:environment.hash" + for request in transport.requests + ) + + +@mark.asyncio +async def test_an_etag_in_a_plain_dict_is_sent_back_on_the_next_fetch(tmp_path): + transport = FakeAsyncTransport(headers={"ETag": ETAG_VALUE}) + + async with build_client( + tmp_path, + transport, + metrics=MetricsOptions(disabled=True, disable_registration=True), + ) as client: + await client._connector._fetch_and_load() + + first, second = transport.requests + assert "If-None-Match" not in first.headers + assert second.headers["If-None-Match"] == ETAG_VALUE + + +@pytest_asyncio.fixture +async def server(): + fake = FakeUnleash() + await fake.start(API_PREFIX) + fake.on("GET", API_PREFIX + FEATURES_URL, payload=FEATURES, repeat=True) + try: + yield fake + finally: + await fake.close() + + +@mark.asyncio +async def test_the_client_never_closes_a_session_passed_in(tmp_path, server): + async with aiohttp.ClientSession() as session: + async with build_client( + tmp_path, + AiohttpTransport(session), + url=server.base_url, + metrics=MetricsOptions(disabled=True, disable_registration=True), + ) as client: + assert client.is_enabled("testFlag") + + assert not session.closed + + +@mark.asyncio +async def test_the_client_never_closes_a_transport_passed_in(tmp_path, server): + transport = AiohttpTransport() + + async with build_client( + tmp_path, + transport, + url=server.base_url, + metrics=MetricsOptions(disabled=True, disable_registration=True), + ): + pass + + try: + response = await transport.execute( + Request(method="GET", url=server.base_url + FEATURES_URL, headers={}) + ) + finally: + await transport.aclose() + assert response.status == 200 + + +@mark.asyncio +async def test_a_session_closed_too_early_is_a_failed_fetch(tmp_path, server): + session = aiohttp.ClientSession() + await session.close() + + async with build_client( + tmp_path, + AiohttpTransport(session), + url=server.base_url, + metrics=MetricsOptions(disabled=True, disable_registration=True), + ) as client: + assert not client.is_enabled("testFlag") + + assert server.requests == [] + + +def test_http_and_transport_together_raise(tmp_path): + with pytest.raises(ValueError, match="either http or transport"): + build_client(tmp_path, FakeAsyncTransport(), http=HttpOptions(timeout=5)) + + +class SyncTransport: + def execute(self, request): + raise AssertionError("never sent") + + +@mark.asyncio +@pytest.mark.parametrize( + "make_transport,message", + ( + (aiohttp.ClientSession, "AiohttpTransport\\(session\\)"), + (object, "execute"), + (SyncTransport, "not async"), + ), + ids=("a-bare-session", "no-execute", "sync-execute"), +) +async def test_a_transport_that_cannot_work_raises(tmp_path, make_transport, message): + transport = make_transport() + try: + with pytest.raises(TypeError, match=message): + build_client(tmp_path, transport) + finally: + if isinstance(transport, aiohttp.ClientSession): + await transport.close() + + +def traced(execute): + """A sync decorator, like a tracing one, that returns the coroutine.""" + + @functools.wraps(execute) + def wrapper(self, request): + return execute(self, request) + + return wrapper + + +def run_async(execute): + """An async decorator around a sync method.""" + + @functools.wraps(execute) + async def wrapper(self, request): + return execute(self, request) + + return wrapper + + +class TracedTransport(FakeAsyncTransport): + @traced + async def execute(self, request: Request) -> Response: + return await super().execute(request) + + +class SyncMadeAsyncTransport(FakeAsyncTransport): + @run_async + def execute(self, request: Request) -> Response: + self.requests.append(request) + body = json.dumps(FEATURES).encode() + return Response(status=200, headers={}, body=body) + + +@mark.asyncio +@pytest.mark.parametrize( + "make_transport", + (TracedTransport, SyncMadeAsyncTransport), + ids=("sync-decorator-on-async-execute", "async-decorator-on-sync-execute"), +) +async def test_a_decorated_async_execute_is_accepted(tmp_path, make_transport): + transport = make_transport() + + async with build_client( + tmp_path, + transport, + metrics=MetricsOptions(disabled=True, disable_registration=True), + ) as client: + assert client.is_enabled("testFlag") + + assert transport.requests + + +def test_a_rejected_transport_does_not_register_the_instance(tmp_path): + with pytest.raises(ValueError): + build_client(tmp_path, FakeAsyncTransport(), http=HttpOptions()) + + # BLOCK would raise if the rejected client had been registered + build_client( + tmp_path, + FakeAsyncTransport(), + multiple_instance_mode=InstanceAllowType.BLOCK, + ) + + +# the examples in examples/ + + +@mark.asyncio +async def test_the_fake_async_transport_example_works(tmp_path): + from examples import fake_async_transport # noqa: PLC0415 + + await fake_async_transport.test_new_checkout_is_on(tmp_path) + + +@mark.asyncio +async def test_the_httpx_async_example_works(tmp_path): + httpx = pytest.importorskip("httpx") + from examples.httpx_async_transport import HttpxAsyncTransport # noqa: PLC0415 + + sent = [] + + def server(request): + sent.append(request) + if request.method == "GET": + return httpx.Response(200, json=FEATURES, headers={"ETag": ETAG_VALUE}) + return httpx.Response(202) + + async with httpx.AsyncClient(transport=httpx.MockTransport(server)) as http_client: + async with build_client( + tmp_path, + HttpxAsyncTransport(http_client), + mode=PollingOptions(interval=3600, project_name="checkout"), + ) as client: + assert client.is_enabled("testFlag") + await client._connector._fetch_and_load() + + register = next(r for r in sent if r.url.path.endswith("/register")) + assert json.loads(register.content)["appName"] == APP_NAME + fetches = [r for r in sent if r.method == "GET"] + assert fetches[0].url.params["project"] == "checkout" + assert fetches[1].headers["If-None-Match"] == ETAG_VALUE diff --git a/tests/unit_tests/test_aiohttp_transport.py b/tests/unit_tests/test_aiohttp_transport.py index a6b80b22..84bc0a94 100644 --- a/tests/unit_tests/test_aiohttp_transport.py +++ b/tests/unit_tests/test_aiohttp_transport.py @@ -177,3 +177,112 @@ async def test_execute_after_aclose_raises_without_an_earlier_request(server): async def test_closing_a_closed_transport_does_nothing(transport): await transport.aclose() await transport.aclose() + + +@mark.asyncio +async def test_execute_sends_on_the_session_passed_in(server): + server.on("GET", PATH, headers={"Set-Cookie": "a=1"}) + server.on("GET", PATH) + # unsafe: the fake server is on an IP address, whose cookies aiohttp + # otherwise ignores + async with aiohttp.ClientSession( + headers={"X-From-Session": "yes"}, + cookie_jar=aiohttp.CookieJar(unsafe=True), + ) as session: + transport = AiohttpTransport(session) + + await transport.execute( + Request(method="GET", url=server.base_url + PATH, headers={}) + ) + await transport.execute( + Request(method="GET", url=server.base_url + PATH, headers={}) + ) + + first, second = server.calls("GET", PATH) + assert first.headers["X-From-Session"] == "yes" + # the same session both times: it sends the cookie back + assert second.headers["Cookie"] == "a=1" + + +@mark.asyncio +async def test_aclose_leaves_the_session_passed_in_open(server): + async with aiohttp.ClientSession() as session: + transport = AiohttpTransport(session) + + await transport.aclose() + + assert not session.closed + + +@mark.asyncio +async def test_a_closed_transport_raises_even_with_a_session_passed_in(server): + async with aiohttp.ClientSession() as session: + transport = AiohttpTransport(session) + await transport.aclose() + + with pytest.raises(AlreadyClosedError): + await transport.execute( + Request(method="GET", url=server.base_url + PATH, headers={}) + ) + + assert server.requests == [] + + +@mark.asyncio +async def test_a_session_that_raises_for_status_still_returns_the_response(server): + server.on("GET", PATH, status=500) + async with aiohttp.ClientSession(raise_for_status=True) as session: + response = await AiohttpTransport(session).execute( + Request(method="GET", url=server.base_url + PATH, headers={}) + ) + + assert response.status == 500 + + +# headers + + +@mark.asyncio +async def test_execute_sends_the_transport_headers(server): + server.on("GET", PATH) + transport = AiohttpTransport(headers={"X-Team": "checkout"}) + + try: + await transport.execute( + Request( + method="GET", url=server.base_url + PATH, headers={"X-Test": "value"} + ) + ) + finally: + await transport.aclose() + + sent = server.calls("GET", PATH)[0] + assert sent.headers["X-Team"] == "checkout" + assert sent.headers["X-Test"] == "value" + + +@mark.asyncio +async def test_the_request_headers_win_over_the_transport_headers_whatever_their_case( + server, +): + server.on("GET", PATH) + transport = AiohttpTransport(headers={"UNLEASH-APPNAME": "spoofed"}) + + try: + await transport.execute( + Request( + method="GET", + url=server.base_url + PATH, + headers={"unleash-appname": "my-app"}, + ) + ) + finally: + await transport.aclose() + + assert server.calls("GET", PATH)[0].headers.getall("unleash-appname") == ["my-app"] + + +@pytest.mark.parametrize("name", ["Authorization", "authorization"]) +def test_an_authorization_header_raises(name): + with pytest.raises(ValueError, match="api_key"): + AiohttpTransport(headers={name: "token"})