From 556ca49273663491b55dc21fc71f9e792a278fd1 Mon Sep 17 00:00:00 2001 From: Lea Konvalinka Date: Thu, 22 Oct 2026 10:47:05 +0200 Subject: [PATCH 1/4] feat: introduce SSE (ref AI-984) --- .gitignore | 1 + README.md | 3 + pyproject.toml | 7 + src/dynatrace/openfeature/_cdn.py | 6 + src/dynatrace/openfeature/_cdn_config.py | 40 +++ src/dynatrace/openfeature/_polling.py | 245 ++++++++++---- src/dynatrace/openfeature/_sse.py | 202 +++++++++++ src/dynatrace/openfeature/_sse_message.py | 49 +++ src/dynatrace/openfeature/provider.py | 137 +++++++- tests/conftest.py | 12 + tests/e2e/conftest.py | 7 +- tests/e2e/mock_server.py | 40 ++- tests/e2e/{test_cdn_e2e.py => test_e2e.py} | 376 ++++++++++++++++++++- tests/sse_server.py | 168 +++++++++ tests/test_cdn_config.py | 118 +++++++ tests/test_polling.py | 219 ++++++++++++ tests/test_provider_sse.py | 365 ++++++++++++++++++++ tests/test_sse.py | 204 +++++++++++ tests/test_sse_message.py | 87 +++++ uv.lock | 18 +- 20 files changed, 2213 insertions(+), 91 deletions(-) create mode 100644 src/dynatrace/openfeature/_cdn_config.py create mode 100644 src/dynatrace/openfeature/_sse.py create mode 100644 src/dynatrace/openfeature/_sse_message.py create mode 100644 tests/conftest.py rename tests/e2e/{test_cdn_e2e.py => test_e2e.py} (54%) create mode 100644 tests/sse_server.py create mode 100644 tests/test_cdn_config.py create mode 100644 tests/test_provider_sse.py create mode 100644 tests/test_sse.py create mode 100644 tests/test_sse_message.py diff --git a/.gitignore b/.gitignore index e38959c..5eb0ae0 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,4 @@ coverage.xml .mypy_cache/ .ruff_cache/ .pytest_cache/ +.idea/ 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/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..ddb58b5 100644 --- a/src/dynatrace/openfeature/_cdn.py +++ b/src/dynatrace/openfeature/_cdn.py @@ -51,6 +51,12 @@ 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: + """``last_modified`` as Unix epoch seconds, or None when absent or unparseable.""" + 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..adc1d98 --- /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``, or None (ADR-0008). + 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..c8c1aea 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 + # announced epoch seconds of a deferred SSE re-fetch, or None when none is pending + self._pending_sse_refetch_at: float | None = None + # announcement the running fetch was issued for (None for a poll tick), kept so a skipped re-fetch can put it back + self._in_flight_sse_refetch_at: 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_refetch_at = None + # leave _in_flight alone: only the fetch thread clears it; clearing here would let a restart double up + 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_refetch_at = ( + announced + if self._pending_sse_refetch_at is None + else max(self._pending_sse_refetch_at, 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: + 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_refetch_at = 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_refetch_at + self._in_flight_sse_refetch_at = 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_refetch_at = ( + issued_for + if self._pending_sse_refetch_at is None + else max(self._pending_sse_refetch_at, issued_for) + ) + log.debug("SSE re-fetch made no request; keeping it pending") + return False + + announced = self._pending_sse_refetch_at + if announced is None or self._stop_event.is_set(): + return False + self._pending_sse_refetch_at = None + + observed = completed.last_modified_epoch_s if completed is not None else None + if ( + announced != _LAST_MODIFIED_UNKNOWN + and observed is not None + and observed >= announced + ): + log.debug( + "dropping deferred SSE re-fetch: completed fetch already returned " + "Last-Modified %s >= announced %s", + observed, + announced, + ) + return False + log.debug("issuing deferred SSE re-fetch") + self._in_flight_sse_refetch_at = announced + 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..f4b9414 --- /dev/null +++ b/src/dynatrace/openfeature/_sse.py @@ -0,0 +1,202 @@ +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 = 100 * 60.0 +# 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. + There a blocked recv() is not woken, but the stop flag keeps the reader inert.""" + 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 + # TODO + # no logger is handed to the client: it logs the full URL, credentials included + 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 54% rename from tests/e2e/test_cdn_e2e.py rename to tests/e2e/test_e2e.py index a07ecc7..4c5b15d 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,21 @@ 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 +# payload a shut-down provider must not react to +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 +112,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 +132,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 +148,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 +161,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,6 +173,23 @@ 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) + ), + ) + + @step( parsers.re( r'an initialized, READY provider serving the "(?P[^"]+)" flag configuration' @@ -140,6 +206,7 @@ def _an_initialized_ready_provider_serving_the_flag_configuration( _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) @@ -243,8 +310,90 @@ 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_TIMEOUT) + + +@step( + "polling triggers a configuration refetch after the SSE_DISCONNECTED_POLL_INTERVAL" +) +def _polling_triggers_a_configuration_refetch_after_the_sse_disconnected_poll_interval( + scenario: Scenario, +) -> None: + _await_cdn_request_within(scenario, SSE_DISCONNECTED_POLL_INTERVAL) + + +@step("polling triggers a configuration refetch after the SSE_CONNECTED_POLL_INTERVAL") +def _polling_triggers_a_configuration_refetch_after_the_sse_connected_poll_interval( + scenario: Scenario, +) -> None: + _await_cdn_request_within(scenario, SSE_CONNECTED_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" + + +# a sustained outage: connections stay refused until the stream is restored or reset +@step("the SSE stream drops and stays down") +def _the_sse_stream_drops_and_stays_down(scenario: Scenario) -> None: + scenario.mock.drop_sse() + # the provider switches cadence as soon as it sees the fault; the Then steps assert what it did + 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("\\\\", "\\")) + # essentially a sleep, cut short once the message has triggered a CDN request + _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 +454,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), ( @@ -454,6 +715,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 +727,94 @@ 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. + + So a scenario starting from a READY provider never races the stream it is + about to drop or message. No-op when SSE is off, as it is throughout the + CDN spec. + """ + 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, 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 +862,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..0bd5d1c --- /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._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 _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..c4156f5 --- /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_origin(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..e92167f 100644 --- a/tests/test_polling.py +++ b/tests/test_polling.py @@ -245,3 +245,222 @@ 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) + 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_unless_proven_satisfied( + 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(None) + 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_refetch_at == _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..88b258c --- /dev/null +++ b/tests/test_provider_sse.py @@ -0,0 +1,365 @@ +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 _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=_config(), status=200) + p = _make_provider(**kwargs) # type: ignore[arg-type] + p.initialize() + fake = _FakeSse() + with p._lock: + p._sse_manager = fake # type: ignore[assignment] + return p, 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=_config(sse_server.url), status=200) + p = _make_provider() + p.initialize() + try: + assert wait_until(lambda: sse_server.client_count == 1) + assert wait_until(lambda: _cdn_calls() == 2), "no catch-up fetch on connect" + assert p._sse_connected + assert p._poller is not None + assert p._poller.interval_s == _CONNECTED_S + finally: + p.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) + p = _make_provider() + p.initialize() + try: + assert p._sse_manager is not None + assert p._sse_manager._sse_url == "https://cdn.us.fm.dynatrace.com/sse" + finally: + p.shutdown() + + @rsps_lib.activate + def test_no_advertised_stream_runs_polling_only(self) -> None: + rsps_lib.add(rsps_lib.GET, _CDN_URL, body=_config(), status=200) + p = _make_provider() + p.initialize() + try: + assert p._sse_manager is None + assert p._poller is not None + assert p._poller.interval_s == _DISCONNECTED_S + finally: + p.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=_config(sse_server.url), status=200) + p = _make_provider(disable_sse=True) + p.initialize() + try: + assert p._sse_manager is None + assert not wait_until(lambda: sse_server.client_count > 0, timeout=0.3) + finally: + p.shutdown() + + @rsps_lib.activate + def test_unreachable_stream_stays_ready_and_polls_fast(self) -> None: + rsps_lib.add( + rsps_lib.GET, _CDN_URL, body=_config("http://127.0.0.1:1/sse"), status=200 + ) + p = _make_provider() + p.initialize() + try: + assert p._sse_manager is not None + # faults before a first connect are not disconnects + assert not wait_until(lambda: p._state != ProviderStatus.READY, timeout=0.5) + assert not p._sse_connected + assert p._poller is not None + assert p._poller.interval_s == _DISCONNECTED_S + finally: + p.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=_config(sse_server.url), status=200) + p = _make_provider(poll_s=0.05) + with pytest.raises(GeneralError): + p.initialize() + try: + assert p._sse_manager is None + assert wait_until(lambda: sse_server.client_count == 1) + assert wait_until(lambda: p._state == ProviderStatus.READY) + finally: + p.shutdown() + + def test_superseded_init_opens_no_stream(self, sse_server: SseServer) -> None: + fetcher = _BlockingFetcher(_config(sse_server.url)) + p = DynatraceProvider(_KEY, fetcher=fetcher, poll_interval_s=600.0) + init_thread = threading.Thread(target=p.initialize) + init_thread.start() + assert fetcher.fetch_started.wait(timeout=5.0) + p.shutdown() + fetcher.release.set() + init_thread.join(timeout=5.0) + assert p._sse_manager is None + assert not wait_until(lambda: sse_server.client_count > 0, timeout=0.3) + + +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: + p, fake = _ready_provider_with_fake_sse() + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) + try: + p._on_sse_connected(fake) # type: ignore[arg-type] + assert p._poller is not None + assert p._poller.interval_s == _CONNECTED_S + assert wait_until(lambda: _cdn_calls() == 2) + finally: + p.shutdown() + + @rsps_lib.activate + def test_disconnect_falls_back_to_polling_without_a_state_change(self) -> None: + p, fake = _ready_provider_with_fake_sse() + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) + events = _record_events(p) + try: + p._on_sse_connected(fake) # type: ignore[arg-type] + assert wait_until(lambda: _cdn_calls() == 2) + p._on_sse_disconnected(fake) # type: ignore[arg-type] + p._on_sse_disconnected(fake) # type: ignore[arg-type] # repeat fault while retrying + assert not p._sse_connected + assert p._poller is not None + assert p._poller.interval_s == _DISCONNECTED_S + # SSE only prompts fetches: losing it changes the cadence, nothing else + assert p._state == ProviderStatus.READY + assert p._grace_timer is None + assert not wait_until(lambda: _cdn_calls() > 2, timeout=0.2) + assert events == [] + finally: + p.shutdown() + + @rsps_lib.activate + def test_reconnect_returns_to_sse_mode_and_pulls_catch_up(self) -> None: + p, fake = _ready_provider_with_fake_sse() + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) + events = _record_events(p) + try: + p._on_sse_connected(fake) # type: ignore[arg-type] + assert wait_until(lambda: _cdn_calls() == 2) + p._on_sse_disconnected(fake) # type: ignore[arg-type] + p._on_sse_connected(fake) # type: ignore[arg-type] + assert wait_until(lambda: _cdn_calls() == 3), ( + "no catch-up after a reconnect" + ) + assert p._sse_connected + assert p._poller is not None + assert p._poller.interval_s == _CONNECTED_S + assert events == [] + finally: + p.shutdown() + + @rsps_lib.activate + def test_stale_provider_recovers_only_through_a_successful_fetch(self) -> None: + p, fake = _ready_provider_with_fake_sse() + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) + events = _record_events(p) + try: + with p._lock: + p._sse_connected = True + p._on_fetch_failure("HTTP 500") + # a disconnect neither worsens nor resolves staleness + p._on_sse_disconnected(fake) # type: ignore[arg-type] + assert p._state == ProviderStatus.STALE + # the reconnect catch-up is an ordinary fetch; its success is what recovers READY + p._on_sse_connected(fake) # type: ignore[arg-type] + assert wait_until(lambda: p._state == ProviderStatus.READY) + assert events == ["stale", "ready"] + finally: + p.shutdown() + + @rsps_lib.activate + def test_callbacks_from_a_replaced_stream_are_ignored(self) -> None: + p, _ = _ready_provider_with_fake_sse() + stale_stream = _FakeSse() + try: + p._on_sse_connected(stale_stream) # type: ignore[arg-type] + assert not p._sse_connected + p._on_sse_message(stale_stream, SseMessage(type="refetchConfig")) # type: ignore[arg-type] + assert not wait_until(lambda: _cdn_calls() > 1, timeout=0.2) + finally: + p.shutdown() + + +class TestPollFailureWithSse: + @pytest.mark.parametrize("sse_connected", [True, False], ids=["sse-up", "sse-down"]) + @rsps_lib.activate + def test_poll_failure_goes_stale_whatever_the_sse_state( + self, sse_connected: bool + ) -> None: + p, _ = _ready_provider_with_fake_sse() + try: + with p._lock: + p._sse_connected = sse_connected + p._on_fetch_failure("HTTP 500") + assert p._state == ProviderStatus.STALE + finally: + p.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=_config(), status=200, headers={"ETag": '"v1"'} + ) + p = _make_provider() + p.initialize() + fake = _FakeSse() + with p._lock: + p._sse_manager = fake # type: ignore[assignment] + rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) + try: + p._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: + p.shutdown() + + @rsps_lib.activate + def test_other_types_are_ignored(self) -> None: + p, fake = _ready_provider_with_fake_sse() + try: + p._on_sse_message(fake, SseMessage(type="unknownType")) # type: ignore[arg-type] + assert not wait_until(lambda: _cdn_calls() > 1, timeout=0.2) + finally: + p.shutdown() + + +class TestSseShutdown: + @rsps_lib.activate + def test_shutdown_closes_sse_before_polling(self) -> None: + order: list[str] = [] + p, _ = _ready_provider_with_fake_sse() + fake = _FakeSse(order) + with p._lock: + p._sse_manager = fake # type: ignore[assignment] + assert p._poller is not None + stop_poller = p._poller.stop + + def recording_stop() -> None: + order.append("poller") + stop_poller() + + p._poller.stop = recording_stop # type: ignore[method-assign] + p.shutdown() + assert order == ["sse", "poller"] + assert p._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=_config(sse_server.url), status=200) + p = _make_provider() + p.initialize() + assert wait_until(lambda: sse_server.client_count == 1) + assert wait_until(lambda: _cdn_calls() == 2) + p.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.3) diff --git a/tests/test_sse.py b/tests/test_sse.py new file mode 100644 index 0000000..9eb83c2 --- /dev/null +++ b/tests/test_sse.py @@ -0,0 +1,204 @@ +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() + # no event to wake the reader first: the socket itself must close, on Windows too + assert wait_until(lambda: sse_server.client_count == 0) + sse_server.emit('{"type":"refetchConfig"}') + assert not wait_until(lambda: rec.messages != [], timeout=0.3) + + 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.3) + + 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("boom") + + 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 + ours = [r for r in caplog.records if r.name.startswith("dynatrace.")] + assert ours + assert not any("secret" in r.getMessage() for r in ours) 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" From 94ce94948b32000e94c98593c91eaf1ce77f1461 Mon Sep 17 00:00:00 2001 From: Lea Konvalinka Date: Tue, 27 Oct 2026 12:08:25 +0100 Subject: [PATCH 2/4] small fixups --- fm-provider-test-harness | 2 +- src/dynatrace/openfeature/_cdn.py | 1 - src/dynatrace/openfeature/_cdn_config.py | 2 +- src/dynatrace/openfeature/_polling.py | 44 ++++++++++++------------ src/dynatrace/openfeature/_sse.py | 9 ++--- tests/e2e/test_e2e.py | 25 ++++++++++---- tests/test_polling.py | 2 +- 7 files changed, 48 insertions(+), 37 deletions(-) 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/src/dynatrace/openfeature/_cdn.py b/src/dynatrace/openfeature/_cdn.py index ddb58b5..9abc4bc 100644 --- a/src/dynatrace/openfeature/_cdn.py +++ b/src/dynatrace/openfeature/_cdn.py @@ -53,7 +53,6 @@ def is_success(self) -> bool: @property def last_modified_epoch_s(self) -> float | None: - """``last_modified`` as Unix epoch seconds, or None when absent or unparseable.""" parsed = _parse_http_date(self.last_modified) return parsed.timestamp() if parsed is not None else None diff --git a/src/dynatrace/openfeature/_cdn_config.py b/src/dynatrace/openfeature/_cdn_config.py index adc1d98..aa76be3 100644 --- a/src/dynatrace/openfeature/_cdn_config.py +++ b/src/dynatrace/openfeature/_cdn_config.py @@ -12,7 +12,7 @@ def _non_blank(value: typing.Any) -> str | 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``, or None (ADR-0008). + """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 diff --git a/src/dynatrace/openfeature/_polling.py b/src/dynatrace/openfeature/_polling.py index c8c1aea..977f2bd 100644 --- a/src/dynatrace/openfeature/_polling.py +++ b/src/dynatrace/openfeature/_polling.py @@ -43,10 +43,10 @@ def __init__( 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 - # announced epoch seconds of a deferred SSE re-fetch, or None when none is pending - self._pending_sse_refetch_at: float | None = None - # announcement the running fetch was issued for (None for a poll tick), kept so a skipped re-fetch can put it back - self._in_flight_sse_refetch_at: float | None = None + # 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: @@ -68,8 +68,7 @@ def stop(self) -> None: # stop event doubles as the fetch cancel signal, so an in-flight poll abandons its backoff self._stop_event.set() with self._cond: - self._pending_sse_refetch_at = None - # leave _in_flight alone: only the fetch thread clears it; clearing here would let a restart double up + 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) @@ -97,10 +96,10 @@ def trigger_immediate_fetch( ) with self._cond: if self._in_flight: - self._pending_sse_refetch_at = ( + self._pending_sse_fetch_last_modified = ( announced - if self._pending_sse_refetch_at is None - else max(self._pending_sse_refetch_at, 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)", @@ -114,6 +113,7 @@ def _ticker_loop(self) -> None: 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() @@ -137,7 +137,7 @@ def _on_tick_locked(self) -> None: self._submit_locked(None) def _submit_locked(self, sse_refetch_at: float | None) -> None: - self._in_flight_sse_refetch_at = sse_refetch_at + self._in_flight_sse_fetch_last_modified = sse_refetch_at self._in_flight = True threading.Thread( target=self._run_fetches, @@ -159,8 +159,8 @@ def _run_fetches(self) -> None: 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_refetch_at - self._in_flight_sse_refetch_at = None + 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 @@ -168,34 +168,34 @@ def _on_fetch_complete_locked(self, completed: FetchResult | None) -> bool: # 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_refetch_at = ( + self._pending_sse_fetch_last_modified = ( issued_for - if self._pending_sse_refetch_at is None - else max(self._pending_sse_refetch_at, 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 - announced = self._pending_sse_refetch_at - if announced is None or self._stop_event.is_set(): + pending = self._pending_sse_fetch_last_modified + if pending is None or self._stop_event.is_set(): return False - self._pending_sse_refetch_at = None + self._pending_sse_fetch_last_modified = None observed = completed.last_modified_epoch_s if completed is not None else None if ( - announced != _LAST_MODIFIED_UNKNOWN + pending != _LAST_MODIFIED_UNKNOWN and observed is not None - and observed >= announced + and observed >= pending ): log.debug( "dropping deferred SSE re-fetch: completed fetch already returned " "Last-Modified %s >= announced %s", observed, - announced, + pending, ) return False log.debug("issuing deferred SSE re-fetch") - self._in_flight_sse_refetch_at = announced + self._in_flight_sse_fetch_last_modified = pending self._in_flight = True return True diff --git a/src/dynatrace/openfeature/_sse.py b/src/dynatrace/openfeature/_sse.py index f4b9414..3b1b474 100644 --- a/src/dynatrace/openfeature/_sse.py +++ b/src/dynatrace/openfeature/_sse.py @@ -18,7 +18,7 @@ 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 = 100 * 60.0 +SSE_READ_TIMEOUT_S: float = 150 * 60.0 # 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 @@ -47,7 +47,10 @@ def urlopen( def shutdown_stream(self) -> None: """Shut the stream's socket down so the server sees it close, even on Windows. - There a blocked recv() is not woken, but the stop flag keeps the reader inert.""" + ld_eventsource's own close() only releases the connection the re, 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 @@ -94,8 +97,6 @@ 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 - # TODO - # no logger is handed to the client: it logs the full URL, credentials included self._client = SSEClient( connect=ConnectStrategy.http( self._sse_url, diff --git a/tests/e2e/test_e2e.py b/tests/e2e/test_e2e.py index 4c5b15d..c8c6ab7 100644 --- a/tests/e2e/test_e2e.py +++ b/tests/e2e/test_e2e.py @@ -190,10 +190,18 @@ def _the_cdn_serves_the_flag_configuration_advertising_an_sse_stream_at_url( ) +# a header value is double-quoted, or single-quoted when it holds double quotes itself (an ETag: '"v1"') +_HEADER_VALUE = r"(?P\"[^\"]*\"|'[^']*')" + + +def _unquote(value: str) -> str: + return value[1:-1] + + @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( @@ -201,7 +209,7 @@ 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) @@ -264,7 +272,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}, ) @@ -273,7 +281,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) + ), ) @@ -370,7 +380,7 @@ def _the_server_emits_an_sse_message(scenario: Scenario, payload: str) -> None: _poll_until(SETTLE, lambda: scenario.mock.count_cdn_requests() > baseline) -@step("the grace period expires") +@step("the GRACE_PERIOD expires") def _the_grace_period_expires(scenario: Scenario) -> None: _poll_until( GRACE_PERIOD + SETTLE, @@ -575,7 +585,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}, ) @@ -586,7 +597,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" ) diff --git a/tests/test_polling.py b/tests/test_polling.py index e92167f..7f804d8 100644 --- a/tests/test_polling.py +++ b/tests/test_polling.py @@ -433,7 +433,7 @@ def fetch(cancel=None) -> FetchResult: 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_refetch_at == _LM_EPOCH + 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) From a41859d1fbc285afa9daa1e4cd8bc83d87cd4e1f Mon Sep 17 00:00:00 2001 From: Lea Konvalinka Date: Tue, 27 Oct 2026 12:19:25 +0100 Subject: [PATCH 3/4] remove duplicate method def --- tests/e2e/test_e2e.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/tests/e2e/test_e2e.py b/tests/e2e/test_e2e.py index c8c6ab7..bdf972a 100644 --- a/tests/e2e/test_e2e.py +++ b/tests/e2e/test_e2e.py @@ -194,10 +194,6 @@ def _the_cdn_serves_the_flag_configuration_advertising_an_sse_stream_at_url( _HEADER_VALUE = r"(?P\"[^\"]*\"|'[^']*')" -def _unquote(value: str) -> str: - return value[1:-1] - - @step( parsers.re( r'an initialized, READY provider serving the "(?P[^"]+)" flag configuration' From aab31afc4d376337b5097b4fc6e079d421873c7f Mon Sep 17 00:00:00 2001 From: Lea Konvalinka Date: Thu, 5 Nov 2026 11:38:19 +0100 Subject: [PATCH 4/4] small fixes --- .gitignore | 3 + src/dynatrace/openfeature/_polling.py | 2 +- src/dynatrace/openfeature/_sse.py | 4 +- tests/e2e/test_e2e.py | 31 +--- tests/sse_server.py | 4 +- tests/test_cdn_config.py | 2 +- tests/test_polling.py | 5 +- tests/test_provider_sse.py | 252 +++++++++++++------------- tests/test_sse.py | 13 +- 9 files changed, 148 insertions(+), 168 deletions(-) diff --git a/.gitignore b/.gitignore index 5eb0ae0..b21110d 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,6 @@ coverage.xml .ruff_cache/ .pytest_cache/ .idea/ + +# macOS +.DS_Store diff --git a/src/dynatrace/openfeature/_polling.py b/src/dynatrace/openfeature/_polling.py index 977f2bd..c6a95e1 100644 --- a/src/dynatrace/openfeature/_polling.py +++ b/src/dynatrace/openfeature/_polling.py @@ -189,7 +189,7 @@ def _on_fetch_complete_locked(self, completed: FetchResult | None) -> bool: ): log.debug( "dropping deferred SSE re-fetch: completed fetch already returned " - "Last-Modified %s >= announced %s", + "Last-Modified %s >= pending %s", observed, pending, ) diff --git a/src/dynatrace/openfeature/_sse.py b/src/dynatrace/openfeature/_sse.py index 3b1b474..f3e39c3 100644 --- a/src/dynatrace/openfeature/_sse.py +++ b/src/dynatrace/openfeature/_sse.py @@ -18,7 +18,7 @@ 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 * 60.0 +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 @@ -47,7 +47,7 @@ def urlopen( 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 the re, because shutdown() does not + 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.""" diff --git a/tests/e2e/test_e2e.py b/tests/e2e/test_e2e.py index bdf972a..5cdcb8c 100644 --- a/tests/e2e/test_e2e.py +++ b/tests/e2e/test_e2e.py @@ -93,7 +93,6 @@ 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 -# payload a shut-down provider must not react to REFETCH_MESSAGE = '{"type":"refetchConfig","lastModified":1704153600}' scenarios(str(HARNESS_DIR / "gherkin" / "cdn.feature")) @@ -316,23 +315,7 @@ 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_TIMEOUT) - - -@step( - "polling triggers a configuration refetch after the SSE_DISCONNECTED_POLL_INTERVAL" -) -def _polling_triggers_a_configuration_refetch_after_the_sse_disconnected_poll_interval( - scenario: Scenario, -) -> None: - _await_cdn_request_within(scenario, SSE_DISCONNECTED_POLL_INTERVAL) - - -@step("polling triggers a configuration refetch after the SSE_CONNECTED_POLL_INTERVAL") -def _polling_triggers_a_configuration_refetch_after_the_sse_connected_poll_interval( - scenario: Scenario, -) -> None: - _await_cdn_request_within(scenario, SSE_CONNECTED_POLL_INTERVAL) + _await_cdn_request_within(scenario, POLL_INTERVAL) # a blip: the backend drops the stream and refuses connections for a while, then accepts them again @@ -356,11 +339,9 @@ def _the_sse_stream_drops_and_reconnects(scenario: Scenario, seconds: int) -> No ), f"SSE did not reconnect within {reconnect_timeout}s of a {seconds}s outage" -# a sustained outage: connections stay refused until the stream is restored or reset @step("the SSE stream drops and stays down") def _the_sse_stream_drops_and_stays_down(scenario: Scenario) -> None: scenario.mock.drop_sse() - # the provider switches cadence as soon as it sees the fault; the Then steps assert what it did assert _poll_until(SETTLE, lambda: scenario.mock.sse_client_count() == 0), ( "the mock server did not drop the SSE stream" ) @@ -372,7 +353,6 @@ def _the_server_emits_an_sse_message(scenario: Scenario, payload: str) -> None: # 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("\\\\", "\\")) - # essentially a sleep, cut short once the message has triggered a CDN request _poll_until(SETTLE, lambda: scenario.mock.count_cdn_requests() > baseline) @@ -769,12 +749,7 @@ def _consume_event(scenario: Scenario, expected: ProviderEvent) -> EventDetails: def _await_sse_settled(scenario: Scenario) -> None: - """Wait for the SSE stream and the catch-up fetch its connect pulls. - - So a scenario starting from a READY provider never races the stream it is - about to drop or message. No-op when SSE is off, as it is throughout the - CDN spec. - """ + """Wait for the SSE stream and the catch-up fetch its connect pulls.""" if not scenario.sse_enabled: return assert _poll_until( @@ -789,7 +764,7 @@ def _await_sse_settled(scenario: Scenario) -> None: 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, lambda: scenario.mock.count_cdn_requests() > baseline) + _wait_for(interval + 1, lambda: scenario.mock.count_cdn_requests() > baseline) def _advertising_sse_url(scenario: Scenario, config_json: str, url: str) -> str: diff --git a/tests/sse_server.py b/tests/sse_server.py index 0bd5d1c..a0e8438 100644 --- a/tests/sse_server.py +++ b/tests/sse_server.py @@ -57,7 +57,7 @@ def _stream(self, events: queue.Queue[bytes | None]) -> None: try: item = events.get(timeout=_IDLE_CHECK_S) except queue.Empty: - if self._client_closed(): + if self._check_client_closed(): return continue if item is None: @@ -69,7 +69,7 @@ def _stream(self, events: queue.Queue[bytes | None]) -> None: self.wfile.write(b"%x\r\n%s\r\n" % (len(item), item)) self.wfile.flush() - def _client_closed(self) -> bool: + def _check_client_closed(self) -> bool: readable, _, _ = select.select([self.connection], [], [], 0) if not readable: return False diff --git a/tests/test_cdn_config.py b/tests/test_cdn_config.py index c4156f5..1b1ebe1 100644 --- a/tests/test_cdn_config.py +++ b/tests/test_cdn_config.py @@ -113,6 +113,6 @@ def test_resolves(self, config: str, expected: str) -> None: def test_no_usable_entry(self, config: str | None) -> None: assert extract_sse_url(config, _ORIGIN) is None - def test_endpoint_without_any_origin(self) -> 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 7f804d8..f9e5834 100644 --- a/tests/test_polling.py +++ b/tests/test_polling.py @@ -343,6 +343,7 @@ def test_contended_refetches_are_deferred_and_coalesced(self) -> None: 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: @@ -373,7 +374,7 @@ def test_deferred_refetch_dropped_when_completed_fetch_is_as_new(self) -> None: ], ids=["completed-older", "completed-unknown", "announced-unknown"], ) - def test_deferred_refetch_issued_unless_proven_satisfied( + def test_deferred_refetch_issued_( self, completed_last_modified: str | None, announced: int | None ) -> None: completed = FetchResult(200, '{"flags":{}}', None, completed_last_modified) @@ -405,7 +406,7 @@ def fetch(cancel=None) -> FetchResult: pm, calls = _make_pm(fetch_fn=fetch, interval_s=_LONG_INTERVAL) pm.start() try: - pm.trigger_immediate_fetch(None) + pm.trigger_immediate_fetch(_LM_EPOCH) assert started.wait(1.0) pm.trigger_immediate_fetch(_LM_EPOCH) gate.set() diff --git a/tests/test_provider_sse.py b/tests/test_provider_sse.py index 88b258c..baa4949 100644 --- a/tests/test_provider_sse.py +++ b/tests/test_provider_sse.py @@ -21,7 +21,7 @@ _DISCONNECTED_S = 300.0 -def _config(sse_url: str | None = None) -> str: +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}] @@ -75,13 +75,13 @@ 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=_config(), status=200) - p = _make_provider(**kwargs) # type: ignore[arg-type] - p.initialize() + 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 p._lock: - p._sse_manager = fake # type: ignore[assignment] - return p, fake + with provider._lock: + provider._sse_manager = fake # type: ignore[assignment] + return provider, fake class TestSseStartup: @@ -89,72 +89,73 @@ class TestSseStartup: 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=_config(sse_server.url), status=200) - p = _make_provider() - p.initialize() + 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 p._sse_connected - assert p._poller is not None - assert p._poller.interval_s == _CONNECTED_S + assert provider._sse_connected + assert provider._poller is not None + assert provider._poller.interval_s == _CONNECTED_S finally: - p.shutdown() + 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) - p = _make_provider() - p.initialize() + provider = _make_provider() + provider.initialize() try: - assert p._sse_manager is not None - assert p._sse_manager._sse_url == "https://cdn.us.fm.dynatrace.com/sse" + assert provider._sse_manager is not None + assert provider._sse_manager._sse_url == "https://cdn.us.fm.dynatrace.com/sse" finally: - p.shutdown() + provider.shutdown() @rsps_lib.activate def test_no_advertised_stream_runs_polling_only(self) -> None: - rsps_lib.add(rsps_lib.GET, _CDN_URL, body=_config(), status=200) - p = _make_provider() - p.initialize() + rsps_lib.add(rsps_lib.GET, _CDN_URL, body=_flag_config(), status=200) + provider = _make_provider() + provider.initialize() try: - assert p._sse_manager is None - assert p._poller is not None - assert p._poller.interval_s == _DISCONNECTED_S + assert provider._sse_manager is None + assert provider._poller is not None + assert provider._poller.interval_s == _DISCONNECTED_S finally: - p.shutdown() + 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=_config(sse_server.url), status=200) - p = _make_provider(disable_sse=True) - p.initialize() + 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 p._sse_manager is None - assert not wait_until(lambda: sse_server.client_count > 0, timeout=0.3) + assert provider._sse_manager is None + assert not wait_until(lambda: sse_server.client_count > 0, timeout=0.5) finally: - p.shutdown() + 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=_config("http://127.0.0.1:1/sse"), status=200 + rsps_lib.GET, _CDN_URL, body=_flag_config("http://127.0.0.1:1/sse"), status=200 ) - p = _make_provider() - p.initialize() + provider = _make_provider() + provider.initialize() try: - assert p._sse_manager is not None + assert provider._sse_manager is not None # faults before a first connect are not disconnects - assert not wait_until(lambda: p._state != ProviderStatus.READY, timeout=0.5) - assert not p._sse_connected - assert p._poller is not None - assert p._poller.interval_s == _DISCONNECTED_S + 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: - p.shutdown() + provider.shutdown() @rsps_lib.activate def test_failed_init_starts_sse_from_the_first_stored_config( @@ -162,28 +163,28 @@ def test_failed_init_starts_sse_from_the_first_stored_config( ) -> 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=_config(sse_server.url), status=200) - p = _make_provider(poll_s=0.05) + 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): - p.initialize() + provider.initialize() try: - assert p._sse_manager is None + assert provider._sse_manager is None assert wait_until(lambda: sse_server.client_count == 1) - assert wait_until(lambda: p._state == ProviderStatus.READY) + assert wait_until(lambda: provider._state == ProviderStatus.READY) finally: - p.shutdown() + provider.shutdown() def test_superseded_init_opens_no_stream(self, sse_server: SseServer) -> None: - fetcher = _BlockingFetcher(_config(sse_server.url)) - p = DynatraceProvider(_KEY, fetcher=fetcher, poll_interval_s=600.0) - init_thread = threading.Thread(target=p.initialize) + 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) - p.shutdown() + provider.shutdown() fetcher.release.set() init_thread.join(timeout=5.0) - assert p._sse_manager is None - assert not wait_until(lambda: sse_server.client_count > 0, timeout=0.3) + assert provider._sse_manager is None + assert not wait_until(lambda: sse_server.client_count > 0, timeout=0.5) class _BlockingFetcher: @@ -204,162 +205,163 @@ def invalidate_cache(self) -> None: class TestSseConnectionLifecycle: @rsps_lib.activate def test_connect_switches_cadence_and_pulls_catch_up(self) -> None: - p, fake = _ready_provider_with_fake_sse() + provider, fake = _ready_provider_with_fake_sse() rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) try: - p._on_sse_connected(fake) # type: ignore[arg-type] - assert p._poller is not None - assert p._poller.interval_s == _CONNECTED_S + 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: - p.shutdown() + provider.shutdown() @rsps_lib.activate def test_disconnect_falls_back_to_polling_without_a_state_change(self) -> None: - p, fake = _ready_provider_with_fake_sse() + provider, fake = _ready_provider_with_fake_sse() rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) - events = _record_events(p) + events = _record_events(provider) try: - p._on_sse_connected(fake) # type: ignore[arg-type] + provider._on_sse_connected(fake) # type: ignore[arg-type] assert wait_until(lambda: _cdn_calls() == 2) - p._on_sse_disconnected(fake) # type: ignore[arg-type] - p._on_sse_disconnected(fake) # type: ignore[arg-type] # repeat fault while retrying - assert not p._sse_connected - assert p._poller is not None - assert p._poller.interval_s == _DISCONNECTED_S + 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 p._state == ProviderStatus.READY - assert p._grace_timer is None - assert not wait_until(lambda: _cdn_calls() > 2, timeout=0.2) + 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: - p.shutdown() + provider.shutdown() @rsps_lib.activate def test_reconnect_returns_to_sse_mode_and_pulls_catch_up(self) -> None: - p, fake = _ready_provider_with_fake_sse() + provider, fake = _ready_provider_with_fake_sse() rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) - events = _record_events(p) + events = _record_events(provider) try: - p._on_sse_connected(fake) # type: ignore[arg-type] + provider._on_sse_connected(fake) # type: ignore[arg-type] assert wait_until(lambda: _cdn_calls() == 2) - p._on_sse_disconnected(fake) # type: ignore[arg-type] - p._on_sse_connected(fake) # type: ignore[arg-type] + 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 p._sse_connected - assert p._poller is not None - assert p._poller.interval_s == _CONNECTED_S + assert provider._sse_connected + assert provider._poller is not None + assert provider._poller.interval_s == _CONNECTED_S assert events == [] finally: - p.shutdown() + provider.shutdown() @rsps_lib.activate def test_stale_provider_recovers_only_through_a_successful_fetch(self) -> None: - p, fake = _ready_provider_with_fake_sse() + provider, fake = _ready_provider_with_fake_sse() rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) - events = _record_events(p) + events = _record_events(provider) try: - with p._lock: - p._sse_connected = True - p._on_fetch_failure("HTTP 500") + with provider._lock: + provider._sse_connected = True + provider._on_fetch_failure("HTTP 500") # a disconnect neither worsens nor resolves staleness - p._on_sse_disconnected(fake) # type: ignore[arg-type] - assert p._state == ProviderStatus.STALE + 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 - p._on_sse_connected(fake) # type: ignore[arg-type] - assert wait_until(lambda: p._state == ProviderStatus.READY) + provider._on_sse_connected(fake) # type: ignore[arg-type] + assert wait_until(lambda: provider._state == ProviderStatus.READY) assert events == ["stale", "ready"] finally: - p.shutdown() + provider.shutdown() @rsps_lib.activate def test_callbacks_from_a_replaced_stream_are_ignored(self) -> None: - p, _ = _ready_provider_with_fake_sse() + provider, _ = _ready_provider_with_fake_sse() stale_stream = _FakeSse() try: - p._on_sse_connected(stale_stream) # type: ignore[arg-type] - assert not p._sse_connected - p._on_sse_message(stale_stream, SseMessage(type="refetchConfig")) # type: ignore[arg-type] + 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: - p.shutdown() + 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_whatever_the_sse_state( + def test_poll_failure_goes_stale_regardless_of_sse_state( self, sse_connected: bool ) -> None: - p, _ = _ready_provider_with_fake_sse() + provider, _ = _ready_provider_with_fake_sse() try: - with p._lock: - p._sse_connected = sse_connected - p._on_fetch_failure("HTTP 500") - assert p._state == ProviderStatus.STALE + with provider._lock: + provider._sse_connected = sse_connected + provider._on_fetch_failure("HTTP 500") + assert provider._state == ProviderStatus.STALE finally: - p.shutdown() + 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=_config(), status=200, headers={"ETag": '"v1"'} + rsps_lib.GET, _CDN_URL, body=_flag_config(), status=200, headers={"ETag": '"v1"'} ) - p = _make_provider() - p.initialize() + provider = _make_provider() + provider.initialize() fake = _FakeSse() - with p._lock: - p._sse_manager = fake # type: ignore[assignment] + with provider._lock: + provider._sse_manager = fake # type: ignore[assignment] rsps_lib.add(rsps_lib.GET, _CDN_URL, status=304) try: - p._on_sse_message(fake, SseMessage(type="refetchConfig", last_modified=1)) # type: ignore[arg-type] + 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: - p.shutdown() + provider.shutdown() @rsps_lib.activate def test_other_types_are_ignored(self) -> None: - p, fake = _ready_provider_with_fake_sse() + provider, fake = _ready_provider_with_fake_sse() try: - p._on_sse_message(fake, SseMessage(type="unknownType")) # type: ignore[arg-type] - assert not wait_until(lambda: _cdn_calls() > 1, timeout=0.2) + provider._on_sse_message(fake, SseMessage(type="unknownType")) # type: ignore[arg-type] + assert not wait_until(lambda: _cdn_calls() > 1, timeout=0.5) finally: - p.shutdown() + provider.shutdown() class TestSseShutdown: @rsps_lib.activate def test_shutdown_closes_sse_before_polling(self) -> None: order: list[str] = [] - p, _ = _ready_provider_with_fake_sse() + provider, _ = _ready_provider_with_fake_sse() fake = _FakeSse(order) - with p._lock: - p._sse_manager = fake # type: ignore[assignment] - assert p._poller is not None - stop_poller = p._poller.stop + 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() - p._poller.stop = recording_stop # type: ignore[method-assign] - p.shutdown() + provider._poller.stop = recording_stop # type: ignore[method-assign] + provider.shutdown() assert order == ["sse", "poller"] - assert p._sse_manager is None + 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=_config(sse_server.url), status=200) - p = _make_provider() - p.initialize() + 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) - p.shutdown() + 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.3) + assert not wait_until(lambda: _cdn_calls() > 2, timeout=0.5) diff --git a/tests/test_sse.py b/tests/test_sse.py index 9eb83c2..e31ce6e 100644 --- a/tests/test_sse.py +++ b/tests/test_sse.py @@ -133,10 +133,9 @@ def test_stop_closes_the_stream_and_silences_callbacks( manager = _start(sse_server.url, rec) assert wait_until(lambda: sse_server.client_count == 1) manager.stop() - # no event to wake the reader first: the socket itself must close, on Windows too assert wait_until(lambda: sse_server.client_count == 0) sse_server.emit('{"type":"refetchConfig"}') - assert not wait_until(lambda: rec.messages != [], timeout=0.3) + 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() @@ -163,7 +162,7 @@ def test_stop_returns_promptly_while_the_reader_is_blocked( 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.3) + assert not wait_until(lambda: rec.messages != [], timeout=0.5) def test_a_failing_callback_does_not_kill_the_stream( self, sse_server: SseServer @@ -173,7 +172,7 @@ def test_a_failing_callback_does_not_kill_the_stream( def on_message(msg: SseMessage) -> None: received.append(msg) if len(received) == 1: - raise RuntimeError("boom") + raise RuntimeError("on_message failed") manager = SseManager(sse_server.url, lambda: None, lambda: None, on_message) manager.start() @@ -199,6 +198,6 @@ def test_logs_never_carry_the_query_string( 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 - ours = [r for r in caplog.records if r.name.startswith("dynatrace.")] - assert ours - assert not any("secret" in r.getMessage() for r in ours) + 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)