diff --git a/contract-tests/async_client_entity.py b/contract-tests/async_client_entity.py index 2569a156..87cd85a7 100644 --- a/contract-tests/async_client_entity.py +++ b/contract-tests/async_client_entity.py @@ -301,9 +301,9 @@ def _build_async_data_system(datasystem_config: dict, opts: dict) -> AsyncDataSy """Build an AsyncDataSystemConfig from the harness's dataSystem config. Wires the FDv2 initializers, the ordered synchronizer chain, the FDv1 - fallback synchronizer, the payload filter, and an optional async - persistent store. The async client injects its shared aiohttp session - into these builders when it starts. + fallback synchronizer, and an optional async persistent store. The async + client injects its shared aiohttp session into these builders when it + starts. """ initializers: Optional[list] = None init_configs = datasystem_config.get('initializers') @@ -347,9 +347,6 @@ def _build_async_data_system(datasystem_config: dict, opts: dict) -> AsyncDataSy _set_optional_time(fdv1_fallback_config, "pollIntervalMs", fallback_builder.poll_interval) fdv1_fallback_synchronizer = fallback_builder - if datasystem_config.get("payloadFilter") is not None: - opts["payload_filter_key"] = datasystem_config["payloadFilter"] - ds_kwargs: dict = { "initializers": initializers, "synchronizers": synchronizers, diff --git a/contract-tests/client_entity.py b/contract-tests/client_entity.py index f93b5d85..b3544020 100644 --- a/contract-tests/client_entity.py +++ b/contract-tests/client_entity.py @@ -91,9 +91,6 @@ def __init__(self, tag, config): _set_optional_time(fdv1_fallback_config, "pollIntervalMs", fallback_builder.poll_interval) datasystem.fdv1_compatible_synchronizer(fallback_builder) - if datasystem_config.get("payloadFilter") is not None: - opts["payload_filter_key"] = datasystem_config["payloadFilter"] - # Handle persistent data store configuration for dataSystem store_config = datasystem_config.get("store") if store_config is not None: diff --git a/ldclient/config.py b/ldclient/config.py index 81271ee8..b6a476a0 100644 --- a/ldclient/config.py +++ b/ldclient/config.py @@ -221,10 +221,6 @@ def initial_reconnect_delay(self) -> float: def poll_interval(self) -> float: """The interval between polling requests.""" - @property - def payload_filter_key(self) -> Optional[str]: - """The payload filter key applied as a query parameter.""" - # Set by the client after construction, so it is a read-write attribute # rather than a read-only property. _instance_id: Optional[str] diff --git a/ldclient/impl/datasourcev2/async_polling.py b/ldclient/impl/datasourcev2/async_polling.py index e3b6a3a3..38c90712 100644 --- a/ldclient/impl/datasourcev2/async_polling.py +++ b/ldclient/impl/datasourcev2/async_polling.py @@ -181,17 +181,9 @@ async def fetch(self, selector: Optional[Selector]) -> PollingResult: Returns a Result containing a tuple of ChangeSet and any request headers, or an error if the data could not be retrieved. """ - query_params = {} - if self._config.payload_filter_key is not None: - query_params["filter"] = self._config.payload_filter_key - - if selector is not None and selector.is_defined(): - query_params["basis"] = selector.state - uri = self._poll_uri - if len(query_params) > 0: - filter_query = parse.urlencode(query_params) - uri += f"?{filter_query}" + if selector is not None and selector.is_defined(): + uri += "?%s" % parse.urlencode({"basis": selector.state}) hdrs = _headers(self._config, ASYNC_USER_AGENT) hdrs["Accept-Encoding"] = "gzip" @@ -379,14 +371,7 @@ async def fetch(self, selector: Optional[Selector]) -> PollingResult: Returns a Result containing a tuple of ChangeSet and any request headers, or an error if the data could not be retrieved. """ - query_params = {} - if self._config.payload_filter_key is not None: - query_params["filter"] = self._config.payload_filter_key - uri = self._poll_uri - if len(query_params) > 0: - filter_query = parse.urlencode(query_params) - uri += f"?{filter_query}" hdrs = _headers(self._config, ASYNC_USER_AGENT) hdrs["Accept-Encoding"] = "gzip" diff --git a/ldclient/impl/datasourcev2/async_streaming.py b/ldclient/impl/datasourcev2/async_streaming.py index 0f8428ab..4d117ea8 100644 --- a/ldclient/impl/datasourcev2/async_streaming.py +++ b/ldclient/impl/datasourcev2/async_streaming.py @@ -6,7 +6,6 @@ import json from time import time from typing import AsyncGenerator, Callable, Optional, Tuple -from urllib import parse import aiohttp from ld_eventsource import AsyncSSEClient @@ -70,8 +69,6 @@ def create_sse_client( caller owns it and ``None`` is returned in its place. """ uri = base_uri + STREAMING_ENDPOINT - if config.payload_filter_key is not None: - uri += "?%s" % parse.urlencode({"filter": config.payload_filter_key}) def query_params() -> dict: selector = ss.selector() diff --git a/ldclient/impl/datasourcev2/polling.py b/ldclient/impl/datasourcev2/polling.py index a96d5ed0..a7e1cf4d 100644 --- a/ldclient/impl/datasourcev2/polling.py +++ b/ldclient/impl/datasourcev2/polling.py @@ -160,17 +160,9 @@ def fetch(self, selector: Optional[Selector]) -> PollingResult: Returns a Result containing a tuple of ChangeSet and any request headers, or an error if the data could not be retrieved. """ - query_params = {} - if self._config.payload_filter_key is not None: - query_params["filter"] = self._config.payload_filter_key - - if selector is not None and selector.is_defined(): - query_params["basis"] = selector.state - uri = self._poll_uri - if len(query_params) > 0: - filter_query = parse.urlencode(query_params) - uri += f"?{filter_query}" + if selector is not None and selector.is_defined(): + uri += "?%s" % parse.urlencode({"basis": selector.state}) hdrs = _headers(self._config) hdrs["Accept-Encoding"] = "gzip" @@ -334,14 +326,7 @@ def fetch(self, selector: Optional[Selector]) -> PollingResult: Returns a Result containing a tuple of ChangeSet and any request headers, or an error if the data could not be retrieved. """ - query_params = {} - if self._config.payload_filter_key is not None: - query_params["filter"] = self._config.payload_filter_key - uri = self._poll_uri - if len(query_params) > 0: - filter_query = parse.urlencode(query_params) - uri += f"?{filter_query}" hdrs = _headers(self._config) hdrs["Accept-Encoding"] = "gzip" diff --git a/ldclient/impl/datasourcev2/streaming.py b/ldclient/impl/datasourcev2/streaming.py index 5e9ab99a..f896d405 100644 --- a/ldclient/impl/datasourcev2/streaming.py +++ b/ldclient/impl/datasourcev2/streaming.py @@ -6,7 +6,6 @@ import json from time import time from typing import Callable, Generator, Optional, Tuple -from urllib import parse from ld_eventsource import SSEClient from ld_eventsource.actions import Event, Fault, Start @@ -73,8 +72,6 @@ def create_sse_client( to the LaunchDarkly streaming endpoint. """ uri = base_uri + STREAMING_ENDPOINT - if config.payload_filter_key is not None: - uri += "?%s" % parse.urlencode({"filter": config.payload_filter_key}) # We don't want the stream to use the same read timeout as the rest of the SDK. base_headers = _base_headers(config) diff --git a/ldclient/testing/impl/datasourcev2/test_payload_filter.py b/ldclient/testing/impl/datasourcev2/test_payload_filter.py new file mode 100644 index 00000000..471f02c7 --- /dev/null +++ b/ldclient/testing/impl/datasourcev2/test_payload_filter.py @@ -0,0 +1,114 @@ +""" +The FDv2 data system does not support payload filtering. These tests show that its data sources do +not send the payload filter key, also when the config has one. This includes the FDv1 fallback +requesters, so that a fallback does not change the client to a filtered subset of the data. +""" + +from unittest import mock +from urllib.parse import parse_qs, urlsplit + +import pytest + +from ldclient.config import Config +from ldclient.impl.datasourcev2 import async_streaming, streaming +from ldclient.impl.datasourcev2.async_polling import ( + AiohttpFDv1PollingRequester, + AiohttpPollingRequester +) +from ldclient.impl.datasourcev2.polling import ( + Urllib3FDv1PollingRequester, + Urllib3PollingRequester +) +from ldclient.interfaces import Selector +from ldclient.testing.http_util import start_server +from ldclient.testing.mock_components import MockSelectorStore + +FILTER_KEY = 'microservice-1' +SELECTOR = Selector.new_selector('p:SOMETHING:300', 300) + + +def path_and_query(request): + parts = urlsplit(request.path) + return parts.path, parse_qs(parts.query) + + +def make_config(base_uri: str = 'http://localhost') -> Config: + return Config('sdk-key', base_uri=base_uri, stream_uri=base_uri, payload_filter_key=FILTER_KEY) + + +@pytest.mark.parametrize('selector, expected_query', [ + (None, {}), + (SELECTOR, {'basis': [SELECTOR.state]}), +]) +def test_polling_requester_does_not_send_the_filter(selector, expected_query): + with start_server() as server: + config = make_config(server.uri) + requester = Urllib3PollingRequester(config, config.base_uri, config.http) + requester.fetch(selector) + assert path_and_query(server.require_request()) == ('/sdk/poll', expected_query) + + +def test_fdv1_fallback_polling_requester_does_not_send_the_filter(): + with start_server() as server: + config = make_config(server.uri) + requester = Urllib3FDv1PollingRequester(config, config.base_uri, config.http) + requester.fetch(SELECTOR) + assert path_and_query(server.require_request()) == ('/sdk/latest-all', {}) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('selector, expected_query', [ + (None, {}), + (SELECTOR, {'basis': [SELECTOR.state]}), +]) +async def test_async_polling_requester_does_not_send_the_filter(selector, expected_query): + with start_server() as server: + config = make_config(server.uri) + requester = AiohttpPollingRequester(config, config.base_uri, config.http) + try: + await requester.fetch(selector) + finally: + await requester._http.close() + assert path_and_query(server.require_request()) == ('/sdk/poll', expected_query) + + +@pytest.mark.asyncio +async def test_async_fdv1_fallback_polling_requester_does_not_send_the_filter(): + with start_server() as server: + config = make_config(server.uri) + requester = AiohttpFDv1PollingRequester(config, config.base_uri, config.http) + try: + await requester.fetch(SELECTOR) + finally: + await requester._http.close() + assert path_and_query(server.require_request()) == ('/sdk/latest-all', {}) + + +@pytest.mark.parametrize('selector, expected_params', [ + (Selector.no_selector(), {}), + (SELECTOR, {'basis': SELECTOR.state}), +]) +def test_streaming_does_not_send_the_filter(selector, expected_params): + config = make_config() + with mock.patch.object(streaming.ConnectStrategy, 'http', wraps=streaming.ConnectStrategy.http) as http: + streaming.create_sse_client(config.stream_base_uri, config.http, config.initial_reconnect_delay, config, MockSelectorStore(selector)) + + kwargs = http.call_args.kwargs + assert kwargs['url'] == 'http://localhost/sdk/stream' + assert kwargs['query_params']() == expected_params + + +@pytest.mark.parametrize('selector, expected_params', [ + (Selector.no_selector(), {}), + (SELECTOR, {'basis': SELECTOR.state}), +]) +def test_async_streaming_does_not_send_the_filter(selector, expected_params): + config = make_config() + with mock.patch.object(async_streaming, 'AsyncSSEFactory') as factory_cls: + async_streaming.create_sse_client( + config.stream_base_uri, config.http, config.initial_reconnect_delay, config, MockSelectorStore(selector), session=object() + ) + + args, kwargs = factory_cls.return_value.create.call_args + assert args[0] == 'http://localhost/sdk/stream' + assert kwargs['query_params']() == expected_params