fix: Stop sending the payload filter from the FDv2 data system

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,

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:

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]

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"

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()

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"

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)

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