From 0b907623e06106a980ce82fb38dab4245e136af9 Mon Sep 17 00:00:00 2001 From: Brian Marks Date: Wed, 5 Aug 2026 10:45:55 -0400 Subject: [PATCH 1/2] fix(parametric): wait for post-restart telemetry --- tests/parametric/conftest.py | 44 ++++-- tests/parametric/test_config_consistency.py | 36 +++-- tests/test_the_test/test_test_agent.py | 162 ++++++++++++++++++++ utils/docker_fixtures/_test_agent.py | 94 +++++++----- 4 files changed, 273 insertions(+), 63 deletions(-) create mode 100644 tests/test_the_test/test_test_agent.py diff --git a/tests/parametric/conftest.py b/tests/parametric/conftest.py index 0c8d37874a2..f00305e06c8 100644 --- a/tests/parametric/conftest.py +++ b/tests/parametric/conftest.py @@ -1,9 +1,10 @@ import base64 -from collections.abc import Generator +from collections.abc import Generator, Mapping import json from pathlib import Path import shutil import subprocess +from typing import Any import uuid import pytest @@ -188,21 +189,42 @@ def nodejs_telemetry_value(test_agent: TestAgentAPI, dd_key: str) -> str | int | return entries[0].get("value") -def assert_nodejs_telemetry_config(test_agent: TestAgentAPI, expected: dict) -> None: - """Assert expected dd_* config values against the nodejs telemetry configuration.""" - configuration_by_name = test_agent.wait_for_telemetry_configurations() +def _nodejs_telemetry_config_matches( + configurations: dict[str, list[dict[str, Any]]], expected: Mapping[str, object] +) -> bool: for dd_key, expected_value in expected.items(): - name = _telemetry_name(dd_key) - entries = configuration_by_name.get(name) - assert entries, f"No telemetry configuration '{name}'" + entries = configurations.get(_telemetry_name(dd_key)) + if not entries: + return False actual = entries[0].get("value") if dd_key == "dd_tags": actual_tags = "" if actual is None else str(actual) expected_tags = expected_value if isinstance(expected_value, list) else str(expected_value).split(",") - for tag in expected_tags: - assert tag in actual_tags, f"Expected tag '{tag}' not found in telemetry tags: {actual_tags}" - else: - assert str(actual).lower() == str(expected_value).lower(), f"Expected {name}={expected_value}, got {actual}" + if any(tag not in actual_tags for tag in expected_tags): + return False + elif str(actual).lower() != str(expected_value).lower(): + return False + return True + + +def assert_nodejs_telemetry_config( + test_agent: TestAgentAPI, expected: Mapping[str, object], *, runtime_id: str | None = None +) -> None: + """Assert expected dd_* config values against the nodejs telemetry configuration.""" + if runtime_id is None: + runtime_id = test_agent.wait_for_telemetry_runtime_id() + test_agent.wait_for_telemetry_configurations( + runtime_id=runtime_id, + validator=lambda configurations: _nodejs_telemetry_config_matches(configurations, expected), + ) + + +def restart_and_get_runtime_id(test_agent: TestAgentAPI, test_library: APMLibrary) -> str | None: + previous_runtime_id = test_agent.wait_for_telemetry_runtime_id() if test_library.lang == "nodejs" else None + test_library.container_restart() + if previous_runtime_id is None: + return None + return test_agent.wait_for_telemetry_runtime_id(exclude={previous_runtime_id}) def nodejs_startup_config(test_library: APMLibrary) -> dict: diff --git a/tests/parametric/test_config_consistency.py b/tests/parametric/test_config_consistency.py index 1868052ea2a..ac8823c4f7e 100644 --- a/tests/parametric/test_config_consistency.py +++ b/tests/parametric/test_config_consistency.py @@ -13,7 +13,13 @@ ) from utils.docker_fixtures.spec.trace import find_span_in_traces, find_only_span from utils.docker_fixtures import TestAgentAPI -from .conftest import APMLibrary, StableConfigWriter, assert_nodejs_telemetry_config, nodejs_startup_config +from .conftest import ( + APMLibrary, + StableConfigWriter, + assert_nodejs_telemetry_config, + nodejs_startup_config, + restart_and_get_runtime_id, +) parametrize = pytest.mark.parametrize @@ -537,9 +543,9 @@ def test_default_config( path, test_library, ) - test_library.container_restart() + runtime_id = restart_and_get_runtime_id(test_agent, test_library) if test_library.lang == "nodejs": - assert_nodejs_telemetry_config(test_agent, expected) + assert_nodejs_telemetry_config(test_agent, expected, runtime_id=runtime_id) else: config = test_library.config() assert expected.items() <= config.items() @@ -597,9 +603,9 @@ def test_extended_configs( path, test_library, ) - test_library.container_restart() + runtime_id = restart_and_get_runtime_id(test_agent, test_library) if test_library.lang == "nodejs": - assert_nodejs_telemetry_config(test_agent, expected) + assert_nodejs_telemetry_config(test_agent, expected, runtime_id=runtime_id) return config = test_library.config() @@ -656,9 +662,9 @@ def test_unknown_key_skipped(self, test_agent: TestAgentAPI, test_library: APMLi path, test_library, ) - test_library.container_restart() + runtime_id = restart_and_get_runtime_id(test_agent, test_library) if test_library.lang == "nodejs": - assert_nodejs_telemetry_config(test_agent, test["expected"]) + assert_nodejs_telemetry_config(test_agent, test["expected"], runtime_id=runtime_id) else: config = test_library.config() assert test["expected"].items() <= config.items() @@ -677,9 +683,9 @@ def test_invalid_files(self, test_agent: TestAgentAPI, test_library: APMLibrary, path, test_library, ) - test_library.container_restart() + runtime_id = restart_and_get_runtime_id(test_agent, test_library) if test_library.lang == "nodejs": - assert_nodejs_telemetry_config(test_agent, SDK_DEFAULT_STABLE_CONFIG) + assert_nodejs_telemetry_config(test_agent, SDK_DEFAULT_STABLE_CONFIG, runtime_id=runtime_id) else: config = test_library.config() assert SDK_DEFAULT_STABLE_CONFIG.items() <= config.items() @@ -750,9 +756,9 @@ def test_config_precedence( test_library, ) - test_library.container_restart() + runtime_id = restart_and_get_runtime_id(test_agent, test_library) if test_library.lang == "nodejs": - assert_nodejs_telemetry_config(test_agent, expected) + assert_nodejs_telemetry_config(test_agent, expected, runtime_id=runtime_id) return config = test_library.config() @@ -790,9 +796,9 @@ def test_targeting_rules(self, test_agent: TestAgentAPI, test_library: APMLibrar path, test_library, ) - test_library.container_restart() + runtime_id = restart_and_get_runtime_id(test_agent, test_library) if test_library.lang == "nodejs": - assert_nodejs_telemetry_config(test_agent, {"dd_service": "my-service"}) + assert_nodejs_telemetry_config(test_agent, {"dd_service": "my-service"}, runtime_id=runtime_id) else: config = test_library.config() assert config["dd_service"] == "my-service", ( @@ -826,9 +832,9 @@ def test_process_arguments(self, test_agent: TestAgentAPI, test_library: APMLibr # Use custom dumper for this specific test stable_config_content = yaml.dump(config, Dumper=CustomDumper) self.write_stable_config_content(stable_config_content, path, test_library) - test_library.container_restart() + runtime_id = restart_and_get_runtime_id(test_agent, test_library) if test_library.lang == "nodejs": - assert_nodejs_telemetry_config(test_agent, {"dd_service": "value"}) + assert_nodejs_telemetry_config(test_agent, {"dd_service": "value"}, runtime_id=runtime_id) else: lib_config = test_library.config() assert lib_config["dd_service"] == "value", ( diff --git a/tests/test_the_test/test_test_agent.py b/tests/test_the_test/test_test_agent.py new file mode 100644 index 00000000000..1f6c3667865 --- /dev/null +++ b/tests/test_the_test/test_test_agent.py @@ -0,0 +1,162 @@ +from collections.abc import Callable + +import pytest + +from tests.parametric.conftest import assert_nodejs_telemetry_config +from utils import scenarios +from utils.docker_fixtures import TestAgentAPI + + +def _configuration_event( + *, runtime_id: str, tracer_time: int, configurations: list[dict[str, object]] +) -> dict[str, object]: + return { + "request_type": "app-started", + "runtime_id": runtime_id, + "tracer_time": tracer_time, + "application": { + "language_name": "nodejs", + "language_version": "24.4.1", + "service_name": "parametric", + "tracer_version": "5.62.0", + }, + "payload": {"configuration": configurations}, + "seq_id": 1, + } + + +def _config(name: str, value: str, *, seq_id: int = 0) -> dict[str, object]: + return {"name": name, "origin": "local_stable_config", "seq_id": seq_id, "value": value} + + +def _telemetry(events: list[dict[str, object]]) -> Callable[..., list[dict[str, object]]]: + def get_events(*, clear: bool = False) -> list[dict[str, object]]: + assert clear is False + return events + + return get_events + + +@scenarios.test_the_test +def test_nodejs_telemetry_assertion_waits_for_post_restart_runtime(monkeypatch: pytest.MonkeyPatch) -> None: + stale_event = _configuration_event( + runtime_id="before-restart", + tracer_time=1, + configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "datadog,tracecontext,baggage")], + ) + current_event = _configuration_event( + runtime_id="after-restart", + tracer_time=2, + configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "tracecontext")], + ) + telemetry_responses = [[stale_event], [stale_event, current_event]] + + test_agent = object.__new__(TestAgentAPI) + + def telemetry(*, clear: bool = False) -> list[dict[str, object]]: + assert clear is False + if len(telemetry_responses) > 1: + return telemetry_responses.pop(0) + return telemetry_responses[0] + + monkeypatch.setattr(test_agent, "telemetry", telemetry) + + assert_nodejs_telemetry_config( + test_agent, + {"dd_trace_propagation_style": "tracecontext"}, + runtime_id="after-restart", + ) + + +@scenarios.test_the_test +def test_nodejs_telemetry_assertion_rejects_stale_matching_value(monkeypatch: pytest.MonkeyPatch) -> None: + events = [ + _configuration_event( + runtime_id="before-restart", + tracer_time=1, + configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "tracecontext")], + ), + _configuration_event( + runtime_id="after-restart", + tracer_time=2, + configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "datadog")], + ), + ] + test_agent = object.__new__(TestAgentAPI) + monkeypatch.setattr(test_agent, "telemetry", _telemetry(events)) + monkeypatch.setattr("utils.docker_fixtures._test_agent.time.sleep", lambda _seconds: None) + + with pytest.raises(AssertionError): + assert_nodejs_telemetry_config( + test_agent, + {"dd_trace_propagation_style": "tracecontext"}, + runtime_id="after-restart", + ) + + +@scenarios.test_the_test +def test_nodejs_telemetry_assertion_requires_one_runtime_to_match_all_values( + monkeypatch: pytest.MonkeyPatch, +) -> None: + events = [ + _configuration_event( + runtime_id="before-restart", + tracer_time=1, + configurations=[_config("DD_SERVICE", "expected"), _config("DD_ENV", "wrong")], + ), + _configuration_event( + runtime_id="after-restart", + tracer_time=2, + configurations=[_config("DD_SERVICE", "wrong"), _config("DD_ENV", "expected")], + ), + ] + test_agent = object.__new__(TestAgentAPI) + monkeypatch.setattr(test_agent, "telemetry", _telemetry(events)) + monkeypatch.setattr("utils.docker_fixtures._test_agent.time.sleep", lambda _seconds: None) + + with pytest.raises(AssertionError): + assert_nodejs_telemetry_config( + test_agent, + {"dd_service": "expected", "dd_env": "expected"}, + runtime_id="after-restart", + ) + + +@scenarios.test_the_test +def test_nodejs_telemetry_assertion_uses_latest_configuration_sequence(monkeypatch: pytest.MonkeyPatch) -> None: + events = [ + _configuration_event( + runtime_id="after-restart", + tracer_time=2, + configurations=[ + _config("DD_SERVICE", "expected", seq_id=0), + _config("DD_SERVICE", "wrong", seq_id=1), + ], + ) + ] + test_agent = object.__new__(TestAgentAPI) + monkeypatch.setattr(test_agent, "telemetry", _telemetry(events)) + monkeypatch.setattr("utils.docker_fixtures._test_agent.time.sleep", lambda _seconds: None) + + with pytest.raises(AssertionError): + assert_nodejs_telemetry_config(test_agent, {"dd_service": "expected"}, runtime_id="after-restart") + + +@scenarios.test_the_test +def test_wait_for_telemetry_runtime_id_ignores_excluded_runtime(monkeypatch: pytest.MonkeyPatch) -> None: + stale_event = _configuration_event(runtime_id="before-restart", tracer_time=1, configurations=[]) + current_event = _configuration_event(runtime_id="after-restart", tracer_time=2, configurations=[]) + telemetry_responses = [[stale_event], [stale_event, current_event]] + test_agent = object.__new__(TestAgentAPI) + + def telemetry(*, clear: bool = False) -> list[dict[str, object]]: + assert clear is False + if len(telemetry_responses) > 1: + return telemetry_responses.pop(0) + return telemetry_responses[0] + + monkeypatch.setattr(test_agent, "telemetry", telemetry) + + runtime_id = test_agent.wait_for_telemetry_runtime_id(exclude={"before-restart"}) + + assert runtime_id == "after-restart" diff --git a/utils/docker_fixtures/_test_agent.py b/utils/docker_fixtures/_test_agent.py index c6c89355e8b..c1c3b673063 100644 --- a/utils/docker_fixtures/_test_agent.py +++ b/utils/docker_fixtures/_test_agent.py @@ -1,5 +1,5 @@ import base64 -from collections.abc import Generator +from collections.abc import Callable, Generator import contextlib import datetime import gzip @@ -705,56 +705,76 @@ def wait_for_telemetry_event(self, event_name: str, *, clear: bool = False, wait raise AssertionError(f"Telemetry event {event_name} not found") def wait_for_telemetry_configurations( - self, *, service: str | None = None, clear: bool = False, wait_loops: int = 100 - ) -> dict[str, list[dict]]: + self, + *, + service: str | None = None, + runtime_id: str | None = None, + validator: Callable[[dict[str, list[dict[str, Any]]]], bool] | None = None, + clear: bool = False, + wait_loops: int = 100, + ) -> dict[str, list[dict[str, Any]]]: """Waits for and returns configurations captured in telemetry events. Telemetry events can be found in `app-started` or `app-client-configuration-change` events. - The function ensures that at least one telemetry event is captured before processing. + When provided, runtime_id filters events and validator defines the readiness condition. Returns a dictionary where keys are configuration names and values are lists of configuration dictionaries, allowing for multiple entries per configuration name with different origins. """ - events = [] - configurations: dict[str, list[dict]] = {} - # Poll until telemetry is captured instead of sleeping a fixed delay: returns as soon as - # app-started arrives (usually within a heartbeat) and retries the empty-read window that - # a single fixed-delay read can land in. + events: list[dict[str, Any]] = [] + configurations: dict[str, list[dict[str, Any]]] = {} for _ in range(wait_loops): with contextlib.suppress(requests.exceptions.RequestException): events = self.telemetry(clear=False) if events: - break + configurations = {} + events.sort(key=lambda event: event["tracer_time"]) + for event in events: + if runtime_id is not None and event.get("runtime_id") != runtime_id: + continue + if service is not None and event["application"]["service_name"] != service: + continue + for event_type in ["app-started", "app-client-configuration-change"]: + telemetry_event = self._get_telemetry_event(event, event_type) + if telemetry_event: + for config in telemetry_event.get("payload", {}).get("configuration", []): + configurations.setdefault(config["name"], []).append(config) + for payload in configurations.values(): + payload.sort(key=lambda item: item.get("seq_id") or 0, reverse=True) + if validator is None or validator(configurations): + if clear: + self.clear() + return configurations time.sleep(0.05) - else: - raise AssertionError("No telemetry events were found. Ensure the application is sending telemetry events.") - # Sort events by tracer_time to ensure configurations are processed in order - events.sort(key=lambda r: r["tracer_time"]) + if not events: + raise AssertionError("No telemetry events were found. Ensure the application is sending telemetry events.") + raise AssertionError(f"No telemetry configurations matched the validator. Last observed: {configurations}") - # Extract configuration data from relevant telemetry events - for event in events: - if service is not None and event["application"]["service_name"] != service: - continue - for event_type in ["app-started", "app-client-configuration-change"]: - telemetry_event = self._get_telemetry_event(event, event_type) - if telemetry_event: - for config in telemetry_event.get("payload", {}).get("configuration", []): - # Store all configurations, allowing multiple entries per name with different origins - config_name = config["name"] - if config_name not in configurations: - configurations[config_name] = [] - configurations[config_name].append(config) - if configurations: - # Checking if we need to sort due to multiple sources being sent for the same config - sample_key = next(iter(configurations)) - if "seq_id" in configurations[sample_key][0] and configurations[sample_key][0]["seq_id"] is not None: - # Sort seq_id for each config from highest to lowest - for payload in configurations.values(): - payload.sort(key=lambda item: item["seq_id"], reverse=True) - if clear: - self.clear() - return configurations + def wait_for_telemetry_runtime_id(self, *, exclude: set[str] | None = None, wait_loops: int = 200) -> str: + excluded_runtime_ids = exclude or set() + observed_runtime_ids: set[str] = set() + for _ in range(wait_loops): + try: + events = self.telemetry(clear=False) + except requests.exceptions.RequestException: + events = [] + else: + events.sort(key=lambda event: event["tracer_time"], reverse=True) + for event in events: + telemetry_event = self._get_telemetry_event(event, "app-started") + if telemetry_event is None: + continue + runtime_id = telemetry_event.get("runtime_id") or event.get("runtime_id") + if runtime_id is None: + continue + observed_runtime_ids.add(runtime_id) + if runtime_id not in excluded_runtime_ids: + return runtime_id + time.sleep(0.01) + raise AssertionError( + f"No telemetry runtime ID found excluding {excluded_runtime_ids}. Observed: {observed_runtime_ids}" + ) def get_telemetry_config_by_origin( self, From 98cfc5823617766bcfafdee57b8f69a704e0dba0 Mon Sep 17 00:00:00 2001 From: Brian Marks Date: Thu, 6 Aug 2026 15:30:17 -0400 Subject: [PATCH 2/2] refactor(parametric): simplify telemetry readiness --- tests/parametric/conftest.py | 40 +++---- tests/test_the_test/test_test_agent.py | 160 ++++++++----------------- utils/docker_fixtures/_test_agent.py | 79 ++++++------ 3 files changed, 98 insertions(+), 181 deletions(-) diff --git a/tests/parametric/conftest.py b/tests/parametric/conftest.py index f00305e06c8..4aa203eeca1 100644 --- a/tests/parametric/conftest.py +++ b/tests/parametric/conftest.py @@ -1,10 +1,9 @@ import base64 -from collections.abc import Generator, Mapping +from collections.abc import Generator import json from pathlib import Path import shutil import subprocess -from typing import Any import uuid import pytest @@ -189,34 +188,23 @@ def nodejs_telemetry_value(test_agent: TestAgentAPI, dd_key: str) -> str | int | return entries[0].get("value") -def _nodejs_telemetry_config_matches( - configurations: dict[str, list[dict[str, Any]]], expected: Mapping[str, object] -) -> bool: +def assert_nodejs_telemetry_config(test_agent: TestAgentAPI, expected: dict, *, runtime_id: str | None = None) -> None: + """Assert expected dd_* config values against the nodejs telemetry configuration.""" + if runtime_id is None: + runtime_id = test_agent.wait_for_telemetry_runtime_id() + configurations = test_agent.wait_for_telemetry_configurations(runtime_id=runtime_id) for dd_key, expected_value in expected.items(): - entries = configurations.get(_telemetry_name(dd_key)) - if not entries: - return False + name = _telemetry_name(dd_key) + entries = configurations.get(name) + assert entries, f"No telemetry configuration '{name}' for runtime '{runtime_id}'" actual = entries[0].get("value") if dd_key == "dd_tags": actual_tags = "" if actual is None else str(actual) expected_tags = expected_value if isinstance(expected_value, list) else str(expected_value).split(",") - if any(tag not in actual_tags for tag in expected_tags): - return False - elif str(actual).lower() != str(expected_value).lower(): - return False - return True - - -def assert_nodejs_telemetry_config( - test_agent: TestAgentAPI, expected: Mapping[str, object], *, runtime_id: str | None = None -) -> None: - """Assert expected dd_* config values against the nodejs telemetry configuration.""" - if runtime_id is None: - runtime_id = test_agent.wait_for_telemetry_runtime_id() - test_agent.wait_for_telemetry_configurations( - runtime_id=runtime_id, - validator=lambda configurations: _nodejs_telemetry_config_matches(configurations, expected), - ) + for tag in expected_tags: + assert tag in actual_tags, f"Expected tag '{tag}' not found in telemetry tags: {actual_tags}" + else: + assert str(actual).lower() == str(expected_value).lower(), f"Expected {name}={expected_value}, got {actual}" def restart_and_get_runtime_id(test_agent: TestAgentAPI, test_library: APMLibrary) -> str | None: @@ -224,7 +212,7 @@ def restart_and_get_runtime_id(test_agent: TestAgentAPI, test_library: APMLibrar test_library.container_restart() if previous_runtime_id is None: return None - return test_agent.wait_for_telemetry_runtime_id(exclude={previous_runtime_id}) + return test_agent.wait_for_telemetry_runtime_id(exclude=previous_runtime_id) def nodejs_startup_config(test_library: APMLibrary) -> dict: diff --git a/tests/test_the_test/test_test_agent.py b/tests/test_the_test/test_test_agent.py index 1f6c3667865..35f970829fd 100644 --- a/tests/test_the_test/test_test_agent.py +++ b/tests/test_the_test/test_test_agent.py @@ -1,5 +1,3 @@ -from collections.abc import Callable - import pytest from tests.parametric.conftest import assert_nodejs_telemetry_config @@ -14,27 +12,24 @@ def _configuration_event( "request_type": "app-started", "runtime_id": runtime_id, "tracer_time": tracer_time, - "application": { - "language_name": "nodejs", - "language_version": "24.4.1", - "service_name": "parametric", - "tracer_version": "5.62.0", - }, "payload": {"configuration": configurations}, - "seq_id": 1, } def _config(name: str, value: str, *, seq_id: int = 0) -> dict[str, object]: - return {"name": name, "origin": "local_stable_config", "seq_id": seq_id, "value": value} + return {"name": name, "seq_id": seq_id, "value": value} + +def _test_agent(monkeypatch: pytest.MonkeyPatch, responses: list[list[dict[str, object]]]) -> TestAgentAPI: + test_agent = object.__new__(TestAgentAPI) -def _telemetry(events: list[dict[str, object]]) -> Callable[..., list[dict[str, object]]]: - def get_events(*, clear: bool = False) -> list[dict[str, object]]: + def telemetry(*, clear: bool = False) -> list[dict[str, object]]: assert clear is False - return events + return responses.pop(0) if len(responses) > 1 else responses[0] - return get_events + monkeypatch.setattr(test_agent, "telemetry", telemetry) + monkeypatch.setattr("utils.docker_fixtures._test_agent.time.sleep", lambda _seconds: None) + return test_agent @scenarios.test_the_test @@ -49,114 +44,57 @@ def test_nodejs_telemetry_assertion_waits_for_post_restart_runtime(monkeypatch: tracer_time=2, configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "tracecontext")], ) - telemetry_responses = [[stale_event], [stale_event, current_event]] - - test_agent = object.__new__(TestAgentAPI) - - def telemetry(*, clear: bool = False) -> list[dict[str, object]]: - assert clear is False - if len(telemetry_responses) > 1: - return telemetry_responses.pop(0) - return telemetry_responses[0] - - monkeypatch.setattr(test_agent, "telemetry", telemetry) + test_agent = _test_agent(monkeypatch, [[stale_event], [stale_event, current_event]]) + runtime_id = test_agent.wait_for_telemetry_runtime_id(exclude="before-restart") assert_nodejs_telemetry_config( test_agent, {"dd_trace_propagation_style": "tracecontext"}, - runtime_id="after-restart", + runtime_id=runtime_id, ) @scenarios.test_the_test -def test_nodejs_telemetry_assertion_rejects_stale_matching_value(monkeypatch: pytest.MonkeyPatch) -> None: - events = [ - _configuration_event( - runtime_id="before-restart", - tracer_time=1, - configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "tracecontext")], +@pytest.mark.parametrize( + ("events", "expected"), + [ + ( + [ + _configuration_event( + runtime_id="before-restart", + tracer_time=1, + configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "tracecontext")], + ), + _configuration_event( + runtime_id="after-restart", + tracer_time=2, + configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "datadog")], + ), + ], + {"dd_trace_propagation_style": "tracecontext"}, ), - _configuration_event( - runtime_id="after-restart", - tracer_time=2, - configurations=[_config("DD_TRACE_PROPAGATION_STYLE", "datadog")], + ( + [ + _configuration_event( + runtime_id="after-restart", + tracer_time=2, + configurations=[ + _config("DD_SERVICE", "expected", seq_id=0), + _config("DD_SERVICE", "wrong", seq_id=1), + ], + ) + ], + {"dd_service": "expected"}, ), - ] - test_agent = object.__new__(TestAgentAPI) - monkeypatch.setattr(test_agent, "telemetry", _telemetry(events)) - monkeypatch.setattr("utils.docker_fixtures._test_agent.time.sleep", lambda _seconds: None) - - with pytest.raises(AssertionError): - assert_nodejs_telemetry_config( - test_agent, - {"dd_trace_propagation_style": "tracecontext"}, - runtime_id="after-restart", - ) - - -@scenarios.test_the_test -def test_nodejs_telemetry_assertion_requires_one_runtime_to_match_all_values( + ], + ids=["stale-match", "superseded-sequence"], +) +def test_nodejs_telemetry_assertion_rejects_invalid_runtime_snapshot( monkeypatch: pytest.MonkeyPatch, + events: list[dict[str, object]], + expected: dict[str, object], ) -> None: - events = [ - _configuration_event( - runtime_id="before-restart", - tracer_time=1, - configurations=[_config("DD_SERVICE", "expected"), _config("DD_ENV", "wrong")], - ), - _configuration_event( - runtime_id="after-restart", - tracer_time=2, - configurations=[_config("DD_SERVICE", "wrong"), _config("DD_ENV", "expected")], - ), - ] - test_agent = object.__new__(TestAgentAPI) - monkeypatch.setattr(test_agent, "telemetry", _telemetry(events)) - monkeypatch.setattr("utils.docker_fixtures._test_agent.time.sleep", lambda _seconds: None) - - with pytest.raises(AssertionError): - assert_nodejs_telemetry_config( - test_agent, - {"dd_service": "expected", "dd_env": "expected"}, - runtime_id="after-restart", - ) - - -@scenarios.test_the_test -def test_nodejs_telemetry_assertion_uses_latest_configuration_sequence(monkeypatch: pytest.MonkeyPatch) -> None: - events = [ - _configuration_event( - runtime_id="after-restart", - tracer_time=2, - configurations=[ - _config("DD_SERVICE", "expected", seq_id=0), - _config("DD_SERVICE", "wrong", seq_id=1), - ], - ) - ] - test_agent = object.__new__(TestAgentAPI) - monkeypatch.setattr(test_agent, "telemetry", _telemetry(events)) - monkeypatch.setattr("utils.docker_fixtures._test_agent.time.sleep", lambda _seconds: None) + test_agent = _test_agent(monkeypatch, [events]) with pytest.raises(AssertionError): - assert_nodejs_telemetry_config(test_agent, {"dd_service": "expected"}, runtime_id="after-restart") - - -@scenarios.test_the_test -def test_wait_for_telemetry_runtime_id_ignores_excluded_runtime(monkeypatch: pytest.MonkeyPatch) -> None: - stale_event = _configuration_event(runtime_id="before-restart", tracer_time=1, configurations=[]) - current_event = _configuration_event(runtime_id="after-restart", tracer_time=2, configurations=[]) - telemetry_responses = [[stale_event], [stale_event, current_event]] - test_agent = object.__new__(TestAgentAPI) - - def telemetry(*, clear: bool = False) -> list[dict[str, object]]: - assert clear is False - if len(telemetry_responses) > 1: - return telemetry_responses.pop(0) - return telemetry_responses[0] - - monkeypatch.setattr(test_agent, "telemetry", telemetry) - - runtime_id = test_agent.wait_for_telemetry_runtime_id(exclude={"before-restart"}) - - assert runtime_id == "after-restart" + assert_nodejs_telemetry_config(test_agent, expected, runtime_id="after-restart") diff --git a/utils/docker_fixtures/_test_agent.py b/utils/docker_fixtures/_test_agent.py index c1c3b673063..35c15dab8fb 100644 --- a/utils/docker_fixtures/_test_agent.py +++ b/utils/docker_fixtures/_test_agent.py @@ -1,5 +1,5 @@ import base64 -from collections.abc import Callable, Generator +from collections.abc import Generator import contextlib import datetime import gzip @@ -709,14 +709,12 @@ def wait_for_telemetry_configurations( *, service: str | None = None, runtime_id: str | None = None, - validator: Callable[[dict[str, list[dict[str, Any]]]], bool] | None = None, clear: bool = False, wait_loops: int = 100, ) -> dict[str, list[dict[str, Any]]]: """Waits for and returns configurations captured in telemetry events. Telemetry events can be found in `app-started` or `app-client-configuration-change` events. - When provided, runtime_id filters events and validator defines the readiness condition. Returns a dictionary where keys are configuration names and values are lists of configuration dictionaries, allowing for multiple entries per configuration name with different origins. @@ -727,54 +725,47 @@ def wait_for_telemetry_configurations( with contextlib.suppress(requests.exceptions.RequestException): events = self.telemetry(clear=False) if events: - configurations = {} - events.sort(key=lambda event: event["tracer_time"]) - for event in events: - if runtime_id is not None and event.get("runtime_id") != runtime_id: - continue - if service is not None and event["application"]["service_name"] != service: - continue - for event_type in ["app-started", "app-client-configuration-change"]: - telemetry_event = self._get_telemetry_event(event, event_type) - if telemetry_event: - for config in telemetry_event.get("payload", {}).get("configuration", []): - configurations.setdefault(config["name"], []).append(config) - for payload in configurations.values(): - payload.sort(key=lambda item: item.get("seq_id") or 0, reverse=True) - if validator is None or validator(configurations): - if clear: - self.clear() - return configurations + break time.sleep(0.05) - - if not events: + else: raise AssertionError("No telemetry events were found. Ensure the application is sending telemetry events.") - raise AssertionError(f"No telemetry configurations matched the validator. Last observed: {configurations}") - def wait_for_telemetry_runtime_id(self, *, exclude: set[str] | None = None, wait_loops: int = 200) -> str: - excluded_runtime_ids = exclude or set() + events.sort(key=lambda event: event["tracer_time"]) + for event in events: + if runtime_id is not None and event.get("runtime_id") != runtime_id: + continue + if service is not None and event["application"]["service_name"] != service: + continue + for event_type in ["app-started", "app-client-configuration-change"]: + telemetry_event = self._get_telemetry_event(event, event_type) + if telemetry_event: + for config in telemetry_event.get("payload", {}).get("configuration", []): + configurations.setdefault(config["name"], []).append(config) + for payload in configurations.values(): + payload.sort(key=lambda item: item.get("seq_id") or 0, reverse=True) + if clear: + self.clear() + return configurations + + def wait_for_telemetry_runtime_id(self, *, exclude: str | None = None, wait_loops: int = 200) -> str: observed_runtime_ids: set[str] = set() for _ in range(wait_loops): - try: + events: list[dict[str, Any]] = [] + with contextlib.suppress(requests.exceptions.RequestException): events = self.telemetry(clear=False) - except requests.exceptions.RequestException: - events = [] - else: - events.sort(key=lambda event: event["tracer_time"], reverse=True) - for event in events: - telemetry_event = self._get_telemetry_event(event, "app-started") - if telemetry_event is None: - continue - runtime_id = telemetry_event.get("runtime_id") or event.get("runtime_id") - if runtime_id is None: - continue - observed_runtime_ids.add(runtime_id) - if runtime_id not in excluded_runtime_ids: - return runtime_id + events.sort(key=lambda event: event["tracer_time"], reverse=True) + for event in events: + telemetry_event = self._get_telemetry_event(event, "app-started") + if telemetry_event is None: + continue + runtime_id = telemetry_event.get("runtime_id") or event.get("runtime_id") + if runtime_id is None: + continue + observed_runtime_ids.add(runtime_id) + if runtime_id != exclude: + return runtime_id time.sleep(0.01) - raise AssertionError( - f"No telemetry runtime ID found excluding {excluded_runtime_ids}. Observed: {observed_runtime_ids}" - ) + raise AssertionError(f"No telemetry runtime ID found excluding {exclude}. Observed: {observed_runtime_ids}") def get_telemetry_config_by_origin( self,