chore: Drop streaming builder shim and dedupe initializer check

Two follow-up cleanups from review of the FDv1 Fallback Directive work:

- Drop the test-only shim in StreamingDataSource.sync() that accepted
  either a bare SSEClient or a (client, pool) tuple from the injected
  builder. The SseClientBuilder type alias already declares the tuple
  return; annotating the field as SseClientBuilder lets pyright align,
  and the lone test fixture (list_sse_client) now returns
  (client, None) like the production create_sse_client does.
- Hoist basis.change_set.selector.is_defined() into a local in
  FDv2._run_initializers() so it isn't computed twice on the
  success-with-fallback path. Also fixes a typo in the adjacent
  comment ("if an only if" -> "if and only if").

Author: keelerm84

ldclient/impl/datasourcev2/streaming.py

@@ -163,7 +163,7 @@ def __init__(self,
         self.__http_options = http_options
         self.__initial_reconnect_delay = initial_reconnect_delay
 
-        self._sse_client_builder = create_sse_client
+        self._sse_client_builder: SseClientBuilder = create_sse_client
         self._config = config
         self._sse: Optional[SSEClient] = None
         self._sse_pool: Optional[urllib3.PoolManager] = None
@@ -187,19 +187,13 @@ def sync(self, ss: SelectorStore) -> Generator[Update, None, None]:
         Update objects until the connection is closed or an unrecoverable error
         occurs.
         """
-        builder_result = self._sse_client_builder(
+        self._sse, self._sse_pool = self._sse_client_builder(
             self.__uri,
             self.__http_options,
             self.__initial_reconnect_delay,
             self._config,
             ss
         )
-        # Tests may inject a builder that returns either an SSEClient directly
-        # or a (client, pool) tuple. Accept both.
-        if isinstance(builder_result, tuple):
-            self._sse, self._sse_pool = builder_result
-        else:
-            self._sse, self._sse_pool = builder_result, None
 
         if self._sse is None:
             log.error("Failed to create SSE client for streaming updates.")

ldclient/impl/datasystem/fdv2.py

@@ -486,8 +486,9 @@ def _run_initializers(self, set_on_ready: Event) -> bool:
                 # Apply the basis to the store
                 self._store.apply(basis.change_set, basis.persist)
 
-                # Set ready event if an only if a selector is defined for the changeset
-                if basis.change_set.selector.is_defined():
+                # Set ready event if and only if a selector is defined for the changeset
+                selector_defined = basis.change_set.selector.is_defined()
+                if selector_defined:
                     set_on_ready.set()
 
                 if basis.fallback_to_fdv1:
@@ -497,7 +498,7 @@ def _run_initializers(self, set_on_ready: Event) -> bool:
                     )
                     return True
 
-                if basis.change_set.selector.is_defined():
+                if selector_defined:
                     return False
             except Exception as e:
                 log.error("Initializer failed with exception: %s", e)

ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py

@@ -13,7 +13,6 @@
 from ldclient.config import Config, HTTPConfig
 from ldclient.impl.datasourcev2.streaming import (
     STREAMING_ENDPOINT,
-    SSEClient,
     SseClientBuilder,
     StreamingDataSource
 )
@@ -48,8 +47,8 @@ def builder(
         initial_reconnect_delay: float,
         config: Config,  # pylint: disable=unused-argument
         ss: SelectorStore  # pylint: disable=unused-argument
-    ) -> SSEClient:
-        return ListBasedSseClient(events)
+    ):
+        return ListBasedSseClient(events), None
 
     return builder
 
@@ -743,7 +742,7 @@ def builder(*_args, **_kwargs):
         ]), tracking_pool
 
     synchronizer = make_streaming_data_source()
-    synchronizer._sse_client_builder = builder
+    synchronizer._sse_client_builder = builder  # type: ignore[assignment]
     updates = list(synchronizer.sync(MockSelectorStore(Selector.no_selector())))
 
     assert len(updates) == 1