diff --git a/.gitignore b/.gitignore index e38959c..b21110d 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,7 @@ coverage.xml .mypy_cache/ .ruff_cache/ .pytest_cache/ +.idea/ + +# macOS +.DS_Store diff --git a/README.md b/README.md index 7783708..da528ec 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,8 @@ The official [Dynatrace](https://dynatrace.com) [OpenFeature](https://openfeatur The provider evaluates flags in-process: on initialization it fetches your flag configuration from the Dynatrace CDN and refreshes it in the background, so evaluations are local and do not make a network call. +Changes are picked up through a server-sent events (SSE) stream advertised in the flag configuration: a change notification triggers an immediate re-fetch, and polling every 10 minutes acts as a safety net. While the stream is unavailable or disabled, the provider polls every 10 seconds instead. + > **Status:** Early alpha — not yet ready for production use. ## Installation @@ -52,6 +54,7 @@ api.set_provider( | --- | --- | --- | --- | | `config_origin` | `str` | derived from SDK key | Override the flag configuration origin. | | `metrics_origin` | `str` | derived from SDK key | Override the evaluation metrics ingest origin. | +| `disable_sse` | `bool` | `False` | Don't attempt an SSE connection and poll only, every 10 seconds. | ## Running the e2e tests diff --git a/fm-provider-test-harness b/fm-provider-test-harness index 5c76973..ec236c7 160000 --- a/fm-provider-test-harness +++ b/fm-provider-test-harness @@ -1 +1 @@ -Subproject commit 5c7697375cfa64ddaec9d3297f435276ab07f5e3 +Subproject commit ec236c732f0fc8bb87cd56b077e04da51a28ed02 diff --git a/pyproject.toml b/pyproject.toml index 6084a19..12fe1fc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,8 @@ dependencies = [ "requests>=2.27.0", # 2.3 added HTTPResponse.shutdown, which _cdn uses to enforce the fetch deadline "urllib3>=2.3.0", + # SSE client (ld_eventsource); the floor is the release _sse was built and verified against + "launchdarkly-eventsource>=1.7.3,<2.0.0", ] requires-python = ">=3.10" @@ -63,6 +65,11 @@ pretty = true strict = true disallow_any_generics = false +[[tool.mypy.overrides]] +# ld_eventsource ships no type stubs or py.typed marker +module = ["ld_eventsource", "ld_eventsource.*"] +ignore_missing_imports = true + [tool.ruff] target-version = "py310" exclude = [".git", ".venv", "__pycache__", "venv"] diff --git a/src/dynatrace/openfeature/_cdn.py b/src/dynatrace/openfeature/_cdn.py index f58be96..9abc4bc 100644 --- a/src/dynatrace/openfeature/_cdn.py +++ b/src/dynatrace/openfeature/_cdn.py @@ -51,6 +51,11 @@ def is_rate_limited(self) -> bool: def is_success(self) -> bool: return 200 <= self.status < 300 + @property + def last_modified_epoch_s(self) -> float | None: + parsed = _parse_http_date(self.last_modified) + return parsed.timestamp() if parsed is not None else None + @dataclass class _CacheState: diff --git a/src/dynatrace/openfeature/_cdn_config.py b/src/dynatrace/openfeature/_cdn_config.py new file mode 100644 index 0000000..aa76be3 --- /dev/null +++ b/src/dynatrace/openfeature/_cdn_config.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import json +import typing + +#: ``eventStreams[].type`` value identifying the SSE change-notification stream +STREAM_TYPE_SSE = "sse" + + +def _non_blank(value: typing.Any) -> str | None: + return value if isinstance(value, str) and value.strip() else None + + +def extract_sse_url(config_json: str | None, fallback_origin: str | None) -> str | None: + """Return the SSE URL of the first ``sse`` entry in ``eventStreams``. + A non-blank ``url`` wins; else ``endpoint.requestUri`` on ``endpoint.origin`` or ``fallback_origin``.""" + if not config_json or not config_json.strip(): + return None + try: + config: typing.Any = json.loads(config_json) + except ValueError: + return None + streams = config.get("eventStreams") if isinstance(config, dict) else None + if not isinstance(streams, list): + return None + for stream in streams: + if not isinstance(stream, dict) or stream.get("type") != STREAM_TYPE_SSE: + continue + url = _non_blank(stream.get("url")) + if url is not None: + return url + endpoint = stream.get("endpoint") + if not isinstance(endpoint, dict): + return None + request_uri = _non_blank(endpoint.get("requestUri")) + if request_uri is None: + return None + origin = _non_blank(endpoint.get("origin")) or _non_blank(fallback_origin) + return origin + request_uri if origin is not None else None + return None diff --git a/src/dynatrace/openfeature/_polling.py b/src/dynatrace/openfeature/_polling.py index fcc5d66..c6a95e1 100644 --- a/src/dynatrace/openfeature/_polling.py +++ b/src/dynatrace/openfeature/_polling.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +import math import threading import time from collections.abc import Callable @@ -9,17 +10,15 @@ log = logging.getLogger(__name__) +# stands in for an SSE announcement that carried no lastModified: greater than any real one, so never dropped +_LAST_MODIFIED_UNKNOWN = math.inf + class PollingManager: - """ - Polls on a fixed-rate schedule anchored to request initiation, not completion. - - A ticker thread fires every interval_s regardless of how long the previous - fetch took. At most one fetch is in flight at a time (providers.md section - 2.1.1): a tick firing during a poll is skipped and emits nothing, so it does - not reset the staleness grace timer. The section 2.1 bound should keep that - from happening at all, but the bound is a design invariant rather than - something the runtime enforces, so the case is handled. + """Polls on a fixed-rate schedule anchored to request initiation, not completion. + + At most one fetch runs at a time (providers.md 2.1.1): a contended tick is skipped and emits + nothing; a contended SSE re-fetch is deferred and coalesced. set_interval switches cadence. """ def __init__( @@ -36,11 +35,23 @@ def __init__( self._emit_change = emit_change self._emit_success = emit_success self._emit_failure = emit_failure - self._interval_s = interval_s self._stop_event = threading.Event() - self._in_flight = False - self._in_flight_lock = threading.Lock() self._ticker_thread: threading.Thread | None = None + # guards everything below, and wakes the ticker when the interval changes or polling stops + self._cond = threading.Condition() + self._interval_s = interval_s + self._rescheduled = False + # set from submit until the fetch has settled (not until its thread exits), so no trigger defers behind a settled fetch + self._in_flight = False + # last-modified value of a deferred SSE re-fetch, or None when none is pending + self._pending_sse_fetch_last_modified: float | None = None + # last-modified value of the current (in-flight) SSE-triggered fetch, or None if poll-triggered + self._in_flight_sse_fetch_last_modified: float | None = None + + @property + def interval_s(self) -> float: + with self._cond: + return self._interval_s def start(self) -> None: if self._ticker_thread is not None and self._ticker_thread.is_alive(): @@ -56,85 +67,185 @@ def start(self) -> None: def stop(self) -> None: # stop event doubles as the fetch cancel signal, so an in-flight poll abandons its backoff self._stop_event.set() - # leave _in_flight alone: only _run_poll's finally clears it; clearing here would let a restart double up + with self._cond: + self._pending_sse_fetch_last_modified = None + self._cond.notify_all() if self._ticker_thread is not None: self._ticker_thread.join(timeout=2.0) - def _ticker_loop(self) -> None: - next_tick = time.monotonic() + self._interval_s - while not self._stop_event.is_set(): - wait = max(0.0, next_tick - time.monotonic()) - if self._stop_event.wait(wait): - break - next_tick += self._interval_s - now = time.monotonic() - if next_tick < now: - next_tick = now + self._interval_s - self._dispatch() - - def _dispatch(self) -> None: + def set_interval(self, interval_s: float) -> None: + """Change the poll cadence and re-anchor the schedule to now. No-op if unchanged.""" + with self._cond: + if interval_s == self._interval_s: + return + self._interval_s = interval_s + self._rescheduled = True + self._cond.notify_all() + + def trigger_immediate_fetch( + self, announced_last_modified: float | None = None + ) -> None: + """Fetch now if the slot is free, else defer and coalesce; never disturbs the cadence. + A deferred re-fetch is dropped only if a completed fetch is at least as new as announced.""" if self._stop_event.is_set(): return - with self._in_flight_lock: + announced = ( + _LAST_MODIFIED_UNKNOWN + if announced_last_modified is None + else announced_last_modified + ) + with self._cond: if self._in_flight: - # Neither failed nor succeeded: the tick learned nothing. - log.debug("poll tick skipped: a fetch is already in flight") + self._pending_sse_fetch_last_modified = ( + announced + if self._pending_sse_fetch_last_modified is None + else max(self._pending_sse_fetch_last_modified, announced) + ) + log.debug( + "fetch in flight; deferring SSE re-fetch (lastModified=%s)", + announced_last_modified, + ) return - self._in_flight = True + self._submit_locked(announced) + + def _ticker_loop(self) -> None: + with self._cond: + next_tick = time.monotonic() + self._interval_s + while not self._stop_event.is_set(): + if self._rescheduled: + # the polling interval changed + self._rescheduled = False + next_tick = time.monotonic() + self._interval_s + wait = next_tick - time.monotonic() + if wait > 0: + # woken early by set_interval or stop; the loop re-evaluates either way + self._cond.wait(wait) + continue + next_tick += self._interval_s + now = time.monotonic() + if next_tick < now: + next_tick = now + self._interval_s + self._on_tick_locked() + + def _on_tick_locked(self) -> None: + if self._stop_event.is_set(): + return + if self._in_flight: + # Neither failed nor succeeded: the tick learned nothing. + log.debug("poll tick skipped: a fetch is already in flight") + return + self._submit_locked(None) + + def _submit_locked(self, sse_refetch_at: float | None) -> None: + self._in_flight_sse_fetch_last_modified = sse_refetch_at + self._in_flight = True threading.Thread( - target=self._run_poll, + target=self._run_fetches, name="dt-provider-poller", daemon=True, ).start() - def _run_poll(self) -> None: - try: - self._poll() - finally: - with self._in_flight_lock: - self._in_flight = False + def _run_fetches(self) -> None: + # a deferred re-fetch runs on this same thread, so it queues behind the fetch rather than overlapping it + again = True + while again: + result: FetchResult | None = None + try: + result = self._poll() + finally: + with self._cond: + again = self._on_fetch_complete_locked(result) + + def _on_fetch_complete_locked(self, completed: FetchResult | None) -> bool: + """Settle the SSE re-fetch bookkeeping; return True to issue the pending re-fetch now. + ``completed`` is the finished fetch's result, or None if it raised.""" + issued_for = self._in_flight_sse_fetch_last_modified + self._in_flight_sse_fetch_last_modified = None + # frees the slot before anything below can re-claim it + self._in_flight = False - def _poll(self) -> None: + if completed is not None and completed.skipped: + # no request was made, so a re-fetch we issued is still unsatisfied; put it back rather than re-issue it, + # which would just be skipped again (the Retry-After window or fetch lock still holds) and spin + if issued_for is not None: + self._pending_sse_fetch_last_modified = ( + issued_for + if self._pending_sse_fetch_last_modified is None + else max(self._pending_sse_fetch_last_modified, issued_for) + ) + log.debug("SSE re-fetch made no request; keeping it pending") + return False + + pending = self._pending_sse_fetch_last_modified + if pending is None or self._stop_event.is_set(): + return False + self._pending_sse_fetch_last_modified = None + + observed = completed.last_modified_epoch_s if completed is not None else None + if ( + pending != _LAST_MODIFIED_UNKNOWN + and observed is not None + and observed >= pending + ): + log.debug( + "dropping deferred SSE re-fetch: completed fetch already returned " + "Last-Modified %s >= pending %s", + observed, + pending, + ) + return False + log.debug("issuing deferred SSE re-fetch") + self._in_flight_sse_fetch_last_modified = pending + self._in_flight = True + return True + + def _poll(self) -> FetchResult | None: + """Run one fetch and emit its outcome; return its result, or None if the fetch raised.""" cancel = self._stop_event + result: FetchResult | None = None try: result = self._fetch_flags(cancel) if cancel.is_set(): log.debug("provider stopped during fetch; discarding result") - return + return result - if result.skipped: - log.debug("fetch skipped (single-flight); emitting nothing") - return + self._handle_result(result) - if result.is_not_modified: - log.debug("flags not modified") - self._emit_success() - return + except Exception as exc: + if cancel.is_set(): + log.debug("poll abandoned during shutdown") + return result + log.warning("poll failed: %s", type(exc).__name__) + self._emit_failure(type(exc).__name__) + return result - if result.is_rate_limited: - log.debug("rate limited; serving cached values") - return + def _handle_result(self, result: FetchResult) -> None: + if result.skipped: + log.debug("fetch skipped (single-flight); emitting nothing") + return - if not result.is_success: - self._emit_failure(f"HTTP {result.status}") - return + if result.is_not_modified: + log.debug("flags not modified") + self._emit_success() + return - body = result.body - if not body: - self._emit_failure("empty response body") - return + if result.is_rate_limited: + log.debug("rate limited; serving cached values") + return - changed_keys = self._store_flags(body) - self._emit_success() + if not result.is_success: + self._emit_failure(f"HTTP {result.status}") + return - if changed_keys: - log.info("flag configuration changed: %s", changed_keys) - self._emit_change(changed_keys) + body = result.body + if not body: + self._emit_failure("empty response body") + return - except Exception as exc: - if cancel.is_set(): - log.debug("poll abandoned during shutdown") - return - log.warning("poll failed: %s", type(exc).__name__) - self._emit_failure(type(exc).__name__) + changed_keys = self._store_flags(body) + self._emit_success() + + if changed_keys: + log.info("flag configuration changed: %s", changed_keys) + self._emit_change(changed_keys) diff --git a/src/dynatrace/openfeature/_sse.py b/src/dynatrace/openfeature/_sse.py new file mode 100644 index 0000000..f3e39c3 --- /dev/null +++ b/src/dynatrace/openfeature/_sse.py @@ -0,0 +1,203 @@ +from __future__ import annotations + +import contextlib +import logging +import socket +import threading +import typing +from collections.abc import Callable + +import urllib3 +from ld_eventsource import SSEClient +from ld_eventsource.actions import Event, Fault, Start +from ld_eventsource.config import ConnectStrategy, ErrorStrategy, RetryDelayStrategy +from urllib3.util import Timeout as Urllib3Timeout + +from ._sse_message import SseMessage + +log = logging.getLogger(__name__) + +# the stream is long-lived and idle between notifications, so only a dead connection should trip this +SSE_READ_TIMEOUT_S: float = 150 +# reconnect backoff: doubles from the initial delay up to the max, with jitter +SSE_INITIAL_RETRY_DELAY_S: float = 1.0 +SSE_MAX_RETRY_DELAY_S: float = 30.0 +_SSE_RETRY_JITTER = 0.5 + + +def mask_url(url: str) -> str: + """Mask the query string, which carries the Ably credentials, for safe logging.""" + q = url.find("?") + return url[:q] + "?" if q >= 0 else url + + +class _StreamPool(urllib3.PoolManager): + """Pool that remembers the latest stream response, so stop() can reach its socket.""" + + def __init__(self) -> None: + super().__init__() + self.response: urllib3.BaseHTTPResponse | None = None + + def urlopen( + self, *args: typing.Any, **kwargs: typing.Any + ) -> urllib3.BaseHTTPResponse: + response = super().urlopen(*args, **kwargs) + self.response = response + return response + + def shutdown_stream(self) -> None: + """Shut the stream's socket down so the server sees it close, even on Windows. + ld_eventsource's own close() only releases the connection there, because shutdown() does not + wake the reader thread blocked in recv(), and closing under it would deadlock on its buffer lock. + So the socket stays open until GC and the server keeps a stale subscriber; shutting it down here + sends the FIN. The reader may stay blocked, but the stop flag keeps it inert: it makes no callbacks.""" + sock = _socket_of(self.response) + if sock is None: + return + # already closed or never connected + with contextlib.suppress(OSError): + sock.shutdown(socket.SHUT_RDWR) + + +def _socket_of(response: urllib3.BaseHTTPResponse | None) -> socket.socket | None: + # a kept-alive connection holds the socket; a close-delimited response takes it over, and then it + # is only reachable through the response's file object (http.client's fp over a SocketIO) + sock = getattr(getattr(response, "connection", None), "sock", None) + if sock is None: + fp = getattr(getattr(response, "_fp", None), "fp", None) + sock = getattr(getattr(fp, "raw", None), "_sock", None) + return sock if isinstance(sock, socket.socket) else None + + +class SseManager: + """Manages the SSE connection to the Ably event stream (an opaque HTTP endpoint, no Ably SDK). + + One daemon thread routes ``Start``, ``Fault`` and parsed ``Event`` actions to the callbacks; + reconnection and backoff are delegated to ``ld_eventsource``. + """ + + def __init__( + self, + sse_url: str, + on_connected: Callable[[], None], + on_disconnected: Callable[[], None], + on_message: Callable[[SseMessage], None], + ) -> None: + self._sse_url = sse_url + self._on_connected = on_connected + self._on_disconnected = on_disconnected + self._on_message = on_message + self._stopped = threading.Event() + self._client: SSEClient | None = None + # ours rather than the library's, so the library leaves closing it to us + self._pool = _StreamPool() + self._thread: threading.Thread | None = None + + def start(self) -> None: + """Open the stream on a daemon thread; returns without waiting for it to connect.""" + if self._thread is not None: + return + self._client = SSEClient( + connect=ConnectStrategy.http( + self._sse_url, + pool=self._pool, + urllib3_request_options={ + "timeout": Urllib3Timeout(read=SSE_READ_TIMEOUT_S) + }, + ), + initial_retry_delay=SSE_INITIAL_RETRY_DELAY_S, + retry_delay_strategy=RetryDelayStrategy.default( + max_delay=SSE_MAX_RETRY_DELAY_S, jitter_multiplier=_SSE_RETRY_JITTER + ), + # never give up: every failure surfaces as a Fault and the client reconnects + error_strategy=ErrorStrategy.always_continue(), + ) + self._thread = threading.Thread( + target=self._run, name="dt-provider-sse", daemon=True + ) + self._thread.start() + log.info("SSE manager started (url=%s)", mask_url(self._sse_url)) + + def stop(self) -> None: + """Close the stream without blocking, so it is safe under a caller's lock. + Callbacks stop at once; the close itself runs on its own daemon thread.""" + if self._stopped.is_set(): + return + self._stopped.set() + client = self._client + if client is not None: + # drops the stream at once on every OS; the library's close() alone cannot on Windows + self._pool.shutdown_stream() + # on Windows ld_eventsource cannot interrupt a blocked recv(), and its close() then waits + # on the reader's buffer lock until the server sends something, possibly forever + threading.Thread( + target=self._close_client, + args=(client,), + name="dt-provider-sse-close", + daemon=True, + ).start() + log.info("SSE manager stopped") + + def _close_client(self, client: SSEClient) -> None: + try: + client.close() + self._pool.clear() + except Exception as exc: + log.warning("error closing SSE connection: %s", type(exc).__name__) + + def _run(self) -> None: + client = self._client + if client is None: + return + try: + for action in client.all: + # the client reconnects even after close() if it was sleeping out a backoff + if self._stopped.is_set(): + break + try: + self._dispatch(action) + except Exception: + # a failing callback must not kill the stream + log.warning("SSE callback failed", exc_info=True) + except Exception as exc: + if not self._stopped.is_set(): + # always_continue keeps the iterator alive, so only a library fault lands here + log.warning("SSE reader stopped unexpectedly: %s", type(exc).__name__) + self._on_disconnected() + finally: + # closes a connection opened after stop() raced a pending reconnect; the library's + # close() alone only releases it on Windows, leaving the stream open + self._pool.shutdown_stream() + self._close_client(client) + + def _dispatch(self, action: object) -> None: + if isinstance(action, Start): + log.info("SSE connection opened (url=%s)", mask_url(self._sse_url)) + self._on_connected() + elif isinstance(action, Fault): + if action.error is None: + log.info("SSE connection closed") + else: + # type only: the message of a connect error can carry the credentialed URL + log.warning( + "SSE error (url=%s): %s", + mask_url(self._sse_url), + type(action.error).__name__, + ) + self._on_disconnected() + elif isinstance(action, Event): + parsed = SseMessage.parse(action.data) + if parsed is None: + log.debug( + "SSE event received (event=%s) but payload unparseable; ignoring", + action.event, + ) + return + log.debug( + "SSE event received (event=%s, type=%s, lastModified=%s)", + action.event, + parsed.type, + parsed.last_modified, + ) + self._on_message(parsed) + # comments are keep-alives; ignore diff --git a/src/dynatrace/openfeature/_sse_message.py b/src/dynatrace/openfeature/_sse_message.py new file mode 100644 index 0000000..e7510ba --- /dev/null +++ b/src/dynatrace/openfeature/_sse_message.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import json +import typing +from dataclasses import dataclass + +#: server-side signal to re-fetch the flagd config JSON from the CDN +TYPE_REFETCH_CONFIG = "refetchConfig" + + +@dataclass(frozen=True) +class SseMessage: + """Parsed SSE message envelope per the SSE message format spec. + + Callers must ignore unknown ``type``s; ``etag`` is parsed for completeness only. + ``metadata`` is not exposed: providers must not forward it to OFREP clients. + """ + + type: str + etag: str | None = None + last_modified: int | None = None + + @classmethod + def parse(cls, payload: str | None) -> SseMessage | None: + """Parse an Ably envelope (message JSON-encoded in ``data``) or a flat message. + Returns None for invalid JSON or a missing/blank ``type``.""" + if not payload or not payload.strip(): + return None + try: + outer: typing.Any = json.loads(payload) + data = outer.get("data") if isinstance(outer, dict) else None + inner: typing.Any = json.loads(data) if isinstance(data, str) else outer + except ValueError: + return None + if not isinstance(inner, dict): + return None + msg_type = inner.get("type") + if not isinstance(msg_type, str) or not msg_type.strip(): + return None + etag = inner.get("etag") + last_modified = inner.get("lastModified") + return cls( + type=msg_type, + etag=etag if isinstance(etag, str) else None, + # a malformed lastModified reads as unknown, so the re-fetch it announces is never dropped + last_modified=last_modified + if isinstance(last_modified, int) and not isinstance(last_modified, bool) + else None, + ) diff --git a/src/dynatrace/openfeature/provider.py b/src/dynatrace/openfeature/provider.py index 2213b38..75b40ad 100644 --- a/src/dynatrace/openfeature/provider.py +++ b/src/dynatrace/openfeature/provider.py @@ -5,6 +5,7 @@ from collections.abc import Mapping, Sequence from importlib.metadata import PackageNotFoundError from importlib.metadata import version as _pkg_version +from urllib.parse import urlsplit import requests @@ -18,6 +19,7 @@ from ._cdn import MAX_FETCH_DURATION_S as _MAX_FETCH_DURATION_S from ._cdn import CdnFetcher +from ._cdn_config import extract_sse_url from ._metrics_aggregator import ( HARD_CAP_DISTINCT_FLAGS, SOFT_CAP_DISTINCT_FLAGS, @@ -26,12 +28,16 @@ from ._metrics_flusher import EvaluationMetricsFlusher from ._metrics_hook import DynatraceMetricsHook from ._polling import PollingManager +from ._sse import SseManager +from ._sse_message import TYPE_REFETCH_CONFIG, SseMessage from .options import DynatraceProviderOptions log = logging.getLogger(__name__) _PROVIDER_NAME = "dynatrace" _POLL_INTERVAL_S: float = 10.0 +#: poll cadence while SSE is connected: SSE delivers notifications, polling is a safety net +_SSE_CONNECTED_POLL_INTERVAL_S: float = 600.0 _GRACE_PERIOD_S: float = 60.0 _METRICS_FLUSH_INTERVAL_S: float = 30.0 _VERSION_METADATA_KEY = "version" @@ -55,6 +61,8 @@ def __init__( # noqa: PLR0913 - public constructor with optional injectable sea fetcher: CdnFetcher | None = None, metrics_flusher: EvaluationMetricsFlusher | None = None, poll_interval_s: float = _POLL_INTERVAL_S, + sse_connected_poll_interval_s: float = _SSE_CONNECTED_POLL_INTERVAL_S, + disable_sse: bool = False, grace_period_s: float = _GRACE_PERIOD_S, metrics_flush_interval_s: float = _METRICS_FLUSH_INTERVAL_S, ) -> None: @@ -62,11 +70,16 @@ def __init__( # noqa: PLR0913 - public constructor with optional injectable sea sdk_key, config_origin=config_origin, metrics_origin=metrics_origin ) # providers.md 2.1: a poll must not outlive its interval; the bound is the whole fetch, not one request - if fetcher is None and poll_interval_s <= _MAX_FETCH_DURATION_S: - raise ValueError( - f"poll_interval_s ({poll_interval_s}s) must be greater than the " - f"maximum fetch duration ({_MAX_FETCH_DURATION_S:.2f}s)" - ) + if fetcher is None: + for name, interval in ( + ("poll_interval_s", poll_interval_s), + ("sse_connected_poll_interval_s", sse_connected_poll_interval_s), + ): + if interval <= _MAX_FETCH_DURATION_S: + raise ValueError( + f"{name} ({interval}s) must be greater than the " + f"maximum fetch duration ({_MAX_FETCH_DURATION_S:.2f}s)" + ) if metrics_flush_interval_s <= 0: raise ValueError( f"metrics_flush_interval_s ({metrics_flush_interval_s}s) must be positive" @@ -75,8 +88,13 @@ def __init__( # noqa: PLR0913 - public constructor with optional injectable sea self._flagd_core = flagd_core or FlagdCore() self._fetcher = fetcher or CdnFetcher(options) self._poll_interval_s = poll_interval_s + self._sse_connected_poll_interval_s = sse_connected_poll_interval_s + self._sse_enabled = not disable_sse self._grace_period_s = grace_period_s self._poller: PollingManager | None = None + self._sse_manager: SseManager | None = None + # true while the SSE stream is open, i.e. the provider is in SSE mode rather than polling mode + self._sse_connected = False self._grace_timer: threading.Timer | None = None self._state = ProviderStatus.NOT_READY self._lock = threading.RLock() @@ -105,6 +123,8 @@ def get_provider_hooks(self) -> list[Hook]: def initialize(self, evaluation_context: EvaluationContext | None = None) -> None: log.info("initializing Dynatrace provider (key=%s)", self._options.public_key) + if not self._sse_enabled: + log.debug("SSE disabled by configuration; running in polling-only mode") with self._lock: self._generation += 1 generation = self._generation @@ -131,6 +151,7 @@ def initialize(self, evaluation_context: EvaluationContext | None = None) -> Non return self._cancel_grace_timer() self._state = ProviderStatus.READY + self._start_sse_locked(body) self._start_poller_locked() self._metrics_flusher.start() log.info("Dynatrace provider initialized successfully") @@ -176,10 +197,51 @@ def _start_poller_locked(self) -> None: emit_change=self._on_configuration_changed, emit_success=self._on_fetch_success, emit_failure=self._on_fetch_failure, - interval_s=self._poll_interval_s, + interval_s=self._sse_connected_poll_interval_s + if self._sse_connected + else self._poll_interval_s, ) self._poller.start() + def _start_sse_locked(self, body: str) -> None: + """Open the SSE stream advertised in ``body``, if SSE is enabled. Caller holds ``_lock``.""" + if not self._sse_enabled or self._sse_manager is not None: + return + cdn = urlsplit(self._options.cdn_url) + sse_url = extract_sse_url(body, f"{cdn.scheme}://{cdn.netloc}") + if sse_url is None: + # the server always sends eventStreams, so this is not worth logging + return + + manager: SseManager | None = None + + def on_connected() -> None: + self._on_sse_connected(manager) + + def on_disconnected() -> None: + self._on_sse_disconnected(manager) + + def on_message(msg: SseMessage) -> None: + self._on_sse_message(manager, msg) + + manager = SseManager(sse_url, on_connected, on_disconnected, on_message) + try: + manager.start() + except Exception as exc: + log.warning( + "failed to start SSE connection; falling back to polling-only mode: %s", + type(exc).__name__, + ) + return + self._sse_manager = manager + + def _stop_sse_locked(self) -> None: + """Close the SSE stream and forget its state. Caller holds ``_lock``.""" + self._sse_connected = False + if self._sse_manager is not None: + self._sse_manager.stop() + self._sse_manager = None + def _fail_initialization(self, generation: int, reason: str) -> None: """Enter ERROR and start polling so the provider can still recover. @@ -212,6 +274,8 @@ def shutdown(self) -> None: ) if not already_idle: log.info("shutting down Dynatrace provider") + # providers.md 6.1: close SSE first, so no notification triggers a fetch mid-shutdown + self._stop_sse_locked() self._state = ProviderStatus.NOT_READY self._cancel_grace_timer() if self._poller is not None: @@ -289,6 +353,10 @@ def _store_flags(self, body: str) -> list[str]: self._fetcher.invalidate_cache() raise self._update_metrics_version_from_flag_set() + with self._lock: + # covers the failed-init case, where there was no config to read an SSE URL from; a shut-down provider has no poller + if self._poller is not None: + self._start_sse_locked(body) return changed def _update_metrics_version_from_flag_set(self) -> None: @@ -327,6 +395,63 @@ def _on_configuration_changed(self, changed_keys: list[str]) -> None: ProviderEventDetails(flags_changed=changed_keys) ) + # SSE callbacks: each names the manager it came from, so a stream already closed by shutdown cannot act. + # SSE only prompts fetches, so its state picks the poll cadence and never touches READY/STALE/ERROR. + + def _on_sse_connected(self, source: SseManager | None) -> None: + """Switch to SSE mode: safety-net cadence plus a catch-up fetch for anything missed while down.""" + with self._lock: + if source is None or source is not self._sse_manager: + return + self._sse_connected = True + poller = self._poller + if poller is None: + return + poller.set_interval(self._sse_connected_poll_interval_s) + log.info( + "SSE connected; poll interval switched to %ss", + self._sse_connected_poll_interval_s, + ) + # no announced lastModified, so a deferred catch-up is never dropped + poller.trigger_immediate_fetch(None) + + def _on_sse_disconnected(self, source: SseManager | None) -> None: + """Fall back to polling mode; fetch failures from here on are handled like any poll failure.""" + with self._lock: + if source is None or source is not self._sse_manager: + return + if not self._sse_connected: + # never connected, or a repeat fault while the library retries + return + self._sse_connected = False + log.warning( + "SSE disconnected; falling back to polling every %ss", + self._poll_interval_s, + ) + if self._poller is not None: + self._poller.set_interval(self._poll_interval_s) + + def _on_sse_message(self, source: SseManager | None, msg: SseMessage) -> None: + """Trigger a conditional re-fetch on ``refetchConfig``; ignore every other type. + lastModified is forwarded only so a deferred re-fetch can be dropped (providers.md 2.1.1).""" + if msg.type != TYPE_REFETCH_CONFIG: + log.debug( + "ignoring SSE message of type %r (server provider only handles %r)", + msg.type, + TYPE_REFETCH_CONFIG, + ) + return + with self._lock: + if source is None or source is not self._sse_manager: + return + poller = self._poller + log.debug( + "SSE refetchConfig received; triggering CDN re-fetch (lastModified=%s)", + msg.last_modified, + ) + if poller is not None: + poller.trigger_immediate_fetch(msg.last_modified) + def _start_grace_timer(self) -> None: self._cancel_grace_timer() t = threading.Timer(self._grace_period_s, self._grace_expired) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..9c6e939 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,12 @@ +from collections.abc import Iterator + +import pytest +from sse_server import SseServer + + +@pytest.fixture +def sse_server() -> Iterator[SseServer]: + """A live SSE endpoint on an ephemeral localhost port, stopped after the test.""" + server = SseServer() + yield server + server.stop() diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index 3774fd5..314b230 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -3,14 +3,11 @@ import pytest from mock_server import MockServer -# needs an SSE-triggered re-fetch; SSE support is on hold -_SKIPPED_TAGS = {"single-flight"} - def pytest_bdd_apply_tag(tag: str, function: Callable[..., object]) -> bool: # gherkin tags are not registered pytest markers, so skip or ignore them rather than mark - if tag in _SKIPPED_TAGS: - pytest.mark.skip(reason=f"@{tag} is not supported yet")(function) + # if tag in _SKIPPED_TAGS: + # pytest.mark.skip(reason=f"@{tag} is not supported yet")(function) return True diff --git a/tests/e2e/mock_server.py b/tests/e2e/mock_server.py index cf168fa..dbc3ad9 100644 --- a/tests/e2e/mock_server.py +++ b/tests/e2e/mock_server.py @@ -18,6 +18,8 @@ # harness submodule at the repo root HARNESS_DIR = Path(__file__).resolve().parents[2] / "fm-provider-test-harness" +# the only path the mock server serves an SSE stream on +SSE_PATH = "/sse" # image repository; the tag comes from the pinned harness's version.txt IMAGE_REPO = "ghcr.io/dynatrace/fm-provider-mock-server" @@ -147,12 +149,46 @@ def count_cdn_requests(self) -> int: """Return how many CDN requests were received since the last reset.""" return len(self.fetch_cdn_requests()) + def sse_client_count(self) -> int: + """Return how many SSE subscribers are currently connected.""" + payload = self.send_control_request("GET", "/__control__/sse/clients").json() + return int(payload["clients"]) + + def drop_sse(self, outage: float | None = None) -> None: + """Drop every SSE subscriber and refuse new connections. + + With no outage they stay refused until restore_sse() or reset(); + otherwise for outage seconds, after which reconnects succeed again. + """ + body = None if outage is None else {"reconnectSeconds": outage} + self.send_control_request("POST", "/__control__/sse/disconnect", body) + + def restore_sse(self) -> None: + """End an outage opened by drop_sse() so the next reconnect attempt succeeds.""" + self.send_control_request( + "POST", "/__control__/sse/disconnect", {"reconnectSeconds": 0} + ) + + def emit_sse(self, payload: str) -> None: + """Broadcast payload verbatim as the data: line of an SSE event to every subscriber.""" + # raw rather than JSON-encoded, so an unparseable payload reaches the wire as written + self.send_control_request("POST", "/__control__/sse/emit", raw=payload) + def send_control_request( - self, method: str, path: str, body: typing.Any = None + self, + method: str, + path: str, + body: typing.Any = None, + *, + raw: str | None = None, ) -> requests.Response: """Call a control-plane endpoint, failing on any non-2xx status.""" res = self._http.request( - method, self.origin + path, json=body, timeout=CONTROL_REQUEST_TIMEOUT + method, + self.origin + path, + json=body, + data=None if raw is None else raw.encode(), + timeout=CONTROL_REQUEST_TIMEOUT, ) res.raise_for_status() return res diff --git a/tests/e2e/test_cdn_e2e.py b/tests/e2e/test_e2e.py similarity index 55% rename from tests/e2e/test_cdn_e2e.py rename to tests/e2e/test_e2e.py index a07ecc7..5cdcb8c 100644 --- a/tests/e2e/test_cdn_e2e.py +++ b/tests/e2e/test_e2e.py @@ -1,4 +1,4 @@ -"""Step definitions for the shared CDN gherkin spec (fm-provider-test-harness/gherkin/cdn.feature). +"""Step definitions for the shared gherkin specs (fm-provider-test-harness/gherkin/). Black-box: each scenario binds a fresh provider to its own OpenFeature domain and observes it only through the SDK client, against the containerized mock server. @@ -7,14 +7,18 @@ from __future__ import annotations import itertools +import json +import threading import time from collections.abc import Callable, Iterator from dataclasses import dataclass, field from http import HTTPStatus +from urllib.parse import urlsplit import pytest from mock_server import ( HARNESS_DIR, + SSE_PATH, CdnResponse, MockServer, RecordedRequest, @@ -22,6 +26,7 @@ ) from openfeature import api from openfeature.client import OpenFeatureClient +from openfeature.event import EventDetails, ProviderEvent from openfeature.exception import ErrorCode from openfeature.provider import ProviderStatus from pytest_bdd import parsers, scenarios, step @@ -34,8 +39,16 @@ BASE_KEY = "dt01.server_us_abcdef1234.de848e97a9cc4cc78aae568e65f49a9d_a1b2c3d4e5" # shared flag configuration fixtures from the harness submodule FLAGS_DIR = HARNESS_DIR / "flags" +# timings the harness shortens for every provider; the rest are production defaults +TIMINGS = json.loads((HARNESS_DIR / "gherkin" / "timings.json").read_text()) +SSE_CONNECTED_POLL_INTERVAL = TIMINGS["SSE_CONNECTED_POLL_INTERVAL_MS"] / 1000 +GRACE_PERIOD = TIMINGS["GRACE_PERIOD_MS"] / 1000 +# the cadence while SSE is disabled or disconnected; the CDN spec runs with SSE disabled +SSE_DISCONNECTED_POLL_INTERVAL = _POLL_INTERVAL_S +# scheduling slack on top of the event source's worst-case reconnect after a blip +BLIP_RECONNECT_SLACK = 1.0 # read from the provider, so scenarios assert against the real cadence -POLL_INTERVAL = _POLL_INTERVAL_S +POLL_INTERVAL = SSE_DISCONNECTED_POLL_INTERVAL # bound on one CDN attempt REQUEST_TIMEOUT = REQUEST_TIMEOUT_S # longer than the poll cadence, so a refresh is observed @@ -70,8 +83,20 @@ WAIT_POLL_STEP = 0.1 # sleep between checks while asserting a condition keeps holding HOLD_CHECK_STEP = 0.05 +# slack for something the step just caused, and the window for asserting its absence +SETTLE = 2.0 +# how long to wait for a provider event the step just caused +EVENT_TIMEOUT = 5.0 +# how long to wait for the provider to open its SSE stream +SSE_CONNECT_TIMEOUT = 5.0 +# placeholder for the mock server's authority in a spec'd URL +MOCK_SERVER_PLACEHOLDER = "" +# how far before a tick an SSE re-fetch is triggered, so it is still in flight then +SHORTLY_BEFORE = 0.5 +REFETCH_MESSAGE = '{"type":"refetchConfig","lastModified":1704153600}' scenarios(str(HARNESS_DIR / "gherkin" / "cdn.feature")) +scenarios(str(HARNESS_DIR / "gherkin" / "sse.feature")) # gives every scenario a unique OpenFeature domain so providers never leak across scenarios _SCENARIO_SEQ = itertools.count(1) @@ -86,12 +111,19 @@ class Scenario: sdk_key: str = "" provider: DynatraceProvider | None = None init_error: object | None = None + # SSE is off throughout the CDN spec; the SSE spec's background turns it on + sse_enabled: bool = False # counts read as "since the last thing that happened" count_baseline: int = 0 # polling timer starts when initialization returns; tick boundaries derive from it polling_started_at: float = 0.0 cdn_responses: list[CdnResponse] = field(default_factory=list) retry_after: float = 0.0 + # provider events seen on this scenario's client, in arrival order + events: list[tuple[ProviderEvent, EventDetails]] = field(default_factory=list) + events_lock: threading.Lock = field(default_factory=threading.Lock) + # index into events; each "a PROVIDER_* event is emitted" step consumes one + event_cursor: int = 0 mock: MockServer @@ -99,8 +131,14 @@ class Scenario: def scenario(mock_server: MockServer) -> Iterator[Scenario]: """Give each scenario a unique domain and a reset mock; shut the SDK down afterwards.""" mock_server.reset() + # reset clears the SSE outage gate but not a stream the previous scenario's provider is + # still tearing down; wait it out so "opens no SSE connection" sees a clean slate + _poll_until(SETTLE, lambda: mock_server.sse_client_count() == 0) domain = f"e2e-test-{next(_SCENARIO_SEQ)}" - yield Scenario(domain=domain, client=api.get_client(domain), mock=mock_server) + s = Scenario(domain=domain, client=api.get_client(domain), mock=mock_server) + for event in ProviderEvent: + s.client.add_handler(event, _event_recorder(s, event)) + yield s api.shutdown() @@ -109,7 +147,12 @@ def scenario(mock_server: MockServer) -> Iterator[Scenario]: @step("a mock server is running") def _a_mock_server_is_running(scenario: Scenario) -> None: - pass + scenario.sse_enabled = False + + +@step("a mock server with SSE enabled is running") +def _a_mock_server_with_sse_enabled_is_running(scenario: Scenario) -> None: + scenario.sse_enabled = True @step(parsers.re(r'the SDK key "(?P[^"]*)"')) @@ -117,6 +160,11 @@ def _the_sdk_key(scenario: Scenario, key: str) -> None: scenario.sdk_key = key +@step("SSE is disabled by configuration") +def _sse_is_disabled_by_configuration(scenario: Scenario) -> None: + scenario.sse_enabled = False + + @step(parsers.re(r'the CDN serves the "(?P[^"]+)" flag configuration')) def _the_cdn_serves_the_flag_configuration(scenario: Scenario, name: str) -> None: _set_cdn_responses( @@ -124,10 +172,31 @@ def _the_cdn_serves_the_flag_configuration(scenario: Scenario, name: str) -> Non ) +@step( + parsers.re( + r'the CDN serves the "(?P[^"]+)" flag configuration' + r' advertising an SSE stream at url "(?P[^"]+)"' + ) +) +def _the_cdn_serves_the_flag_configuration_advertising_an_sse_stream_at_url( + scenario: Scenario, name: str, url: str +) -> None: + _set_cdn_responses( + scenario, + build_cdn_response( + HTTPStatus.OK, _advertising_sse_url(scenario, _load_flag_config(name), url) + ), + ) + + +# a header value is double-quoted, or single-quoted when it holds double quotes itself (an ETag: '"v1"') +_HEADER_VALUE = r"(?P\"[^\"]*\"|'[^']*')" + + @step( parsers.re( r'an initialized, READY provider serving the "(?P[^"]+)" flag configuration' - r'(?: with "(?P
[^"]+)" header "(?P[^"]*)")?' + r'(?: with "(?P
[^"]+)" header ' + _HEADER_VALUE + r")?" ) ) def _an_initialized_ready_provider_serving_the_flag_configuration( @@ -135,11 +204,12 @@ def _an_initialized_ready_provider_serving_the_flag_configuration( ) -> None: response = build_cdn_response(HTTPStatus.OK, _load_flag_config(name)) if header: - response.with_header(header, value or "") + response.with_header(header, _unquote(value) if value else "") scenario.sdk_key = BASE_KEY _set_cdn_responses(scenario, response) _register_provider_with_sdk(scenario) assert scenario.init_error is None, scenario.init_error + _await_sse_settled(scenario) # end of arrangement: counts and tick boundaries run from here scenario.polling_started_at = time.time() _reset_count_baseline(scenario) @@ -197,7 +267,7 @@ def _the_cdn_responds_with_status_and_body( @step( parsers.re( r'the CDN responds with status (?P\d+) and the "(?P[^"]+)" flag configuration' - r' with "(?P
[^"]+)" header "(?P[^"]*)"' + r' with "(?P
[^"]+)" header ' + _HEADER_VALUE ), converters={"status": int}, ) @@ -206,7 +276,9 @@ def _the_cdn_responds_with_status_and_the_flag_configuration_with_header( ) -> None: _set_cdn_responses( scenario, - build_cdn_response(status, _load_flag_config(name)).with_header(header, value), + build_cdn_response(status, _load_flag_config(name)).with_header( + header, _unquote(value) + ), ) @@ -243,8 +315,71 @@ def _the_cdn_responds_with_status_and_retry_after_seconds( @step("polling triggers a configuration refetch") def _polling_triggers_a_configuration_refetch(scenario: Scenario) -> None: + _await_cdn_request_within(scenario, POLL_INTERVAL) + + +# a blip: the backend drops the stream and refuses connections for a while, then accepts them again +@step( + parsers.re(r"the SSE stream drops for (?P\d+) seconds and reconnects"), + converters={"seconds": int}, +) +def _the_sse_stream_drops_and_reconnects(scenario: Scenario, seconds: int) -> None: + # the bound is the event source's doubling backoff, whose retry after an outage can come up to + # twice the outage later. It must stay under the fallback poll interval, or a poll lands mid-blip. + reconnect_timeout = 3 * seconds + BLIP_RECONNECT_SLACK + assert reconnect_timeout < POLL_INTERVAL, ( + f"a {seconds}s blip cannot be told apart from polling at {POLL_INTERVAL}s" + ) + scenario.mock.drop_sse(seconds) + # the drop is asynchronous: wait for the old stream to go, so its count is not mistaken for the reconnect + _poll_until(SETTLE, lambda: scenario.mock.sse_client_count() == 0) + assert _poll_until( + reconnect_timeout, + lambda: scenario.mock.sse_client_count() >= 1, + ), f"SSE did not reconnect within {reconnect_timeout}s of a {seconds}s outage" + + +@step("the SSE stream drops and stays down") +def _the_sse_stream_drops_and_stays_down(scenario: Scenario) -> None: + scenario.mock.drop_sse() + assert _poll_until(SETTLE, lambda: scenario.mock.sse_client_count() == 0), ( + "the mock server did not drop the SSE stream" + ) + + +@step(parsers.re(r"the server emits an SSE message (?P.*)")) +def _the_server_emits_an_sse_message(scenario: Scenario, payload: str) -> None: baseline = scenario.mock.count_cdn_requests() - _wait_for(POLL_TIMEOUT, lambda: scenario.mock.count_cdn_requests() > baseline) + # pytest-bdd doubles backslashes when it substitutes an outline cell into the step text, + # so "\"v2\"" would reach the wire as invalid JSON; Cucumber passes the cell as written + scenario.mock.emit_sse(_unquote(payload).replace("\\\\", "\\")) + _poll_until(SETTLE, lambda: scenario.mock.count_cdn_requests() > baseline) + + +@step("the GRACE_PERIOD expires") +def _the_grace_period_expires(scenario: Scenario) -> None: + _poll_until( + GRACE_PERIOD + SETTLE, + lambda: scenario.client.get_provider_status() == ProviderStatus.ERROR, + ) + + +@step("the provider is shut down") +def _the_provider_is_shut_down(scenario: Scenario) -> None: + api.shutdown() + + +# occupies the fetch slot just before a tick, so the tick has something to contend with +@step("an SSE re-fetch is triggered shortly before the next poll tick") +def _an_sse_refetch_is_triggered_shortly_before_the_next_poll_tick( + scenario: Scenario, +) -> None: + _sleep_until(_compute_next_tick_at(scenario) - SHORTLY_BEFORE) + before = scenario.mock.count_cdn_requests() + _trigger_sse_refetch(scenario) + # the re-fetch runs on a poller thread, so wait for its request to reach the CDN; otherwise it + # lands after the next step takes its baseline and is miscounted as the contended tick's + _wait_for(SHORTLY_BEFORE, lambda: scenario.mock.count_cdn_requests() > before) # baseline is taken before sleeping, so the next assertion counts what these intervals produced @@ -305,6 +440,118 @@ def _the_provider_state_is(scenario: Scenario, state: str) -> None: ) +@step( + parsers.re( + r'a PROVIDER_CONFIGURATION_CHANGED event is emitted with changed flag "(?P[^"]+)"' + ) +) +def _a_configuration_changed_event_is_emitted_with_changed_flag( + scenario: Scenario, flag: str +) -> None: + details = _consume_event(scenario, ProviderEvent.PROVIDER_CONFIGURATION_CHANGED) + changed = details.flags_changed + assert changed is not None, ( + "PROVIDER_CONFIGURATION_CHANGED carried no changed flags" + ) + assert flag in changed, ( + f"PROVIDER_CONFIGURATION_CHANGED changed flags {changed} do not include {flag}" + ) + + +@step(parsers.re(r"a (?PPROVIDER_[A-Z_]+) event is emitted")) +def _a_provider_event_is_emitted(scenario: Scenario, event_name: str) -> None: + _consume_event(scenario, ProviderEvent[event_name]) + + +@step(parsers.re(r"no (?PPROVIDER_[A-Z_]+) event is emitted")) +def _no_provider_event_is_emitted(scenario: Scenario, event_name: str) -> None: + unwanted = ProviderEvent[event_name] + + def none_seen() -> bool: + with scenario.events_lock: + return all(e != unwanted for e, _ in scenario.events) + + # hold it: an event the earlier steps caused may still be on its way to the client + _assert_holds_for( + SETTLE, none_seen, lambda: f"provider unexpectedly emitted {event_name}" + ) + + +@step(parsers.re(r'the provider has an SSE connection to "(?P[^"]+)"')) +def _the_provider_has_an_sse_connection_to(scenario: Scenario, path: str) -> None: + # the backend serves a single stream path, so a live subscriber is a stream to that path + assert path == SSE_PATH, f"the mock server only serves an SSE stream at {SSE_PATH}" + assert _poll_until( + SSE_CONNECT_TIMEOUT, lambda: scenario.mock.sse_client_count() == 1 + ), f"provider did not open an SSE connection to {path}" + + +@step("the provider opens no SSE connection") +def _the_provider_opens_no_sse_connection(scenario: Scenario) -> None: + # hold the assertion: a connection attempt in flight must not be mistaken for none at all + _assert_holds_for( + SETTLE, + lambda: scenario.mock.sse_client_count() == 0, + lambda: "provider unexpectedly opened an SSE connection", + ) + + +@step("the provider closes the SSE connection") +def _the_provider_closes_the_sse_connection(scenario: Scenario) -> None: + assert _poll_until(SETTLE, lambda: scenario.mock.sse_client_count() == 0), ( + "provider did not close the SSE connection" + ) + + +@step("a message after shutdown triggers no CDN request") +def _a_message_after_shutdown_triggers_no_cdn_request(scenario: Scenario) -> None: + baseline = scenario.mock.count_cdn_requests() + scenario.mock.emit_sse(REFETCH_MESSAGE) + time.sleep(STABLE_WINDOW) + assert scenario.mock.count_cdn_requests() == baseline, ( + "a shut-down provider still re-fetched on an SSE message" + ) + + +# the cadence itself isn't observable black-box, so it is asserted by its consequence: with SSE +# connected no fetch may arrive for longer than the aggressive cadence would allow +@step("the active poll interval is the SSE_CONNECTED_POLL_INTERVAL") +def _the_active_poll_interval_is_the_sse_connected_interval( + scenario: Scenario, +) -> None: + assert POLL_TIMEOUT < SSE_CONNECTED_POLL_INTERVAL, ( + f"observation window must be shorter than the {SSE_CONNECTED_POLL_INTERVAL}s interval under test" + ) + # the connect catch-up fetch must be in before the baseline, or it lands mid-observation + _await_sse_settled(scenario) + baseline = scenario.mock.count_cdn_requests() + time.sleep(POLL_TIMEOUT) + assert scenario.mock.count_cdn_requests() == baseline, ( + f"provider kept fetching within {POLL_TIMEOUT}s, so polling did not relax to the " + f"{SSE_CONNECTED_POLL_INTERVAL}s interval" + ) + + +# asserted by its consequence too: a fetch must arrive within the aggressive cadence +@step("the active poll interval is the SSE_DISCONNECTED_POLL_INTERVAL") +def _the_active_poll_interval_is_the_sse_disconnected_interval( + scenario: Scenario, +) -> None: + _await_cdn_request_within(scenario, SSE_DISCONNECTED_POLL_INTERVAL) + + +@step( + parsers.re(r"the CDN received (?P\d+) requests"), + converters={"count": int}, +) +def _the_cdn_received_requests(scenario: Scenario, count: int) -> None: + # a fetch the provider was told to make (SSE connect, SSE message, poll) is in flight by the + # time this step runs, so wait for the expected count before insisting on it + _poll_until(SETTLE, lambda: scenario.mock.count_cdn_requests() >= count) + seen = scenario.mock.count_cdn_requests() + assert seen == count, f"expected {count} CDN requests, saw {seen}" + + @step(parsers.re(r'the initial CDN request has no "(?P
[^"]+)" header')) def _the_initial_cdn_request_has_no_header(scenario: Scenario, header: str) -> None: assert not _get_nth_cdn_request(scenario, 0).has_header(header), ( @@ -314,7 +561,8 @@ def _the_initial_cdn_request_has_no_header(scenario: Scenario, header: str) -> N @step( parsers.re( - r'the (?P\d+)\. CDN request has the "(?P
[^"]+)" header "(?P[^"]*)"' + r'the (?P\d+)\. CDN request has the "(?P
[^"]+)" header ' + + _HEADER_VALUE ), converters={"index": int}, ) @@ -325,7 +573,7 @@ def _the_cdn_request_has_the_header( assert request.has_header(header), ( f"{index}. CDN request missing expected header: {header}" ) - assert request.get_header(header) == value, ( + assert request.get_header(header) == _unquote(value), ( f"{index}. CDN request {header} header mismatch" ) @@ -454,6 +702,10 @@ def _register_provider_with_sdk(scenario: Scenario) -> None: config_origin=scenario.mock.origin, metrics_origin=scenario.mock.origin, metrics_flush_interval_s=METRICS_FLUSH_INTERVAL, + # the harness's shortened timings, so SSE scenarios finish in reasonable time + sse_connected_poll_interval_s=SSE_CONNECTED_POLL_INTERVAL, + grace_period_s=GRACE_PERIOD, + disable_sse=not scenario.sse_enabled, ) api.set_provider(scenario.provider, scenario.domain) # set_provider runs initialize inline and reports failure as PROVIDER_ERROR rather than raising @@ -462,6 +714,89 @@ def _register_provider_with_sdk(scenario: Scenario) -> None: scenario.init_error = f"provider is {status} after initialization" +def _event_recorder( + scenario: Scenario, event: ProviderEvent +) -> Callable[[EventDetails], None]: + """Return a client handler that appends ``event`` to the scenario's event log.""" + + def record(details: EventDetails) -> None: + with scenario.events_lock: + scenario.events.append((event, details)) + + return record + + +def _consume_event(scenario: Scenario, expected: ProviderEvent) -> EventDetails: + """Return the next recorded event of type ``expected`` and advance the cursor past it. + + The cursor makes repeated steps (READY at startup, READY again after + recovery) consume distinct events. + """ + deadline = time.time() + EVENT_TIMEOUT + while True: + with scenario.events_lock: + for i in range(scenario.event_cursor, len(scenario.events)): + event, details = scenario.events[i] + if event == expected: + scenario.event_cursor = i + 1 + return details + seen = [e.name for e, _ in scenario.events[scenario.event_cursor :]] + if time.time() >= deadline: + pytest.fail( + f"no {expected.name} event emitted within {EVENT_TIMEOUT}s; saw {seen}" + ) + time.sleep(WAIT_POLL_STEP) + + +def _await_sse_settled(scenario: Scenario) -> None: + """Wait for the SSE stream and the catch-up fetch its connect pulls.""" + if not scenario.sse_enabled: + return + assert _poll_until( + SSE_CONNECT_TIMEOUT, lambda: scenario.mock.sse_client_count() >= 1 + ), "provider did not open its SSE stream" + # fetch 1 comes from initialize, fetch 2 from the successful connect + assert _poll_until(SETTLE, lambda: scenario.mock.count_cdn_requests() >= 2), ( + "SSE connected but pulled no catch-up fetch" + ) + + +def _await_cdn_request_within(scenario: Scenario, interval: float) -> None: + """Wait for a CDN request to arrive within ``interval`` seconds.""" + baseline = scenario.mock.count_cdn_requests() + _wait_for(interval + 1, lambda: scenario.mock.count_cdn_requests() > baseline) + + +def _advertising_sse_url(scenario: Scenario, config_json: str, url: str) -> str: + """Rewrite a flag configuration fixture to advertise a direct SSE url.""" + resolved = url.replace( + MOCK_SERVER_PLACEHOLDER, urlsplit(scenario.mock.origin).netloc + ) + config = json.loads(config_json) + config["eventStreams"] = [{"type": "sse", "url": resolved}] + return json.dumps(config) + + +def _trigger_sse_refetch(scenario: Scenario) -> None: + """Request an SSE-triggered re-fetch as the SSE manager would, without a live stream. + + Connecting a real one would switch the provider onto the SSE-connected + cadence, changing the timing the single-flight scenario exists to measure. + """ + assert scenario.provider is not None, "provider has not been initialized" + poller = scenario.provider._poller + assert poller is not None, "provider has no polling manager" + poller.trigger_immediate_fetch(None) + + +def _unquote(value: str) -> str: + """Strip one matching pair of surrounding single or double quotes, if present.""" + trimmed = value.strip() + if len(trimmed) >= 2 and trimmed[0] == trimmed[-1] and trimmed[0] in "'\"": + return trimmed[1:-1] + return trimmed + + def _reset_count_baseline(scenario: Scenario) -> None: """Start counting further CDN requests from now.""" scenario.count_baseline = scenario.mock.count_cdn_requests() @@ -509,14 +844,19 @@ def _load_flag_config(name: str) -> str: return (FLAGS_DIR / f"{name}.json").read_text() -def _wait_for(timeout: float, condition: Callable[[], bool]) -> None: - """Poll until condition holds, failing once timeout passes.""" +def _poll_until(timeout: float, condition: Callable[[], bool]) -> bool: + """Poll until condition holds or timeout passes; report whether it held.""" deadline = time.time() + timeout while time.time() < deadline: if condition(): - return + return True time.sleep(WAIT_POLL_STEP) - assert condition(), f"condition not met within {timeout}s" + return condition() + + +def _wait_for(timeout: float, condition: Callable[[], bool]) -> None: + """Poll until condition holds, failing once timeout passes.""" + assert _poll_until(timeout, condition), f"condition not met within {timeout}s" def _assert_holds_for( diff --git a/tests/sse_server.py b/tests/sse_server.py new file mode 100644 index 0000000..a0e8438 --- /dev/null +++ b/tests/sse_server.py @@ -0,0 +1,168 @@ +"""A real-socket SSE endpoint for unit tests. + +``responses`` patches ``requests`` and cannot stream ``text/event-stream``, and +the SSE client talks to ``urllib3`` directly anyway, so SSE tests run against a +real socket: a ThreadingHTTPServer serving one stream path. +""" + +from __future__ import annotations + +import queue +import select +import socket +import threading +import time +from collections.abc import Callable +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +SSE_PATH = "/sse" +# how often an idle stream handler checks whether the client went away +_IDLE_CHECK_S = 0.05 + + +class _SseHandler(BaseHTTPRequestHandler): + # chunked, like a real SSE server: urllib3 hands a chunked body over chunk by chunk, but blocks on a + # fixed-size read of an unframed one, so events would sit in its buffer + protocol_version = "HTTP/1.1" + server: _OwnedServer + + def do_GET(self) -> None: # BaseHTTPRequestHandler dispatch name + owner = self.server.owner + owner.record_path(self.path) + if self.path.split("?", 1)[0] != SSE_PATH: + self.send_error(404) + return + if owner.refuse: + self.send_error(503) + self.close_connection = True + return + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.send_header("Transfer-Encoding", "chunked") + self.send_header("Connection", "close") + self.end_headers() + self.wfile.flush() + events: queue.Queue[bytes | None] = queue.Queue() + owner.register(events) + try: + self._stream(events) + except OSError: + pass # client went away mid-write + finally: + owner.unregister(events) + + def _stream(self, events: queue.Queue[bytes | None]) -> None: + while True: + try: + item = events.get(timeout=_IDLE_CHECK_S) + except queue.Empty: + if self._check_client_closed(): + return + continue + if item is None: + # dropped by the test: end the body cleanly, then close + self.wfile.write(b"0\r\n\r\n") + self.wfile.flush() + self.close_connection = True + return + self.wfile.write(b"%x\r\n%s\r\n" % (len(item), item)) + self.wfile.flush() + + def _check_client_closed(self) -> bool: + readable, _, _ = select.select([self.connection], [], [], 0) + if not readable: + return False + try: + return self.connection.recv(1, socket.MSG_PEEK) == b"" + except OSError: + return True + + def log_message(self, format: str, *args: object) -> None: # noqa: A002 - signature of the overridden method + pass + + +class _OwnedServer(ThreadingHTTPServer): + daemon_threads = True + owner: SseServer + + +class SseServer: + """A real SSE endpoint the tests can push events through, drop, and refuse.""" + + def __init__(self) -> None: + self._httpd = _OwnedServer(("127.0.0.1", 0), _SseHandler) + self._httpd.owner = self + self._lock = threading.Lock() + self._clients: list[queue.Queue[bytes | None]] = [] + self.paths: list[str] = [] + #: answer new connections with 503 while set + self.refuse = False + self._thread = threading.Thread(target=self._httpd.serve_forever, daemon=True) + self._thread.start() + + @property + def origin(self) -> str: + host, port = self._httpd.server_address[:2] + return f"http://{host!s}:{port}" + + @property + def url(self) -> str: + return self.origin + SSE_PATH + + @property + def client_count(self) -> int: + with self._lock: + return len(self._clients) + + def record_path(self, path: str) -> None: + with self._lock: + self.paths.append(path) + + def register(self, events: queue.Queue[bytes | None]) -> None: + with self._lock: + self._clients.append(events) + + def unregister(self, events: queue.Queue[bytes | None]) -> None: + with self._lock: + if events in self._clients: + self._clients.remove(events) + + def emit(self, data: str, event: str | None = None) -> None: + """Send ``data`` as one SSE event to every connected client.""" + frame = "".join(f"data: {line}\n" for line in data.split("\n")) + if event is not None: + frame = f"event: {event}\n" + frame + self._broadcast((frame + "\n").encode()) + + def comment(self, text: str) -> None: + self._broadcast(f": {text}\n\n".encode()) + + def drop(self) -> None: + """End every open stream; clients may reconnect unless ``refuse`` is set.""" + with self._lock: + clients = list(self._clients) + for events in clients: + events.put(None) + + def _broadcast(self, frame: bytes) -> None: + with self._lock: + clients = list(self._clients) + for events in clients: + events.put(frame) + + def stop(self) -> None: + self.refuse = True + self.drop() + self._httpd.shutdown() + self._httpd.server_close() + + +def wait_until(condition: Callable[[], bool], timeout: float = 5.0) -> bool: + """Poll ``condition`` until it holds or ``timeout`` passes; report whether it held.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if condition(): + return True + time.sleep(0.02) + return condition() diff --git a/tests/test_cdn_config.py b/tests/test_cdn_config.py new file mode 100644 index 0000000..1b1ebe1 --- /dev/null +++ b/tests/test_cdn_config.py @@ -0,0 +1,118 @@ +import json + +import pytest + +from dynatrace.openfeature._cdn_config import extract_sse_url + +_ORIGIN = "https://cdn.us.fm.dynatrace.com" + + +def _config(*streams: object) -> str: + return json.dumps({"flags": {}, "eventStreams": list(streams)}) + + +class TestExtractSseUrl: + @pytest.mark.parametrize( + ("config", "expected"), + [ + ( + _config({"type": "sse", "url": "https://sse.example/s?k=1"}), + "https://sse.example/s?k=1", + ), + ( + _config({"type": "sse", "endpoint": {"requestUri": "/sse"}}), + _ORIGIN + "/sse", + ), + ( + _config( + { + "type": "sse", + "endpoint": { + "origin": "https://rt.example", + "requestUri": "/sse", + }, + } + ), + "https://rt.example/sse", + ), + # a non-blank url wins over the endpoint + ( + _config( + { + "type": "sse", + "url": "https://direct.example/sse", + "endpoint": {"requestUri": "/other"}, + } + ), + "https://direct.example/sse", + ), + # a blank url falls through to the endpoint, a blank origin to the fallback + ( + _config( + { + "type": "sse", + "url": " ", + "endpoint": {"origin": "", "requestUri": "/sse"}, + } + ), + _ORIGIN + "/sse", + ), + # entries of other types are skipped, unknown attributes ignored + ( + _config( + {"type": "websocket", "url": "wss://ws.example"}, + {"type": "sse", "url": "https://sse.example", "future": 1}, + ), + "https://sse.example", + ), + ], + ids=[ + "direct-url", + "endpoint-fallback-origin", + "endpoint-explicit-origin", + "url-wins", + "blank-url-and-origin", + "skips-other-types", + ], + ) + def test_resolves(self, config: str, expected: str) -> None: + assert extract_sse_url(config, _ORIGIN) == expected + + @pytest.mark.parametrize( + "config", + [ + None, + "", + "not-json", + "[]", + json.dumps({"flags": {}}), + json.dumps({"eventStreams": {"type": "sse"}}), + _config(), + _config("sse", None), + _config({"type": "websocket", "url": "wss://ws.example"}), + _config({"type": "sse"}), + _config({"type": "sse", "endpoint": {"origin": _ORIGIN}}), + # only the first sse entry counts, even when it is unusable + _config({"type": "sse"}, {"type": "sse", "url": "https://late.example"}), + ], + ids=[ + "none", + "empty", + "not-json", + "array", + "no-event-streams", + "event-streams-not-a-list", + "empty-list", + "non-object-entries", + "no-sse-entry", + "sse-without-url-or-endpoint", + "endpoint-without-request-uri", + "first-sse-entry-unusable", + ], + ) + def test_no_usable_entry(self, config: str | None) -> None: + assert extract_sse_url(config, _ORIGIN) is None + + def test_endpoint_without_any_fallback_sorigin(self) -> None: + config = _config({"type": "sse", "endpoint": {"requestUri": "/sse"}}) + assert extract_sse_url(config, None) is None diff --git a/tests/test_polling.py b/tests/test_polling.py index 3df11e4..f9e5834 100644 --- a/tests/test_polling.py +++ b/tests/test_polling.py @@ -245,3 +245,223 @@ def fetch(cancel=None) -> FetchResult: t0 = time.monotonic() pm.stop() assert time.monotonic() - t0 < 0.5, "stop() should return promptly" + + +_LONG_INTERVAL = 600.0 +_LM_HEADER = "Tue, 02 Jan 2024 00:00:00 GMT" +# _LM_HEADER as epoch seconds +_LM_EPOCH = 1704153600 + + +def _wait_until(condition, timeout: float = 1.0) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if condition(): + return True + time.sleep(0.01) + return condition() + + +class _GatedFetch: + """Fetch stub whose first call blocks until released; later calls walk ``results``.""" + + def __init__(self, results: list[FetchResult]) -> None: + self._results = results + self.calls = 0 + self.first_started = threading.Event() + self.release = threading.Event() + self._lock = threading.Lock() + + def __call__(self, cancel: threading.Event | None = None) -> FetchResult: + with self._lock: + self.calls += 1 + n = self.calls + if n == 1: + self.first_started.set() + self.release.wait(2.0) + return self._results[min(n, len(self._results)) - 1] + + +class TestPollingManagerSetInterval: + def test_set_interval_re_anchors_the_ticker(self) -> None: + fired = threading.Event() + pm, _ = _make_pm( + fetch_fn=lambda cancel=None: (_304, fired.set())[0], + interval_s=_LONG_INTERVAL, + ) + pm.start() + try: + pm.set_interval(_FAST_INTERVAL) + # the new cadence counts from now, not from the old 10-minute tick + assert fired.wait(timeout=1.0), "ticker did not adopt the new interval" + assert pm.interval_s == _FAST_INTERVAL + finally: + pm.stop() + + def test_slowing_down_postpones_the_next_tick(self) -> None: + calls: list[int] = [] + pm, _ = _make_pm( + fetch_fn=lambda cancel=None: (_304, calls.append(1))[0], + interval_s=_FAST_INTERVAL, + ) + pm.start() + try: + pm.set_interval(_LONG_INTERVAL) + time.sleep(_FAST_INTERVAL * 4) + assert calls == [] + finally: + pm.stop() + + def test_unchanged_interval_does_not_re_anchor(self) -> None: + pm, _ = _make_pm(interval_s=_LONG_INTERVAL) + pm.set_interval(_LONG_INTERVAL) + assert pm._rescheduled is False + + +class TestPollingManagerSseRefetch: + def test_free_slot_fetches_immediately(self) -> None: + fetch = _GatedFetch([_304]) + fetch.release.set() + pm, calls = _make_pm(fetch_fn=fetch, interval_s=_LONG_INTERVAL) + pm.start() + try: + pm.trigger_immediate_fetch(_LM_EPOCH) + assert _wait_until(lambda: calls["success"] == [True]) + assert fetch.calls == 1 + finally: + pm.stop() + + def test_contended_refetches_are_deferred_and_coalesced(self) -> None: + fetch = _GatedFetch([_304]) + pm, _ = _make_pm(fetch_fn=fetch, interval_s=_LONG_INTERVAL) + pm.start() + try: + pm.trigger_immediate_fetch(None) + assert fetch.first_started.wait(1.0) + pm.trigger_immediate_fetch(_LM_EPOCH) + pm.trigger_immediate_fetch(_LM_EPOCH + 60) + assert fetch.calls == 1, "a contended re-fetch must not overlap" + fetch.release.set() + assert _wait_until(lambda: fetch.calls == 2) + # ensure the fetches were coalesced and not both are executed + time.sleep(0.1) + assert fetch.calls == 2, "deferred re-fetches must coalesce into one" + finally: + pm.stop() + + def test_deferred_refetch_dropped_when_completed_fetch_is_as_new(self) -> None: + fresh = FetchResult(200, '{"flags":{}}', None, _LM_HEADER) + fetch = _GatedFetch([fresh, _304]) + pm, _ = _make_pm(fetch_fn=fetch, interval_s=_LONG_INTERVAL) + pm.start() + try: + pm.trigger_immediate_fetch(None) + assert fetch.first_started.wait(1.0) + pm.trigger_immediate_fetch(_LM_EPOCH) + fetch.release.set() + time.sleep(0.2) + assert fetch.calls == 1 + finally: + pm.stop() + + @pytest.mark.parametrize( + ("completed_last_modified", "announced"), + [ + ("Mon, 01 Jan 2024 00:00:00 GMT", _LM_EPOCH), + (None, _LM_EPOCH), + # an announcement without lastModified is never dropped + (_LM_HEADER, None), + ], + ids=["completed-older", "completed-unknown", "announced-unknown"], + ) + def test_deferred_refetch_issued_( + self, completed_last_modified: str | None, announced: int | None + ) -> None: + completed = FetchResult(200, '{"flags":{}}', None, completed_last_modified) + fetch = _GatedFetch([completed, _304]) + pm, _ = _make_pm(fetch_fn=fetch, interval_s=_LONG_INTERVAL) + pm.start() + try: + pm.trigger_immediate_fetch(None) + assert fetch.first_started.wait(1.0) + pm.trigger_immediate_fetch(announced) + fetch.release.set() + assert _wait_until(lambda: fetch.calls == 2) + finally: + pm.stop() + + def test_deferred_refetch_issued_after_a_failed_fetch(self) -> None: + started = threading.Event() + gate = threading.Event() + count = [0] + + def fetch(cancel=None) -> FetchResult: + count[0] += 1 + if count[0] == 1: + started.set() + gate.wait(2.0) + raise OSError("network") + return _304 + + pm, calls = _make_pm(fetch_fn=fetch, interval_s=_LONG_INTERVAL) + pm.start() + try: + pm.trigger_immediate_fetch(_LM_EPOCH) + assert started.wait(1.0) + pm.trigger_immediate_fetch(_LM_EPOCH) + gate.set() + assert _wait_until(lambda: calls["success"] == [True]) + assert calls["failure"] == ["OSError"] + finally: + pm.stop() + + def test_skipped_refetch_stays_pending_until_a_fetch_reaches_the_cdn( + self, + ) -> None: + results = iter([_SKIPPED, _304, _304]) + seen: list[FetchResult] = [] + + def fetch(cancel=None) -> FetchResult: + r = next(results) + seen.append(r) + return r + + pm, _ = _make_pm(fetch_fn=fetch, interval_s=_LONG_INTERVAL) + pm.start() + try: + pm.trigger_immediate_fetch(_LM_EPOCH) + assert _wait_until(lambda: len(seen) == 1) + time.sleep(0.1) + # re-issuing a skipped re-fetch at once would only be skipped again and spin + assert len(seen) == 1 + assert pm._pending_sse_fetch_last_modified == _LM_EPOCH + # the next fetch that makes a request issues the kept one after it + pm.trigger_immediate_fetch(None) + assert _wait_until(lambda: len(seen) == 3) + finally: + pm.stop() + + def test_contended_tick_neither_fetches_nor_emits(self) -> None: + fetch = _GatedFetch([_304]) + pm, calls = _make_pm(fetch_fn=fetch, interval_s=_FAST_INTERVAL) + pm.start() + try: + assert fetch.first_started.wait(1.0) + pm.trigger_immediate_fetch(_LM_EPOCH) + # ticks keep firing while the first fetch is blocked + time.sleep(_FAST_INTERVAL * 4) + assert fetch.calls == 1 + assert calls["success"] == [] + finally: + fetch.release.set() + pm.stop() + + def test_trigger_after_stop_is_ignored(self) -> None: + fetch = _GatedFetch([_304]) + fetch.release.set() + pm, _ = _make_pm(fetch_fn=fetch, interval_s=_LONG_INTERVAL) + pm.start() + pm.stop() + pm.trigger_immediate_fetch(_LM_EPOCH) + time.sleep(0.1) + assert fetch.calls == 0 diff --git a/tests/test_provider_sse.py b/tests/test_provider_sse.py new file mode 100644 index 0000000..baa4949 --- /dev/null +++ b/tests/test_provider_sse.py @@ -0,0 +1,367 @@ +import json +import threading +from pathlib import Path + +import pytest +import responses as rsps_lib +from openfeature.exception import GeneralError +from openfeature.provider import ProviderStatus +from sse_server import SseServer, wait_until + +from dynatrace.openfeature import DynatraceProvider +from dynatrace.openfeature._cdn import CdnFetcher, FetchResult +from dynatrace.openfeature._sse_message import SseMessage +from dynatrace.openfeature.options import DynatraceProviderOptions + +_KEY = "dt01.server_us_abcdef1234.de848e97a9cc4cc78aae568e65f49a9d_a1b2c3d4e5" +_CDN_URL = "https://cdn.us.fm.dynatrace.com/server/" + _KEY + ".json" +_FIXTURE = Path(__file__).parent / "fixtures" / "flag-config.json" + +_CONNECTED_S = 600.0 +_DISCONNECTED_S = 300.0 + + +def _flag_config(sse_url: str | None = None) -> str: + config = json.loads(_FIXTURE.read_text()) + if sse_url is not None: + config["eventStreams"] = [{"type": "sse", "url": sse_url}] + return json.dumps(config) + + +def _make_provider( + *, poll_s: float = _DISCONNECTED_S, grace_s: float = 60.0, disable_sse: bool = False +) -> DynatraceProvider: + return DynatraceProvider( + _KEY, + fetcher=CdnFetcher(DynatraceProviderOptions.from_sdk_key(_KEY)), + poll_interval_s=poll_s, + sse_connected_poll_interval_s=_CONNECTED_S, + disable_sse=disable_sse, + grace_period_s=grace_s, + ) + + +def _record_events(p: DynatraceProvider) -> list[str]: + events: list[str] = [] + for name in ("ready", "stale", "error"): + original = getattr(p, f"emit_provider_{name}") + + def emit(details: object, _name: str = name, _original=original) -> None: + _original(details) + events.append(_name) + + setattr(p, f"emit_provider_{name}", emit) + return events + + +def _cdn_calls() -> int: + return len(rsps_lib.calls) + + +class _FakeSse: + """Stands in for an SseManager when a test drives the callbacks directly.""" + + def __init__(self, log: list[str] | None = None) -> None: + self.stopped = False + self._log = log + + def stop(self) -> None: + self.stopped = True + if self._log is not None: + self._log.append("sse") + + +def _ready_provider_with_fake_sse( + **kwargs: object, +) -> tuple[DynatraceProvider, _FakeSse]: + """An initialized, READY provider whose SSE manager is a fake the test drives.""" + rsps_lib.add(rsps_lib.GET, _CDN_URL, body=_flag_config(), status=200) + provider = _make_provider(**kwargs) # type: ignore[arg-type] + provider.initialize() + fake = _FakeSse() + with provider._lock: + provider._sse_manager = fake # type: ignore[assignment] + return provider, fake + + +class TestSseStartup: + @rsps_lib.activate + def test_advertised_stream_connects_and_pulls_a_catch_up_fetch( + self, sse_server: SseServer + ) -> None: + rsps_lib.add(rsps_lib.GET, _CDN_URL, body=_flag_config(sse_server.url), status=200) + provider = _make_provider() + provider.initialize() + try: + assert wait_until(lambda: sse_server.client_count == 1) + # first on init, second on sse connection + assert wait_until(lambda: _cdn_calls() == 2), "no catch-up fetch on connect" + assert provider._sse_connected + assert provider._poller is not None + assert provider._poller.interval_s == _CONNECTED_S + finally: + provider.shutdown() + + @rsps_lib.activate + def test_structured_endpoint_falls_back_to_the_cdn_origin(self) -> None: + config = json.loads(_FIXTURE.read_text()) + config["eventStreams"] = [{"type": "sse", "endpoint": {"requestUri": "/sse"}}] + rsps_lib.add(rsps_lib.GET, _CDN_URL, body=json.dumps(config), status=200) + provider = _make_provider() + provider.initialize() + try: + assert provider._sse_manager is not None + assert provider._sse_manager._sse_url == "https://cdn.us.fm.dynatrace.com/sse" + finally: + provider.shutdown() + + @rsps_lib.activate + def test_no_advertised_stream_runs_polling_only(self) -> None: + rsps_lib.add(rsps_lib.GET, _CDN_URL, body=_flag_config(), status=200) + provider = _make_provider() + provider.initialize() + try: + assert provider._sse_manager is None + assert provider._poller is not None + assert provider._poller.interval_s == _DISCONNECTED_S + finally: + provider.shutdown() + + @rsps_lib.activate + def test_disable_sse_ignores_an_advertised_stream( + self, sse_server: SseServer + ) -> None: + rsps_lib.add(rsps_lib.GET, _CDN_URL, body=_flag_config(sse_server.url), status=200) + provider = _make_provider(disable_sse=True) + provider.initialize() + try: + assert provider._sse_manager is None + assert not wait_until(lambda: sse_server.client_count > 0, timeout=0.5) + finally: + provider.shutdown() + + @rsps_lib.activate + def test_unreachable_stream_stays_ready_and_polls_fast(self) -> None: + rsps_lib.add( + rsps_lib.GET, _CDN_URL, body=_flag_config("http://127.0.0.1:1/sse"), status=200 + ) + provider = _make_provider() + provider.initialize() + try: + assert provider._sse_manager is not None + # faults before a first connect are not disconnects + assert not wait_until(lambda: provider._state != ProviderStatus.READY, timeout=0.5) + assert not provider._sse_connected + assert provider._poller is not None + assert provider._poller.interval_s == _DISCONNECTED_S + finally: + provider.shutdown() + + @rsps_lib.activate + def test_failed_init_starts_sse_from_the_first_stored_config( + self, sse_server: SseServer + ) -> None: + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=500) + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=500) + rsps_lib.add(rsps_lib.GET, _CDN_URL, body=_flag_config(sse_server.url), status=200) + provider = _make_provider(poll_s=0.05) + with pytest.raises(GeneralError): + provider.initialize() + try: + assert provider._sse_manager is None + assert wait_until(lambda: sse_server.client_count == 1) + assert wait_until(lambda: provider._state == ProviderStatus.READY) + finally: + provider.shutdown() + + def test_superseded_init_opens_no_stream(self, sse_server: SseServer) -> None: + fetcher = _BlockingFetcher(_flag_config(sse_server.url)) + provider = DynatraceProvider(_KEY, fetcher=fetcher, poll_interval_s=600.0) + init_thread = threading.Thread(target=provider.initialize) + init_thread.start() + assert fetcher.fetch_started.wait(timeout=5.0) + provider.shutdown() + fetcher.release.set() + init_thread.join(timeout=5.0) + assert provider._sse_manager is None + assert not wait_until(lambda: sse_server.client_count > 0, timeout=0.5) + + +class _BlockingFetcher: + def __init__(self, body: str) -> None: + self._body = body + self.fetch_started = threading.Event() + self.release = threading.Event() + + def fetch(self, cancel: threading.Event | None = None) -> FetchResult: + self.fetch_started.set() + self.release.wait(timeout=5.0) + return FetchResult(200, self._body, None, None) + + def invalidate_cache(self) -> None: + pass + + +class TestSseConnectionLifecycle: + @rsps_lib.activate + def test_connect_switches_cadence_and_pulls_catch_up(self) -> None: + provider, fake = _ready_provider_with_fake_sse() + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) + try: + provider._on_sse_connected(fake) # type: ignore[arg-type] + assert provider._poller is not None + assert provider._poller.interval_s == _CONNECTED_S + assert wait_until(lambda: _cdn_calls() == 2) + finally: + provider.shutdown() + + @rsps_lib.activate + def test_disconnect_falls_back_to_polling_without_a_state_change(self) -> None: + provider, fake = _ready_provider_with_fake_sse() + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) + events = _record_events(provider) + try: + provider._on_sse_connected(fake) # type: ignore[arg-type] + assert wait_until(lambda: _cdn_calls() == 2) + assert provider._poller.interval_s == _CONNECTED_S + provider._on_sse_disconnected(fake) # type: ignore[arg-type] + provider._on_sse_disconnected(fake) # type: ignore[arg-type] # repeat fault while retrying + assert not provider._sse_connected + assert provider._poller is not None + assert provider._poller.interval_s == _DISCONNECTED_S + # SSE only prompts fetches: losing it changes the cadence, nothing else + assert provider._state == ProviderStatus.READY + assert provider._grace_timer is None + assert not wait_until(lambda: _cdn_calls() > 2, timeout=0.5) + assert events == [] + finally: + provider.shutdown() + + @rsps_lib.activate + def test_reconnect_returns_to_sse_mode_and_pulls_catch_up(self) -> None: + provider, fake = _ready_provider_with_fake_sse() + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) + events = _record_events(provider) + try: + provider._on_sse_connected(fake) # type: ignore[arg-type] + assert wait_until(lambda: _cdn_calls() == 2) + provider._on_sse_disconnected(fake) # type: ignore[arg-type] + provider._on_sse_connected(fake) # type: ignore[arg-type] + assert wait_until(lambda: _cdn_calls() == 3), ( + "no catch-up after a reconnect" + ) + assert provider._sse_connected + assert provider._poller is not None + assert provider._poller.interval_s == _CONNECTED_S + assert events == [] + finally: + provider.shutdown() + + @rsps_lib.activate + def test_stale_provider_recovers_only_through_a_successful_fetch(self) -> None: + provider, fake = _ready_provider_with_fake_sse() + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) + events = _record_events(provider) + try: + with provider._lock: + provider._sse_connected = True + provider._on_fetch_failure("HTTP 500") + # a disconnect neither worsens nor resolves staleness + provider._on_sse_disconnected(fake) # type: ignore[arg-type] + assert provider._state == ProviderStatus.STALE + # the reconnect catch-up is an ordinary fetch; its success is what recovers READY + provider._on_sse_connected(fake) # type: ignore[arg-type] + assert wait_until(lambda: provider._state == ProviderStatus.READY) + assert events == ["stale", "ready"] + finally: + provider.shutdown() + + @rsps_lib.activate + def test_callbacks_from_a_replaced_stream_are_ignored(self) -> None: + provider, _ = _ready_provider_with_fake_sse() + stale_stream = _FakeSse() + try: + provider._on_sse_connected(stale_stream) # type: ignore[arg-type] + assert not provider._sse_connected + provider._on_sse_message(stale_stream, SseMessage(type="refetchConfig")) # type: ignore[arg-type] + assert not wait_until(lambda: _cdn_calls() > 1, timeout=0.2) + finally: + provider.shutdown() + + +class TestPollFailureWithSse: + @pytest.mark.parametrize("sse_connected", [True, False], ids=["sse-up", "sse-down"]) + @rsps_lib.activate + def test_poll_failure_goes_stale_regardless_of_sse_state( + self, sse_connected: bool + ) -> None: + provider, _ = _ready_provider_with_fake_sse() + try: + with provider._lock: + provider._sse_connected = sse_connected + provider._on_fetch_failure("HTTP 500") + assert provider._state == ProviderStatus.STALE + finally: + provider.shutdown() + + +class TestSseMessages: + @rsps_lib.activate + def test_refetch_config_triggers_a_conditional_refetch(self) -> None: + rsps_lib.add( + rsps_lib.GET, _CDN_URL, body=_flag_config(), status=200, headers={"ETag": '"v1"'} + ) + provider = _make_provider() + provider.initialize() + fake = _FakeSse() + with provider._lock: + provider._sse_manager = fake # type: ignore[assignment] + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) + try: + provider._on_sse_message(fake, SseMessage(type="refetchConfig", last_modified=1)) # type: ignore[arg-type] + assert wait_until(lambda: _cdn_calls() == 2) + assert rsps_lib.calls[1].request.headers["If-None-Match"] == '"v1"' + finally: + provider.shutdown() + + @rsps_lib.activate + def test_other_types_are_ignored(self) -> None: + provider, fake = _ready_provider_with_fake_sse() + try: + provider._on_sse_message(fake, SseMessage(type="unknownType")) # type: ignore[arg-type] + assert not wait_until(lambda: _cdn_calls() > 1, timeout=0.5) + finally: + provider.shutdown() + + +class TestSseShutdown: + @rsps_lib.activate + def test_shutdown_closes_sse_before_polling(self) -> None: + order: list[str] = [] + provider, _ = _ready_provider_with_fake_sse() + fake = _FakeSse(order) + with provider._lock: + provider._sse_manager = fake # type: ignore[assignment] + assert provider._poller is not None + stop_poller = provider._poller.stop + + def recording_stop() -> None: + order.append("poller") + stop_poller() + + provider._poller.stop = recording_stop # type: ignore[method-assign] + provider.shutdown() + assert order == ["sse", "poller"] + assert provider._sse_manager is None + + @rsps_lib.activate + def test_shutdown_closes_the_live_stream(self, sse_server: SseServer) -> None: + rsps_lib.add(rsps_lib.GET, _CDN_URL, body=_flag_config(sse_server.url), status=200) + provider = _make_provider() + provider.initialize() + assert wait_until(lambda: sse_server.client_count == 1) + assert wait_until(lambda: _cdn_calls() == 2) + provider.shutdown() + sse_server.emit('{"type":"refetchConfig"}') + assert wait_until(lambda: sse_server.client_count == 0) + assert not wait_until(lambda: _cdn_calls() > 2, timeout=0.5) diff --git a/tests/test_sse.py b/tests/test_sse.py new file mode 100644 index 0000000..e31ce6e --- /dev/null +++ b/tests/test_sse.py @@ -0,0 +1,203 @@ +import logging +import threading +import time + +import pytest +from sse_server import SseServer, wait_until + +from dynatrace.openfeature._sse import SseManager, mask_url +from dynatrace.openfeature._sse_message import SseMessage + + +class _Recorder: + def __init__(self) -> None: + self.connected = 0 + self.disconnected = 0 + self.messages: list[SseMessage] = [] + self._lock = threading.Lock() + + def on_connected(self) -> None: + with self._lock: + self.connected += 1 + + def on_disconnected(self) -> None: + with self._lock: + self.disconnected += 1 + + def on_message(self, msg: SseMessage) -> None: + with self._lock: + self.messages.append(msg) + + +def _start(url: str, rec: _Recorder) -> SseManager: + manager = SseManager(url, rec.on_connected, rec.on_disconnected, rec.on_message) + manager.start() + return manager + + +class TestMaskUrl: + @pytest.mark.parametrize( + ("url", "masked"), + [ + ( + "https://sse.example/sse?key=secret", + "https://sse.example/sse?", + ), + ("https://sse.example/sse", "https://sse.example/sse"), + ], + ) + def test_masks_query_string(self, url: str, masked: str) -> None: + assert mask_url(url) == masked + + +class TestSseManager: + def test_connect_reports_connected(self, sse_server: SseServer) -> None: + rec = _Recorder() + manager = _start(sse_server.url, rec) + try: + assert wait_until(lambda: rec.connected == 1) + assert sse_server.client_count == 1 + finally: + manager.stop() + + def test_parsed_events_reach_on_message(self, sse_server: SseServer) -> None: + rec = _Recorder() + manager = _start(sse_server.url, rec) + try: + assert wait_until(lambda: sse_server.client_count == 1) + sse_server.emit('{"type":"refetchConfig","lastModified":1704153600}') + sse_server.emit('{"data":"{\\"type\\":\\"refetchConfig\\"}"}', event="msg") + assert wait_until(lambda: len(rec.messages) == 2) + assert rec.messages == [ + SseMessage(type="refetchConfig", last_modified=1704153600), + SseMessage(type="refetchConfig"), + ] + finally: + manager.stop() + + def test_unparseable_events_and_comments_are_dropped( + self, sse_server: SseServer + ) -> None: + rec = _Recorder() + manager = _start(sse_server.url, rec) + try: + assert wait_until(lambda: sse_server.client_count == 1) + sse_server.emit("not-json") + sse_server.comment("keep-alive") + sse_server.emit('{"type":"marker"}') + assert wait_until(lambda: len(rec.messages) == 1) + assert rec.messages == [SseMessage(type="marker")] + assert rec.disconnected == 0 + finally: + manager.stop() + + def test_drop_reports_disconnect_and_library_reconnects( + self, sse_server: SseServer + ) -> None: + rec = _Recorder() + manager = _start(sse_server.url, rec) + try: + assert wait_until(lambda: rec.connected == 1) + sse_server.drop() + assert wait_until(lambda: rec.disconnected >= 1) + # the reconnect itself is ld_eventsource's, after its initial backoff + assert wait_until(lambda: rec.connected == 2, timeout=5.0) + finally: + manager.stop() + + def test_unreachable_endpoint_reports_faults_but_never_connects(self) -> None: + rec = _Recorder() + # port 1 is never listening, so every attempt is refused + manager = _start("http://127.0.0.1:1/sse", rec) + try: + assert wait_until(lambda: rec.disconnected >= 1) + assert rec.connected == 0 + finally: + manager.stop() + + def test_refused_stream_is_retried(self, sse_server: SseServer) -> None: + rec = _Recorder() + sse_server.refuse = True + manager = _start(sse_server.url, rec) + try: + assert wait_until(lambda: rec.disconnected >= 1) + sse_server.refuse = False + assert wait_until(lambda: rec.connected == 1, timeout=5.0) + finally: + manager.stop() + + def test_stop_closes_the_stream_and_silences_callbacks( + self, sse_server: SseServer + ) -> None: + rec = _Recorder() + manager = _start(sse_server.url, rec) + assert wait_until(lambda: sse_server.client_count == 1) + manager.stop() + assert wait_until(lambda: sse_server.client_count == 0) + sse_server.emit('{"type":"refetchConfig"}') + assert not wait_until(lambda: rec.messages != [], timeout=0.5) + + def test_a_reconnect_racing_stop_is_closed(self, sse_server: SseServer) -> None: + rec = _Recorder() + sse_server.refuse = True + manager = _start(sse_server.url, rec) + # the first attempt is refused, so the reader sleeps out a backoff of up to a second + assert wait_until(lambda: rec.disconnected >= 1) + manager.stop() + sse_server.refuse = False + # the library reconnects once the sleep ends, despite the stop; that stream must close again + assert wait_until(lambda: len(sse_server.paths) >= 2, timeout=3.0) + assert wait_until(lambda: sse_server.client_count == 0) + assert rec.connected == 0 + + def test_stop_returns_promptly_while_the_reader_is_blocked( + self, sse_server: SseServer + ) -> None: + rec = _Recorder() + manager = _start(sse_server.url, rec) + assert wait_until(lambda: rec.connected == 1) + # the server sends nothing, so the reader stays in recv(); on Windows the library's close() + # would block on it, so stop() must not wait for the close + t0 = time.monotonic() + manager.stop() + assert time.monotonic() - t0 < 0.5, "stop() blocked on the SSE close" + sse_server.emit('{"type":"refetchConfig"}') + assert not wait_until(lambda: rec.messages != [], timeout=0.5) + + def test_a_failing_callback_does_not_kill_the_stream( + self, sse_server: SseServer + ) -> None: + received: list[SseMessage] = [] + + def on_message(msg: SseMessage) -> None: + received.append(msg) + if len(received) == 1: + raise RuntimeError("on_message failed") + + manager = SseManager(sse_server.url, lambda: None, lambda: None, on_message) + manager.start() + try: + assert wait_until(lambda: sse_server.client_count == 1) + sse_server.emit('{"type":"a"}') + sse_server.emit('{"type":"b"}') + assert wait_until(lambda: len(received) == 2) + finally: + manager.stop() + + def test_logs_never_carry_the_query_string( + self, sse_server: SseServer, caplog: pytest.LogCaptureFixture + ) -> None: + rec = _Recorder() + caplog.set_level(logging.DEBUG) + manager = _start(sse_server.url + "?key=secret", rec) + try: + assert wait_until(lambda: rec.connected == 1) + sse_server.drop() + assert wait_until(lambda: rec.disconnected >= 1) + finally: + manager.stop() + assert sse_server.paths[0] == "/sse?key=secret" + # urllib3's own DEBUG request log is out of our hands, as it is for CDN fetches + dt_logs = [r for r in caplog.records if r.name.startswith("dynatrace.")] + assert dt_logs + assert not any("secret" in r.getMessage() for r in dt_logs) diff --git a/tests/test_sse_message.py b/tests/test_sse_message.py new file mode 100644 index 0000000..b7e922d --- /dev/null +++ b/tests/test_sse_message.py @@ -0,0 +1,87 @@ +import pytest + +from dynatrace.openfeature._sse_message import SseMessage + + +class TestSseMessageParse: + @pytest.mark.parametrize( + ("payload", "expected"), + [ + ( + '{"type":"refetchConfig","etag":"\\"v2\\"","lastModified":1704153600}', + SseMessage(type="refetchConfig", etag='"v2"', last_modified=1704153600), + ), + ( + '{"data":"{\\"type\\":\\"refetchConfig\\",\\"lastModified\\":1704153600}"}', + SseMessage(type="refetchConfig", last_modified=1704153600), + ), + # unknown types parse; ignoring them is the caller's call + ('{"type":"unknownType"}', SseMessage(type="unknownType")), + # unknown fields, including metadata, are dropped + ( + '{"type":"refetchConfig","metadata":{"a":1},"extra":true}', + SseMessage(type="refetchConfig"), + ), + # a malformed lastModified reads as unknown rather than rejecting the message + ( + '{"type":"refetchConfig","lastModified":"soon"}', + SseMessage(type="refetchConfig"), + ), + ( + '{"type":"refetchConfig","lastModified":true}', + SseMessage(type="refetchConfig"), + ), + ], + ids=[ + "flat", + "ably-envelope", + "unknown-type", + "unknown-fields", + "bad-last-modified", + "bool-last-modified", + ], + ) + def test_parses(self, payload: str, expected: SseMessage) -> None: + assert SseMessage.parse(payload) == expected + + @pytest.mark.parametrize( + "payload", + [ + None, + "", + " ", + "not-json", + "{}", + "[]", + '"refetchConfig"', + '{"lastModified":1704153600}', + '{"type":""}', + '{"type":" "}', + '{"type":42}', + '{"data":"not-json"}', + '{"data":"[1]"}', + ], + ids=[ + "none", + "empty", + "blank", + "not-json", + "empty-object", + "array", + "string", + "missing-type", + "empty-type", + "blank-type", + "non-string-type", + "envelope-not-json", + "envelope-not-object", + ], + ) + def test_rejects(self, payload: str | None) -> None: + assert SseMessage.parse(payload) is None + + def test_non_string_data_is_read_as_a_flat_message(self) -> None: + # only a JSON-encoded string is an Ably envelope + assert SseMessage.parse('{"type":"refetchConfig","data":{"x":1}}') == ( + SseMessage(type="refetchConfig") + ) diff --git a/uv.lock b/uv.lock index 8b90168..1d54b7a 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 5 requires-python = ">=3.10" resolution-markers = [ "python_full_version >= '3.15'", @@ -284,6 +284,7 @@ name = "dynatrace-openfeature-provider-python" version = "0.2.0" source = { editable = "." } dependencies = [ + { name = "launchdarkly-eventsource" }, { name = "openfeature-flagd-core" }, { name = "openfeature-sdk" }, { name = "requests" }, @@ -306,6 +307,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "launchdarkly-eventsource", specifier = ">=1.7.3,<2.0.0" }, { name = "openfeature-flagd-core", specifier = ">=1.0.0,<2.0.0" }, { name = "openfeature-sdk", specifier = ">=0.8.2" }, { name = "requests", specifier = ">=2.27.0" }, @@ -331,7 +333,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -383,6 +385,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, ] +[[package]] +name = "launchdarkly-eventsource" +version = "1.7.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "urllib3" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/f5/b6/be13d4bff6ed3db625bcee9e4aef97496a014d271767be28816fc9502330/launchdarkly_eventsource-1.7.3.tar.gz", hash = "sha256:e47e32a4752e60795038deec463f63a6c6e4c20d944df155fa63036edff20450", size = 48482, upload-time = "2026-09-17T16:00:30.059Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/24/e0/68172b381d599a87675c58783387f12fa184a1da54f9f9de7de5cb08e42e/launchdarkly_eventsource-1.7.3-py3-none-any.whl", hash = "sha256:f83f6c8289d1f4b4945522cbd2b8630bd0673e6952eb2af5d32dfa0aa068ee8f", size = 29547, upload-time = "2026-09-17T16:00:29.108Z" }, +] + [[package]] name = "librt" version = "0.11.0"