diff --git a/test/launchdarkly/mzcompose.py b/test/launchdarkly/mzcompose.py index 773ad142e1ab5..9b1c9e3d7dac5 100644 --- a/test/launchdarkly/mzcompose.py +++ b/test/launchdarkly/mzcompose.py @@ -11,14 +11,16 @@ Test the LaunchDarkly integration, get configuration flags from LD. """ +import json from itertools import chain from os import environ from textwrap import dedent -from time import sleep +from time import monotonic, sleep from typing import Any from uuid import uuid1 import launchdarkly_api # type: ignore +import requests from launchdarkly_api.api import feature_flags_api # type: ignore from launchdarkly_api.model.client_side_availability_post import ( # type: ignore ClientSideAvailabilityPost, @@ -96,7 +98,8 @@ def context_rule( }, external_metadata_store=True, ), - Testdrive(no_reset=True, seed=1), + # Include streaming reconnect backoff in the live-update assertion window. + Testdrive(no_reset=True, seed=1, default_timeout="120s"), ] @@ -143,10 +146,6 @@ def workflow_default(c: Composition) -> None: on=True, ) - # 3 seconds should be enough to avoid race conditions between the update - # above and the query below. - sleep(3) - # Assert that the value is as expected after the initial parameter sync. with c.override( Materialized( @@ -622,7 +621,7 @@ def update_targeting( ) -> Any: with launchdarkly_api.ApiClient(self.configuration) as api_client: api = feature_flags_api.FeatureFlagsApi(api_client) - return api.patch_feature_flag( + flag = api.patch_feature_flag( project_key=self.project_key, feature_flag_key=feature_flag_key, patch_with_comment=PatchWithComment( @@ -667,6 +666,73 @@ def update_targeting( ) ), ) + self.wait_for_propagation( + feature_flag_key, flag.environments[self.environment_key].version + ) + return flag + + def wait_for_propagation(self, feature_flag_key: str, version: int) -> None: + """Probe fresh SDK stream snapshots until one contains the updated flag.""" + assert LAUNCHDARKLY_SDK_KEY is not None + # A successful management API write does not imply SDK visibility. In + # particular, the boot-only assertion has no sync loop to repair a stale + # initial read. Probe the same initial snapshot endpoint that boot uses, + # not the polling endpoint or later patches on an already-open stream. + # These versions are environment-specific, not the flag's top-level + # management API version. + deadline = monotonic() + 120 + observed: int | str | None = "no snapshot" + while monotonic() < deadline: + try: + with requests.get( + "https://stream.launchdarkly.com/all", + headers={"Authorization": LAUNCHDARKLY_SDK_KEY}, + stream=True, + timeout=10, + ) as response: + response.raise_for_status() + response.encoding = "utf-8" + event = "" + data: list[str] = [] + for line in response.iter_lines(decode_unicode=True): + if monotonic() >= deadline: + break + if line.startswith("event:"): + event = line[len("event:") :].strip() + elif line.startswith("data:"): + data.append(line[len("data:") :].lstrip()) + elif not line and data: + snapshot = json.loads("\n".join(data)) + if event != "put" or snapshot["path"] != "/": + raise UIError( + "Expected an initial root put from SDK stream" + ) + flag = snapshot["data"]["flags"].get(feature_flag_key) + snapshot_version = ( + flag["version"] if flag is not None else None + ) + observed = snapshot_version + if ( + snapshot_version is not None + and snapshot_version >= version + ): + print( + f"SDK snapshot contains {feature_flag_key} version " + f"{observed} (wanted {version})" + ) + return + break + except requests.HTTPError as e: + if e.response is None or e.response.status_code < 500: + raise + observed = f"HTTP {e.response.status_code}" + except (requests.ConnectionError, requests.Timeout) as e: + observed = type(e).__name__ + sleep(1) + raise UIError( + f"SDK propagation timed out for {feature_flag_key}: " + f"expected version >= {version}, last observed {observed}" + ) def delete_flag(self, feature_flag_key: str) -> Any: with launchdarkly_api.ApiClient(self.configuration) as api_client: