From c8aaf10414230cf60478e10989eb9c276115b79c Mon Sep 17 00:00:00 2001 From: Tyler Potter Date: Tue, 3 Nov 2026 13:46:17 -0600 Subject: [PATCH 1/3] test(ffe): validate imported dependency telemetry Consume canonical dependent-suite expectations to assert graph evaluation events, exposure rollback, memoization, and short-circuit behavior without local fixture overlays. Environment: Datadog workspace Co-Authored-By: Codex GPT-5 --- .../test_ffe/test_dynamic_evaluation.py | 172 +++++++++++++----- tests/parametric/test_ffe/utils.py | 65 ++++++- 2 files changed, 193 insertions(+), 44 deletions(-) diff --git a/tests/parametric/test_ffe/test_dynamic_evaluation.py b/tests/parametric/test_ffe/test_dynamic_evaluation.py index b35d3280840..904bd336a6a 100644 --- a/tests/parametric/test_ffe/test_dynamic_evaluation.py +++ b/tests/parametric/test_ffe/test_dynamic_evaluation.py @@ -15,12 +15,14 @@ from utils.dd_constants import RemoteConfigApplyState from utils.docker_fixtures import TestAgentAPI from tests.parametric.conftest import APMLibrary -from tests.parametric.test_ffe.utils import assert_feature_map, ffe_events_from_requests +from tests.parametric.test_ffe.utils import assert_feature_map, events_match, ffe_events_from_requests RC_PRODUCT = "FFE_FLAGS" RC_PATH = f"datadog/2/{RC_PRODUCT}" FFE_READY_RETRY_ATTEMPTS = 10 FFE_READY_RETRY_INTERVAL_SECONDS = 0.2 +FFE_EVENT_WAIT_ATTEMPTS = 300 +FFE_EVENT_WAIT_INTERVAL_SECONDS = 0.1 FFE_SYSTEM_TEST_DATA_DIR = Path(__file__).parent / "ffe-system-test-data" FFE_EVALUATION_CASES_DIR = FFE_SYSTEM_TEST_DATA_DIR / "evaluation-cases" DEPENDENT_FLAGS_DIR = FFE_SYSTEM_TEST_DATA_DIR / "evaluation-suites" / "dependent-flags" @@ -30,56 +32,54 @@ # Load the UFC fixture file at module level -def _load_ufc_fixture() -> dict[str, Any]: - """Load the UFC fixture file.""" - fixture_path = FFE_SYSTEM_TEST_DATA_DIR / "ufc-config.json" - - with fixture_path.open() as f: - return json.load(f) - - -def _get_test_case_files() -> list[str]: - """Get all test case files from the fixtures directory.""" - case_files = sorted(path.name for path in FFE_EVALUATION_CASES_DIR.glob("*.json")) +def _load_evaluation_suite( + suite_directory: Path, + cases_directory: Path, + *, + required: bool, +) -> tuple[dict[str, Any], list[Path]]: + """Load one canonical rules-based evaluation suite.""" + case_files = sorted(cases_directory.glob("*.json")) if not case_files: - raise FileNotFoundError(f"No canonical evaluation cases found in {FFE_EVALUATION_CASES_DIR}") - return case_files + if required: + raise FileNotFoundError(f"No canonical evaluation cases found in {cases_directory}") + return {}, [] + return json.loads((suite_directory / "ufc-config.json").read_text()), case_files -def _filter_test_cases_for_platform(test_cases: list[dict[str, Any]], platform: str) -> list[dict[str, Any]]: +def _filter_for_platform(items: list[dict[str, Any]], platform: str) -> list[dict[str, Any]]: """Apply the canonical corpus _skip/_include selection rules.""" - applicable_cases = [] - for test_case in test_cases: - if "_skip" in test_case and "_include" in test_case: - raise ValueError("Evaluation cases cannot define both _skip and _include") - skipped_platforms = {entry["platform"] for entry in test_case.get("_skip", [])} - included_platforms = {entry["platform"] for entry in test_case.get("_include", [])} + applicable_items = [] + for item in items: + if "_skip" in item and "_include" in item: + raise ValueError("Corpus entries cannot define both _skip and _include") + skipped_platforms = {entry["platform"] for entry in item.get("_skip", [])} + included_platforms = {entry["platform"] for entry in item.get("_include", [])} if platform in skipped_platforms: continue if included_platforms and platform not in included_platforms: continue - applicable_cases.append(test_case) - return applicable_cases - - -def _load_dependent_flags_suite() -> tuple[dict[str, Any], list[Path]]: - """Load the dependent-flags suite when it is present in the corpus.""" - case_files = sorted(DEPENDENT_FLAGS_CASES_DIR.glob("*.json")) - if not case_files: - return {}, [] - return json.loads((DEPENDENT_FLAGS_DIR / "ufc-config.json").read_text()), case_files + applicable_items.append(item) + return applicable_items # Load fixture at module level for reuse across tests -UFC_FIXTURE_DATA = _load_ufc_fixture() -ALL_TEST_CASE_FILES = _get_test_case_files() +UFC_FIXTURE_DATA, ALL_TEST_CASE_FILES = _load_evaluation_suite( + FFE_SYSTEM_TEST_DATA_DIR, + FFE_EVALUATION_CASES_DIR, + required=True, +) SPLIT_FEATURE_CASES = [ case - for filename in ALL_TEST_CASE_FILES - for case in json.loads((FFE_EVALUATION_CASES_DIR / filename).read_text()) + for case_file in ALL_TEST_CASE_FILES + for case in json.loads(case_file.read_text()) if "features" in case["result"] ] -DEPENDENT_FLAGS_UFC_DATA, DEPENDENT_FLAGS_TEST_CASE_FILES = _load_dependent_flags_suite() +DEPENDENT_FLAGS_UFC_DATA, DEPENDENT_FLAGS_TEST_CASE_FILES = _load_evaluation_suite( + DEPENDENT_FLAGS_DIR, + DEPENDENT_FLAGS_CASES_DIR, + required=False, +) DEFAULT_ENVVARS = { "DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED": "true", @@ -172,7 +172,7 @@ def _assert_evaluation_cases( assert test_cases, f"No evaluation cases in {test_case_file.name}" platform = context.library.name - test_cases = _filter_test_cases_for_platform(test_cases, platform) + test_cases = _filter_for_platform(test_cases, platform) if not test_cases: pytest.skip(f"No cases in {test_case_file.name} apply to platform {platform}") @@ -211,6 +211,90 @@ def _assert_evaluation_cases( f"flag='{test_case['flag']}' unexpectedly returned errorCode={result.get('errorCode')!r}" ) + telemetry_cases = [test_case for test_case in test_cases if "expectations" in test_case] + if telemetry_cases: + _assert_evaluation_telemetry(test_agent, test_library, telemetry_cases) + + +def _expected_event_matchers(expectations: dict[str, Any], event_type: str, platform: str) -> list[dict[str, Any]]: + matchers = expectations.get(event_type, []) + assert isinstance(matchers, list), f"expectations.{event_type} must be an array" + expanded: list[dict[str, Any]] = [] + for matcher in matchers: + assert isinstance(matcher, dict), f"{event_type} matcher must be an object" + if not _filter_for_platform([matcher], platform): + continue + count = matcher.get("_count", 1) + assert isinstance(count, int), f"{event_type} matcher count must be an integer" + assert count >= 1, f"{event_type} matcher count must be positive" + expanded.extend([matcher] * count) + return expanded + + +def _assert_evaluation_telemetry( + test_agent: TestAgentAPI, test_library: APMLibrary, test_cases: list[dict[str, Any]] +) -> None: + platform = context.library.name + marker_cases = [ + test_case + for test_case in test_cases + if _expected_event_matchers(test_case["expectations"], "exposures", platform) + ] + assert marker_cases, "Telemetry expectations require a case that emits an exposure drain marker" + marker_case = marker_cases[0] + marker_targeting_key = f"{marker_case['targetingKey']}-telemetry-drain-marker" + marker_result = _ffe_evaluate_with_rc_retry( + test_library, + flag=marker_case["flag"], + variation_type=marker_case["variationType"], + default_value=marker_case["defaultValue"], + targeting_key=marker_targeting_key, + attributes=marker_case.get("attributes", {}), + ) + assert marker_result.get("reason") != "ERROR", f"telemetry drain marker evaluation failed: {marker_result}" + + captured: dict[str, tuple[list[dict[str, Any]], list[dict[str, Any]]]] = {} + + for _ in range(FFE_EVENT_WAIT_ATTEMPTS): + requests = test_agent.requests() + captured = {} + for test_case in test_cases: + case_events = ffe_events_from_requests(requests, targeting_key=test_case["targetingKey"]) + captured[test_case["targetingKey"]] = (case_events["EXPOSURE"], case_events["EVALUATION"]) + + # Each writer preserves event order. Seeing the post-batch marker from both writers + # proves that all telemetry from the evaluation cases has drained before exact checks. + marker_events = ffe_events_from_requests(requests, targeting_key=marker_targeting_key) + complete = all(marker_events.values()) + for test_case in test_cases: + expectations = test_case["expectations"] + assert isinstance(expectations, dict) + exposures, evaluations = captured[test_case["targetingKey"]] + exact = expectations.get("noUnmatchedEvents") is True + complete = complete and events_match( + exposures, _expected_event_matchers(expectations, "exposures", platform), exact=exact + ) + complete = complete and events_match( + evaluations, _expected_event_matchers(expectations, "evaluationEvents", platform), exact=exact + ) + + if complete: + break + time.sleep(FFE_EVENT_WAIT_INTERVAL_SECONDS) + else: + raise AssertionError(f"evaluation telemetry did not stabilize: {captured}") + + for test_case in test_cases: + expectations = test_case["expectations"] + exposures, evaluations = captured[test_case["targetingKey"]] + exact = expectations.get("noUnmatchedEvents") is True + assert events_match(exposures, _expected_event_matchers(expectations, "exposures", platform), exact=exact), ( + f"unexpected exposures for {test_case['flag']}: {exposures}" + ) + assert events_match( + evaluations, _expected_event_matchers(expectations, "evaluationEvents", platform), exact=exact + ), f"unexpected evaluation events for {test_case['flag']}: {evaluations}" + @scenarios.parametric @features.feature_flags_dynamic_evaluation @@ -231,8 +315,14 @@ def test_ffe_remote_config(self, test_agent: TestAgentAPI, test_library: APMLibr assert apply_state["product"] == RC_PRODUCT @parametrize("library_env", [{**DEFAULT_ENVVARS}]) + @parametrize("agent_env", [{"SYSTEM_TESTS_FFE_INTAKE": "true"}]) @parametrize("test_case_file", ALL_TEST_CASE_FILES) - def test_ffe_flag_evaluation(self, test_case_file: str, test_agent: TestAgentAPI, test_library: APMLibrary) -> None: + def test_ffe_flag_evaluation( + self, + test_case_file: Path, + test_agent: TestAgentAPI, + test_library: APMLibrary, + ) -> None: """Test FFE flag evaluation logic with various targeting scenarios. This is the core FFE test that validates the OpenFeature provider correctly: @@ -242,10 +332,10 @@ def test_ffe_flag_evaluation(self, test_case_file: str, test_agent: TestAgentAPI 4. Handles user targeting, attribute matching, and rollout percentages """ - test_case_path = FFE_EVALUATION_CASES_DIR / test_case_file - _assert_evaluation_cases(test_case_path, UFC_FIXTURE_DATA, test_agent, test_library) + _assert_evaluation_cases(test_case_file, UFC_FIXTURE_DATA, test_agent, test_library) @parametrize("library_env", [{**DEFAULT_ENVVARS}]) + @parametrize("agent_env", [{"SYSTEM_TESTS_FFE_INTAKE": "true"}]) @parametrize("test_case_file", DEPENDENT_FLAGS_TEST_CASE_FILES) def test_ffe_dependent_flags( self, @@ -280,7 +370,7 @@ class Test_Feature_Flag_Split_Features: def test_split_feature_destinations( self, case: dict[str, Any], test_agent: TestAgentAPI, test_library: APMLibrary ) -> None: - if not _filter_test_cases_for_platform([case], context.library.name): + if not _filter_for_platform([case], context.library.name): pytest.skip(f"Case does not apply to {context.library.name}") expected = case["result"] _set_and_wait_ffe_rc(test_agent, UFC_FIXTURE_DATA) diff --git a/tests/parametric/test_ffe/utils.py b/tests/parametric/test_ffe/utils.py index b82b1c64c0a..bfb7243b729 100644 --- a/tests/parametric/test_ffe/utils.py +++ b/tests/parametric/test_ffe/utils.py @@ -157,8 +157,17 @@ def assert_feature_map(actual: object, expected: dict[str, Any], destination: st assert actual_value == expected_value, f"{destination}.{key}: expected {expected_value!r}, got {actual_value!r}" -def ffe_events_from_requests(requests: Sequence[Mapping[str, Any]], flag: str) -> dict[str, list[dict[str, Any]]]: - """Read real EVP payloads recorded by the parametric test agent.""" +def _targeting_key_matches(actual: object, expected: str) -> bool: + hashed = f"sha256_{hashlib.sha256(expected.encode()).hexdigest()}" + return actual in (expected, hashed) + + +def ffe_events_from_requests( + requests: Sequence[Mapping[str, Any]], + flag: str | None = None, + targeting_key: str | None = None, +) -> dict[str, list[dict[str, Any]]]: + """Read matching EVP payloads recorded by the parametric test agent.""" events: dict[str, list[dict[str, Any]]] = {"EXPOSURE": [], "EVALUATION": []} for request in requests: path = request["url"].split("?", 1)[0] @@ -173,5 +182,55 @@ def ffe_events_from_requests(requests: Sequence[Mapping[str, Any]], flag: str) - if body.startswith(b"\x1f\x8b"): body = gzip.decompress(body) payload = json.loads(body) - events[destination].extend(event for event in payload[field] if event.get("flag", {}).get("key") == flag) + for event in payload[field]: + if flag is not None and event.get("flag", {}).get("key") != flag: + continue + if targeting_key is not None: + if destination == "EXPOSURE": + subject = event.get("subject") + actual_targeting_key = subject.get("id") if isinstance(subject, dict) else None + if actual_targeting_key != targeting_key: + continue + elif not _targeting_key_matches(event.get("targeting_key"), targeting_key): + continue + events[destination].append(event) return events + + +def ffe_event_matches(event: dict[str, Any], matcher: dict[str, Any]) -> bool: + """Match a canonical corpus expectation against an EVP event.""" + nested_key_fields = { + "allocation": "allocation", + "errorCode": "error", + "flag": "flag", + "variant": "variant", + } + + def event_field(field: str) -> object: + if field in nested_key_fields: + value = event.get(nested_key_fields[field]) + if not isinstance(value, dict): + return None + return value.get("message" if field == "errorCode" else "key") + if field == "runtimeDefaultUsed": + return event.get("runtime_default_used", False) + return event.get(field) + + fields_match = all( + event_field(field) == expected for field, expected in matcher.items() if not field.startswith("_") + ) + expected_error_code = matcher.get("errorCode") + return fields_match and (expected_error_code is None or event.get("runtime_default_used") is True) + + +def events_match(actual: list[dict[str, Any]], expected: list[dict[str, Any]], *, exact: bool) -> bool: + """Consume each expected matcher from a distinct emitted event.""" + unmatched = actual.copy() + for matcher in expected: + matching_index = next( + (index for index, event in enumerate(unmatched) if ffe_event_matches(event, matcher)), None + ) + if matching_index is None: + return False + unmatched.pop(matching_index) + return not exact or not unmatched From 1da52800e9a38ca3389ac2fad0252aec541ec4e0 Mon Sep 17 00:00:00 2001 From: Tyler Potter Date: Tue, 3 Nov 2026 13:46:19 -0600 Subject: [PATCH 2/3] test(ffe): harden shared telemetry harness Keep feature suites lazy and use corpus-owned drain markers with unambiguous telemetry correlation. Environment: Datadog workspace Co-Authored-By: Codex GPT-5 --- .../test_ffe/test_dynamic_evaluation.py | 235 +++++++++++++----- tests/parametric/test_ffe/utils.py | 34 ++- 2 files changed, 208 insertions(+), 61 deletions(-) diff --git a/tests/parametric/test_ffe/test_dynamic_evaluation.py b/tests/parametric/test_ffe/test_dynamic_evaluation.py index 904bd336a6a..e80a01d08d5 100644 --- a/tests/parametric/test_ffe/test_dynamic_evaluation.py +++ b/tests/parametric/test_ffe/test_dynamic_evaluation.py @@ -15,18 +15,24 @@ from utils.dd_constants import RemoteConfigApplyState from utils.docker_fixtures import TestAgentAPI from tests.parametric.conftest import APMLibrary -from tests.parametric.test_ffe.utils import assert_feature_map, events_match, ffe_events_from_requests +from tests.parametric.test_ffe.utils import ( + assert_feature_map, + events_match, + ffe_events_for_target, + ffe_events_from_requests, +) RC_PRODUCT = "FFE_FLAGS" RC_PATH = f"datadog/2/{RC_PRODUCT}" FFE_READY_RETRY_ATTEMPTS = 10 FFE_READY_RETRY_INTERVAL_SECONDS = 0.2 -FFE_EVENT_WAIT_ATTEMPTS = 300 -FFE_EVENT_WAIT_INTERVAL_SECONDS = 0.1 +FFE_EVENT_WAIT_ATTEMPTS = 60 +FFE_EVENT_WAIT_INTERVAL_SECONDS = 0.5 FFE_SYSTEM_TEST_DATA_DIR = Path(__file__).parent / "ffe-system-test-data" FFE_EVALUATION_CASES_DIR = FFE_SYSTEM_TEST_DATA_DIR / "evaluation-cases" DEPENDENT_FLAGS_DIR = FFE_SYSTEM_TEST_DATA_DIR / "evaluation-suites" / "dependent-flags" DEPENDENT_FLAGS_CASES_DIR = DEPENDENT_FLAGS_DIR / "cases" +TELEMETRY_DRAIN_TARGETING_KEY = "system-tests-telemetry-drain" parametrize = pytest.mark.parametrize @@ -37,14 +43,17 @@ def _load_evaluation_suite( cases_directory: Path, *, required: bool, -) -> tuple[dict[str, Any], list[Path]]: - """Load one canonical rules-based evaluation suite.""" +) -> tuple[Path, list[Path]]: + """Discover one canonical rules-based evaluation suite without parsing its UFC.""" case_files = sorted(cases_directory.glob("*.json")) if not case_files: if required: raise FileNotFoundError(f"No canonical evaluation cases found in {cases_directory}") - return {}, [] - return json.loads((suite_directory / "ufc-config.json").read_text()), case_files + return suite_directory / "ufc-config.json", [] + config_path = suite_directory / "ufc-config.json" + if not config_path.is_file(): + raise FileNotFoundError(f"No canonical UFC configuration found at {config_path}") + return config_path, case_files def _filter_for_platform(items: list[dict[str, Any]], platform: str) -> list[dict[str, Any]]: @@ -64,22 +73,40 @@ def _filter_for_platform(items: list[dict[str, Any]], platform: str) -> list[dic # Load fixture at module level for reuse across tests -UFC_FIXTURE_DATA, ALL_TEST_CASE_FILES = _load_evaluation_suite( +UFC_FIXTURE_PATH, ALL_TEST_CASE_FILES = _load_evaluation_suite( FFE_SYSTEM_TEST_DATA_DIR, FFE_EVALUATION_CASES_DIR, required=True, ) +UFC_FIXTURE_DATA = json.loads(UFC_FIXTURE_PATH.read_text()) SPLIT_FEATURE_CASES = [ case for case_file in ALL_TEST_CASE_FILES for case in json.loads(case_file.read_text()) if "features" in case["result"] ] -DEPENDENT_FLAGS_UFC_DATA, DEPENDENT_FLAGS_TEST_CASE_FILES = _load_evaluation_suite( +DEPENDENT_FLAGS_UFC_PATH, DEPENDENT_FLAGS_TEST_CASE_FILES = _load_evaluation_suite( DEPENDENT_FLAGS_DIR, DEPENDENT_FLAGS_CASES_DIR, required=False, ) +DEPENDENT_FLAGS_CASE_PARAMETERS = [ + pytest.param(case_file, id=case_file.stem) for case_file in DEPENDENT_FLAGS_TEST_CASE_FILES +] or [pytest.param(None, id="missing-suite")] + + +def _case_file_has_telemetry_expectations(case_file: Path) -> bool: + return any("expectations" in case for case in json.loads(case_file.read_text())) + + +CORE_CASE_PARAMETERS = [ + pytest.param( + case_file, + {"SYSTEM_TESTS_FFE_INTAKE": "true"} if _case_file_has_telemetry_expectations(case_file) else {}, + id=case_file.stem, + ) + for case_file in ALL_TEST_CASE_FILES +] DEFAULT_ENVVARS = { "DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED": "true", @@ -162,7 +189,7 @@ def _ffe_evaluate_with_rc_retry( def _assert_evaluation_cases( test_case_file: Path, - ufc_data: dict[str, Any], + ufc_path: Path, test_agent: TestAgentAPI, test_library: APMLibrary, ) -> None: @@ -176,6 +203,10 @@ def _assert_evaluation_cases( if not test_cases: pytest.skip(f"No cases in {test_case_file.name} apply to platform {platform}") + ufc_data = json.loads(ufc_path.read_text()) + telemetry_cases = [test_case for test_case in test_cases if "expectations" in test_case] + telemetry_drain_case = _find_telemetry_drain_case(ufc_data) if telemetry_cases else None + _set_and_wait_ffe_rc(test_agent, ufc_data) assert test_library.ffe_start(ufc_data), "Failed to start FFE provider" @@ -211,9 +242,69 @@ def _assert_evaluation_cases( f"flag='{test_case['flag']}' unexpectedly returned errorCode={result.get('errorCode')!r}" ) - telemetry_cases = [test_case for test_case in test_cases if "expectations" in test_case] if telemetry_cases: - _assert_evaluation_telemetry(test_agent, test_library, telemetry_cases) + assert telemetry_drain_case is not None + _assert_evaluation_telemetry(test_agent, test_library, test_cases, telemetry_drain_case) + + +def _find_telemetry_drain_case(ufc_data: dict[str, Any]) -> dict[str, Any]: + """Find a corpus-owned static logging flag suitable for draining both writers.""" + flags = ufc_data.get("flags") + assert isinstance(flags, dict), "UFC flags must be an object" + defaults: dict[str, Any] = { + "BOOLEAN": False, + "INTEGER": 0, + "JSON": {}, + "NUMERIC": 0.0, + "STRING": "", + } + for flag_key, flag in flags.items(): + if not isinstance(flag, dict) or flag.get("key") != flag_key or flag.get("enabled") is not True: + continue + allocations = flag.get("allocations") + if not isinstance(allocations, list) or len(allocations) != 1: + continue + allocation = allocations[0] + if not isinstance(allocation, dict) or allocation.get("doLog") is not True: + continue + if allocation.get("rules", []) != []: + continue + splits = allocation.get("splits") + if ( + not isinstance(splits, list) + or len(splits) != 1 + or not isinstance(splits[0], dict) + or splits[0].get("shards") != [] + ): + continue + variation_type = flag.get("variationType") + variations = flag.get("variations") + variation_key = splits[0].get("variationKey") + if variation_type not in defaults or not isinstance(variations, dict): + continue + variation = variations.get(variation_key) + if not isinstance(variation, dict) or variation.get("key") != variation_key or "value" not in variation: + continue + value = variation["value"] + if variation_type == "BOOLEAN": + valid_value = type(value) is bool + elif variation_type == "INTEGER": + valid_value = type(value) is int + elif variation_type == "NUMERIC": + valid_value = type(value) in (int, float) + elif variation_type == "STRING": + valid_value = isinstance(value, str) + else: + valid_value = isinstance(value, dict) + if not valid_value: + continue + return { + "flag": flag_key, + "variationType": variation_type, + "defaultValue": defaults[variation_type], + "value": value, + } + raise AssertionError("Telemetry expectations require a static doLog=true flag for the drain barrier") def _expected_event_matchers(expectations: dict[str, Any], event_type: str, platform: str) -> list[dict[str, Any]]: @@ -231,69 +322,98 @@ def _expected_event_matchers(expectations: dict[str, Any], event_type: str, plat return expanded +def _expected_events_for_cases( + cases: list[dict[str, Any]], platform: str +) -> tuple[list[dict[str, Any]], list[dict[str, Any]], bool] | None: + expectation_cases = [case for case in cases if "expectations" in case] + if not expectation_cases: + return None + exposures = [ + matcher + for case in expectation_cases + for matcher in _expected_event_matchers(case["expectations"], "exposures", platform) + ] + evaluations = [ + matcher + for case in expectation_cases + for matcher in _expected_event_matchers(case["expectations"], "evaluationEvents", platform) + ] + exact = len(expectation_cases) == len(cases) and all( + case["expectations"].get("noUnmatchedEvents") is True for case in expectation_cases + ) + return exposures, evaluations, exact + + def _assert_evaluation_telemetry( - test_agent: TestAgentAPI, test_library: APMLibrary, test_cases: list[dict[str, Any]] + test_agent: TestAgentAPI, + test_library: APMLibrary, + test_cases: list[dict[str, Any]], + drain_case: dict[str, Any], ) -> None: platform = context.library.name - marker_cases = [ - test_case - for test_case in test_cases - if _expected_event_matchers(test_case["expectations"], "exposures", platform) - ] - assert marker_cases, "Telemetry expectations require a case that emits an exposure drain marker" - marker_case = marker_cases[0] - marker_targeting_key = f"{marker_case['targetingKey']}-telemetry-drain-marker" + assert all(test_case["targetingKey"] != TELEMETRY_DRAIN_TARGETING_KEY for test_case in test_cases), ( + f"Evaluation cases may not use reserved targeting key {TELEMETRY_DRAIN_TARGETING_KEY}" + ) + telemetry_targeting_keys = [test_case["targetingKey"] for test_case in test_cases if "expectations" in test_case] + assert all( + sum(case["targetingKey"] == targeting_key for case in test_cases) == 1 + for targeting_key in telemetry_targeting_keys + ), "Telemetry expectations require a targeting key unique within the case file" marker_result = _ffe_evaluate_with_rc_retry( test_library, - flag=marker_case["flag"], - variation_type=marker_case["variationType"], - default_value=marker_case["defaultValue"], - targeting_key=marker_targeting_key, - attributes=marker_case.get("attributes", {}), + flag=drain_case["flag"], + variation_type=drain_case["variationType"], + default_value=drain_case["defaultValue"], + targeting_key=TELEMETRY_DRAIN_TARGETING_KEY, + attributes={}, + ) + assert marker_result.get("value") == drain_case["value"], ( + f"telemetry drain marker evaluation failed: {marker_result}" ) assert marker_result.get("reason") != "ERROR", f"telemetry drain marker evaluation failed: {marker_result}" - captured: dict[str, tuple[list[dict[str, Any]], list[dict[str, Any]]]] = {} + grouped_cases: dict[str | None, list[dict[str, Any]]] = {} + for test_case in test_cases: + grouped_cases.setdefault(test_case["targetingKey"], []).append(test_case) + captured: dict[str | None, dict[str, list[dict[str, Any]]]] = {} for _ in range(FFE_EVENT_WAIT_ATTEMPTS): requests = test_agent.requests() - captured = {} - for test_case in test_cases: - case_events = ffe_events_from_requests(requests, targeting_key=test_case["targetingKey"]) - captured[test_case["targetingKey"]] = (case_events["EXPOSURE"], case_events["EVALUATION"]) + all_events = ffe_events_from_requests(requests) + captured = {targeting_key: ffe_events_for_target(all_events, targeting_key) for targeting_key in grouped_cases} # Each writer preserves event order. Seeing the post-batch marker from both writers # proves that all telemetry from the evaluation cases has drained before exact checks. - marker_events = ffe_events_from_requests(requests, targeting_key=marker_targeting_key) + marker_events = ffe_events_for_target(all_events, TELEMETRY_DRAIN_TARGETING_KEY) complete = all(marker_events.values()) - for test_case in test_cases: - expectations = test_case["expectations"] - assert isinstance(expectations, dict) - exposures, evaluations = captured[test_case["targetingKey"]] - exact = expectations.get("noUnmatchedEvents") is True + for targeting_key, cases in grouped_cases.items(): + expected = _expected_events_for_cases(cases, platform) + if expected is None: + continue + expected_exposures, expected_evaluations, exact = expected + complete = complete and events_match(captured[targeting_key]["EXPOSURE"], expected_exposures, exact=exact) complete = complete and events_match( - exposures, _expected_event_matchers(expectations, "exposures", platform), exact=exact - ) - complete = complete and events_match( - evaluations, _expected_event_matchers(expectations, "evaluationEvents", platform), exact=exact + captured[targeting_key]["EVALUATION"], expected_evaluations, exact=exact ) if complete: break time.sleep(FFE_EVENT_WAIT_INTERVAL_SECONDS) else: - raise AssertionError(f"evaluation telemetry did not stabilize: {captured}") + raise AssertionError(f"evaluation telemetry did not stabilize: marker={marker_events}, captured={captured}") - for test_case in test_cases: - expectations = test_case["expectations"] - exposures, evaluations = captured[test_case["targetingKey"]] - exact = expectations.get("noUnmatchedEvents") is True - assert events_match(exposures, _expected_event_matchers(expectations, "exposures", platform), exact=exact), ( - f"unexpected exposures for {test_case['flag']}: {exposures}" + for targeting_key, cases in grouped_cases.items(): + expected = _expected_events_for_cases(cases, platform) + if expected is None: + continue + expected_exposures, expected_evaluations, exact = expected + events = captured[targeting_key] + assert events_match(events["EXPOSURE"], expected_exposures, exact=exact), ( + f"unexpected exposures for targeting key {targeting_key!r}: {events['EXPOSURE']}" + ) + assert events_match(events["EVALUATION"], expected_evaluations, exact=exact), ( + f"unexpected evaluation events for targeting key {targeting_key!r}: {events['EVALUATION']}" ) - assert events_match( - evaluations, _expected_event_matchers(expectations, "evaluationEvents", platform), exact=exact - ), f"unexpected evaluation events for {test_case['flag']}: {evaluations}" @scenarios.parametric @@ -315,8 +435,7 @@ def test_ffe_remote_config(self, test_agent: TestAgentAPI, test_library: APMLibr assert apply_state["product"] == RC_PRODUCT @parametrize("library_env", [{**DEFAULT_ENVVARS}]) - @parametrize("agent_env", [{"SYSTEM_TESTS_FFE_INTAKE": "true"}]) - @parametrize("test_case_file", ALL_TEST_CASE_FILES) + @parametrize("test_case_file,agent_env", CORE_CASE_PARAMETERS) def test_ffe_flag_evaluation( self, test_case_file: Path, @@ -332,19 +451,21 @@ def test_ffe_flag_evaluation( 4. Handles user targeting, attribute matching, and rollout percentages """ - _assert_evaluation_cases(test_case_file, UFC_FIXTURE_DATA, test_agent, test_library) + _assert_evaluation_cases(test_case_file, UFC_FIXTURE_PATH, test_agent, test_library) @parametrize("library_env", [{**DEFAULT_ENVVARS}]) @parametrize("agent_env", [{"SYSTEM_TESTS_FFE_INTAKE": "true"}]) - @parametrize("test_case_file", DEPENDENT_FLAGS_TEST_CASE_FILES) + @parametrize("test_case_file", DEPENDENT_FLAGS_CASE_PARAMETERS) def test_ffe_dependent_flags( self, - test_case_file: Path, + test_case_file: Path | None, test_agent: TestAgentAPI, test_library: APMLibrary, ) -> None: """Run the imported dependent-flags feature suite through the shared evaluator harness.""" - _assert_evaluation_cases(test_case_file, DEPENDENT_FLAGS_UFC_DATA, test_agent, test_library) + if test_case_file is None: + raise FileNotFoundError(f"No canonical evaluation cases found in {DEPENDENT_FLAGS_CASES_DIR}") + _assert_evaluation_cases(test_case_file, DEPENDENT_FLAGS_UFC_PATH, test_agent, test_library) @scenarios.parametric diff --git a/tests/parametric/test_ffe/utils.py b/tests/parametric/test_ffe/utils.py index bfb7243b729..c46fc55b57c 100644 --- a/tests/parametric/test_ffe/utils.py +++ b/tests/parametric/test_ffe/utils.py @@ -157,7 +157,16 @@ def assert_feature_map(actual: object, expected: dict[str, Any], destination: st assert actual_value == expected_value, f"{destination}.{key}: expected {expected_value!r}, got {actual_value!r}" -def _targeting_key_matches(actual: object, expected: str) -> bool: +class _TargetingKeyUnset: + pass + + +TARGETING_KEY_UNSET = _TargetingKeyUnset() + + +def _targeting_key_matches(actual: object, expected: str | None) -> bool: + if expected is None: + return actual is None hashed = f"sha256_{hashlib.sha256(expected.encode()).hexdigest()}" return actual in (expected, hashed) @@ -165,7 +174,7 @@ def _targeting_key_matches(actual: object, expected: str) -> bool: def ffe_events_from_requests( requests: Sequence[Mapping[str, Any]], flag: str | None = None, - targeting_key: str | None = None, + targeting_key: str | None | _TargetingKeyUnset = TARGETING_KEY_UNSET, ) -> dict[str, list[dict[str, Any]]]: """Read matching EVP payloads recorded by the parametric test agent.""" events: dict[str, list[dict[str, Any]]] = {"EXPOSURE": [], "EVALUATION": []} @@ -185,11 +194,11 @@ def ffe_events_from_requests( for event in payload[field]: if flag is not None and event.get("flag", {}).get("key") != flag: continue - if targeting_key is not None: + if not isinstance(targeting_key, _TargetingKeyUnset): if destination == "EXPOSURE": subject = event.get("subject") actual_targeting_key = subject.get("id") if isinstance(subject, dict) else None - if actual_targeting_key != targeting_key: + if not _targeting_key_matches(actual_targeting_key, targeting_key): continue elif not _targeting_key_matches(event.get("targeting_key"), targeting_key): continue @@ -197,6 +206,23 @@ def ffe_events_from_requests( return events +def ffe_events_for_target( + events: Mapping[str, Sequence[dict[str, Any]]], targeting_key: str | None +) -> dict[str, list[dict[str, Any]]]: + """Filter already decoded FFE events for one raw or hashed targeting key.""" + result: dict[str, list[dict[str, Any]]] = {"EXPOSURE": [], "EVALUATION": []} + for destination, destination_events in events.items(): + for event in destination_events: + if destination == "EXPOSURE": + subject = event.get("subject") + actual_targeting_key = subject.get("id") if isinstance(subject, dict) else None + else: + actual_targeting_key = event.get("targeting_key") + if _targeting_key_matches(actual_targeting_key, targeting_key): + result[destination].append(event) + return result + + def ffe_event_matches(event: dict[str, Any], matcher: dict[str, Any]) -> bool: """Match a canonical corpus expectation against an EVP event.""" nested_key_fields = { From 8355d0736e25023569176448a30b54f9e7e5c746 Mon Sep 17 00:00:00 2001 From: Tyler Potter Date: Tue, 3 Nov 2026 13:46:22 -0600 Subject: [PATCH 3/3] fix(ffe): isolate telemetry assertions Honor logical evaluation counts and drain readiness retries before capturing corpus telemetry. Environment: Datadog workspace Co-Authored-By: Codex GPT-5 --- .../test_ffe/test_dynamic_evaluation.py | 98 ++++++++++++++----- tests/parametric/test_ffe/utils.py | 12 ++- 2 files changed, 80 insertions(+), 30 deletions(-) diff --git a/tests/parametric/test_ffe/test_dynamic_evaluation.py b/tests/parametric/test_ffe/test_dynamic_evaluation.py index e80a01d08d5..78a187b68dd 100644 --- a/tests/parametric/test_ffe/test_dynamic_evaluation.py +++ b/tests/parametric/test_ffe/test_dynamic_evaluation.py @@ -1,10 +1,12 @@ """Test FFE (Feature Flags & Experimentation) functionality via parametric tests.""" +# ruff: noqa: UP045 - Repository guidance requires explicit Optional annotations. + import json from utils import pytest import time from pathlib import Path -from typing import Any +from typing import Any, Optional from utils import ( context, @@ -33,6 +35,7 @@ DEPENDENT_FLAGS_DIR = FFE_SYSTEM_TEST_DATA_DIR / "evaluation-suites" / "dependent-flags" DEPENDENT_FLAGS_CASES_DIR = DEPENDENT_FLAGS_DIR / "cases" TELEMETRY_DRAIN_TARGETING_KEY = "system-tests-telemetry-drain" +TELEMETRY_READINESS_TARGETING_KEY = "system-tests-telemetry-readiness" parametrize = pytest.mark.parametrize @@ -117,7 +120,7 @@ def _case_file_has_telemetry_expectations(case_file: Path) -> bool: def _set_and_wait_ffe_rc( - test_agent: TestAgentAPI, ufc_data: dict[str, Any], config_id: str | None = None + test_agent: TestAgentAPI, ufc_data: dict[str, Any], config_id: Optional[str] = None ) -> dict[str, Any]: """Set FFE Remote Config and wait for it to be acknowledged. @@ -162,8 +165,9 @@ def _ffe_evaluate_with_rc_retry( flag: str, variation_type: str, default_value: bool | str | float | dict[str, Any], - targeting_key: str | None, - attributes: dict[str, Any] | None = None, + targeting_key: Optional[str], + attributes: Optional[dict[str, Any]] = None, + retry_if_waiting: bool = True, ) -> dict[str, Any]: result = test_library.ffe_evaluate( flag=flag, @@ -172,7 +176,8 @@ def _ffe_evaluate_with_rc_retry( targeting_key=targeting_key, attributes=attributes, ) - for _ in range(FFE_READY_RETRY_ATTEMPTS - 1): + retry_attempts = FFE_READY_RETRY_ATTEMPTS - 1 if retry_if_waiting else 0 + for _ in range(retry_attempts): if not _is_ffe_waiting_for_rc(result): return result time.sleep(FFE_READY_RETRY_INTERVAL_SECONDS) @@ -209,6 +214,9 @@ def _assert_evaluation_cases( _set_and_wait_ffe_rc(test_agent, ufc_data) assert test_library.ffe_start(ufc_data), "Failed to start FFE provider" + if telemetry_cases: + assert telemetry_drain_case is not None + _prepare_telemetry_capture(test_agent, test_library, telemetry_drain_case) for index, test_case in enumerate(test_cases): expected_result = test_case["result"] @@ -219,6 +227,7 @@ def _assert_evaluation_cases( default_value=test_case["defaultValue"], targeting_key=test_case["targetingKey"], attributes=test_case.get("attributes", {}), + retry_if_waiting=not telemetry_cases, ) assert not _is_ffe_waiting_for_rc(result), ( f"Test case {index} in {test_case_file.name} failed: FFE provider did not load RC data after " @@ -324,7 +333,7 @@ def _expected_event_matchers(expectations: dict[str, Any], event_type: str, plat def _expected_events_for_cases( cases: list[dict[str, Any]], platform: str -) -> tuple[list[dict[str, Any]], list[dict[str, Any]], bool] | None: +) -> Optional[tuple[list[dict[str, Any]], list[dict[str, Any]], bool]]: expectation_cases = [case for case in cases if "expectations" in case] if not expectation_cases: return None @@ -344,6 +353,53 @@ def _expected_events_for_cases( return exposures, evaluations, exact +def _logical_evaluation_events(events: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Expand aggregated evaluation rows into the logical evaluations they represent.""" + expanded: list[dict[str, Any]] = [] + for event in events: + count = event.get("evaluation_count", 1) + assert type(count) is int, f"evaluation_count must be an integer: {count!r}" + assert count >= 1, f"evaluation_count must be positive: {count!r}" + expanded.extend([event] * count) + return expanded + + +def _evaluate_telemetry_marker( + test_library: APMLibrary, + drain_case: dict[str, Any], + targeting_key: str, +) -> dict[str, Any]: + result = _ffe_evaluate_with_rc_retry( + test_library, + flag=drain_case["flag"], + variation_type=drain_case["variationType"], + default_value=drain_case["defaultValue"], + targeting_key=targeting_key, + attributes={}, + ) + assert result.get("value") == drain_case["value"], f"telemetry marker evaluation failed: {result}" + assert result.get("reason") != "ERROR", f"telemetry marker evaluation failed: {result}" + return result + + +def _prepare_telemetry_capture( + test_agent: TestAgentAPI, + test_library: APMLibrary, + drain_case: dict[str, Any], +) -> None: + """Establish provider readiness, drain retry telemetry, then start with an empty capture.""" + _evaluate_telemetry_marker(test_library, drain_case, TELEMETRY_READINESS_TARGETING_KEY) + for _ in range(FFE_EVENT_WAIT_ATTEMPTS): + events = ffe_events_for_target( + ffe_events_from_requests(test_agent.requests()), TELEMETRY_READINESS_TARGETING_KEY + ) + if all(events.values()): + test_agent.clear() + return + time.sleep(FFE_EVENT_WAIT_INTERVAL_SECONDS) + raise AssertionError(f"telemetry readiness marker did not drain: {events}") + + def _assert_evaluation_telemetry( test_agent: TestAgentAPI, test_library: APMLibrary, @@ -351,31 +407,21 @@ def _assert_evaluation_telemetry( drain_case: dict[str, Any], ) -> None: platform = context.library.name - assert all(test_case["targetingKey"] != TELEMETRY_DRAIN_TARGETING_KEY for test_case in test_cases), ( - f"Evaluation cases may not use reserved targeting key {TELEMETRY_DRAIN_TARGETING_KEY}" + reserved_targeting_keys = {TELEMETRY_DRAIN_TARGETING_KEY, TELEMETRY_READINESS_TARGETING_KEY} + assert all(test_case["targetingKey"] not in reserved_targeting_keys for test_case in test_cases), ( + f"Evaluation cases may not use reserved targeting keys {sorted(reserved_targeting_keys)}" ) telemetry_targeting_keys = [test_case["targetingKey"] for test_case in test_cases if "expectations" in test_case] assert all( sum(case["targetingKey"] == targeting_key for case in test_cases) == 1 for targeting_key in telemetry_targeting_keys ), "Telemetry expectations require a targeting key unique within the case file" - marker_result = _ffe_evaluate_with_rc_retry( - test_library, - flag=drain_case["flag"], - variation_type=drain_case["variationType"], - default_value=drain_case["defaultValue"], - targeting_key=TELEMETRY_DRAIN_TARGETING_KEY, - attributes={}, - ) - assert marker_result.get("value") == drain_case["value"], ( - f"telemetry drain marker evaluation failed: {marker_result}" - ) - assert marker_result.get("reason") != "ERROR", f"telemetry drain marker evaluation failed: {marker_result}" + _evaluate_telemetry_marker(test_library, drain_case, TELEMETRY_DRAIN_TARGETING_KEY) - grouped_cases: dict[str | None, list[dict[str, Any]]] = {} + grouped_cases: dict[Optional[str], list[dict[str, Any]]] = {} for test_case in test_cases: grouped_cases.setdefault(test_case["targetingKey"], []).append(test_case) - captured: dict[str | None, dict[str, list[dict[str, Any]]]] = {} + captured: dict[Optional[str], dict[str, list[dict[str, Any]]]] = {} for _ in range(FFE_EVENT_WAIT_ATTEMPTS): requests = test_agent.requests() @@ -393,7 +439,9 @@ def _assert_evaluation_telemetry( expected_exposures, expected_evaluations, exact = expected complete = complete and events_match(captured[targeting_key]["EXPOSURE"], expected_exposures, exact=exact) complete = complete and events_match( - captured[targeting_key]["EVALUATION"], expected_evaluations, exact=exact + _logical_evaluation_events(captured[targeting_key]["EVALUATION"]), + expected_evaluations, + exact=exact, ) if complete: @@ -411,7 +459,7 @@ def _assert_evaluation_telemetry( assert events_match(events["EXPOSURE"], expected_exposures, exact=exact), ( f"unexpected exposures for targeting key {targeting_key!r}: {events['EXPOSURE']}" ) - assert events_match(events["EVALUATION"], expected_evaluations, exact=exact), ( + assert events_match(_logical_evaluation_events(events["EVALUATION"]), expected_evaluations, exact=exact), ( f"unexpected evaluation events for targeting key {targeting_key!r}: {events['EVALUATION']}" ) @@ -458,7 +506,7 @@ def test_ffe_flag_evaluation( @parametrize("test_case_file", DEPENDENT_FLAGS_CASE_PARAMETERS) def test_ffe_dependent_flags( self, - test_case_file: Path | None, + test_case_file: Optional[Path], test_agent: TestAgentAPI, test_library: APMLibrary, ) -> None: diff --git a/tests/parametric/test_ffe/utils.py b/tests/parametric/test_ffe/utils.py index c46fc55b57c..8eb9a0a095b 100644 --- a/tests/parametric/test_ffe/utils.py +++ b/tests/parametric/test_ffe/utils.py @@ -9,13 +9,15 @@ Example: [100, 108, 128, 130] -> deltas [100, 8, 20, 2] -> varint bytes -> base64 """ +# ruff: noqa: UP007, UP045 - Repository guidance requires explicit Optional annotations. + from collections.abc import Mapping, Sequence import base64 import gzip import hashlib import json from pathlib import Path -from typing import Any +from typing import Any, Optional, Union # Varint encoding constants VARINT_DATA_BITS = 0x7F # Lower 7 bits contain data @@ -164,7 +166,7 @@ class _TargetingKeyUnset: TARGETING_KEY_UNSET = _TargetingKeyUnset() -def _targeting_key_matches(actual: object, expected: str | None) -> bool: +def _targeting_key_matches(actual: object, expected: Optional[str]) -> bool: if expected is None: return actual is None hashed = f"sha256_{hashlib.sha256(expected.encode()).hexdigest()}" @@ -173,8 +175,8 @@ def _targeting_key_matches(actual: object, expected: str | None) -> bool: def ffe_events_from_requests( requests: Sequence[Mapping[str, Any]], - flag: str | None = None, - targeting_key: str | None | _TargetingKeyUnset = TARGETING_KEY_UNSET, + flag: Optional[str] = None, + targeting_key: Optional[Union[str, _TargetingKeyUnset]] = TARGETING_KEY_UNSET, ) -> dict[str, list[dict[str, Any]]]: """Read matching EVP payloads recorded by the parametric test agent.""" events: dict[str, list[dict[str, Any]]] = {"EXPOSURE": [], "EVALUATION": []} @@ -207,7 +209,7 @@ def ffe_events_from_requests( def ffe_events_for_target( - events: Mapping[str, Sequence[dict[str, Any]]], targeting_key: str | None + events: Mapping[str, Sequence[dict[str, Any]]], targeting_key: Optional[str] ) -> dict[str, list[dict[str, Any]]]: """Filter already decoded FFE events for one raw or hashed targeting key.""" result: dict[str, list[dict[str, Any]]] = {"EXPOSURE": [], "EVALUATION": []}