From fccfbb4e324c7e362407e11c1aa62bb25ed0e533 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 11 Aug 2026 22:57:51 +0000 Subject: [PATCH 1/6] feat: Propagate environment ID to evaluation hooks Co-Authored-By: rlamb@launchdarkly.com <4955475+kinyoklion@users.noreply.github.com> --- ldclient/client.py | 2 +- ldclient/hook.py | 3 +- ldclient/impl/datasource/datasource_common.py | 29 +++++- ldclient/impl/datasource/feature_requester.py | 7 +- ldclient/impl/datasource/polling.py | 26 +++++- ldclient/impl/datasource/status.py | 10 ++ ldclient/impl/datasource/streaming.py | 9 +- ldclient/impl/datasystem/__init__.py | 14 ++- ldclient/impl/datasystem/fdv1.py | 4 + ldclient/impl/datasystem/fdv2.py | 27 +++++- ldclient/interfaces.py | 6 ++ .../impl/datasource/test_feature_requester.py | 13 +++ .../impl/datasource/test_polling_processor.py | 35 +++++++ .../testing/impl/datasource/test_streaming.py | 52 +++++++++++ .../impl/datasystem/test_fdv2_datasystem.py | 93 ++++++++++++++++++- ldclient/testing/stub_util.py | 10 +- ldclient/testing/test_ldclient_end_to_end.py | 29 ++++++ ldclient/testing/test_ldclient_hooks.py | 15 +++ 18 files changed, 368 insertions(+), 16 deletions(-) diff --git a/ldclient/client.py b/ldclient/client.py index 2e05749e..f04b4d36 100644 --- a/ldclient/client.py +++ b/ldclient/client.py @@ -709,7 +709,7 @@ def __evaluate_with_hooks(self, key: str, context: Context, default_value: Any, hooks = self.__hooks.copy() - series_context = EvaluationSeriesContext(key=key, context=context, default_value=default_value, method=method) + series_context = EvaluationSeriesContext(key=key, context=context, default_value=default_value, method=method, environment_id=self._data_system.environment_id) hook_data = self.__execute_before_evaluation(hooks, series_context) evaluation_result = block() self.__execute_after_evaluation(hooks, series_context, hook_data, evaluation_result.evaluation_detail) diff --git a/ldclient/hook.py b/ldclient/hook.py index e9cd63e2..49ddba8d 100644 --- a/ldclient/hook.py +++ b/ldclient/hook.py @@ -1,6 +1,6 @@ from abc import ABC, abstractmethod from dataclasses import dataclass -from typing import Any +from typing import Any, Optional from ldclient.context import Context from ldclient.evaluation import EvaluationDetail @@ -17,6 +17,7 @@ class EvaluationSeriesContext: context: Context #: The context used during evaluation. default_value: Any #: The default value provided to the evaluation method method: str #: The string version of the method which triggered the evaluation series. + environment_id: Optional[str] = None #: The environment ID the SDK is connected to, if available. @dataclass diff --git a/ldclient/impl/datasource/datasource_common.py b/ldclient/impl/datasource/datasource_common.py index d5be8697..aa4da878 100644 --- a/ldclient/impl/datasource/datasource_common.py +++ b/ldclient/impl/datasource/datasource_common.py @@ -5,8 +5,9 @@ # currently excluded from documentation - see docs/README.md from collections import namedtuple -from typing import Optional +from typing import Mapping, Optional, Protocol, runtime_checkable +from ldclient.impl.util import _LD_ENVID_HEADER from ldclient.interfaces import DataSourceUpdateSink, FeatureStore from ldclient.versioned_data_kind import FEATURES, SEGMENTS @@ -34,6 +35,32 @@ def sink_or_store(sink: Optional[DataSourceUpdateSink], store: FeatureStore): return sink +@runtime_checkable +class EnvironmentIdSink(Protocol): + """ + Implemented by data source update sinks which can record the environment ID + reported by LaunchDarkly. This is separate from + :class:`ldclient.interfaces.DataSourceUpdateSink` so that externally + implemented sinks remain compatible. + """ + + def set_environment_id(self, environment_id: str) -> None: + ... + + +def record_environment_id(sink, headers: Optional[Mapping[str, str]]): + """ + Records the environment ID from a set of LaunchDarkly response headers, if + both the headers and the sink provide one. + """ + if headers is None or not isinstance(sink, EnvironmentIdSink): + return + + environment_id = headers.get(_LD_ENVID_HEADER) + if isinstance(environment_id, str) and environment_id != '': + sink.set_environment_id(environment_id) + + def parse_path(path: str): for kind in [FEATURES, SEGMENTS]: if path.startswith(kind.stream_api_path): diff --git a/ldclient/impl/datasource/feature_requester.py b/ldclient/impl/datasource/feature_requester.py index cf2becdf..4dd4e2f0 100644 --- a/ldclient/impl/datasource/feature_requester.py +++ b/ldclient/impl/datasource/feature_requester.py @@ -4,6 +4,7 @@ import json from collections import namedtuple +from typing import Mapping, Optional, Tuple from urllib import parse import urllib3 @@ -27,6 +28,10 @@ def __init__(self, config): self._poll_uri += '?%s' % parse.urlencode({'filter': config.payload_filter_key}) def get_all_data(self): + (data, _) = self.get_all_data_with_headers() + return data + + def get_all_data_with_headers(self) -> Tuple[dict, Optional[Mapping[str, str]]]: uri = self._poll_uri hdrs = _headers(self._config) cache_entry = self._cache.get(uri) @@ -47,4 +52,4 @@ def get_all_data(self): self._cache[uri] = CacheEntry(data=data, etag=etag) log.debug("%s response status:[%d] From cache? [%s] ETag:[%s]", uri, r.status, from_cache, etag) - return {FEATURES: data['flags'], SEGMENTS: data['segments']} + return ({FEATURES: data['flags'], SEGMENTS: data['segments']}, r.headers) diff --git a/ldclient/impl/datasource/polling.py b/ldclient/impl/datasource/polling.py index 42b3c743..171df9eb 100644 --- a/ldclient/impl/datasource/polling.py +++ b/ldclient/impl/datasource/polling.py @@ -6,10 +6,13 @@ import time from threading import Event -from typing import Optional +from typing import Any, Mapping, Optional, Protocol, Tuple, runtime_checkable from ldclient.config import Config -from ldclient.impl.datasource.datasource_common import sink_or_store +from ldclient.impl.datasource.datasource_common import ( + record_environment_id, + sink_or_store +) from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.util import ( UnsuccessfulResponseException, @@ -28,6 +31,12 @@ ) +@runtime_checkable +class _FeatureRequesterWithHeaders(Protocol): + def get_all_data_with_headers(self) -> Tuple[Any, Optional[Mapping[str, str]]]: + ... + + class PollingUpdateProcessor(UpdateProcessor): def __init__(self, config: Config, requester: FeatureRequester, store: FeatureStore, ready: Event): self._config = config @@ -58,7 +67,8 @@ def __stop_with_error_info(self, error: Optional[DataSourceErrorInfo]): def _poll(self): try: - all_data = self._requester.get_all_data() + (all_data, headers) = self._get_all_data_with_headers() + record_environment_id(self._data_source_update_sink, headers) 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") @@ -84,3 +94,13 @@ def _poll(self): 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(e))) + + def _get_all_data_with_headers(self) -> Tuple[Any, Optional[Mapping[str, str]]]: + """ + Externally provided feature requesters are not required to surface + response headers, so fall back to the data-only method. + """ + if isinstance(self._requester, _FeatureRequesterWithHeaders): + return self._requester.get_all_data_with_headers() + + return (self._requester.get_all_data(), None) diff --git a/ldclient/impl/datasource/status.py b/ldclient/impl/datasource/status.py index 820278f5..c4d046d7 100644 --- a/ldclient/impl/datasource/status.py +++ b/ldclient/impl/datasource/status.py @@ -26,12 +26,22 @@ def __init__(self, store: FeatureStore, status_listeners: Listeners, flag_change self.__lock = ReadWriteLock() self.__status = DataSourceStatus(DataSourceState.INITIALIZING, time.time(), None) + self.__environment_id: Optional[str] = None @property def status(self) -> DataSourceStatus: with self.__lock.read(): return self.__status + @property + def environment_id(self) -> Optional[str]: + with self.__lock.read(): + return self.__environment_id + + def set_environment_id(self, environment_id: str) -> None: + with self.__lock.write(): + self.__environment_id = environment_id + def init(self, all_data: Mapping[VersionedDataKind, Mapping[str, dict]]): old_data = None diff --git a/ldclient/impl/datasource/streaming.py b/ldclient/impl/datasource/streaming.py index 47b7ffdb..6b13e336 100644 --- a/ldclient/impl/datasource/streaming.py +++ b/ldclient/impl/datasource/streaming.py @@ -5,7 +5,7 @@ from urllib import parse from ld_eventsource import SSEClient -from ld_eventsource.actions import Event, Fault +from ld_eventsource.actions import Event, Fault, Start from ld_eventsource.config import ( ConnectStrategy, ErrorStrategy, @@ -16,6 +16,7 @@ from ldclient.impl.datasource.datasource_common import ( STREAM_ALL_PATH, parse_path, + record_environment_id, sink_or_store ) from ldclient.impl.http import HTTPFactory, _http_factory @@ -62,7 +63,9 @@ def run(self): self._sse = self._create_sse_client() self._connection_attempt_start_time = time.time() for action in self._sse.all: - if isinstance(action, Event): + if isinstance(action, Start): + record_environment_id(self._data_source_update_sink, action.headers) + elif isinstance(action, Event): message_ok = False try: message_ok = self._process_message(sink_or_store(self._data_source_update_sink, self._store), action) @@ -91,6 +94,8 @@ def run(self): log.info("StreamingUpdateProcessor initialized ok.") self._ready.set() elif isinstance(action, Fault): + record_environment_id(self._data_source_update_sink, action.headers) + # 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: diff --git a/ldclient/impl/datasystem/__init__.py b/ldclient/impl/datasystem/__init__.py index d04bdd9b..1180e8e5 100644 --- a/ldclient/impl/datasystem/__init__.py +++ b/ldclient/impl/datasystem/__init__.py @@ -6,7 +6,7 @@ from abc import abstractmethod from enum import Enum from threading import Event -from typing import TYPE_CHECKING, Protocol, runtime_checkable +from typing import TYPE_CHECKING, Optional, Protocol, runtime_checkable if TYPE_CHECKING: from ldclient.impl.aio.concurrency import AsyncEvent @@ -146,6 +146,18 @@ def store(self) -> ReadOnlyStore: """ raise NotImplementedError + @property + @abstractmethod + def environment_id(self) -> Optional[str]: + """ + Returns the environment ID reported by LaunchDarkly, if it is known. + + This is only available once a connection to LaunchDarkly has provided + it, and it will be None when the SDK is offline, using an unsupported + data source, or connected to a service which does not report it. + """ + raise NotImplementedError + class AsyncDataSystem(Protocol): """ diff --git a/ldclient/impl/datasystem/fdv1.py b/ldclient/impl/datasystem/fdv1.py index ee1656ea..f21c3e6c 100644 --- a/ldclient/impl/datasystem/fdv1.py +++ b/ldclient/impl/datasystem/fdv1.py @@ -98,6 +98,10 @@ def stop(self): def store(self) -> ReadOnlyStore: return self._store_wrapper + @property + def environment_id(self) -> Optional[str]: + return self._data_source_update_sink.environment_id + def set_diagnostic_accumulator(self, diagnostic_accumulator: DiagnosticAccumulator): """ Sets the diagnostic accumulator for streaming initialization metrics. diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index 9df91f84..fa763094 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -21,7 +21,12 @@ from ldclient.impl.listeners import Listeners from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.rwlock import ReadWriteLock -from ldclient.impl.util import _LD_FD_FALLBACK_HEADER, _Fail, log +from ldclient.impl.util import ( + _LD_ENVID_HEADER, + _LD_FD_FALLBACK_HEADER, + _Fail, + log +) from ldclient.interfaces import ( DataSourceErrorInfo, DataSourceErrorKind, @@ -95,6 +100,7 @@ def __init__( self._lock = ReadWriteLock() self._active_synchronizer: Optional[Synchronizer] = None self._threads: List[Thread] = [] + self._environment_id: Optional[str] = None # Track configuration self._configured_with_data_sources = ( @@ -216,6 +222,8 @@ def _run_initializers(self, set_on_ready: Event) -> bool: if isinstance(basis_result, _Fail): log.warning("Initializer %s failed: %s", initializer.name, basis_result.error) + if basis_result.headers is not None: + self._record_environment_id(basis_result.headers.get(_LD_ENVID_HEADER)) # An error response can still carry the FDv1 fallback directive. if basis_result.headers is not None and \ basis_result.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': @@ -240,6 +248,8 @@ def _run_initializers(self, set_on_ready: Event) -> bool: basis = basis_result.value log.info("Initialized via %s", initializer.name) + self._record_environment_id(basis.environment_id) + # Apply the basis to the store self._store.apply(basis.change_set, basis.persist) @@ -412,6 +422,8 @@ def reader(self: 'FDv2'): if self._stop_event.is_set(): return ConditionDirective.FALLBACK + self._record_environment_id(update.environment_id) + # Handle the update if update.change_set is not None: self._store.apply(update.change_set, True) @@ -496,6 +508,19 @@ def _persistent_store_outage_recovery(self, data_store_status: DataStoreStatus): if err is not None: log.error("Failed to reinitialize data store", exc_info=err) + def _record_environment_id(self, environment_id: Optional[str]): + if environment_id is None: + return + + with self._lock.write(): + self._environment_id = environment_id + + @property + def environment_id(self) -> Optional[str]: + """Get the environment ID reported by LaunchDarkly, if known.""" + with self._lock.read(): + return self._environment_id + @property def store(self) -> ReadOnlyStore: """Get the underlying store for flag evaluation.""" diff --git a/ldclient/interfaces.py b/ldclient/interfaces.py index 7c2e17bb..6f6a03d2 100644 --- a/ldclient/interfaces.py +++ b/ldclient/interfaces.py @@ -570,6 +570,12 @@ def get_all(self): """ pass + def get_all_data(self) -> Mapping[VersionedDataKind, Mapping[str, dict]]: + """ + Fetches all feature flag and segment data. + """ + raise NotImplementedError + class AsyncFeatureRequester(ABC): """ diff --git a/ldclient/testing/impl/datasource/test_feature_requester.py b/ldclient/testing/impl/datasource/test_feature_requester.py index 8a25ed4a..badf566e 100644 --- a/ldclient/testing/impl/datasource/test_feature_requester.py +++ b/ldclient/testing/impl/datasource/test_feature_requester.py @@ -25,6 +25,19 @@ def test_get_all_data_returns_data(): assert result == expected_data +def test_get_all_data_with_headers_returns_response_headers(): + with start_server() as server: + config = Config(sdk_key='sdk-key', base_uri=server.uri) + fr = FeatureRequesterImpl(config) + + resp_data = {'flags': {}, 'segments': {}} + server.for_path('/sdk/latest-all', JsonResponse(resp_data, {'X-LD-EnvID': 'env-abc-123'})) + + (data, headers) = fr.get_all_data_with_headers() + assert data == {FEATURES: {}, SEGMENTS: {}} + assert headers.get('X-LD-EnvID') == 'env-abc-123' + + def test_get_all_data_sends_headers(): with start_server() as server: config = Config(sdk_key='sdk-key', base_uri=server.uri) diff --git a/ldclient/testing/impl/datasource/test_polling_processor.py b/ldclient/testing/impl/datasource/test_polling_processor.py index 80790563..726442fd 100644 --- a/ldclient/testing/impl/datasource/test_polling_processor.py +++ b/ldclient/testing/impl/datasource/test_polling_processor.py @@ -145,3 +145,38 @@ def verify_recoverable_http_error(http_status_code, ignore_mock): assert status.state == DataSourceState.INITIALIZING assert status.error.kind == DataSourceErrorKind.ERROR_RESPONSE assert status.error.status_code == http_status_code + + +class MockFeatureRequesterWithHeaders(MockFeatureRequester): + def __init__(self, headers): + super().__init__() + self.headers = headers + + def get_all_data_with_headers(self): + return (self.get_all_data(), self.headers) + + +def test_records_environment_id_from_polling_headers(): + global mock_requester + mock_requester = MockFeatureRequesterWithHeaders({'X-LD-EnvID': 'env-abc-123'}) + mock_requester.all_data = {FEATURES: {}, SEGMENTS: {}} + + config = Config("SDK_KEY") + sink = DataSourceUpdateSinkImpl(store, Listeners(), Listeners()) + config._data_source_update_sink = sink + setup_processor(config) + ready.wait() + + assert sink.environment_id == 'env-abc-123' + + +def test_environment_id_is_none_when_requester_provides_no_headers(): + mock_requester.all_data = {FEATURES: {}, SEGMENTS: {}} + + config = Config("SDK_KEY") + sink = DataSourceUpdateSinkImpl(store, Listeners(), Listeners()) + config._data_source_update_sink = sink + setup_processor(config) + ready.wait() + + assert sink.environment_id is None diff --git a/ldclient/testing/impl/datasource/test_streaming.py b/ldclient/testing/impl/datasource/test_streaming.py index 17b9143e..66519c12 100644 --- a/ldclient/testing/impl/datasource/test_streaming.py +++ b/ldclient/testing/impl/datasource/test_streaming.py @@ -438,6 +438,58 @@ def test_failure_transitions_from_valid(): assert spy.statuses[1].error.status_code == 401 +def test_records_environment_id_from_stream_headers(): + store = InMemoryFeatureStore() + ready = Event() + + with start_server() as server: + with stream_content(make_put_event(), headers={'X-LD-EnvID': 'env-abc-123'}) as stream: + config = Config(sdk_key='sdk-key', stream_uri=server.uri) + sink = DataSourceUpdateSinkImpl(store, Listeners(), Listeners()) + config._data_source_update_sink = sink + server.for_path('/all', stream) + + with StreamingUpdateProcessor(config, store, ready, None) as sp: + sp.start() + ready.wait(start_wait) + assert sink.environment_id == 'env-abc-123' + + +def test_environment_id_is_none_when_not_provided(): + store = InMemoryFeatureStore() + ready = Event() + + with start_server() as server: + with stream_content(make_put_event()) as stream: + config = Config(sdk_key='sdk-key', stream_uri=server.uri) + sink = DataSourceUpdateSinkImpl(store, Listeners(), Listeners()) + config._data_source_update_sink = sink + server.for_path('/all', stream) + + with StreamingUpdateProcessor(config, store, ready, None) as sp: + sp.start() + ready.wait(start_wait) + assert sink.environment_id is None + + +def test_records_environment_id_from_error_response_headers(): + store = InMemoryFeatureStore() + ready = Event() + + with start_server() as server: + with stream_content(make_put_event()) as stream: + error_then_success = SequentialHandler(BasicResponse(503, None, {'X-LD-EnvID': 'env-from-error'}), stream) + config = Config(sdk_key='sdk-key', stream_uri=server.uri, initial_reconnect_delay=brief_delay) + sink = DataSourceUpdateSinkImpl(store, Listeners(), Listeners()) + config._data_source_update_sink = sink + server.for_path('/all', error_then_success) + + with StreamingUpdateProcessor(config, store, ready, None) as sp: + sp.start() + ready.wait(start_wait) + assert sink.environment_id == 'env-from-error' + + def expect_item(store, kind, item): assert store.get(kind, item['key'], lambda x: x) == item diff --git a/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py b/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py index aa2f01f5..43445a64 100644 --- a/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py +++ b/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py @@ -16,7 +16,12 @@ from ldclient.datasystem import file_ds_builder from ldclient.impl.datasystem import DataAvailability from ldclient.impl.datasystem.fdv2 import FDv2 -from ldclient.impl.util import _LD_FD_FALLBACK_HEADER, _Fail, _Success +from ldclient.impl.util import ( + _LD_ENVID_HEADER, + _LD_FD_FALLBACK_HEADER, + _Fail, + _Success +) from ldclient.integrations.test_datav2 import TestDataV2 from ldclient.interfaces import ( Basis, @@ -755,3 +760,89 @@ def listener(flag_change: FlagChange): # The Valid update's payload must be applied before the handoff. assert payload_flag_seen.wait(1), "FDv2 payload was not applied before fallback" assert fdv1_flag_seen.wait(1), "FDv1 fallback synchronizer did not run after directive" + + +def test_environment_id_from_initializer_basis(): + builder = ChangeSetBuilder() + builder.start(IntentCode.TRANSFER_FULL) + change_set = builder.finish(Selector(state="initializer-state", version=1)) + init = _StaticInitializer( + "envid-initializer", + _Success(value=Basis(change_set=change_set, persist=True, environment_id="env-from-initializer")), + ) + + fdv2 = FDv2( + Config(sdk_key="dummy"), + DataSystemConfig(initializers=[_InitializerBuilder(init)], synchronizers=None), + ) + + assert fdv2.environment_id is None + + set_on_ready = Event() + fdv2.start(set_on_ready) + assert set_on_ready.wait(1), "Data system did not become ready in time" + + assert fdv2.environment_id == "env-from-initializer" + fdv2.stop() + + +def test_environment_id_from_initializer_error_headers(): + init = _StaticInitializer( + "failing-initializer", + _Fail(error="boom", exception=None, headers={_LD_ENVID_HEADER: 'env-from-error'}), + ) + + fdv2 = FDv2( + Config(sdk_key="dummy"), + DataSystemConfig(initializers=[_InitializerBuilder(init)], synchronizers=None), + ) + + set_on_ready = Event() + fdv2.start(set_on_ready) + assert set_on_ready.wait(1), "Data system did not become ready in time" + + assert fdv2.environment_id == 'env-from-error' + fdv2.stop() + + +def test_environment_id_from_synchronizer_update(): + sync_mock: Synchronizer = Mock() + sync_mock.name = "envid-sync" + sync_mock.stop = Mock() + sync_mock.sync.return_value = iter([ + Update(state=DataSourceState.VALID, environment_id="env-from-sync"), + ]) + + fdv2 = FDv2( + Config(sdk_key="dummy"), + DataSystemConfig(initializers=None, synchronizers=[MockDataSourceBuilder(sync_mock)]), + ) + + set_on_ready = Event() + fdv2.start(set_on_ready) + assert set_on_ready.wait(1), "Data system did not become ready in time" + + assert fdv2.environment_id == "env-from-sync" + fdv2.stop() + + +def test_environment_id_is_retained_when_updates_omit_it(): + sync_mock: Synchronizer = Mock() + sync_mock.name = "envid-sync" + sync_mock.stop = Mock() + sync_mock.sync.return_value = iter([ + Update(state=DataSourceState.VALID, environment_id="env-from-sync"), + Update(state=DataSourceState.INTERRUPTED), + ]) + + fdv2 = FDv2( + Config(sdk_key="dummy"), + DataSystemConfig(initializers=None, synchronizers=[MockDataSourceBuilder(sync_mock)]), + ) + + set_on_ready = Event() + fdv2.start(set_on_ready) + assert set_on_ready.wait(1), "Data system did not become ready in time" + + assert fdv2.environment_id == "env-from-sync" + fdv2.stop() diff --git a/ldclient/testing/stub_util.py b/ldclient/testing/stub_util.py index c546bbe7..7f5f6aab 100644 --- a/ldclient/testing/stub_util.py +++ b/ldclient/testing/stub_util.py @@ -43,16 +43,18 @@ def make_delete_event(kind, key, version): return 'event:delete\ndata: %s\n\n' % json.dumps(data) -def stream_content(event=None): - stream = ChunkedResponse({'Content-Type': 'text/event-stream'}) +def stream_content(event=None, headers=None): + stream_headers = {'Content-Type': 'text/event-stream'} + stream_headers.update(headers or {}) + stream = ChunkedResponse(stream_headers) if event: stream.push(event) return stream -def poll_content(flags=[], segments=[]): +def poll_content(flags=[], segments=[], headers=None): data = {"flags": make_items_map(flags), "segments": make_items_map(segments)} - return JsonResponse(data) + return JsonResponse(data, headers) class MockEventProcessor(EventProcessor): diff --git a/ldclient/testing/test_ldclient_end_to_end.py b/ldclient/testing/test_ldclient_end_to_end.py index 61e245d2..8e608d14 100644 --- a/ldclient/testing/test_ldclient_end_to_end.py +++ b/ldclient/testing/test_ldclient_end_to_end.py @@ -5,6 +5,7 @@ from ldclient.client import Context, LDClient from ldclient.config import Config, HTTPConfig +from ldclient.hook import EvaluationSeriesContext, Hook, Metadata from ldclient.testing.http_util import ( BasicResponse, SequentialHandler, @@ -165,3 +166,31 @@ def test_can_connect_with_selfsigned_cert_by_setting_ca_certs(): config = Config(sdk_key='sdk_key', base_uri=server.uri, stream=False, send_events=False, http=HTTPConfig(ca_certs='./ldclient/testing/selfsigned.pem')) with LDClient(config=config) as client: assert client.is_initialized() + + +def test_hooks_receive_environment_id_in_streaming_mode(): + contexts = [] + + class CapturingHook(Hook): + @property + def metadata(self) -> Metadata: + return Metadata(name='capturing-hook') + + def before_evaluation(self, series_context: EvaluationSeriesContext, data: dict) -> dict: + contexts.append(series_context) + return data + + def after_evaluation(self, series_context: EvaluationSeriesContext, data: dict, detail) -> dict: + return data + + with start_server() as stream_server: + with stream_content(make_put_event([always_true_flag]), headers={'X-LD-EnvID': 'env-abc-123'}) as stream_handler: + stream_server.for_path('/all', stream_handler) + config = Config(sdk_key=sdk_key, stream_uri=stream_server.uri, send_events=False, hooks=[CapturingHook()]) + + with LDClient(config=config) as client: + assert client.is_initialized() + client.variation(always_true_flag['key'], user, False) + + assert len(contexts) == 1 + assert contexts[0].environment_id == 'env-abc-123' diff --git a/ldclient/testing/test_ldclient_hooks.py b/ldclient/testing/test_ldclient_hooks.py index 61521177..dcf085db 100644 --- a/ldclient/testing/test_ldclient_hooks.py +++ b/ldclient/testing/test_ldclient_hooks.py @@ -176,3 +176,18 @@ def test_migration_evaluation_detail_default_converts_to_off_if_invalid(): assert len(details) == 1 assert details[0].value == Stage.OFF.value assert details[0].variation_index is None + + +def test_series_context_environment_id_is_none_without_environment_id(): + contexts = [] + hook = MockHook(before_evaluation=lambda series_context, data: contexts.append(series_context), after_evaluation=record('after', [])) + + td = TestData.data_source() + td.update(td.flag('flag-key').variation_for_all(True)) + + config = Config('SDK_KEY', update_processor_class=td, send_events=False, hooks=[hook]) + client = LDClient(config=config) + client.variation('flag-key', user, False) + + assert len(contexts) == 1 + assert contexts[0].environment_id is None From 5ea64e4e55da33e0fe14bc295f7f3702c80a804d Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 12 Aug 2026 17:24:27 +0000 Subject: [PATCH 2/6] fix: Only record environment ID from successful responses Co-Authored-By: rlamb@launchdarkly.com <4955475+kinyoklion@users.noreply.github.com> --- ldclient/impl/datasource/streaming.py | 2 -- ldclient/impl/datasystem/fdv2.py | 14 ++++---------- .../impl/datasource/test_polling_processor.py | 4 ++-- ldclient/testing/impl/datasource/test_streaming.py | 4 ++-- .../impl/datasystem/test_fdv2_datasystem.py | 8 ++++---- 5 files changed, 12 insertions(+), 20 deletions(-) diff --git a/ldclient/impl/datasource/streaming.py b/ldclient/impl/datasource/streaming.py index 6b13e336..e5496147 100644 --- a/ldclient/impl/datasource/streaming.py +++ b/ldclient/impl/datasource/streaming.py @@ -94,8 +94,6 @@ def run(self): log.info("StreamingUpdateProcessor initialized ok.") self._ready.set() elif isinstance(action, Fault): - record_environment_id(self._data_source_update_sink, action.headers) - # 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: diff --git a/ldclient/impl/datasystem/fdv2.py b/ldclient/impl/datasystem/fdv2.py index fa763094..3359be3f 100644 --- a/ldclient/impl/datasystem/fdv2.py +++ b/ldclient/impl/datasystem/fdv2.py @@ -21,12 +21,7 @@ from ldclient.impl.listeners import Listeners from ldclient.impl.repeating_task import RepeatingTask from ldclient.impl.rwlock import ReadWriteLock -from ldclient.impl.util import ( - _LD_ENVID_HEADER, - _LD_FD_FALLBACK_HEADER, - _Fail, - log -) +from ldclient.impl.util import _LD_FD_FALLBACK_HEADER, _Fail, log from ldclient.interfaces import ( DataSourceErrorInfo, DataSourceErrorKind, @@ -222,8 +217,6 @@ def _run_initializers(self, set_on_ready: Event) -> bool: if isinstance(basis_result, _Fail): log.warning("Initializer %s failed: %s", initializer.name, basis_result.error) - if basis_result.headers is not None: - self._record_environment_id(basis_result.headers.get(_LD_ENVID_HEADER)) # An error response can still carry the FDv1 fallback directive. if basis_result.headers is not None and \ basis_result.headers.get(_LD_FD_FALLBACK_HEADER) == 'true': @@ -422,7 +415,8 @@ def reader(self: 'FDv2'): if self._stop_event.is_set(): return ConditionDirective.FALLBACK - self._record_environment_id(update.environment_id) + if update.state == DataSourceState.VALID: + self._record_environment_id(update.environment_id) # Handle the update if update.change_set is not None: @@ -509,7 +503,7 @@ def _persistent_store_outage_recovery(self, data_store_status: DataStoreStatus): log.error("Failed to reinitialize data store", exc_info=err) def _record_environment_id(self, environment_id: Optional[str]): - if environment_id is None: + if not isinstance(environment_id, str) or environment_id == '': return with self._lock.write(): diff --git a/ldclient/testing/impl/datasource/test_polling_processor.py b/ldclient/testing/impl/datasource/test_polling_processor.py index 726442fd..06e92d89 100644 --- a/ldclient/testing/impl/datasource/test_polling_processor.py +++ b/ldclient/testing/impl/datasource/test_polling_processor.py @@ -165,7 +165,7 @@ def test_records_environment_id_from_polling_headers(): sink = DataSourceUpdateSinkImpl(store, Listeners(), Listeners()) config._data_source_update_sink = sink setup_processor(config) - ready.wait() + assert ready.wait(2) assert sink.environment_id == 'env-abc-123' @@ -177,6 +177,6 @@ def test_environment_id_is_none_when_requester_provides_no_headers(): sink = DataSourceUpdateSinkImpl(store, Listeners(), Listeners()) config._data_source_update_sink = sink setup_processor(config) - ready.wait() + assert ready.wait(2) assert sink.environment_id is None diff --git a/ldclient/testing/impl/datasource/test_streaming.py b/ldclient/testing/impl/datasource/test_streaming.py index 66519c12..98c9d02a 100644 --- a/ldclient/testing/impl/datasource/test_streaming.py +++ b/ldclient/testing/impl/datasource/test_streaming.py @@ -472,7 +472,7 @@ def test_environment_id_is_none_when_not_provided(): assert sink.environment_id is None -def test_records_environment_id_from_error_response_headers(): +def test_does_not_record_environment_id_from_error_response_headers(): store = InMemoryFeatureStore() ready = Event() @@ -487,7 +487,7 @@ def test_records_environment_id_from_error_response_headers(): with StreamingUpdateProcessor(config, store, ready, None) as sp: sp.start() ready.wait(start_wait) - assert sink.environment_id == 'env-from-error' + assert sink.environment_id is None def expect_item(store, kind, item): diff --git a/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py b/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py index 43445a64..d4699e32 100644 --- a/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py +++ b/ldclient/testing/impl/datasystem/test_fdv2_datasystem.py @@ -786,7 +786,7 @@ def test_environment_id_from_initializer_basis(): fdv2.stop() -def test_environment_id_from_initializer_error_headers(): +def test_environment_id_is_not_recorded_from_initializer_error_headers(): init = _StaticInitializer( "failing-initializer", _Fail(error="boom", exception=None, headers={_LD_ENVID_HEADER: 'env-from-error'}), @@ -801,7 +801,7 @@ def test_environment_id_from_initializer_error_headers(): fdv2.start(set_on_ready) assert set_on_ready.wait(1), "Data system did not become ready in time" - assert fdv2.environment_id == 'env-from-error' + assert fdv2.environment_id is None fdv2.stop() @@ -826,13 +826,13 @@ def test_environment_id_from_synchronizer_update(): fdv2.stop() -def test_environment_id_is_retained_when_updates_omit_it(): +def test_environment_id_is_not_recorded_from_non_valid_updates(): sync_mock: Synchronizer = Mock() sync_mock.name = "envid-sync" sync_mock.stop = Mock() sync_mock.sync.return_value = iter([ Update(state=DataSourceState.VALID, environment_id="env-from-sync"), - Update(state=DataSourceState.INTERRUPTED), + Update(state=DataSourceState.INTERRUPTED, environment_id="env-from-interrupted"), ]) fdv2 = FDv2( From aefb0df2ee05fa0409b21c065d501bad03af1a60 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 12 Aug 2026 17:32:40 +0000 Subject: [PATCH 3/6] chore: Report environment ID from the contract test hooks Co-Authored-By: rlamb@launchdarkly.com <4955475+kinyoklion@users.noreply.github.com> --- contract-tests/hook.py | 1 + 1 file changed, 1 insertion(+) diff --git a/contract-tests/hook.py b/contract-tests/hook.py index d891504e..b05481d2 100644 --- a/contract-tests/hook.py +++ b/contract-tests/hook.py @@ -33,6 +33,7 @@ def __post(self, stage: str, series_context: EvaluationSeriesContext, data: dict 'context': series_context.context.to_dict(), 'defaultValue': series_context.default_value, 'method': series_context.method, + 'environmentId': series_context.environment_id, }, 'evaluationSeriesData': data, 'stage': stage, From 585459b648b50bd9c896c227e21c4197102f11c9 Mon Sep 17 00:00:00 2001 From: Devin AI Date: Wed, 12 Aug 2026 17:45:55 +0000 Subject: [PATCH 4/6] chore: Advertise the hook-environment-id contract test capability --- contract-tests/service.py | 1 + 1 file changed, 1 insertion(+) diff --git a/contract-tests/service.py b/contract-tests/service.py index 342c3870..a8e93674 100644 --- a/contract-tests/service.py +++ b/contract-tests/service.py @@ -77,6 +77,7 @@ def status(): 'instance-id', 'anonymous-redaction', 'evaluation-hooks', + 'hook-environment-id', 'omit-anonymous-contexts', 'client-prereq-events', 'persistent-data-store-redis', From c167ece64530d697fac4c6e2cfc67273f103bdba Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 12 Aug 2026 21:54:52 +0000 Subject: [PATCH 5/6] ci: Re-run contract tests against the released test harness Co-Authored-By: rlamb@launchdarkly.com <4955475+kinyoklion@users.noreply.github.com> From 7ce7b6a8efffb8c24d657d16bb2114f26ac11c60 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 12 Aug 2026 22:19:24 +0000 Subject: [PATCH 6/6] ci: Re-run contract tests against the released test harness Co-Authored-By: rlamb@launchdarkly.com <4955475+kinyoklion@users.noreply.github.com>