From 2e45ed8a972ee8881b138a5dd851b0c055f40ba3 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Tue, 21 Jul 2026 16:23:19 -0500 Subject: [PATCH 1/2] feat: Add async FDv1 streaming and data source status tracking --- ldclient/impl/datasource/async_status.py | 163 ++++++ ldclient/impl/datasource/async_streaming.py | 238 +++++++++ ldclient/impl/datasource/datasource_common.py | 40 ++ ldclient/impl/datasource/polling.py | 20 +- ldclient/impl/datasource/streaming.py | 29 +- .../impl/datasource/test_async_status.py | 317 ++++++++++++ .../impl/datasource/test_async_streaming.py | 465 ++++++++++++++++++ 7 files changed, 1233 insertions(+), 39 deletions(-) create mode 100644 ldclient/impl/datasource/async_status.py create mode 100644 ldclient/impl/datasource/async_streaming.py create mode 100644 ldclient/impl/datasource/datasource_common.py create mode 100644 ldclient/testing/impl/datasource/test_async_status.py create mode 100644 ldclient/testing/impl/datasource/test_async_streaming.py diff --git a/ldclient/impl/datasource/async_status.py b/ldclient/impl/datasource/async_status.py new file mode 100644 index 00000000..da621abf --- /dev/null +++ b/ldclient/impl/datasource/async_status.py @@ -0,0 +1,163 @@ +import time +from typing import Callable, Dict, Mapping, Optional, Set + +from ldclient.impl.dependency_tracker import DependencyTracker, KindAndKey +from ldclient.impl.listeners import Listeners +from ldclient.impl.rwlock import ReadWriteLock +from ldclient.interfaces import ( + AsyncDataSourceUpdateSink, + AsyncFeatureStore, + DataSourceErrorInfo, + DataSourceErrorKind, + DataSourceState, + DataSourceStatus, + DataSourceStatusProvider, + FlagChange +) +from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind + + +class AsyncDataSourceUpdateSinkImpl(AsyncDataSourceUpdateSink): + def __init__(self, store: AsyncFeatureStore, status_listeners: Listeners, flag_change_listeners: Listeners): + self.__store = store + self.__status_listeners = status_listeners + self.__flag_change_listeners = flag_change_listeners + self.__tracker = DependencyTracker() + + self.__lock = ReadWriteLock() + self.__status = DataSourceStatus(DataSourceState.INITIALIZING, time.time(), None) + + @property + def status(self) -> DataSourceStatus: + with self.__lock.read(): + return self.__status + + async def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]) -> None: + old_data: Optional[Dict[VersionedDataKind, Mapping[str, dict]]] = None + + if self.__flag_change_listeners.has_listeners(): + old_data = {} + for kind in [FEATURES, SEGMENTS]: + old_data[kind] = await self.__store.all(kind) + + try: + await self.__store.init(all_data) + except Exception as e: + error_info = DataSourceErrorInfo(DataSourceErrorKind.STORE_ERROR, 0, time.time(), str(e)) + self.update_status(DataSourceState.INTERRUPTED, error_info) + raise + + self.__reset_tracker_with_new_data(all_data) + + if old_data is None: + return + + self.__send_change_events(self.__compute_changed_items_for_full_data_set(old_data, all_data)) + + async def upsert(self, kind: VersionedDataKind, item: dict) -> None: + key = item.get('key', '') + + try: + updated = await self.__store.upsert(kind, item) + except Exception as e: + error_info = DataSourceErrorInfo(DataSourceErrorKind.STORE_ERROR, 0, time.time(), str(e)) + self.update_status(DataSourceState.INTERRUPTED, error_info) + raise + + # Only update dependency tracking and notify listeners if the store actually applied the + # update. The AsyncFeatureStore contract returns whether it wrote, so a stale + # (version-rejected) upsert produces no spurious flag-change events. + if updated: + self.__update_dependency_for_single_item(kind, key, item) + + async def delete(self, kind: VersionedDataKind, key: str, version: int) -> None: + try: + await self.__store.delete(kind, key, version) + except Exception as e: + error_info = DataSourceErrorInfo(DataSourceErrorKind.STORE_ERROR, 0, time.time(), str(e)) + self.update_status(DataSourceState.INTERRUPTED, error_info) + raise + + self.__update_dependency_for_single_item(kind, key, None) + + def update_status(self, new_state: DataSourceState, new_error: Optional[DataSourceErrorInfo]) -> None: + status_to_broadcast = None + + with self.__lock.write(): + old_status = self.__status + + if new_state == DataSourceState.INTERRUPTED and old_status.state == DataSourceState.INITIALIZING: + new_state = DataSourceState.INITIALIZING + + if new_state == old_status.state and new_error is None: + return + + self.__status = DataSourceStatus( + new_state, + self.__status.since if new_state == self.__status.state else time.time(), + self.__status.error if new_error is None else new_error, + ) + + status_to_broadcast = self.__status + + if status_to_broadcast is not None: + self.__status_listeners.notify(status_to_broadcast) + + def __update_dependency_for_single_item(self, kind: VersionedDataKind, key: str, item): + self.__tracker.update_dependencies_from(kind, key, item) + if self.__flag_change_listeners.has_listeners(): + affected_items: Set[KindAndKey] = set() + self.__tracker.add_affected_items(affected_items, KindAndKey(kind=kind, key=key)) + self.__send_change_events(affected_items) + + def __reset_tracker_with_new_data(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]): + self.__tracker.reset() + for kind, items in all_data.items(): + for key, item in items.items(): + self.__tracker.update_dependencies_from(kind, key, item) + + def __send_change_events(self, affected_items: Set[KindAndKey]): + for item in affected_items: + if item.kind == FEATURES: + self.__flag_change_listeners.notify(FlagChange(item.key)) + + def __compute_changed_items_for_full_data_set( + self, + old_data: Mapping[VersionedDataKind, Mapping[str, dict]], + new_data: Mapping[VersionedDataKind, Mapping[str, dict]], + ) -> Set[KindAndKey]: + affected_items: Set[KindAndKey] = set() + + for kind in [FEATURES, SEGMENTS]: + old_items = old_data.get(kind, {}) + new_items = new_data.get(kind, {}) + + keys: Set[str] = set() + + for key in keys.union(old_items.keys(), new_items.keys()): + old_item = old_items.get(key) + new_item = new_items.get(key) + + if old_item is None and new_item is None: + continue + + if old_item is None or new_item is None or old_item['version'] < new_item['version']: + self.__tracker.add_affected_items(affected_items, KindAndKey(kind=kind, key=key)) + + return affected_items + + +class AsyncDataSourceStatusProviderImpl(DataSourceStatusProvider): + def __init__(self, listeners: Listeners, update_sink: AsyncDataSourceUpdateSinkImpl): + self.__listeners = listeners + self.__update_sink = update_sink + + @property + def status(self) -> DataSourceStatus: + return self.__update_sink.status + + def add_listener(self, listener: Callable[[DataSourceStatus], None]): + self.__listeners.add(listener) + + def remove_listener(self, listener: Callable[[DataSourceStatus], None]): + self.__listeners.remove(listener) diff --git a/ldclient/impl/datasource/async_streaming.py b/ldclient/impl/datasource/async_streaming.py new file mode 100644 index 00000000..801723d5 --- /dev/null +++ b/ldclient/impl/datasource/async_streaming.py @@ -0,0 +1,238 @@ +""" +Default implementation of the streaming component. +""" + +# currently excluded from documentation - see docs/README.md + +import json +import time +from typing import Any, Optional +from urllib import parse + +from ld_eventsource.actions import Event, Fault, Start +from ld_eventsource.errors import HTTPStatusError + +from ldclient.impl.aio.concurrency import AsyncTaskRunner +from ldclient.impl.aio.transport import AsyncSSEFactory, make_client_session +from ldclient.impl.datasource.datasource_common import ( + STREAM_ALL_PATH, + parse_path, + sink_or_store +) +from ldclient.impl.util import ( + http_error_message, + is_http_error_recoverable, + log +) +from ldclient.interfaces import ( + DataSourceErrorInfo, + DataSourceErrorKind, + DataSourceState, + UpdateProcessor +) +from ldclient.versioned_data_kind import FEATURES, SEGMENTS + + +class AsyncStreamingUpdateProcessor(UpdateProcessor): + def __init__(self, config, store, ready, diagnostic_accumulator, sse_factory: Optional[AsyncSSEFactory] = None): + self._uri = config.stream_base_uri + STREAM_ALL_PATH + if config.payload_filter_key is not None: + self._uri += '?%s' % parse.urlencode({'filter': config.payload_filter_key}) + self._config = config + self._data_source_update_sink = config.data_source_update_sink + self._store = store + self._running = False + self._ready = ready + self._diagnostic_accumulator = diagnostic_accumulator + if sse_factory is not None: + # A caller-supplied factory owns whatever session it uses; we don't + # create or close one here. + self._sse_factory = sse_factory + self._owned_session = None + else: + # Build a session configured from the SDK's HTTP options (CA certs, + # client cert, SSL verification, proxy trust) so the streaming + # connection isn't a plain unconfigured ClientSession. The SSE + # client treats the supplied session as externally owned and never + # closes it, so this data source closes it on teardown. + self._owned_session = make_client_session(config) + self._sse_factory = AsyncSSEFactory(config, session=self._owned_session) + self._sse: Any = None + self._connection_attempt_start_time = None + self._runner = AsyncTaskRunner() + self._started = False + + def start(self): + if self._started: + raise RuntimeError("processors can only be started once") + self._started = True + self._runner.spawn("ldclient.datasource.streaming", self._run) + + async def _run(self): + log.info("Starting AsyncStreamingUpdateProcessor connecting to uri: " + self._uri) + self._running = True + self._sse = self._sse_factory.create(self._uri, self._config.initial_reconnect_delay) + self._connection_attempt_start_time = time.time() + async for action in self._sse.all: + if isinstance(action, Start): + # On reconnect after an error the timer was cleared; reset it here. + # For the initial connect the pre-loop timestamp is already set. + if self._connection_attempt_start_time is None: + self._connection_attempt_start_time = time.time() + elif isinstance(action, Event): + message_ok = False + try: + message_ok = await self._process_message(action) + except json.decoder.JSONDecodeError as e: + log.info("Error while handling stream event; will restart stream: %s" % e) + await self._sse.interrupt() + + await self._handle_error(e) + except Exception as e: + log.warning("Error while handling stream event; will restart stream: %s" % e) + await self._sse.interrupt() + + if self._data_source_update_sink is not None: + error_info = DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(e)) + + self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) + + if message_ok: + self._record_stream_init(False) + self._connection_attempt_start_time = None + + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.VALID, None) + + if not self._ready.is_set(): + log.info("AsyncStreamingUpdateProcessor initialized ok.") + self._ready.set() + elif isinstance(action, Fault): + # If the SSE client detects the stream has closed, then it will emit a fault with no-error. We can + # ignore this since we want the connection to continue. + if action.error is None: + continue + + if not await self._handle_error(action.error): + break + await self._sse.close() + await self._close_owned_session() + + async def _close_owned_session(self): + """Close the aiohttp session if the SDK created it. A caller-supplied + factory owns its own session, so ``_owned_session`` is ``None`` and + nothing is closed here. Closing resets the reference to ``None`` so it + isn't closed twice.""" + if self._owned_session is not None: + await self._owned_session.close() + self._owned_session = None + + def _record_stream_init(self, failed: bool): + if self._diagnostic_accumulator and self._connection_attempt_start_time: + current_time = int(time.time() * 1000) + elapsed = current_time - int(self._connection_attempt_start_time * 1000) + self._diagnostic_accumulator.record_stream_init(current_time, elapsed if elapsed >= 0 else 0, failed) + + async def stop(self): + await self.__stop_with_error_info(None) + await self._runner.stop_all() + + async def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]): + log.info("Stopping AsyncStreamingUpdateProcessor") + self._running = False + if self._sse: + await self._sse.close() + await self._close_owned_session() + + if self._data_source_update_sink is None: + return + + self._data_source_update_sink.update_status(DataSourceState.OFF, error) + + def initialized(self): + return self._running and self._ready.is_set() is True and self._store.initialized is True + + # Returns True if we initialized the feature store + async def _process_message(self, msg: Event) -> bool: + """Process a single SSE event. Returns True on a successful ``put``.""" + target = sink_or_store(self._data_source_update_sink, self._store) + if msg.event == 'put': + all_data = json.loads(msg.data) + init_data = {FEATURES: all_data['data']['flags'], SEGMENTS: all_data['data']['segments']} + log.debug("Received put event with %d flags and %d segments", len(init_data[FEATURES]), len(init_data[SEGMENTS])) + await target.init(init_data) + return True + elif msg.event == 'patch': + payload = json.loads(msg.data) + path = payload['path'] + obj = payload['data'] + log.debug("Received patch event for %s, New version: [%d]", path, obj.get("version")) + parsed = parse_path(path) + if parsed is not None: + await target.upsert(parsed.kind, obj) + else: + log.warning("Patch for unknown path: %s", path) + elif msg.event == 'delete': + payload = json.loads(msg.data) + path = payload['path'] + # noinspection PyShadowingNames + version = payload['version'] + log.debug("Received delete event for %s, New version: [%d]", path, version) + parsed = parse_path(path) + if parsed is not None: + await target.delete(parsed.kind, parsed.key, version) + else: + log.warning("Delete for unknown path: %s", path) + else: + log.warning('Unhandled event in stream processor: ' + msg.event) + return False + + # Returns true to continue, false to stop + async def _handle_error(self, error: Exception) -> bool: + if not self._running: + return False # don't retry if we've been deliberately stopped + + if isinstance(error, json.decoder.JSONDecodeError): + error_info = DataSourceErrorInfo(DataSourceErrorKind.INVALID_DATA, 0, time.time(), str(error)) + + log.error("Unexpected error on stream connection: %s, will retry" % error) + self._record_stream_init(True) + self._connection_attempt_start_time = None + + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) + elif isinstance(error, HTTPStatusError): + self._record_stream_init(True) + self._connection_attempt_start_time = None + + error_info = DataSourceErrorInfo(DataSourceErrorKind.ERROR_RESPONSE, error.status, time.time(), str(error)) + + http_error_message_result = http_error_message(error.status, "stream connection") + if not is_http_error_recoverable(error.status): + log.error(http_error_message_result) + self._running = False + self._ready.set() # if client is initializing, make it stop waiting; has no effect if already inited + await self.__stop_with_error_info(error_info) + return False + else: + log.warning(http_error_message_result) + + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, error_info) + else: + log.warning("Unexpected error on stream connection: %s, will retry" % error) + self._record_stream_init(True) + self._connection_attempt_start_time = None + + if self._data_source_update_sink is not None: + self._data_source_update_sink.update_status(DataSourceState.INTERRUPTED, DataSourceErrorInfo(DataSourceErrorKind.UNKNOWN, 0, time.time(), str(error))) + # no stacktrace here because, for a typical connection error, it'll just be a lengthy tour of HTTP client internals + self._connection_attempt_start_time = time.time() + self._sse.next_retry_delay + return True + + # magic methods for "with" statement (used in testing) + async def __aenter__(self): + return self + + async def __aexit__(self, type, value, traceback): + await self.stop() diff --git a/ldclient/impl/datasource/datasource_common.py b/ldclient/impl/datasource/datasource_common.py new file mode 100644 index 00000000..bec4edd2 --- /dev/null +++ b/ldclient/impl/datasource/datasource_common.py @@ -0,0 +1,40 @@ +""" +Pure helpers shared by the FDv1 polling and streaming data sources. +""" + +# currently excluded from documentation - see docs/README.md + +from collections import namedtuple +from typing import Optional + +from ldclient.interfaces import DataSourceUpdateSink, FeatureStore +from ldclient.versioned_data_kind import FEATURES, SEGMENTS + +STREAM_ALL_PATH = '/all' + +ParsedPath = namedtuple('ParsedPath', ['kind', 'key']) + + +def sink_or_store(sink: Optional[DataSourceUpdateSink], store: FeatureStore): + """ + The original implementation of the data sources relied on the feature store + directly, which we are trying to move away from. Customers who might have + instantiated one directly for some reason wouldn't know they have to set + the config's sink manually, so we have to fall back to the store if the + sink isn't present. + + The next major release should be able to simplify this structure and + remove the need for fall back to the data store because the update sink + should always be present. + """ + if sink is None: + return store + + return sink + + +def parse_path(path: str): + for kind in [FEATURES, SEGMENTS]: + if path.startswith(kind.stream_api_path): + return ParsedPath(kind=kind, key=path[len(kind.stream_api_path):]) + return None diff --git a/ldclient/impl/datasource/polling.py b/ldclient/impl/datasource/polling.py index 3392594c..42b3c743 100644 --- a/ldclient/impl/datasource/polling.py +++ b/ldclient/impl/datasource/polling.py @@ -9,6 +9,7 @@ from typing import Optional from ldclient.config import Config +from ldclient.impl.datasource.datasource_common import sink_or_store from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.util import ( UnsuccessfulResponseException, @@ -55,27 +56,10 @@ def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]): self._data_source_update_sink.update_status(DataSourceState.OFF, error) - def _sink_or_store(self): - """ - The original implementation of this class relied on the feature store - directly, which we are trying to move away from. Customers who might have - instantiated this directly for some reason wouldn't know they have to set - the config's sink manually, so we have to fall back to the store if the - sink isn't present. - - The next major release should be able to simplify this structure and - remove the need for fall back to the data store because the update sink - should always be present. - """ - if self._data_source_update_sink is None: - return self._store - - return self._data_source_update_sink - def _poll(self): try: all_data = self._requester.get_all_data() - self._sink_or_store().init(all_data) + sink_or_store(self._data_source_update_sink, self._store).init(all_data) if not self._ready.is_set() and self._store.initialized: log.info("PollingUpdateProcessor initialized ok") self._ready.set() diff --git a/ldclient/impl/datasource/streaming.py b/ldclient/impl/datasource/streaming.py index d12b4043..e980e9d6 100644 --- a/ldclient/impl/datasource/streaming.py +++ b/ldclient/impl/datasource/streaming.py @@ -1,6 +1,5 @@ import json import time -from collections import namedtuple from threading import Thread from typing import Optional from urllib import parse @@ -14,6 +13,11 @@ ) from ld_eventsource.errors import HTTPStatusError +from ldclient.impl.datasource.datasource_common import ( + STREAM_ALL_PATH, + parse_path, + sink_or_store +) from ldclient.impl.http import HTTPFactory, _http_factory from ldclient.impl.util import ( http_error_message, @@ -36,10 +40,6 @@ BACKOFF_RESET_INTERVAL = 60 JITTER_RATIO = 0.5 -STREAM_ALL_PATH = '/all' - -ParsedPath = namedtuple('ParsedPath', ['kind', 'key']) - class StreamingUpdateProcessor(Thread, UpdateProcessor): def __init__(self, config, store, ready, diagnostic_accumulator): @@ -65,7 +65,7 @@ def run(self): if isinstance(action, Event): message_ok = False try: - message_ok = self._process_message(self._sink_or_store(), action) + message_ok = self._process_message(sink_or_store(self._data_source_update_sink, self._store), action) except json.decoder.JSONDecodeError as e: log.info("Error while handling stream event; will restart stream: %s" % e) self._sse.interrupt() @@ -135,12 +135,6 @@ def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]): self._data_source_update_sink.update_status(DataSourceState.OFF, error) - def _sink_or_store(self): - if self._data_source_update_sink is None: - return self._store - - return self._data_source_update_sink - def initialized(self): return self._running and self._ready.is_set() is True and self._store.initialized is True @@ -157,7 +151,7 @@ def _process_message(self, store, msg: Event) -> bool: path = payload['path'] obj = payload['data'] log.debug("Received patch event for %s, New version: [%d]", path, obj.get("version")) - target = StreamingUpdateProcessor._parse_path(path) + target = parse_path(path) if target is not None: store.upsert(target.kind, obj) else: @@ -168,7 +162,7 @@ def _process_message(self, store, msg: Event) -> bool: # noinspection PyShadowingNames version = payload['version'] log.debug("Received delete event for %s, New version: [%d]", path, version) - target = StreamingUpdateProcessor._parse_path(path) + target = parse_path(path) if target is not None: store.delete(target.kind, target.key, version) else: @@ -220,13 +214,6 @@ def _handle_error(self, error: Exception) -> bool: self._connection_attempt_start_time = time.time() + self._sse.next_retry_delay return True - @staticmethod - def _parse_path(path: str): - for kind in [FEATURES, SEGMENTS]: - if path.startswith(kind.stream_api_path): - return ParsedPath(kind=kind, key=path[len(kind.stream_api_path):]) - return None - # magic methods for "with" statement (used in testing) def __enter__(self): return self diff --git a/ldclient/testing/impl/datasource/test_async_status.py b/ldclient/testing/impl/datasource/test_async_status.py new file mode 100644 index 00000000..fecf395e --- /dev/null +++ b/ldclient/testing/impl/datasource/test_async_status.py @@ -0,0 +1,317 @@ +import pytest + +from ldclient.async_feature_store import AsyncInMemoryFeatureStore +from ldclient.impl.datasource.async_status import ( + AsyncDataSourceStatusProviderImpl, + AsyncDataSourceUpdateSinkImpl +) +from ldclient.impl.listeners import Listeners +from ldclient.interfaces import ( + DataSourceErrorInfo, + DataSourceErrorKind, + DataSourceState, + DataSourceStatus, + FlagChange +) +from ldclient.testing.builders import FlagBuilder, SegmentBuilder +from ldclient.versioned_data_kind import FEATURES, SEGMENTS + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def make_sink(store=None): + if store is None: + store = AsyncInMemoryFeatureStore() + status_listeners = Listeners() + flag_change_listeners = Listeners() + sink = AsyncDataSourceUpdateSinkImpl(store, status_listeners, flag_change_listeners) + return sink, status_listeners, flag_change_listeners + + +def make_flag(key, version=1): + return FlagBuilder(key).version(version).build() + + +def make_segment(key, version=1): + return SegmentBuilder(key).version(version).build() + + +class FlagChangeCapture: + """Captures FlagChange events delivered to a listener.""" + + def __init__(self): + self.keys = [] + + def __call__(self, change: FlagChange): + self.keys.append(change.key) + + +class StatusCapture: + """Captures DataSourceStatus events delivered to a status listener.""" + + def __init__(self): + self.statuses = [] + + def __call__(self, status: DataSourceStatus): + self.statuses.append(status) + + +# --------------------------------------------------------------------------- +# init tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_init_notifies_all_flag_change_listeners(): + sink, _, flag_listeners = make_sink() + + capture = FlagChangeCapture() + flag_listeners.add(capture) + + flag_a = make_flag('flag-a') + flag_b = make_flag('flag-b') + + await sink.init({ + FEATURES: { + 'flag-a': flag_a.to_json_dict(), + 'flag-b': flag_b.to_json_dict(), + }, + SEGMENTS: {}, + }) + + assert sorted(capture.keys) == ['flag-a', 'flag-b'] + + +@pytest.mark.asyncio +async def test_init_does_not_notify_flag_listeners_when_none_registered(): + sink, _, flag_listeners = make_sink() + + # No listener added — calling init should not raise and should be a no-op + # for notifications. + flag_a = make_flag('flag-a') + await sink.init({FEATURES: {'flag-a': flag_a.to_json_dict()}, SEGMENTS: {}}) + # No assertions needed — just confirming no error and no listener calls. + + +@pytest.mark.asyncio +async def test_init_notifies_listeners_for_changed_flags_on_reinit(): + sink, _, flag_listeners = make_sink() + capture = FlagChangeCapture() + flag_listeners.add(capture) + + flag_v1 = make_flag('flag-a', version=1) + await sink.init({FEATURES: {'flag-a': flag_v1.to_json_dict()}, SEGMENTS: {}}) + capture.keys.clear() + + flag_v2 = make_flag('flag-a', version=2) + flag_new = make_flag('flag-b', version=1) + await sink.init({ + FEATURES: { + 'flag-a': flag_v2.to_json_dict(), + 'flag-b': flag_new.to_json_dict(), + }, + SEGMENTS: {}, + }) + + assert sorted(capture.keys) == ['flag-a', 'flag-b'] + + +# --------------------------------------------------------------------------- +# upsert tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_upsert_flag_notifies_listener(): + sink, _, flag_listeners = make_sink() + capture = FlagChangeCapture() + flag_listeners.add(capture) + + flag = make_flag('my-flag', version=1) + await sink.upsert(FEATURES, flag.to_json_dict()) + + assert capture.keys == ['my-flag'] + + +@pytest.mark.asyncio +async def test_upsert_segment_does_not_notify_flag_listener(): + sink, _, flag_listeners = make_sink() + capture = FlagChangeCapture() + flag_listeners.add(capture) + + segment = make_segment('my-segment', version=1) + await sink.upsert(SEGMENTS, segment.to_json_dict()) + + assert capture.keys == [] + + +@pytest.mark.asyncio +async def test_upsert_same_version_does_not_notify(): + store = AsyncInMemoryFeatureStore() + sink, _, flag_listeners = make_sink(store) + capture = FlagChangeCapture() + flag_listeners.add(capture) + + flag_v1 = make_flag('my-flag', version=1) + + # Insert the flag at version 1. + await sink.upsert(FEATURES, flag_v1.to_json_dict()) + assert capture.keys == ['my-flag'] + capture.keys.clear() + + # Upsert the same version — store rejects it, no notification expected. + await sink.upsert(FEATURES, flag_v1.to_json_dict()) + assert capture.keys == [] + + +@pytest.mark.asyncio +async def test_upsert_older_version_does_not_notify(): + store = AsyncInMemoryFeatureStore() + sink, _, flag_listeners = make_sink(store) + capture = FlagChangeCapture() + flag_listeners.add(capture) + + flag_v2 = make_flag('my-flag', version=2) + await sink.upsert(FEATURES, flag_v2.to_json_dict()) + capture.keys.clear() + + flag_v1 = make_flag('my-flag', version=1) + await sink.upsert(FEATURES, flag_v1.to_json_dict()) + assert capture.keys == [] + + +@pytest.mark.asyncio +async def test_upsert_newer_version_notifies(): + store = AsyncInMemoryFeatureStore() + sink, _, flag_listeners = make_sink(store) + capture = FlagChangeCapture() + flag_listeners.add(capture) + + flag_v1 = make_flag('my-flag', version=1) + await sink.upsert(FEATURES, flag_v1.to_json_dict()) + capture.keys.clear() + + flag_v2 = make_flag('my-flag', version=2) + await sink.upsert(FEATURES, flag_v2.to_json_dict()) + assert capture.keys == ['my-flag'] + + +# --------------------------------------------------------------------------- +# delete tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_delete_flag_notifies_listener(): + sink, _, flag_listeners = make_sink() + capture = FlagChangeCapture() + flag_listeners.add(capture) + + flag = make_flag('my-flag', version=1) + await sink.upsert(FEATURES, flag.to_json_dict()) + capture.keys.clear() + + await sink.delete(FEATURES, 'my-flag', version=2) + + assert capture.keys == ['my-flag'] + + +@pytest.mark.asyncio +async def test_delete_segment_does_not_notify_flag_listener(): + sink, _, flag_listeners = make_sink() + capture = FlagChangeCapture() + flag_listeners.add(capture) + + segment = make_segment('my-segment', version=1) + await sink.upsert(SEGMENTS, segment.to_json_dict()) + capture.keys.clear() + + await sink.delete(SEGMENTS, 'my-segment', version=2) + + assert capture.keys == [] + + +# --------------------------------------------------------------------------- +# update_status / DataSourceStatusProvider tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_update_status_broadcasts_to_status_listeners(): + sink, status_listeners, _ = make_sink() + capture = StatusCapture() + status_listeners.add(capture) + + sink.update_status(DataSourceState.VALID, None) + + assert len(capture.statuses) == 1 + assert capture.statuses[0].state == DataSourceState.VALID + + +@pytest.mark.asyncio +async def test_update_status_interrupted_during_init_stays_initializing(): + sink, status_listeners, _ = make_sink() + capture = StatusCapture() + status_listeners.add(capture) + + # State starts as INITIALIZING; INTERRUPTED should be coerced back. + sink.update_status(DataSourceState.INTERRUPTED, None) + + assert len(capture.statuses) == 0 # state unchanged → no broadcast + + +@pytest.mark.asyncio +async def test_update_status_no_broadcast_when_unchanged(): + sink, status_listeners, _ = make_sink() + + # Transition to VALID first. + sink.update_status(DataSourceState.VALID, None) + + capture = StatusCapture() + status_listeners.add(capture) + + # Calling with the same state and no error should not broadcast. + sink.update_status(DataSourceState.VALID, None) + + assert capture.statuses == [] + + +@pytest.mark.asyncio +async def test_update_status_broadcasts_error_even_with_same_state(): + sink, status_listeners, _ = make_sink() + sink.update_status(DataSourceState.VALID, None) + + capture = StatusCapture() + status_listeners.add(capture) + + error = DataSourceErrorInfo(DataSourceErrorKind.NETWORK_ERROR, 0, 0.0, "connection lost") + sink.update_status(DataSourceState.VALID, error) + + assert len(capture.statuses) == 1 + assert capture.statuses[0].error is error + + +@pytest.mark.asyncio +async def test_status_provider_delegates_to_sink(): + sink, status_listeners, _ = make_sink() + provider = AsyncDataSourceStatusProviderImpl(status_listeners, sink) + + assert provider.status.state == DataSourceState.INITIALIZING + + sink.update_status(DataSourceState.VALID, None) + + assert provider.status.state == DataSourceState.VALID + + +@pytest.mark.asyncio +async def test_status_provider_add_remove_listener(): + sink, status_listeners, _ = make_sink() + provider = AsyncDataSourceStatusProviderImpl(status_listeners, sink) + + capture = StatusCapture() + provider.add_listener(capture) + + sink.update_status(DataSourceState.VALID, None) + assert len(capture.statuses) == 1 + + provider.remove_listener(capture) + sink.update_status(DataSourceState.OFF, None) + assert len(capture.statuses) == 1 # listener was removed, no new events diff --git a/ldclient/testing/impl/datasource/test_async_streaming.py b/ldclient/testing/impl/datasource/test_async_streaming.py new file mode 100644 index 00000000..cc958840 --- /dev/null +++ b/ldclient/testing/impl/datasource/test_async_streaming.py @@ -0,0 +1,465 @@ +""" +Tests for AsyncStreamingUpdateProcessor. + +These tests inject a mock SSE factory that yields pre-configured actions +rather than making real network connections. +""" + +import asyncio +import json +from unittest import mock + +import pytest + +from ldclient.config import Config +from ldclient.impl.datasource import async_streaming +from ldclient.impl.datasource.async_streaming import ( + AsyncStreamingUpdateProcessor +) +from ldclient.impl.model import ModelEntity +from ldclient.interfaces import DataSourceErrorKind, DataSourceState +from ldclient.testing.builders import FlagBuilder, SegmentBuilder +from ldclient.testing.mock_async_components import MockAsyncFeatureStore +from ldclient.versioned_data_kind import FEATURES, SEGMENTS + + +def _item_dict(item): + """Convert a model entity (FeatureFlag, Segment, etc.) to a plain dict.""" + return item.to_json_dict() if isinstance(item, ModelEntity) else item + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_config(**kwargs): + return Config(sdk_key='sdk-key', **kwargs) + + +def _make_put_data(flags=None, segments=None): + flags = flags or {} + segments = segments or {} + return json.dumps({"data": {"flags": flags, "segments": segments}}) + + +def _make_patch_data(kind, item): + path = '%s%s' % (kind.stream_api_path, item['key']) + return json.dumps({"path": path, "data": item}) + + +def _make_delete_data(kind, key, version): + path = '%s%s' % (kind.stream_api_path, key) + return json.dumps({"path": path, "version": version}) + + +# The processor matches actions with isinstance() against the real +# ld_eventsource action classes, so the fakes must be real instances. +from ld_eventsource.actions import Event as _RealEvent # noqa: E402 +from ld_eventsource.actions import Fault as _RealFault # noqa: E402 +from ld_eventsource.actions import Start as _RealStart # noqa: E402 + + +def _event(event_type: str, data: str) -> _RealEvent: + return _RealEvent(event=event_type, data=data) + + +def _fault(error=None) -> _RealFault: + return _RealFault(error=error) + + +def _start() -> _RealStart: + return _RealStart(headers={}) + + +async def _actions_generator(actions: list): + """Yield a fixed sequence of actions then hang (simulates a live stream).""" + for action in actions: + yield action + # Block forever so the processor's loop doesn't exit until cancelled. + await asyncio.Event().wait() + + +class _MockSSE: + """Stand-in for AsyncSSEClient exposing the surface the processor uses.""" + + def __init__(self, actions: list): + self._actions = actions + self.interrupted = False + self.closed = False + self.next_retry_delay = 0.0 + + async def interrupt(self): + self.interrupted = True + + async def close(self): + self.closed = True + + @property + def all(self): + return _actions_generator(self._actions) + + +class _MockSSEFactory: + """Stand-in for AsyncSSEFactory; create() returns a _MockSSE.""" + + def __init__(self, actions: list): + self._actions = actions + self.created: list = [] + + def create(self, url: str, initial_retry_delay: float) -> _MockSSE: + sse = _MockSSE(self._actions) + self.created.append(sse) + return sse + + +def _make_processor(actions, config=None, store=None, ready_event=None, diag=None): + config = config or _make_config() + store = store or MockAsyncFeatureStore() + ready_event = ready_event or asyncio.Event() + factory = _MockSSEFactory(actions) + proc = AsyncStreamingUpdateProcessor(config, store, ready_event, diag, factory) + return proc, store, ready_event, factory + + +async def _run_with_actions(actions: list, config=None, store=None, ready_event=None, + diag=None, extra_ready_timeout=3.0): + """Run the processor against a fake SSE action sequence. + + Starts the processor and waits for the ready event (up to + *extra_ready_timeout* seconds), then returns + ``(processor, store, ready_event, factory)``. + """ + proc, store, ready, factory = _make_processor(actions, config, store, ready_event, diag) + proc.start() + try: + await asyncio.wait_for(ready.wait(), timeout=extra_ready_timeout) + except asyncio.TimeoutError: + pass # some tests expect ready NOT to be set + return proc, store, ready, factory + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_put_event_initializes_store_and_sets_ready(): + flag = FlagBuilder('f1').version(1).build() + segment = SegmentBuilder('s1').version(1).build() + put_data = _make_put_data( + flags={'f1': _item_dict(flag)}, + segments={'s1': _item_dict(segment)}, + ) + actions = [_start(), _event('put', put_data)] + + proc, store, ready, _ = await _run_with_actions(actions) + + assert ready.is_set() + assert store.initialized + stored_flag = await store.get(FEATURES, 'f1') + assert stored_flag is not None + assert stored_flag['version'] == 1 + + stored_seg = await store.get(SEGMENTS, 's1') + assert stored_seg is not None + assert stored_seg['version'] == 1 + + await proc.stop() + + +@pytest.mark.asyncio +async def test_patch_event_upserts_to_store(): + flagv1 = FlagBuilder('f1').version(1).build() + flagv2 = FlagBuilder('f1').version(2).build() + put_data = _make_put_data(flags={'f1': _item_dict(flagv1)}) + patch_data = _make_patch_data(FEATURES, _item_dict(flagv2)) + actions = [_start(), _event('put', put_data), _event('patch', patch_data)] + + proc, store, ready, _ = await _run_with_actions(actions) + + # Give the event loop a beat for the patch to process after ready fires. + await asyncio.sleep(0.05) + + stored = await store.get(FEATURES, 'f1') + assert stored is not None + assert stored['version'] == 2 + + await proc.stop() + + +@pytest.mark.asyncio +async def test_delete_event_removes_item_from_store(): + flagv1 = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flagv1)}) + delete_data = _make_delete_data(FEATURES, 'f1', 2) + actions = [_start(), _event('put', put_data), _event('delete', delete_data)] + + proc, store, ready, _ = await _run_with_actions(actions) + + await asyncio.sleep(0.05) + + deleted = await store.get(FEATURES, 'f1') + assert deleted is None + + await proc.stop() + + +@pytest.mark.asyncio +async def test_fault_with_error_does_not_set_ready_by_itself(): + """A Fault that arrives before any put must not mark the processor as initialized.""" + from ld_eventsource.errors import HTTPStatusError + + # Provide only a recoverable fault (503) — no put follows. + actions = [_start(), _fault(error=HTTPStatusError(503))] + + # Use a short timeout so the test doesn't hang. + proc, store, ready, _ = await _run_with_actions(actions, extra_ready_timeout=0.2) + + # Ready should NOT have been set by the fault alone. + assert not ready.is_set() + assert not store.initialized + + await proc.stop() + + +@pytest.mark.asyncio +async def test_fault_none_error_is_ignored(): + """A Fault with error=None (clean close) should not update status or stop the processor.""" + flag = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flag)}) + actions = [ + _start(), + _event('put', put_data), + _fault(error=None), # clean close — should be ignored + ] + + proc, store, ready, _ = await _run_with_actions(actions) + + assert ready.is_set() + assert store.initialized + + await proc.stop() + + +@pytest.mark.asyncio +async def test_unrecoverable_http_error_stops_processor(): + """An unrecoverable HTTP status closes the stream and reports OFF with error info.""" + from ld_eventsource.errors import HTTPStatusError + + from ldclient.impl.datasource.async_status import ( + AsyncDataSourceUpdateSinkImpl + ) + from ldclient.impl.listeners import Listeners + + store = MockAsyncFeatureStore() + statuses = [] + listeners = Listeners() + listeners.add(lambda s: statuses.append(s)) + + config = _make_config() + config._data_source_update_sink = AsyncDataSourceUpdateSinkImpl(store, listeners, Listeners()) + + actions = [_start(), _fault(error=HTTPStatusError(401))] + + proc, store, ready, factory = await _run_with_actions(actions, config=config, store=store) + + # The unrecoverable error unblocks initialization without initializing the store. + assert ready.is_set() + assert not proc.initialized() + assert factory.created[0].closed + assert any( + s.state == DataSourceState.OFF + and s.error is not None + and s.error.kind == DataSourceErrorKind.ERROR_RESPONSE + and s.error.status_code == 401 + for s in statuses + ) + + await proc.stop() + + +@pytest.mark.asyncio +async def test_stop_closes_sse_and_finishes_task(): + flag = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flag)}) + actions = [_start(), _event('put', put_data)] + + proc, store, ready, factory = await _run_with_actions(actions) + assert ready.is_set() + + await proc.stop() + + # After stop() the SSE client is closed and no background task remains. + assert factory.created[0].closed + assert len(proc._runner._tasks) == 0 + + +@pytest.mark.asyncio +async def test_second_start_raises(): + actions = [_start()] + proc, store, ready, _ = _make_processor(actions) + proc.start() + try: + with pytest.raises(RuntimeError): + proc.start() + finally: + await proc.stop() + + +@pytest.mark.asyncio +async def test_invalid_json_triggers_invalid_data_status(): + from ldclient.impl.datasource.async_status import ( + AsyncDataSourceUpdateSinkImpl + ) + from ldclient.impl.listeners import Listeners + + store = MockAsyncFeatureStore() + listeners = Listeners() + statuses = [] + listeners.add(lambda s: statuses.append(s)) + + config = _make_config() + config._data_source_update_sink = AsyncDataSourceUpdateSinkImpl(store, listeners, Listeners()) + + # Deliver a put first so the processor initializes, then a bad patch. + flag = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flag)}) + bad_patch_data = 'not valid json' + actions = [ + _start(), + _event('put', put_data), + _event('patch', bad_patch_data), + ] + + proc, store, ready, _ = await _run_with_actions(actions, config=config, store=store) + await asyncio.sleep(0.1) # let patch event propagate + + error_statuses = [s for s in statuses if s.error is not None] + assert any( + s.error.kind == DataSourceErrorKind.INVALID_DATA for s in error_statuses + ), "Expected INVALID_DATA status from bad JSON" + + await proc.stop() + + +@pytest.mark.asyncio +async def test_patch_unknown_path_is_ignored(): + """A patch for an unknown path should log a warning and not crash.""" + flag = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flag)}) + bad_patch = json.dumps({"path": "/unknown/something", "data": {"key": "x", "version": 1}}) + actions = [_start(), _event('put', put_data), _event('patch', bad_patch)] + + proc, store, ready, _ = await _run_with_actions(actions) + await asyncio.sleep(0.05) + + # Store should be unchanged (still has f1 at version 1). + stored = await store.get(FEATURES, 'f1') + assert stored is not None + assert stored['version'] == 1 + + await proc.stop() + + +@pytest.mark.asyncio +async def test_initialized_reflects_store_state(): + proc, store, ready, _ = _make_processor([]) + assert not proc.initialized() + + proc.start() + # Seed the store and set ready manually to simulate post-put state. + flag = FlagBuilder('f1').version(1).build() + await store.init({FEATURES: {'f1': _item_dict(flag)}, SEGMENTS: {}}) + ready.set() + proc._running = True + + assert proc.initialized() + + await proc.stop() + + +# --------------------------------------------------------------------------- +# Session ownership +# --------------------------------------------------------------------------- + + +class _FakeSession: + """Minimal stand-in for an aiohttp.ClientSession that records closure.""" + + def __init__(self): + self.closed = False + + async def close(self): + self.closed = True + + +@pytest.mark.asyncio +async def test_default_construction_builds_configured_session(): + """With no injected factory, a configured session is built via + make_client_session and handed to the SSE factory.""" + config = _make_config() + store = MockAsyncFeatureStore() + ready = asyncio.Event() + fake_session = _FakeSession() + + with mock.patch.object( + async_streaming, "make_client_session", return_value=fake_session + ) as make_session, mock.patch.object( + async_streaming, "AsyncSSEFactory" + ) as factory_cls: + proc = AsyncStreamingUpdateProcessor(config, store, ready, None) + + make_session.assert_called_once_with(config) + assert factory_cls.call_args.kwargs["session"] is fake_session + assert proc._owned_session is fake_session + + +@pytest.mark.asyncio +async def test_default_construction_session_closed_on_stop(): + """The SDK-created session is closed when the processor stops.""" + config = _make_config() + store = MockAsyncFeatureStore() + ready = asyncio.Event() + fake_session = _FakeSession() + + with mock.patch.object( + async_streaming, "make_client_session", return_value=fake_session + ), mock.patch.object(async_streaming, "AsyncSSEFactory"): + proc = AsyncStreamingUpdateProcessor(config, store, ready, None) + + await proc.stop() + + assert fake_session.closed is True + assert proc._owned_session is None + + +@pytest.mark.asyncio +async def test_injected_factory_leaves_session_unowned(): + """When a factory is injected, no session is built and none is owned.""" + with mock.patch.object(async_streaming, "make_client_session") as make_session: + proc, store, ready, factory = _make_processor([]) + + make_session.assert_not_called() + assert proc._owned_session is None + + # stop() must not attempt to close a session it doesn't own. + await proc.stop() + assert proc._owned_session is None + + +@pytest.mark.asyncio +async def test_diagnostics_recorded_on_successful_init(): + from ldclient.impl.events.diagnostics import _DiagnosticAccumulator + + diag = _DiagnosticAccumulator(1) + flag = FlagBuilder('f1').version(1).build() + put_data = _make_put_data(flags={'f1': _item_dict(flag)}) + actions = [_start(), _event('put', put_data)] + + proc, store, ready, _ = await _run_with_actions(actions, diag=diag) + + recorded = diag.create_event_and_reset(0, 0)['streamInits'] + assert len(recorded) == 1 + assert recorded[0]['failed'] is False + + await proc.stop() From 924244bd13543f7a6ce30fd6961476ed515f660d Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Tue, 21 Jul 2026 16:56:32 -0500 Subject: [PATCH 2/2] refactor: Reuse sync data source status provider for async --- ldclient/impl/datasource/async_status.py | 19 +------------------ ldclient/impl/datasource/status.py | 13 +++++++++++-- .../impl/datasource/test_async_status.py | 10 ++++------ 3 files changed, 16 insertions(+), 26 deletions(-) diff --git a/ldclient/impl/datasource/async_status.py b/ldclient/impl/datasource/async_status.py index da621abf..ebe32917 100644 --- a/ldclient/impl/datasource/async_status.py +++ b/ldclient/impl/datasource/async_status.py @@ -1,5 +1,5 @@ import time -from typing import Callable, Dict, Mapping, Optional, Set +from typing import Dict, Mapping, Optional, Set from ldclient.impl.dependency_tracker import DependencyTracker, KindAndKey from ldclient.impl.listeners import Listeners @@ -11,7 +11,6 @@ DataSourceErrorKind, DataSourceState, DataSourceStatus, - DataSourceStatusProvider, FlagChange ) from ldclient.versioned_data_kind import FEATURES, SEGMENTS, VersionedDataKind @@ -145,19 +144,3 @@ def __compute_changed_items_for_full_data_set( self.__tracker.add_affected_items(affected_items, KindAndKey(kind=kind, key=key)) return affected_items - - -class AsyncDataSourceStatusProviderImpl(DataSourceStatusProvider): - def __init__(self, listeners: Listeners, update_sink: AsyncDataSourceUpdateSinkImpl): - self.__listeners = listeners - self.__update_sink = update_sink - - @property - def status(self) -> DataSourceStatus: - return self.__update_sink.status - - def add_listener(self, listener: Callable[[DataSourceStatus], None]): - self.__listeners.add(listener) - - def remove_listener(self, listener: Callable[[DataSourceStatus], None]): - self.__listeners.remove(listener) diff --git a/ldclient/impl/datasource/status.py b/ldclient/impl/datasource/status.py index f59fd49c..820278f5 100644 --- a/ldclient/impl/datasource/status.py +++ b/ldclient/impl/datasource/status.py @@ -1,5 +1,5 @@ import time -from typing import Callable, Mapping, Optional, Set +from typing import Callable, Mapping, Optional, Protocol, Set from ldclient.impl.dependency_tracker import DependencyTracker, KindAndKey from ldclient.impl.listeners import Listeners @@ -132,8 +132,17 @@ def __compute_changed_items_for_full_data_set(self, old_data: Mapping[VersionedD return affected_items +class _DataSourceStatusSource(Protocol): + """The status property a status provider reads from its update sink; + satisfied by both the sync and async sink implementations.""" + + @property + def status(self) -> DataSourceStatus: + ... + + class DataSourceStatusProviderImpl(DataSourceStatusProvider): - def __init__(self, listeners: Listeners, update_sink: DataSourceUpdateSinkImpl): + def __init__(self, listeners: Listeners, update_sink: _DataSourceStatusSource): self.__listeners = listeners self.__update_sink = update_sink diff --git a/ldclient/testing/impl/datasource/test_async_status.py b/ldclient/testing/impl/datasource/test_async_status.py index fecf395e..4304b3c5 100644 --- a/ldclient/testing/impl/datasource/test_async_status.py +++ b/ldclient/testing/impl/datasource/test_async_status.py @@ -1,10 +1,8 @@ import pytest from ldclient.async_feature_store import AsyncInMemoryFeatureStore -from ldclient.impl.datasource.async_status import ( - AsyncDataSourceStatusProviderImpl, - AsyncDataSourceUpdateSinkImpl -) +from ldclient.impl.datasource.async_status import AsyncDataSourceUpdateSinkImpl +from ldclient.impl.datasource.status import DataSourceStatusProviderImpl from ldclient.impl.listeners import Listeners from ldclient.interfaces import ( DataSourceErrorInfo, @@ -292,7 +290,7 @@ async def test_update_status_broadcasts_error_even_with_same_state(): @pytest.mark.asyncio async def test_status_provider_delegates_to_sink(): sink, status_listeners, _ = make_sink() - provider = AsyncDataSourceStatusProviderImpl(status_listeners, sink) + provider = DataSourceStatusProviderImpl(status_listeners, sink) assert provider.status.state == DataSourceState.INITIALIZING @@ -304,7 +302,7 @@ async def test_status_provider_delegates_to_sink(): @pytest.mark.asyncio async def test_status_provider_add_remove_listener(): sink, status_listeners, _ = make_sink() - provider = AsyncDataSourceStatusProviderImpl(status_listeners, sink) + provider = DataSourceStatusProviderImpl(status_listeners, sink) capture = StatusCapture() provider.add_listener(capture)