From 37377d7c6cf25de89464b19b91359b22f9001947 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Thu, 10 Sep 2026 21:11:36 +0000 Subject: [PATCH 1/6] test(ffe): add direct agentless EVP contract Add reusable direct-EVP scenarios, wire assertions, topology proof, and bounded shutdown evidence without activating any language manifest. Environment: Datadog workspace --- tests/ffe/test_exposure_egress.py | 140 +++-- tests/ffe/test_flag_eval_evp.py | 122 ++++- tests/ffe/utils/evp.py | 447 ++++++++++++++++ tests/test_the_test/scenarios.json | 9 + tests/test_the_test/test_ffe_evp_contract.py | 481 ++++++++++++++++++ .../test_mock_ffe_agentless_backend.py | 254 ++++++++- .../_context/_scenarios/agentless_endtoend.py | 450 +++++++++++++++- utils/_context/_scenarios/endtoend.py | 6 +- utils/_context/containers.py | 14 +- utils/_context/weblog_infrastructure.py | 12 +- 10 files changed, 1862 insertions(+), 73 deletions(-) create mode 100644 tests/ffe/utils/evp.py create mode 100644 tests/test_the_test/test_ffe_evp_contract.py diff --git a/tests/ffe/test_exposure_egress.py b/tests/ffe/test_exposure_egress.py index 2bbf7e09153..f2968df841a 100644 --- a/tests/ffe/test_exposure_egress.py +++ b/tests/ffe/test_exposure_egress.py @@ -1,50 +1,25 @@ """Test one exposure contract through every supported deployment topology.""" -from dataclasses import dataclass - -from tests.ffe.utils.exposures import assert_exposure_side_effects_contract, exposure_events_from_data +from tests.ffe.utils.evp import ( + assert_agentless_evp_intake_request, + assert_agentless_evp_topology, + assert_direct_evp_shutdown_evidence, + feature_flagging_evp_egress, + register_expected_evp_capture, + register_shutdown_evp_evaluation, +) +from tests.ffe.utils.exposures import ( + EXPOSURES_PATH, + assert_exposure_side_effects_contract, + exposure_events_from_data, +) from tests.ffe.utils.fixtures import make_ufc_fixture -from utils import context, features, interfaces, remote_config as rc, scenarios, weblog +from utils import context, features, remote_config as rc, scenarios, weblog from utils._context._scenarios.agentless_endtoend import FeatureFlaggingAgentlessEndToEndScenario -from utils.interfaces._core import ProxyBasedInterfaceValidator -from utils.mocked_backend.ffe import EXPECTED_API_KEY RC_PATH = "datadog/2/FFE_FLAGS" -@dataclass(frozen=True) -class ExposureEgress: - interface: ProxyBasedInterfaceValidator - excluded_interfaces: tuple[ProxyBasedInterfaceValidator, ...] = () - expected_api_key: str | None = None - - -def exposure_egress() -> ExposureEgress: - """Return the capture interface and route-only expectations for this topology.""" - scenario = context.scenario - if not isinstance(scenario, FeatureFlaggingAgentlessEndToEndScenario): - assert scenario.name == "FEATURE_FLAGGING_AND_EXPERIMENTATION" - return ExposureEgress(interfaces.agent) - - if scenario.exposure_egress == "sidecar": - assert "serverless-init" in scenario.components - expected_api_key = scenario.serverless_init_container.environment["DD_API_KEY"] - assert expected_api_key is not None - return ExposureEgress( - interfaces.datadog_sidecar, - (interfaces.datadog_direct,), - expected_api_key, - ) - - assert scenario.exposure_egress == "direct" - assert "serverless-init" not in scenario.components - return ExposureEgress( - interfaces.datadog_direct, - (interfaces.datadog_sidecar,), - EXPECTED_API_KEY, - ) - - class ExposureEgressContract: """One exposure contract inherited by each supported topology adapter.""" @@ -52,6 +27,7 @@ class ExposureEgressContract: targeting_key = "exposure-egress-user" def setup_exposure_egress(self) -> None: + register_expected_evp_capture(EXPOSURES_PATH) if not isinstance(context.scenario, FeatureFlaggingAgentlessEndToEndScenario): rc.tracer_rc_state.reset().set_config( f"{RC_PATH}/exposure-egress/config", @@ -73,7 +49,7 @@ def setup_exposure_egress(self) -> None: ] def test_exposure_egress(self) -> None: - egress = exposure_egress() + egress = feature_flagging_evp_egress() matching_requests = assert_exposure_side_effects_contract( egress.interface, self.responses, @@ -83,16 +59,20 @@ def test_exposure_egress(self) -> None: expected_variant="on", ) assert len(matching_requests) == 1 + assert_agentless_evp_topology(egress) if egress.expected_api_key is None: return - request = matching_requests[0] - assert request["host"] == "event-platform-intake.mock-intake.invalid" - assert request["response"]["status_code"] == 202 - - headers = {name.lower(): value for name, value in request["request"]["headers"]} - assert headers["dd-api-key"] in {egress.expected_api_key, "--redacted--"} + for request in matching_requests: + assert_agentless_evp_intake_request( + request, + route=egress.route, + path=EXPOSURES_PATH, + expected_api_key=egress.expected_api_key, + library_name=context.library.name, + library_version=context.library.raw_version, + ) for excluded_interface in egress.excluded_interfaces: assert not any( @@ -113,6 +93,74 @@ class Test_FFE_Exposure_Egress_Agentless_Direct(ExposureEgressContract): pass +@scenarios.feature_flagging_and_experimentation_agentless_direct +@features.feature_flags_exposures +class Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: + """Prove a just-produced exposure is flushed by runtime shutdown, not a pre-stop wait.""" + + flag_key = "empty-targeting-key-flag" + targeting_key = "exposure-shutdown-user" + + def setup_exposure_egress_shutdown(self) -> None: + register_shutdown_evp_evaluation( + signal_path=EXPOSURES_PATH, + request_path="/ffe", + body={ + "flag": self.flag_key, + "variationType": "STRING", + "defaultValue": "default", + "targetingKey": self.targeting_key, + "attributes": {}, + }, + flag_key=self.flag_key, + subject_id=self.targeting_key, + ) + + def test_exposure_egress_shutdown(self) -> None: + egress = feature_flagging_evp_egress() + assert egress.route == "direct" + assert_agentless_evp_topology(egress) + + scenario = context.scenario + assert isinstance(scenario, FeatureFlaggingAgentlessEndToEndScenario) + assert_direct_evp_shutdown_evidence(scenario.direct_evp_shutdown_evidence()) + + matching_requests = [ + data + for data in egress.interface.get_data() + if exposure_events_from_data(data, {self.flag_key}, self.targeting_key) + ] + assert len(matching_requests) == 1 + events = exposure_events_from_data(matching_requests[0], {self.flag_key}, self.targeting_key) + assert len(events) == 1 + event = events[0] + assert event["flag"]["key"] == self.flag_key + assert event["variant"]["key"] == "on" + assert event["allocation"]["key"] == "default-allocation" + assert event["subject"]["id"] == self.targeting_key + batch_context = matching_requests[0]["request"]["content"]["context"] + assert batch_context == { + "env": "system-tests", + "service": "weblog", + "version": "1.0.0", + } + + assert egress.expected_api_key is not None + assert_agentless_evp_intake_request( + matching_requests[0], + route=egress.route, + path=EXPOSURES_PATH, + expected_api_key=egress.expected_api_key, + library_name=context.library.name, + library_version=context.library.raw_version, + ) + for excluded_interface in egress.excluded_interfaces: + assert not any( + exposure_events_from_data(data, {self.flag_key}, self.targeting_key) + for data in excluded_interface.get_data() + ) + + @scenarios.feature_flagging_and_experimentation_agentless_serverless @features.feature_flags_exposures class Test_FFE_Exposure_Egress_Agentless_Sidecar(ExposureEgressContract): diff --git a/tests/ffe/test_flag_eval_evp.py b/tests/ffe/test_flag_eval_evp.py index 6f909598a6b..13b951eb7f9 100644 --- a/tests/ffe/test_flag_eval_evp.py +++ b/tests/ffe/test_flag_eval_evp.py @@ -1,18 +1,27 @@ """Test server-side feature flag evaluation counts via EVP flagevaluation.""" +import hashlib import json from concurrent.futures import ThreadPoolExecutor from typing import cast +from tests.ffe.utils.evp import ( + assert_agentless_evp_intake_request, + assert_agentless_evp_topology, + feature_flagging_evp_egress, + register_expected_evp_capture, +) from tests.ffe.utils.fixtures import JSON, make_ufc_fixture from utils import HttpResponse +from utils import context from utils import features from utils import interfaces from utils import remote_config as rc from utils import scenario_crash from utils import scenarios from utils import weblog +from utils._context._scenarios.agentless_endtoend import FeatureFlaggingAgentlessEndToEndScenario RC_PRODUCT = "FFE_FLAGS" @@ -65,7 +74,9 @@ def evaluate_flag( return weblog.post("/ffe", json=payload) -def evp_flagevaluation_events_from_data(data: JSON, flag_key: str) -> list[tuple[JSON, JSON]]: +def evp_flagevaluation_events_from_data( + data: JSON, flag_key: str, targeting_key: str | None = None +) -> list[tuple[JSON, JSON]]: if data.get("path") != EVP_FLAGEVALUATIONS_PATH: return [] @@ -87,8 +98,15 @@ def evp_flagevaluation_events_from_data(data: JSON, flag_key: str) -> list[tuple continue flag = event.get("flag") - if isinstance(flag, dict) and flag.get("key") == flag_key: - results.append((cast("JSON", content), cast("JSON", event))) + if not isinstance(flag, dict) or flag.get("key") != flag_key: + continue + + if targeting_key is not None: + hashed_targeting_key = f"sha256_{hashlib.sha256(targeting_key.encode()).hexdigest()}" + if event.get("targeting_key") not in (targeting_key, hashed_targeting_key): + continue + + results.append((cast("JSON", content), cast("JSON", event))) return results @@ -236,6 +254,104 @@ def assert_no_duplicate_visible_events(events: list[tuple[JSON, JSON]]) -> None: assert not duplicates, f"found duplicate serialized-visible EVP buckets in one payload: {sorted(duplicates)}" +class FlagevaluationEgressContract: + """One flag-evaluation contract inherited by each supported topology adapter.""" + + # Agentless scenarios preload flags-v1.json before the weblog starts. Use the + # same known fixture key in every topology; the Agent scenario installs an + # equivalent fixture through Remote Config below. + flag_key = "empty-targeting-key-flag" + targeting_key = "evp-egress-user" + evaluation_count = 5 + + def setup_ffe_evp_flagevaluation_egress(self) -> None: + register_expected_evp_capture(EVP_FLAGEVALUATIONS_PATH) + if not isinstance(context.scenario, FeatureFlaggingAgentlessEndToEndScenario): + config_id = "ffe-evp-egress" + rc.tracer_rc_state.reset().set_config( + f"{RC_PATH}/{config_id}/config", + make_ufc_fixture(self.flag_key), + ).apply() + + self.responses = [ + evaluate_flag(self.flag_key, targeting_key=self.targeting_key, attributes={}) + for _ in range(self.evaluation_count) + ] + + def test_ffe_evp_flagevaluation_egress(self) -> None: + for index, response in enumerate(self.responses): + assert response.status_code == 200, f"Request {index + 1} failed: {response.text}" + + egress = feature_flagging_evp_egress() + + def matcher(data: JSON) -> bool: + return bool(evp_flagevaluation_events_from_data(data, self.flag_key, self.targeting_key)) + + assert egress.interface.wait_for( + lambda data: matcher(cast("JSON", data)), + timeout=EVP_WAIT_TIMEOUT_SECONDS, + ), f"Timed out waiting for EVP flagevaluation event for flag {self.flag_key}" + + matching_requests = [ + cast("JSON", data) + for data in egress.interface.get_data(path_filters=EVP_FLAGEVALUATIONS_PATH) + if matcher(cast("JSON", data)) + ] + assert matching_requests, f"Expected flagevaluation requests for {self.flag_key}" + + events = [ + event + for request in matching_requests + for event in evp_flagevaluation_events_from_data(request, self.flag_key, self.targeting_key) + ] + assert events, f"Expected EVP flagevaluation events for flag {self.flag_key}" + for _, event in events: + assert_event_contract(event, self.flag_key) + assert object_key(event.get("variant"), "variant") == "on" + assert object_key(event.get("allocation"), "allocation") == "default-allocation" + + assert_no_duplicate_visible_events(events) + assert_total_evaluation_count(events, self.evaluation_count, self.flag_key) + assert_agentless_evp_topology(egress) + + if egress.expected_api_key is None: + return + + for request in matching_requests: + assert_agentless_evp_intake_request( + request, + route=egress.route, + path=EVP_FLAGEVALUATIONS_PATH, + expected_api_key=egress.expected_api_key, + library_name=context.library.name, + library_version=context.library.raw_version, + ) + + for excluded_interface in egress.excluded_interfaces: + assert not any( + evp_flagevaluation_events_from_data(cast("JSON", data), self.flag_key, self.targeting_key) + for data in excluded_interface.get_data(path_filters=EVP_FLAGEVALUATIONS_PATH) + ) + + +@scenarios.feature_flagging_and_experimentation +@features.feature_flags_evp_flagevaluation +class Test_FFE_EVP_Flagevaluation_Egress_Datadog_Agent(FlagevaluationEgressContract): + pass + + +@scenarios.feature_flagging_and_experimentation_agentless_direct +@features.feature_flags_evp_flagevaluation +class Test_FFE_EVP_Flagevaluation_Egress_Agentless_Direct(FlagevaluationEgressContract): + pass + + +@scenarios.feature_flagging_and_experimentation_agentless_serverless +@features.feature_flags_evp_flagevaluation +class Test_FFE_EVP_Flagevaluation_Egress_Agentless_Sidecar(FlagevaluationEgressContract): + pass + + @scenarios.feature_flagging_and_experimentation @features.feature_flags_evp_flagevaluation class Test_FFE_EVP_Flagevaluation_Basic: diff --git a/tests/ffe/utils/evp.py b/tests/ffe/utils/evp.py new file mode 100644 index 00000000000..970f6091c2b --- /dev/null +++ b/tests/ffe/utils/evp.py @@ -0,0 +1,447 @@ +"""Shared Feature Flags EVP route, wire, identity, and topology assertions.""" + +from collections.abc import Iterable +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any, Literal + +from utils import context, interfaces +from utils._context._scenarios.agentless_endtoend import ( + DIRECT_EVP_AGENT_VARIABLES, + DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH, + DIRECT_EVP_CA_BUNDLE_SOURCE, + DIRECT_EVP_SHUTDOWN_MARKER_EVENT, + DIRECT_EVP_SIGNAL_PATHS, + DIRECT_EVP_STOP_TIMEOUT_SECONDS, + FeatureFlaggingAgentlessEndToEndScenario, +) +from utils.interfaces._core import ProxyBasedInterfaceValidator +from utils.mocked_backend.ffe import EXPECTED_API_KEY + + +EVP_DIRECT_HOST = "event-platform-intake.mock-intake.invalid" +EVP_DIRECT_PORT = 443 +EVP_PROXY_PATH_PREFIXES = ("/evp_proxy/v2/", "/evp_proxy/v4/") +EVP_FORBIDDEN_RUNTIME_COMPONENTS = ( + "datadog-agent", + "datadog-init", + "process-agent", + "serverless-init", + "telemetry-sidecar", + "trace-agent", +) +EVP_LANGUAGE_PID1_EXECUTABLES = { + "dotnet": {"dotnet"}, + "golang": {"weblog"}, + "java": {"java"}, + "nodejs": {"node"}, + "python": {"gunicorn", "python", "python3", "uwsgi"}, + "ruby": {"puma", "ruby"}, +} + +EVP_ORIGINS = { + "dotnet": "dd-trace-dotnet", + "golang": "dd-trace-go", + "java": "dd-trace-java", + "nodejs": "dd-trace-js", + "python": "dd-trace-py", + "ruby": "dd-trace-rb", +} + +EVPRoute = Literal["agent", "sidecar", "direct"] + + +def _is_expected_pid1_executable(library_name: str, executable: str, expected: set[str]) -> bool: + """Accept ordinary language launchers without accepting wrappers by prefix.""" + if executable in expected: + return True + if library_name != "python" or not executable.startswith("python3."): + return False + return executable.removeprefix("python3.").isdigit() + + +@dataclass(frozen=True) +class FeatureFlaggingEVPEgress: + """The capture interface and routing expectations for one FFE topology.""" + + interface: ProxyBasedInterfaceValidator + route: EVPRoute + excluded_interfaces: tuple[ProxyBasedInterfaceValidator, ...] = () + expected_api_key: str | None = None + + +def feature_flagging_evp_egress() -> FeatureFlaggingEVPEgress: + """Return the EVP capture route selected by the current FFE scenario.""" + scenario = context.scenario + if not isinstance(scenario, FeatureFlaggingAgentlessEndToEndScenario): + assert scenario.name == "FEATURE_FLAGGING_AND_EXPERIMENTATION" + return FeatureFlaggingEVPEgress(interfaces.agent, "agent") + + if scenario.exposure_egress == "sidecar": + assert "serverless-init" in scenario.components + expected_api_key = scenario.serverless_init_container.environment["DD_API_KEY"] + assert expected_api_key is not None + return FeatureFlaggingEVPEgress( + interfaces.datadog_sidecar, + "sidecar", + (interfaces.datadog_direct,), + expected_api_key, + ) + + assert scenario.exposure_egress == "direct" + assert "serverless-init" not in scenario.components + return FeatureFlaggingEVPEgress( + interfaces.datadog_direct, + "direct", + (interfaces.datadog_sidecar,), + EXPECTED_API_KEY, + ) + + +def register_expected_evp_capture(path: str) -> None: + """Register a signal for scenario-owned capture waiting before shutdown.""" + scenario = context.scenario + if isinstance(scenario, FeatureFlaggingAgentlessEndToEndScenario) and scenario.exposure_egress is not None: + scenario.register_expected_evp_capture(path) + + +def register_shutdown_evp_evaluation( + *, + signal_path: str, + request_path: str, + body: dict[str, Any], + flag_key: str, + subject_id: str, +) -> None: + """Register one evaluation for the scenario-owned shutdown-flush phase.""" + scenario = context.scenario + if isinstance(scenario, FeatureFlaggingAgentlessEndToEndScenario): + scenario.register_shutdown_evp_evaluation( + signal_path=signal_path, + request_path=request_path, + body=body, + flag_key=flag_key, + subject_id=subject_id, + ) + + +def expected_evp_origin(library_name: str) -> str: + """Return the logical EVP producer identity for a system-tests library.""" + try: + return EVP_ORIGINS[library_name] + except KeyError as error: + raise AssertionError(f"No EVP origin is defined for system-tests library {library_name!r}") from error + + +def expected_evp_origin_version(library_name: str, library_version: str) -> str: + """Return the tracer's canonical wire version for direct EVP identity. + + System-tests deliberately removes Go's leading ``v`` when it parses the + component version for manifest comparisons. dd-trace-go's public version + tag and existing EVP producer identity retain that prefix, so restore it + only for the wire assertion instead of weakening identity checks globally. + """ + if library_name == "golang" and not library_version.startswith("v"): + return f"v{library_version}" + return library_version + + +def _header_values(data: dict[str, Any], name: str) -> list[str]: + headers = data.get("request", {}).get("headers") + assert isinstance(headers, list), f"request headers must be a list: {data}" + return [ + value for header_name, value in headers if isinstance(header_name, str) and header_name.lower() == name.lower() + ] + + +def _assert_single_header(data: dict[str, Any], name: str, expected: str | set[str]) -> None: + values = _header_values(data, name) + assert len(values) == 1, f"expected exactly one {name} header, got {values!r}" + allowed = {expected} if isinstance(expected, str) else expected + assert values[0] in allowed, f"expected {name} in {sorted(allowed)!r}, got {values[0]!r}" + + +def assert_agentless_evp_intake_request( + data: dict[str, Any], + *, + route: EVPRoute, + path: str, + expected_api_key: str, + library_name: str, + library_version: str, +) -> None: + """Assert one captured Agentless EVP request against the shared wire contract.""" + assert path in DIRECT_EVP_SIGNAL_PATHS, f"unsupported EVP signal path {path!r}" + assert data.get("method") == "POST", f"EVP request must use POST: {data}" + assert data.get("path") == path, f"EVP request must use canonical path {path!r}: {data}" + assert data.get("host") == EVP_DIRECT_HOST, f"unexpected EVP intake host: {data}" + assert data.get("port") == EVP_DIRECT_PORT, f"EVP direct request must use HTTPS port 443: {data}" + + response = data.get("response") + assert isinstance(response, dict), f"EVP capture must include a response: {data}" + assert response.get("status_code") == 202, f"EVP intake must return 202: {data}" + + _assert_single_header(data, "DD-API-KEY", {expected_api_key, "--redacted--"}) + + content_types = _header_values(data, "Content-Type") + assert len(content_types) == 1, f"expected exactly one Content-Type header, got {content_types!r}" + assert content_types[0].split(";", 1)[0].strip().lower() == "application/json" + assert not _header_values(data, "X-Datadog-EVP-Subdomain"), ( + "direct-intake requests must not send X-Datadog-EVP-Subdomain" + ) + + assert route in ("direct", "sidecar") + if route == "sidecar": + # datadog_sidecar captures relay-to-intake traffic, not the SDK-to-relay request. Relayed + # producer identity requires Agent 7.84+, while the default fixture may still be earlier; + # SDK local-route identity and credential omission remain covered by SDK unit tests. + return + + _assert_single_header(data, "DD-EVP-ORIGIN", expected_evp_origin(library_name)) + _assert_single_header( + data, + "DD-EVP-ORIGIN-VERSION", + expected_evp_origin_version(library_name, library_version), + ) + + +def assert_no_evp_proxy_requests(captures: Iterable[dict[str, Any]]) -> None: + """Reject local relay paths in a direct-intake capture stream.""" + proxy_paths = [ + path + for data in captures + if isinstance((path := data.get("path")), str) and path.startswith(EVP_PROXY_PATH_PREFIXES) + ] + assert not proxy_paths, f"direct capture unexpectedly contains local EVP proxy requests: {proxy_paths}" + + +def _runtime_text(container: dict[str, Any]) -> str: + process_text = "\n".join( + " ".join(str(value) for value in process.values()) for process in container.get("processes", []) + ) + return "\n".join( + ( + str(container.get("name", "")), + str(container.get("image", "")), + str(container.get("pid1_command", "")), + process_text, + ) + ).lower() + + +def assert_direct_evp_runtime_evidence(evidence: dict[str, Any], *, library_name: str) -> None: + """Assert preserved Docker inspect/top evidence for the live direct scenario.""" + containers = evidence.get("containers") + assert isinstance(containers, list), "Direct EVP runtime evidence has no live container list" + containers_by_name: dict[str, dict[str, Any]] = {} + for container in containers: + assert isinstance(container, dict), f"Malformed live container evidence: {container!r}" + name = container.get("name") + assert isinstance(name, str), f"Live container evidence has no name: {container}" + assert name not in containers_by_name, f"Duplicate live container evidence for {name!r}" + containers_by_name[name] = container + expected_names = {"system-tests-proxy", "system-tests-weblog"} + assert set(containers_by_name) == expected_names, ( + f"Direct EVP runtime network must contain only TLS proxy and weblog, got {sorted(containers_by_name)}" + ) + + network = evidence.get("network") + assert isinstance(network, dict), "Direct EVP runtime evidence has no network inspection" + assert network.get("id"), "Direct EVP runtime network has no Docker ID" + assert network.get("name"), "Direct EVP runtime network has no name" + observed_ids = {container.get("id") for container in containers_by_name.values()} + assert set(network.get("container_ids", [])) == observed_ids, "Network membership and inspected containers disagree" + assert set(network.get("container_names", [])) == expected_names, "Docker network inspection has unexpected members" + + for container in containers_by_name.values(): + assert container.get("running") is True, f"Runtime container was not running: {container}" + assert container.get("status") == "running", f"Unexpected live container status: {container}" + assert container.get("top_error") is None, f"docker top failed: {container.get('top_error')}" + assert container.get("pid1_error") is None, f"PID 1 inspection failed: {container.get('pid1_error')}" + assert container.get("pid1_command"), f"Runtime container has no PID 1 command: {container}" + assert container.get("networks") == [network["name"]], f"Unexpected runtime networks: {container}" + processes = container.get("processes") + assert isinstance(processes, list), f"Runtime process tree is malformed: {container}" + assert processes, f"Runtime process tree is empty: {container}" + assert all(isinstance(process, dict) for process in processes), f"Malformed runtime process rows: {container}" + state_pid = str(container.get("state_pid")) + pid1_processes = [process for process in processes if process.get("pid") == state_pid] + assert len(pid1_processes) == 1, ( + f"docker top must contain exactly one inspected container PID {state_pid}: {container}" + ) + forbidden = [name for name in EVP_FORBIDDEN_RUNTIME_COMPONENTS if name in _runtime_text(container)] + assert not forbidden, f"Forbidden direct-scenario runtime components found: {forbidden}" + + proxy = containers_by_name["system-tests-proxy"] + proxy_runtime = _runtime_text(proxy) + assert "proxy.core" in proxy_runtime or "proxy/core.py" in proxy_runtime, ( + f"The only allowed helper must be the system-tests TLS capture proxy: {proxy}" + ) + + weblog = containers_by_name["system-tests-weblog"] + assert weblog.get("library") == library_name, ( + f"Live weblog library label is {weblog.get('library')!r}, expected {library_name!r}" + ) + expected_executables = EVP_LANGUAGE_PID1_EXECUTABLES.get(library_name) + assert expected_executables, f"No ordinary PID 1 contract is defined for {library_name!r}" + pid1_executable = Path(str(weblog["pid1_command"]).split()[0]).name + assert _is_expected_pid1_executable(library_name, pid1_executable, expected_executables), ( + f"Live weblog PID 1 is {weblog['pid1_command']!r}, expected one of " + f"{sorted(expected_executables)} or a versioned python3 interpreter" + ) + weblog_state_pid = str(weblog.get("state_pid")) + pid1_process = next(process for process in weblog["processes"] if process.get("pid") == weblog_state_pid) + pid1_process_command = pid1_process.get("command") + assert pid1_process_command is not None, f"docker top PID 1 row has no command: {pid1_process}" + assert isinstance(pid1_process_command, str), f"docker top PID 1 command is malformed: {pid1_process}" + assert pid1_process_command, f"docker top PID 1 row has no command: {pid1_process}" + pid1_process_executable = Path(pid1_process_command.split()[0]).name + assert _is_expected_pid1_executable(library_name, pid1_process_executable, expected_executables), ( + f"docker top reports PID 1 as {pid1_process_command!r}, expected one of " + f"{sorted(expected_executables)} or a versioned python3 interpreter" + ) + + environment = evidence.get("weblog_environment") + assert isinstance(environment, dict), "Live weblog environment was not captured" + assert environment.get("agent_variables_present") == [], ( + f"Live direct weblog contains Agent connection variables: {environment.get('agent_variables_present')}" + ) + assert environment.get("api_key_present") is True + assert environment.get("api_key_matches_expected") is True + assert environment.get("configuration_source") == "agentless" + assert environment.get("provider_enabled") == "true" + assert environment.get("shutdown_flush_enabled") == "true" + assert environment.get("site") == "mock-intake.invalid" + + expected_mounts = [ + mount for mount in weblog.get("mounts", []) if mount.get("destination") == DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH + ] + assert len(expected_mounts) == 1, ( + f"Live weblog must have one {DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH} mount: {weblog.get('mounts')}" + ) + assert expected_mounts[0].get("read_write") is False, "Direct EVP CA bundle must be mounted read-only" + + +def assert_direct_evp_shutdown_evidence(evidence: dict[str, Any]) -> None: + """Assert a late evaluation was delivered by the bounded shutdown-flush phase.""" + assert evidence.get("flush_window_primed") is True, "Shutdown target was not placed in a fresh flush window" + assert evidence.get("priming_evaluation_status_code") == 200, "Flush-window priming evaluation did not succeed" + priming_files = evidence.get("priming_capture_files") + assert isinstance(priming_files, list), f"Priming capture files are malformed: {priming_files}" + assert len(priming_files) == 1, f"Flush-window priming must produce exactly one capture: {priming_files}" + priming_request_started = evidence.get("priming_capture_request_started_at") + assert isinstance(priming_request_started, list), ( + f"Priming capture request timestamps are malformed: {priming_request_started}" + ) + assert len(priming_request_started) == 1, ( + f"Flush-window priming must identify exactly one request timestamp: {priming_request_started}" + ) + assert evidence.get("priming_subject_id") != evidence.get("subject_id"), ( + "Flush-window priming must use a subject distinct from the shutdown target" + ) + assert evidence.get("evaluation_status_code") == 200, "Shutdown evaluation did not succeed" + assert evidence.get("explicit_flush") is False, "Shutdown proof must not call the weblog /flush endpoint" + assert evidence.get("captures_before_evaluation") == 0, "Shutdown probe was already captured before evaluation" + assert evidence.get("captures_before_stop") == 0, "Shutdown probe escaped before Docker sent SIGTERM" + assert evidence.get("capture_observed_after_stop") is True, "Shutdown-flushed event was not observed after stop" + assert evidence.get("captures_after_settle") == 1, "Shutdown flush did not deliver exactly one event" + capture_files = evidence.get("capture_files") + assert isinstance(capture_files, list), f"Shutdown capture files are malformed: {capture_files}" + assert len(capture_files) == 1, f"Shutdown evidence must identify exactly one capture file: {capture_files}" + assert capture_files[0], f"Shutdown capture filename is empty: {capture_files}" + + duration = evidence.get("shutdown_duration_seconds") + bound = evidence.get("shutdown_bound_seconds") + assert isinstance(duration, (int, float)) + assert isinstance(bound, (int, float)) + assert duration <= bound, f"Shutdown flush took {duration}s, exceeding the {bound}s bound" + assert evidence.get("stop_timeout_seconds") == DIRECT_EVP_STOP_TIMEOUT_SECONDS + assert evidence.get("stop_error") is None, f"Shutdown returned an error: {evidence.get('stop_error')}" + + stopped = evidence.get("stopped_container") + assert isinstance(stopped, dict), "Post-stop Docker inspection is absent" + assert "inspection_error" not in stopped, f"Post-stop Docker inspection failed: {stopped}" + assert stopped.get("running") is False, "Weblog remained running after bounded stop" + assert stopped.get("status") == "exited", f"Unexpected post-stop state: {stopped}" + assert stopped.get("oom_killed") is False, f"Weblog was OOM-killed: {stopped}" + assert stopped.get("error") in (None, ""), f"Docker reported a stop error: {stopped}" + assert stopped.get("exit_code") == 0, f"Weblog did not complete its graceful shutdown path: {stopped}" + assert stopped.get("finished_at"), f"Post-stop finish timestamp is absent: {stopped}" + + marker_errors = evidence.get("shutdown_marker_errors") + assert marker_errors == [], f"Could not read shutdown markers from post-stop Docker logs: {marker_errors}" + markers = evidence.get("shutdown_markers") + assert isinstance(markers, list), f"Shutdown markers are malformed: {markers}" + assert len(markers) == 1, f"Expected exactly one structured server-close marker, got {markers}" + marker = markers[0] + assert isinstance(marker, dict), f"Shutdown marker is malformed: {marker}" + assert marker.get("event") == DIRECT_EVP_SHUTDOWN_MARKER_EVENT + assert marker.get("stream") == "stdout", f"Server-close marker must be written to stdout: {marker}" + + priming_finished = datetime.fromisoformat(str(evidence.get("priming_evaluation_finished_at"))) + priming_capture_started = datetime.fromisoformat(str(priming_request_started[0])) + evaluation_finished = datetime.fromisoformat(str(evidence.get("evaluation_finished_at"))) + stop_started = datetime.fromisoformat(str(evidence.get("stop_started_at"))) + server_closed = datetime.fromisoformat(str(marker.get("timestamp"))) + stopped_at = datetime.fromisoformat(str(evidence.get("stopped_at"))) + assert priming_finished <= evaluation_finished <= stop_started <= server_closed <= stopped_at, ( + "Shutdown lifecycle timestamps are out of order" + ) + assert priming_capture_started <= evaluation_finished, "Priming flush did not begin before the shutdown target" + capture_started_values = evidence.get("capture_request_started_at") + assert isinstance(capture_started_values, list) + assert len(capture_started_values) == 1 + capture_started = datetime.fromisoformat(str(capture_started_values[0])) + assert capture_started >= server_closed, ( + f"Event request started at {capture_started.isoformat()} before server close at {server_closed.isoformat()}" + ) + + +def assert_agentless_evp_topology(egress: FeatureFlaggingEVPEgress) -> None: + """Assert the effective scenario topology for an Agentless EVP route.""" + if egress.route == "agent": + return + + scenario = context.scenario + assert isinstance(scenario, FeatureFlaggingAgentlessEndToEndScenario) + assert scenario.include_agent is False + assert scenario.get_libraries() is None + + environment = scenario.weblog_infra.library_container.environment + assert environment["DD_FEATURE_FLAGS_CONFIGURATION_SOURCE"] == "agentless" + assert environment["DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED"] == "true" + assert environment["DD_SITE"] == "mock-intake.invalid" + if not scenario.replay: + assert environment["DD_API_KEY"] == EXPECTED_API_KEY + if egress.route == "direct": + assert_direct_evp_runtime_evidence( + scenario.direct_evp_runtime_evidence(), + library_name=context.library.name, + ) + library_container = scenario.weblog_infra.library_container + expected_mount = { + "bind": DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH, + "mode": "ro", + } + mounted_sources = [ + Path(source) for source, mount in library_container.volumes.items() if mount == expected_mount + ] + assert len(mounted_sources) == 1, ( + f"expected exactly one read-only {DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH} mount, " + f"got {library_container.volumes!r}" + ) + mounted_source = mounted_sources[0] + if not mounted_source.is_absolute(): + mounted_source = Path(library_container.host_project_dir) / mounted_source + expected_source = Path(library_container.host_project_dir) / DIRECT_EVP_CA_BUNDLE_SOURCE + assert mounted_source.resolve() == expected_source.resolve() + for name in DIRECT_EVP_AGENT_VARIABLES: + assert name not in environment + assert "serverless-init" not in scenario.components + assert not any(container.name == "ffe-serverless-init" for container in scenario.weblog_infra.get_containers()) + assert_no_evp_proxy_requests(interfaces.datadog_direct.get_data()) + return + + assert egress.route == "sidecar" + assert "serverless-init" in scenario.components diff --git a/tests/test_the_test/scenarios.json b/tests/test_the_test/scenarios.json index 7030b60d222..c153bd73e32 100644 --- a/tests/test_the_test/scenarios.json +++ b/tests/test_the_test/scenarios.json @@ -3390,6 +3390,15 @@ "tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Sidecar::test_exposure_egress": [ "FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS_SERVERLESS" ], + "tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_Egress_Agentless_Direct::test_ffe_evp_flagevaluation_egress": [ + "FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS_DIRECT" + ], + "tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_Egress_Agentless_Sidecar::test_ffe_evp_flagevaluation_egress": [ + "FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS_SERVERLESS" + ], + "tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_Egress_Datadog_Agent::test_ffe_evp_flagevaluation_egress": [ + "FEATURE_FLAGGING_AND_EXPERIMENTATION" + ], "tests/integrations/test_cassandra.py::Test_Cassandra::test_main": [ "INTEGRATIONS" ], diff --git a/tests/test_the_test/test_ffe_evp_contract.py b/tests/test_the_test/test_ffe_evp_contract.py new file mode 100644 index 00000000000..e8c7187baff --- /dev/null +++ b/tests/test_the_test/test_ffe_evp_contract.py @@ -0,0 +1,481 @@ +"""Unit coverage for the shared Feature Flags EVP wire contract.""" + +from copy import deepcopy +from typing import Any, Literal + +import pytest + +from tests.ffe.utils.evp import ( + EVP_ORIGINS, + FeatureFlaggingEVPEgress, + assert_agentless_evp_intake_request, + assert_agentless_evp_topology, + assert_direct_evp_runtime_evidence, + assert_direct_evp_shutdown_evidence, + assert_no_evp_proxy_requests, + expected_evp_origin, +) +from utils import context, features, interfaces, scenarios +from utils._context.component_version import ComponentVersion, Version +from utils._context._scenarios.agentless_endtoend import ( + DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH, + DIRECT_EVP_CA_BUNDLE_SOURCE, + FeatureFlaggingAgentlessEndToEndScenario, +) + + +def _direct_runtime_evidence(library_name: str = "nodejs") -> dict[str, Any]: + executable = { + "dotnet": "dotnet", + "golang": "weblog", + "java": "java", + "nodejs": "node", + "python": "python", + "ruby": "ruby", + }[library_name] + network_name = "system-tests-network" + return { + "captured_at": "2026-09-09T12:00:00+00:00", + "network": { + "id": "network-id", + "name": network_name, + "container_ids": ["proxy-id", "weblog-id"], + "container_names": ["system-tests-proxy", "system-tests-weblog"], + }, + "containers": [ + { + "id": "proxy-id", + "name": "system-tests-proxy", + "image": "system-tests/proxy", + "image_id": "sha256:proxy", + "library": None, + "weblog_variant": None, + "status": "running", + "running": True, + "state_pid": 101, + "pid1_command": "python3 -m utils.proxy.core", + "pid1_error": None, + "processes": [{"pid": "101", "ppid": "0", "command": "python3 -m utils.proxy.core"}], + "top_error": None, + "networks": [network_name], + "mounts": [], + }, + { + "id": "weblog-id", + "name": "system-tests-weblog", + "image": "system-tests/weblog", + "image_id": "sha256:weblog", + "library": library_name, + "weblog_variant": "express4", + "status": "running", + "running": True, + "state_pid": 202, + "pid1_command": f"{executable} app", + "pid1_error": None, + "processes": [{"pid": "202", "ppid": "0", "command": f"{executable} app"}], + "top_error": None, + "networks": [network_name], + "mounts": [ + { + "destination": DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH, + "read_write": False, + "source": "/system-tests/utils/build/docker/agent/ca-certificates.crt", + "type": "bind", + } + ], + }, + ], + "weblog_environment": { + "agent_variables_present": [], + "api_key_matches_expected": True, + "api_key_present": True, + "configuration_source": "agentless", + "custom_configuration_url_present": True, + "provider_enabled": "true", + "shutdown_flush_enabled": "true", + "site": "mock-intake.invalid", + "tls_ca_bundle": None, + }, + } + + +def _direct_shutdown_evidence() -> dict[str, Any]: + return { + "capture_files": ["direct-0001.json"], + "capture_observed_after_stop": True, + "capture_request_started_at": ["2026-09-09T12:00:01.200000+00:00"], + "captures_after_settle": 1, + "captures_before_evaluation": 0, + "captures_before_stop": 0, + "evaluation_finished_at": "2026-09-09T12:00:01+00:00", + "evaluation_started_at": "2026-09-09T12:00:00+00:00", + "evaluation_status_code": 200, + "explicit_flush": False, + "flag_key": "empty-targeting-key-flag", + "flush_window_primed": True, + "priming_capture_files": ["direct-prime-0001.json"], + "priming_capture_request_started_at": ["2026-09-09T11:59:59.800000+00:00"], + "priming_evaluation_finished_at": "2026-09-09T11:59:59.900000+00:00", + "priming_evaluation_started_at": "2026-09-09T11:59:59.700000+00:00", + "priming_evaluation_status_code": 200, + "priming_subject_id": "exposure-shutdown-user-flush-window-prime", + "shutdown_bound_seconds": 12, + "shutdown_duration_seconds": 0.5, + "shutdown_marker_errors": [], + "shutdown_markers": [ + { + "event": "system_tests.ffe.shutdown.server_closed", + "stream": "stdout", + "timestamp": "2026-09-09T12:00:01.150000+00:00", + } + ], + "signal_path": "/api/v2/exposures", + "stop_error": None, + "stop_started_at": "2026-09-09T12:00:01.100000+00:00", + "stop_timeout_seconds": 10, + "stopped_at": "2026-09-09T12:00:01.600000+00:00", + "stopped_container": { + "error": "", + "exit_code": 0, + "finished_at": "2026-09-09T12:00:01.500000000Z", + "oom_killed": False, + "running": False, + "status": "exited", + }, + "subject_id": "exposure-shutdown-user", + } + + +def _direct_capture(*, path: str = "/api/v2/exposures", headers: list[list[str]] | None = None) -> dict[str, Any]: + return { + "method": "POST", + "path": path, + "host": "event-platform-intake.mock-intake.invalid", + "port": 443, + "request": { + "headers": headers + or [ + ["DD-API-KEY", "system-tests-mock-api-key"], + ["DD-EVP-ORIGIN", "dd-trace-js"], + ["DD-EVP-ORIGIN-VERSION", "7.0.0-pre"], + ["Content-Type", "application/json; charset=utf-8"], + ], + "content": {"exposures": []}, + }, + "response": {"status_code": 202, "content": "Ok"}, + } + + +@pytest.mark.parametrize(("library_name", "origin"), sorted(EVP_ORIGINS.items())) +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_evp_origin_contract(library_name: str, origin: str) -> None: + assert expected_evp_origin(library_name) == origin + + +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_agentless_evp_wire_contract_accepts_exact_identity() -> None: + assert_agentless_evp_intake_request( + _direct_capture(), + route="direct", + path="/api/v2/exposures", + expected_api_key="system-tests-mock-api-key", + library_name="nodejs", + library_version="7.0.0-pre", + ) + + +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_agentless_evp_wire_contract_accepts_canonical_go_version() -> None: + capture = _direct_capture( + headers=[ + ["DD-API-KEY", "system-tests-mock-api-key"], + ["DD-EVP-ORIGIN", "dd-trace-go"], + ["DD-EVP-ORIGIN-VERSION", "v2.11.0-dev.1"], + ["Content-Type", "application/json"], + ] + ) + + assert_agentless_evp_intake_request( + capture, + route="direct", + path="/api/v2/exposures", + expected_api_key="system-tests-mock-api-key", + library_name="golang", + library_version="2.11.0-dev.1", + ) + + capture["request"]["headers"][2][1] = "2.11.0-dev.1" + with pytest.raises(AssertionError, match="DD-EVP-ORIGIN-VERSION"): + assert_agentless_evp_intake_request( + capture, + route="direct", + path="/api/v2/exposures", + expected_api_key="system-tests-mock-api-key", + library_name="golang", + library_version="2.11.0-dev.1", + ) + + +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_agentless_evp_wire_contract_rejects_missing_identity() -> None: + capture = _direct_capture( + headers=[ + ["DD-API-KEY", "system-tests-mock-api-key"], + ["Content-Type", "application/json"], + ] + ) + + with pytest.raises(AssertionError, match="DD-EVP-ORIGIN"): + assert_agentless_evp_intake_request( + capture, + route="direct", + path="/api/v2/exposures", + expected_api_key="system-tests-mock-api-key", + library_name="nodejs", + library_version="7.0.0-pre", + ) + + +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_agentless_evp_wire_contract_rejects_proxy_header() -> None: + capture = _direct_capture() + capture["request"]["headers"].append(["X-Datadog-EVP-Subdomain", "event-platform-intake"]) + + with pytest.raises(AssertionError, match="must not send X-Datadog-EVP-Subdomain"): + assert_agentless_evp_intake_request( + capture, + route="direct", + path="/api/v2/exposures", + expected_api_key="system-tests-mock-api-key", + library_name="nodejs", + library_version="7.0.0-pre", + ) + + +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_agentless_evp_wire_contract_scopes_producer_identity_to_direct_route() -> None: + capture = _direct_capture( + headers=[ + ["DD-API-KEY", "--redacted--"], + ["Content-Type", "application/json"], + ["Via", "trace-agent 7.81.2"], + ] + ) + + assert_agentless_evp_intake_request( + capture, + route="sidecar", + path="/api/v2/exposures", + expected_api_key="system-tests-mock-api-key", + library_name="nodejs", + library_version="7.0.0-pre", + ) + + with pytest.raises(AssertionError, match="DD-EVP-ORIGIN"): + assert_agentless_evp_intake_request( + capture, + route="direct", + path="/api/v2/exposures", + expected_api_key="system-tests-mock-api-key", + library_name="nodejs", + library_version="7.0.0-pre", + ) + + +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_direct_capture_rejects_local_proxy_routes() -> None: + with pytest.raises(AssertionError, match="local EVP proxy requests"): + assert_no_evp_proxy_requests( + [ + _direct_capture(), + _direct_capture(path="/evp_proxy/v4/api/v2/exposures"), + ] + ) + + +@pytest.mark.parametrize("route", ["direct", "sidecar"]) +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_agentless_evp_topology_supports_both_routes( + route: Literal["direct", "sidecar"], + monkeypatch: pytest.MonkeyPatch, +) -> None: + scenario = FeatureFlaggingAgentlessEndToEndScenario( + f"MOCK_FFE_AGENTLESS_{route.upper()}_TOPOLOGY", + doc="test", + exposure_egress=route, + ) + scenario.weblog_infra.library_container.environment["DD_API_KEY"] = "system-tests-mock-api-key" + if route == "sidecar": + scenario.components["serverless-init"] = Version("0.0.0") + else: + scenario.weblog_infra.http_container._library = ComponentVersion("java", "0.66.0") # noqa: SLF001 + scenario._last_direct_evp_runtime_evidence = _direct_runtime_evidence("java") # noqa: SLF001 + monkeypatch.setattr(context, "scenario", scenario) + monkeypatch.setattr(interfaces.datadog_direct, "get_data", list) + + selected_interface = interfaces.datadog_direct if route == "direct" else interfaces.datadog_sidecar + assert_agentless_evp_topology(FeatureFlaggingEVPEgress(selected_interface, route)) + + environment = scenario.weblog_infra.library_container.environment + for runtime_trust_variable in ("NODE_EXTRA_CA_CERTS", "REQUESTS_CA_BUNDLE", "SSL_CERT_FILE"): + assert runtime_trust_variable not in environment + + if route == "direct": + library_container = scenario.weblog_infra.library_container + expected_mount = { + "bind": DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH, + "mode": "ro", + } + assert library_container.volumes[DIRECT_EVP_CA_BUNDLE_SOURCE] == expected_mount + + library_container._fix_host_pwd_in_volumes() # noqa: SLF001 - reproduce container start normalization + normalized_source = f"{library_container.host_project_dir}{DIRECT_EVP_CA_BUNDLE_SOURCE[1:]}" + assert library_container.volumes[normalized_source] == expected_mount + assert_agentless_evp_topology(FeatureFlaggingEVPEgress(selected_interface, route)) + else: + assert DIRECT_EVP_CA_BUNDLE_SOURCE not in scenario.weblog_infra.library_container.volumes + + +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_direct_runtime_evidence_accepts_live_minimal_topology() -> None: + assert_direct_evp_runtime_evidence(_direct_runtime_evidence(), library_name="nodejs") + + +@pytest.mark.parametrize("executable", ["python3.11", "python3.12"]) +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_direct_runtime_evidence_accepts_versioned_python(executable: str) -> None: + evidence = _direct_runtime_evidence("python") + evidence["containers"][1]["pid1_command"] = f"/usr/local/bin/{executable} -m gunicorn app:app" + evidence["containers"][1]["processes"][0]["command"] = f"{executable} -m gunicorn app:app" + + assert_direct_evp_runtime_evidence(evidence, library_name="python") + + +@pytest.mark.parametrize("executable", ["python3.x", "python3.11-wrapper"]) +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_direct_runtime_evidence_rejects_python_prefix_wrappers(executable: str) -> None: + evidence = _direct_runtime_evidence("python") + evidence["containers"][1]["pid1_command"] = f"{executable} app" + evidence["containers"][1]["processes"][0]["command"] = f"{executable} app" + + with pytest.raises(AssertionError, match="versioned python3 interpreter"): + assert_direct_evp_runtime_evidence(evidence, library_name="python") + + +@pytest.mark.parametrize( + ("mutation", "message"), + [ + ("agent_container", "only TLS proxy and weblog"), + ("agent_variable", "Agent connection variables"), + ("init_wrapper", "Forbidden direct-scenario runtime components"), + ("shell_pid1", "expected one of"), + ("shell_process", "docker top reports PID 1"), + ("missing_process_command", "PID 1 row has no command"), + ("failed_top", "docker top failed"), + ], +) +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_direct_runtime_evidence_rejects_false_proofs( + mutation: str, + message: str, +) -> None: + evidence = _direct_runtime_evidence() + if mutation == "agent_container": + agent = deepcopy(evidence["containers"][0]) + agent |= {"id": "agent-id", "name": "system-tests-agent", "image": "datadog/agent"} + evidence["containers"].append(agent) + evidence["network"]["container_ids"].append("agent-id") + evidence["network"]["container_names"].append("system-tests-agent") + elif mutation == "agent_variable": + evidence["weblog_environment"]["agent_variables_present"] = ["DD_AGENT_HOST"] + elif mutation == "init_wrapper": + evidence["containers"][1]["pid1_command"] = "datadog-init node app" + elif mutation == "shell_pid1": + evidence["containers"][1]["pid1_command"] = "/bin/sh ./app.sh" + elif mutation == "shell_process": + evidence["containers"][1]["processes"][0]["command"] = "/bin/sh ./app.sh" + elif mutation == "missing_process_command": + del evidence["containers"][1]["processes"][0]["command"] + elif mutation == "failed_top": + evidence["containers"][1]["top_error"] = "APIError: top failed" + else: + raise AssertionError(f"Unhandled mutation {mutation}") + + with pytest.raises(AssertionError, match=message): + assert_direct_evp_runtime_evidence(evidence, library_name="nodejs") + + +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_direct_shutdown_evidence_accepts_bounded_post_stop_delivery() -> None: + assert_direct_evp_shutdown_evidence(_direct_shutdown_evidence()) + + +@pytest.mark.parametrize( + ("mutation", "message"), + [ + ("explicit_flush", "must not call"), + ("request_before_marker", "before server close"), + ("duplicate_capture", "exactly one event"), + ("over_bound", "exceeding the"), + ("stop_error", "Shutdown returned an error"), + ("still_running", "remained running"), + ("oom_killed", "OOM-killed"), + ("signal_exit", "graceful shutdown path"), + ("missing_marker", "exactly one structured server-close marker"), + ("duplicate_marker", "exactly one structured server-close marker"), + ("missing_prime", "fresh flush window"), + ("duplicate_prime", "exactly one capture"), + ("same_prime_subject", "subject distinct"), + ], +) +@scenarios.test_the_test +@features.not_reported +def test_feature_flagging_direct_shutdown_evidence_rejects_false_proofs( + mutation: str, + message: str, +) -> None: + evidence = _direct_shutdown_evidence() + if mutation == "explicit_flush": + evidence["explicit_flush"] = True + elif mutation == "request_before_marker": + evidence["capture_request_started_at"] = ["2026-09-09T12:00:01.125000+00:00"] + elif mutation == "duplicate_capture": + evidence["captures_after_settle"] = 2 + elif mutation == "over_bound": + evidence["shutdown_duration_seconds"] = 12.1 + elif mutation == "stop_error": + evidence["stop_error"] = "RuntimeError: stop failed" + elif mutation == "still_running": + evidence["stopped_container"]["running"] = True + elif mutation == "oom_killed": + evidence["stopped_container"]["oom_killed"] = True + elif mutation == "signal_exit": + evidence["stopped_container"]["exit_code"] = 143 + elif mutation == "missing_marker": + evidence["shutdown_markers"] = [] + elif mutation == "duplicate_marker": + evidence["shutdown_markers"].append(deepcopy(evidence["shutdown_markers"][0])) + elif mutation == "missing_prime": + evidence["flush_window_primed"] = False + elif mutation == "duplicate_prime": + evidence["priming_capture_files"].append("direct-prime-0002.json") + elif mutation == "same_prime_subject": + evidence["priming_subject_id"] = evidence["subject_id"] + else: + raise AssertionError(f"Unhandled mutation {mutation}") + + with pytest.raises(AssertionError, match=message): + assert_direct_evp_shutdown_evidence(evidence) diff --git a/tests/test_the_test/test_mock_ffe_agentless_backend.py b/tests/test_the_test/test_mock_ffe_agentless_backend.py index d3568744b2b..3c0a538dbe2 100644 --- a/tests/test_the_test/test_mock_ffe_agentless_backend.py +++ b/tests/test_the_test/test_mock_ffe_agentless_backend.py @@ -1,13 +1,14 @@ """Unit coverage for the mock FFE agentless backend test fixture.""" +from collections.abc import Callable from pathlib import Path -from typing import Literal +from typing import Any, Literal from unittest.mock import MagicMock import requests import pytest -from utils import features, scenarios +from utils import features, interfaces, scenarios from utils._context.containers import ServerlessInitContainer from utils._context._scenarios import agentless_endtoend as agentless_endtoend_scenarios from utils._context._scenarios import endtoend as endtoend_scenarios @@ -19,7 +20,14 @@ MockFFEAgentlessBackendServer, UFC_RESPONSE_TYPE, ) -from utils._context._scenarios.agentless_endtoend import FeatureFlaggingAgentlessEndToEndScenario +from utils._context._scenarios.agentless_endtoend import ( + DIRECT_EVP_AGENT_VARIABLES, + DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH, + DIRECT_EVP_CA_BUNDLE_SOURCE, + DIRECT_EVP_CAPTURE_SETTLE_SECONDS, + DIRECT_EVP_CAPTURE_WAIT_SECONDS, + FeatureFlaggingAgentlessEndToEndScenario, +) from utils.proxy.ports import ProxyPorts @@ -114,7 +122,8 @@ def test_agentless_end_to_end_scenario_starts_backend_before_weblog() -> None: scenario._start_mock_backend() # noqa: SLF001 - focused lifecycle test environment = scenario.weblog_infra.library_container.environment - assert "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE" not in environment + assert environment["DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED"] == "true" + assert environment["DD_FEATURE_FLAGS_CONFIGURATION_SOURCE"] == "agentless" assert "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT" not in environment base_url = environment["DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL"] assert isinstance(base_url, str) @@ -152,13 +161,21 @@ def test_agentless_exposure_scenario_has_no_agent_and_two_capture_routes( assert environment["DD_SITE"] == "mock-intake.invalid" assert environment["DD_PROXY_HTTPS"] == f"http://proxy:{ProxyPorts.datadog_direct}" assert environment["HTTPS_PROXY"] == f"http://proxy:{ProxyPorts.datadog_direct}" + assert environment["DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED"] == "true" + assert environment["DD_FEATURE_FLAGS_CONFIGURATION_SOURCE"] == "agentless" if exposure_egress == "direct": - for name in ("DD_AGENT_HOST", "DD_DOGSTATSD_HOST", "DD_TRACE_AGENT_PORT", "DD_TRACE_AGENT_URL"): + assert environment["SYSTEM_TESTS_FFE_SHUTDOWN_FLUSH_ENABLED"] == "true" + assert scenario.weblog_infra.library_container.volumes[DIRECT_EVP_CA_BUNDLE_SOURCE] == { + "bind": DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH, + "mode": "ro", + } + for name in DIRECT_EVP_AGENT_VARIABLES: assert name not in environment assert not serverless_init_containers return + assert DIRECT_EVP_CA_BUNDLE_SOURCE not in scenario.weblog_infra.library_container.volumes serverless_init = scenario.serverless_init_container assert serverless_init_containers == (serverless_init,) assert isinstance(serverless_init, ServerlessInitContainer) @@ -170,6 +187,233 @@ def test_agentless_exposure_scenario_has_no_agent_and_two_capture_routes( assert serverless_init.environment["DD_PROXY_HTTP"] == f"http://proxy:{ProxyPorts.datadog_sidecar}" +@scenarios.test_the_test +@features.not_reported +def test_agentless_evp_capture_registry_rejects_missing_and_unknown_paths() -> None: + scenario = FeatureFlaggingAgentlessEndToEndScenario( + "MOCK_FFE_AGENTLESS_DIRECT_CAPTURE_REGISTRY", + doc="test", + exposure_egress="direct", + ) + + with pytest.raises(RuntimeError, match="registered no expected capture paths"): + scenario._wait_for_expected_evp_captures(is_empty_test_run=False) # noqa: SLF001 + + # Empty selections and replay runs have no live setup phase and therefore need no registration. + scenario._wait_for_expected_evp_captures(is_empty_test_run=True) # noqa: SLF001 + scenario.replay = True + scenario._wait_for_expected_evp_captures(is_empty_test_run=False) # noqa: SLF001 + + with pytest.raises(ValueError, match="Unsupported Feature Flags EVP path"): + scenario.register_expected_evp_capture("/api/v2/not-a-signal") + + +@scenarios.test_the_test +@features.not_reported +def test_agentless_evp_capture_registry_waits_for_each_path_before_settling( + monkeypatch: pytest.MonkeyPatch, +) -> None: + scenario = FeatureFlaggingAgentlessEndToEndScenario( + "MOCK_FFE_AGENTLESS_DIRECT_CAPTURE_WAIT", + doc="test", + exposure_egress="direct", + ) + expected_paths = ("/api/v2/exposures", "/api/v2/flagevaluation") + for path in expected_paths: + scenario.register_expected_evp_capture(path) + + captures = iter({"path": path} for path in expected_paths) + wait_timeouts: list[int] = [] + + def wait_for(matcher: Callable[[dict[str, Any]], bool], *, timeout: int) -> bool: + wait_timeouts.append(timeout) + return matcher(next(captures)) + + wait = MagicMock() + monkeypatch.setattr(interfaces.datadog_direct, "wait_for", wait_for) + monkeypatch.setattr(interfaces.datadog_direct, "wait", wait) + + scenario._wait_for_expected_evp_captures(is_empty_test_run=False) # noqa: SLF001 + + assert wait_timeouts == [DIRECT_EVP_CAPTURE_WAIT_SECONDS, DIRECT_EVP_CAPTURE_WAIT_SECONDS] + wait.assert_called_once_with(DIRECT_EVP_CAPTURE_SETTLE_SECONDS) + + +@scenarios.test_the_test +@features.not_reported +def test_agentless_evp_capture_wait_happens_before_container_stop(monkeypatch: pytest.MonkeyPatch) -> None: + scenario = FeatureFlaggingAgentlessEndToEndScenario( + "MOCK_FFE_AGENTLESS_DIRECT_CAPTURE_ORDER", + doc="test", + exposure_egress="direct", + ) + lifecycle: list[str] = [] + + def wait_for_captures(*, is_empty_test_run: bool) -> None: + assert is_empty_test_run is False + lifecycle.append("capture") + + def stop_containers(self: object, *, is_empty_test_run: bool) -> None: + assert self is scenario + assert is_empty_test_run is False + lifecycle.append("stop") + + monkeypatch.setattr(scenario, "_wait_for_expected_evp_captures", wait_for_captures) + monkeypatch.setattr(endtoend_scenarios.DdTraceEndToEndScenario, "_wait_and_stop_containers", stop_containers) + + scenario._wait_and_stop_containers(is_empty_test_run=False) # noqa: SLF001 + + assert lifecycle == ["capture", "stop"] + + +@scenarios.test_the_test +@features.not_reported +def test_direct_evp_shutdown_probe_uses_sigterm_without_explicit_flush( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + monkeypatch.chdir(tmp_path) + scenario = FeatureFlaggingAgentlessEndToEndScenario( + "MOCK_FFE_AGENTLESS_DIRECT_SHUTDOWN", + doc="test", + exposure_egress="direct", + ) + Path(scenario.host_log_folder).mkdir() + scenario.register_shutdown_evp_evaluation( + signal_path="/api/v2/exposures", + request_path="/ffe", + body={ + "flag": "empty-targeting-key-flag", + "variationType": "STRING", + "defaultValue": "default", + "targetingKey": "shutdown-user", + "attributes": {}, + }, + flag_key="empty-targeting-key-flag", + subject_id="shutdown-user", + ) + + lifecycle: list[str] = [] + monkeypatch.setattr( + scenario, + "_capture_direct_evp_runtime_evidence", + lambda: lifecycle.append("runtime-evidence"), + ) + + response = MagicMock() + response.status_code = 200 + + def evaluate(*_args: object, **_kwargs: object) -> MagicMock: + lifecycle.append("evaluate") + return response + + monkeypatch.setattr(agentless_endtoend_scenarios.weblog, "post", evaluate) + + priming_capture = { + "path": "/api/v2/exposures", + "log_filename": "direct-0000.json", + "request": { + "timestamp_start": "2099-09-09T11:59:59+00:00", + "content": { + "exposures": [ + { + "flag": {"key": "empty-targeting-key-flag"}, + "subject": {"id": "shutdown-user-flush-window-prime"}, + } + ] + }, + }, + } + capture = { + "path": "/api/v2/exposures", + "log_filename": "direct-0001.json", + "request": { + "timestamp_start": "2099-09-09T12:00:01+00:00", + "content": { + "exposures": [ + { + "flag": {"key": "empty-targeting-key-flag"}, + "subject": {"id": "shutdown-user"}, + } + ] + }, + }, + } + snapshots = iter( + ( + [priming_capture], + [priming_capture], + [priming_capture], + [priming_capture, capture], + ) + ) + + def get_data() -> list[dict[str, Any]]: + lifecycle.append("capture-snapshot") + return next(snapshots) + + def wait_for(matcher: Callable[[dict[str, Any]], bool], *, timeout: int) -> bool: + lifecycle.append("capture-wait") + assert timeout == DIRECT_EVP_CAPTURE_WAIT_SECONDS + return any(matcher(candidate) for candidate in (priming_capture, capture)) + + settle = MagicMock(side_effect=lambda _: lifecycle.append("settle")) + monkeypatch.setattr(interfaces.datadog_direct, "get_data", get_data) + monkeypatch.setattr(interfaces.datadog_direct, "wait_for", wait_for) + monkeypatch.setattr(interfaces.datadog_direct, "wait", settle) + + runtime_container = MagicMock() + runtime_container.status = "exited" + runtime_container.attrs = { + "State": { + "Error": "", + "ExitCode": 0, + "FinishedAt": "2099-09-09T12:00:02Z", + "OOMKilled": False, + "Running": False, + } + } + runtime_container.logs.side_effect = lambda *, stdout, stderr: ( + b'{"event":"system_tests.ffe.shutdown.server_closed","timestamp":"2099-09-09T12:00:00Z"}\n' + if stdout and not stderr + else b"" + ) + scenario.weblog_infra.library_container._container = runtime_container # noqa: SLF001 + stop = MagicMock(side_effect=lambda **_: lifecycle.append("docker-stop")) + monkeypatch.setattr(scenario.weblog_infra, "stop", stop) + + scenario._stop_weblog(is_empty_test_run=False) # noqa: SLF001 + + stop.assert_called_once_with(flush=False, stop_timeout=10) + settle.assert_called_once_with(DIRECT_EVP_CAPTURE_SETTLE_SECONDS) + assert lifecycle == [ + "runtime-evidence", + "evaluate", + "capture-wait", + "capture-snapshot", + "capture-snapshot", + "evaluate", + "capture-snapshot", + "docker-stop", + "capture-wait", + "settle", + "capture-snapshot", + ] + evidence = scenario.direct_evp_shutdown_evidence() + assert evidence["explicit_flush"] is False + assert evidence["captures_before_evaluation"] == 0 + assert evidence["captures_before_stop"] == 0 + assert evidence["captures_after_settle"] == 1 + assert evidence["flush_window_primed"] is True + assert evidence["priming_subject_id"] == "shutdown-user-flush-window-prime" + assert evidence["priming_capture_files"] == ["direct-0000.json"] + assert evidence["priming_capture_request_started_at"] == ["2099-09-09T11:59:59+00:00"] + assert evidence["priming_evaluation_status_code"] == 200 + assert evidence["shutdown_marker_errors"] == [] + assert len(evidence["shutdown_markers"]) == 1 + assert evidence["stopped_container"]["exit_code"] == 0 + + @scenarios.test_the_test def test_agentless_end_to_end_scenario_closes_backend_when_startup_fails( monkeypatch: pytest.MonkeyPatch, diff --git a/utils/_context/_scenarios/agentless_endtoend.py b/utils/_context/_scenarios/agentless_endtoend.py index d461c21de35..7a02cd32987 100644 --- a/utils/_context/_scenarios/agentless_endtoend.py +++ b/utils/_context/_scenarios/agentless_endtoend.py @@ -1,12 +1,19 @@ +from dataclasses import dataclass +from datetime import UTC, datetime import json from pathlib import Path -from typing import TYPE_CHECKING, Literal, cast +import time +from typing import TYPE_CHECKING, Any, Literal, cast +from docker.models.containers import Container import pytest from utils import interfaces from utils._context.containers import ServerlessInitContainer, TestedContainer +from utils._context.docker import get_docker_client from utils.docker_fixtures._core import extra_hosts_for_environment +from utils._logger import logger +from utils._weblog import weblog from utils.mocked_backend.ffe import ( EXPECTED_API_KEY, MockFFEAgentlessBackendServer, @@ -21,6 +28,39 @@ from utils.interfaces._core import ProxyBasedInterfaceValidator +DIRECT_EVP_CA_BUNDLE_SOURCE = "./utils/build/docker/agent/ca-certificates.crt" +DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH = "/etc/ssl/certs/ca-certificates.crt" +DIRECT_EVP_SIGNAL_PATHS = frozenset({"/api/v2/exposures", "/api/v2/flagevaluation"}) +DIRECT_EVP_CAPTURE_WAIT_SECONDS = 30 +DIRECT_EVP_CAPTURE_SETTLE_SECONDS = 3 +DIRECT_EVP_RUNTIME_EVIDENCE_FILENAME = "direct_evp_runtime.json" +DIRECT_EVP_SHUTDOWN_EVIDENCE_FILENAME = "direct_evp_shutdown.json" +DIRECT_EVP_STOP_TIMEOUT_SECONDS = 10 +DIRECT_EVP_SHUTDOWN_BOUND_SECONDS = DIRECT_EVP_STOP_TIMEOUT_SECONDS + 2 +DIRECT_EVP_SHUTDOWN_MARKER_EVENT = "system_tests.ffe.shutdown.server_closed" +DIRECT_EVP_AGENT_VARIABLES = ( + "DD_AGENT_HOST", + "DD_DOGSTATSD_HOST", + "DD_DOGSTATSD_PORT", + "DD_DOGSTATSD_URL", + "DD_TRACE_AGENT_HOSTNAME", + "DD_TRACE_AGENT_PORT", + "DD_TRACE_AGENT_URL", +) +DIRECT_EVP_SHUTDOWN_FLUSH_ENV = "SYSTEM_TESTS_FFE_SHUTDOWN_FLUSH_ENABLED" + + +@dataclass(frozen=True) +class DirectEVPShutdownEvaluation: + """One evaluation deferred until the scenario-owned shutdown-flush phase.""" + + signal_path: str + request_path: str + body: dict[str, Any] + flag_key: str + subject_id: str + + class AgentlessEndToEndScenario(DdTraceEndToEndScenario): """End-to-end scenario without a Datadog Agent, using agentless delivery mechanisms.""" @@ -129,7 +169,16 @@ def __init__( weblog_env: dict[str, str | None] | None = None, ) -> None: self.exposure_egress = exposure_egress + self._expected_evp_capture_paths: set[str] = set() + self._shutdown_evp_evaluation: DirectEVPShutdownEvaluation | None = None + self._last_direct_evp_runtime_evidence: dict[str, Any] | None = None + self._last_direct_evp_shutdown_evidence: dict[str, Any] | None = None environment: dict[str, str | None] = { + # The shared weblogs use this switch to install their OpenFeature provider. The + # configuration source selects how the provider receives flags; it does not make the + # application adopt the provider on its own. + "DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED": "true", + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE": "agentless", # Both variables are integer seconds across the SDKs: Java parses them with # getInteger, and the shared configuration registry declares them "int" with an # allowed pattern of [1-9]\d*. A fractional value only ever worked on Node, which @@ -169,9 +218,14 @@ def __init__( ) if exposure_egress == "direct": + self.weblog_infra.library_container.environment[DIRECT_EVP_SHUTDOWN_FLUSH_ENV] = "true" + self.weblog_infra.library_container.volumes[DIRECT_EVP_CA_BUNDLE_SOURCE] = { + "bind": DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH, + "mode": "ro", + } # Direct mode uses the proxy only to capture HTTPS intake requests. # Do not advertise the proxy as a local Agent endpoint. - for env_name in ("DD_AGENT_HOST", "DD_DOGSTATSD_HOST", "DD_TRACE_AGENT_PORT", "DD_TRACE_AGENT_URL"): + for env_name in DIRECT_EVP_AGENT_VARIABLES: self.weblog_infra.library_container.environment.pop(env_name, None) def configure(self, config: pytest.Config) -> None: @@ -211,18 +265,394 @@ def _set_components(self) -> None: if self.exposure_egress == "sidecar": self.components["serverless-init"] = self.serverless_init_container.serverless_init_version - def _wait_and_stop_containers(self, *, is_empty_test_run: bool) -> None: - super()._wait_and_stop_containers(is_empty_test_run=is_empty_test_run) + def register_expected_evp_capture(self, path: str) -> None: + """Register a canonical signal path that must be captured before shutdown.""" if self.exposure_egress is None: + raise ValueError("EVP capture expectations require an egress scenario") + if path not in DIRECT_EVP_SIGNAL_PATHS: + raise ValueError(f"Unsupported Feature Flags EVP path {path!r}") + self._expected_evp_capture_paths.add(path) + + def register_shutdown_evp_evaluation( + self, + *, + signal_path: str, + request_path: str, + body: dict[str, Any], + flag_key: str, + subject_id: str, + ) -> None: + """Defer one evaluation to immediately before the bounded shutdown flush.""" + if self.exposure_egress != "direct": + raise ValueError("Shutdown EVP evaluation requires the direct egress scenario") + if signal_path not in DIRECT_EVP_SIGNAL_PATHS: + raise ValueError(f"Unsupported Feature Flags EVP path {signal_path!r}") + if self._shutdown_evp_evaluation is not None: + raise ValueError("Only one shutdown EVP evaluation may be registered") + self._shutdown_evp_evaluation = DirectEVPShutdownEvaluation( + signal_path=signal_path, + request_path=request_path, + body=body, + flag_key=flag_key, + subject_id=subject_id, + ) + + @property + def _direct_evp_runtime_evidence_path(self) -> Path: + return Path(self.host_log_folder) / DIRECT_EVP_RUNTIME_EVIDENCE_FILENAME + + @property + def _direct_evp_shutdown_evidence_path(self) -> Path: + return Path(self.host_log_folder) / DIRECT_EVP_SHUTDOWN_EVIDENCE_FILENAME + + def direct_evp_runtime_evidence(self) -> dict[str, Any]: + """Return preserved live-container evidence for a direct scenario run.""" + if self._last_direct_evp_runtime_evidence is None: + raise AssertionError("Direct EVP runtime evidence was not captured") + return self._last_direct_evp_runtime_evidence + + def direct_evp_shutdown_evidence(self) -> dict[str, Any]: + """Return preserved shutdown-flush evidence for a direct scenario run.""" + if self._last_direct_evp_shutdown_evidence is None: + raise AssertionError("Direct EVP shutdown evidence was not captured") + return self._last_direct_evp_shutdown_evidence + + @staticmethod + def _process_rows(container: Container) -> tuple[list[dict[str, str]], str | None]: + try: + process_table = container.top(ps_args="-eo pid,ppid,comm,args") + titles = [str(title).lower() for title in process_table.get("Titles", [])] + processes = [ + {title: str(value) for title, value in zip(titles, row, strict=True)} + for row in process_table.get("Processes", []) + ] + return processes, None + except BaseException as error: + return [], f"{type(error).__name__}: {error}" + + @staticmethod + def _pid1_command(container: Container) -> tuple[str, str | None]: + try: + result = container.exec_run(["sh", "-c", "tr '\\000' ' ' None: + """Capture live Docker/network/process state without retaining credentials.""" + self._network.reload() + network_members = self._network.attrs.get("Containers") or {} + live_containers = get_docker_client().containers.list( + all=False, + filters={"network": self._network.id}, + ) + containers: list[dict[str, Any]] = [] + weblog_environment: dict[str, Any] | None = None + + for container in sorted(live_containers, key=lambda item: item.name): + container.reload() + attributes = container.attrs + state = attributes.get("State", {}) + config = attributes.get("Config", {}) + labels = config.get("Labels") or {} + environment = dict( + item.split("=", 1) for item in config.get("Env", []) if isinstance(item, str) and "=" in item + ) + processes, top_error = self._process_rows(container) + pid1_command, pid1_error = self._pid1_command(container) + containers.append( + { + "id": container.id, + "name": container.name, + "image": config.get("Image"), + "image_id": attributes.get("Image"), + "library": labels.get("system-tests-library"), + "weblog_variant": labels.get("system-tests-weblog-variant"), + "status": container.status, + "running": state.get("Running"), + "state_pid": state.get("Pid"), + "pid1_command": pid1_command, + "pid1_error": pid1_error, + "processes": processes, + "top_error": top_error, + "networks": sorted(attributes.get("NetworkSettings", {}).get("Networks", {})), + "mounts": [ + { + "destination": mount.get("Destination"), + "read_write": mount.get("RW"), + "source": mount.get("Source"), + "type": mount.get("Type"), + } + for mount in attributes.get("Mounts", []) + ], + } + ) + + if container.name == self.weblog_infra.library_container.container_name: + weblog_environment = { + "agent_variables_present": sorted( + name for name in DIRECT_EVP_AGENT_VARIABLES if name in environment + ), + "api_key_matches_expected": environment.get("DD_API_KEY") == EXPECTED_API_KEY, + "api_key_present": bool(environment.get("DD_API_KEY")), + "configuration_source": environment.get("DD_FEATURE_FLAGS_CONFIGURATION_SOURCE"), + "custom_configuration_url_present": bool( + environment.get("DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL") + ), + "provider_enabled": environment.get("DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED"), + "shutdown_flush_enabled": environment.get(DIRECT_EVP_SHUTDOWN_FLUSH_ENV), + "site": environment.get("DD_SITE"), + "tls_ca_bundle": environment.get("NODE_EXTRA_CA_CERTS") + or environment.get("SSL_CERT_FILE") + or environment.get("REQUESTS_CA_BUNDLE"), + } + + self._last_direct_evp_runtime_evidence = { + "captured_at": datetime.now(UTC).isoformat(), + "containers": containers, + "network": { + "id": self._network.id, + "name": self._network.name, + "container_ids": sorted(network_members), + "container_names": sorted( + member.get("Name") for member in network_members.values() if isinstance(member.get("Name"), str) + ), + }, + "weblog_environment": weblog_environment, + } + self._direct_evp_runtime_evidence_path.write_text( + json.dumps(self._last_direct_evp_runtime_evidence, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + @staticmethod + def _capture_has_shutdown_evaluation(data: dict[str, Any], evaluation: DirectEVPShutdownEvaluation) -> bool: + if data.get("path") != evaluation.signal_path: + return False + content = data.get("request", {}).get("content") + if not isinstance(content, dict): + return False + events = content.get("exposures") + if not isinstance(events, list): + events = content.get("flagEvaluations") + if not isinstance(events, list): + return False + return any( + isinstance(event, dict) + and isinstance(event.get("flag"), dict) + and event["flag"].get("key") == evaluation.flag_key + and isinstance(event.get("subject"), dict) + and event["subject"].get("id") == evaluation.subject_id + for event in events + ) + + @staticmethod + def _stopped_container_state(container: Container) -> dict[str, Any]: + try: + container.reload() + state = container.attrs.get("State", {}) + return { + "error": state.get("Error"), + "exit_code": state.get("ExitCode"), + "finished_at": state.get("FinishedAt"), + "oom_killed": state.get("OOMKilled"), + "running": state.get("Running"), + "status": container.status, + } + except BaseException as error: + return {"inspection_error": f"{type(error).__name__}: {error}"} + + @staticmethod + def _shutdown_markers(container: Container) -> tuple[list[dict[str, str]], list[str]]: + """Read structured shutdown lifecycle markers from the stopped container's logs.""" + markers: list[dict[str, str]] = [] + errors: list[str] = [] + for stream, options in ( + ("stdout", {"stdout": True, "stderr": False}), + ("stderr", {"stdout": False, "stderr": True}), + ): + try: + output = container.logs(**options) + text = output.decode("utf-8") if isinstance(output, bytes) else str(output) + except BaseException as error: + errors.append(f"{stream}: {type(error).__name__}: {error}") + continue + + for line in text.splitlines(): + try: + value = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(value, dict) or value.get("event") != DIRECT_EVP_SHUTDOWN_MARKER_EVENT: + continue + markers.append( + { + "event": str(value.get("event")), + "stream": stream, + "timestamp": str(value.get("timestamp")), + } + ) + return markers, errors + + def _stop_weblog(self, *, is_empty_test_run: bool) -> None: + if self.exposure_egress != "direct" or self.replay or is_empty_test_run: + super()._stop_weblog(is_empty_test_run=is_empty_test_run) + return + + self._capture_direct_evp_runtime_evidence() + evaluation = self._shutdown_evp_evaluation + if evaluation is None: + super()._stop_weblog(is_empty_test_run=is_empty_test_run) return - if self.replay: - self._load_telemetry_interfaces() - elif self.exposure_egress == "sidecar": - self.serverless_init_container.stop() + interface = interfaces.datadog_direct + + def matcher(data: dict[str, Any]) -> bool: + return self._capture_has_shutdown_evaluation(data, evaluation) + + # Timer-based writers may flush the first event after a long idle period immediately. + # Complete one unique exposure first so the target below is produced inside a fresh flush + # window. The priming event is evidence, not the event under test; the target must still + # start its request only after the runtime's truthful server-close marker. + priming_subject_id = f"{evaluation.subject_id}-flush-window-prime" + priming_body = dict(evaluation.body) + priming_body["targetingKey"] = priming_subject_id + priming_evaluation = DirectEVPShutdownEvaluation( + signal_path=evaluation.signal_path, + request_path=evaluation.request_path, + body=priming_body, + flag_key=evaluation.flag_key, + subject_id=priming_subject_id, + ) + + def priming_matcher(data: dict[str, Any]) -> bool: + return self._capture_has_shutdown_evaluation(data, priming_evaluation) + + priming_started_at = datetime.now(UTC) + priming_response = weblog.post(priming_evaluation.request_path, json=priming_evaluation.body) + priming_finished_at = datetime.now(UTC) + priming_observed = interface.wait_for(priming_matcher, timeout=DIRECT_EVP_CAPTURE_WAIT_SECONDS) + priming_captures = list(filter(priming_matcher, interface.get_data())) + if not priming_observed or len(priming_captures) != 1: + raise RuntimeError( + "Could not establish one completed direct-EVP flush immediately before the shutdown target" + ) + + matching_before = list(filter(matcher, interface.get_data())) + evaluation_started_at = datetime.now(UTC) + response = weblog.post(evaluation.request_path, json=evaluation.body) + evaluation_finished_at = datetime.now(UTC) + matching_before_stop = list(filter(matcher, interface.get_data())) + + runtime_container = self.weblog_infra.library_container.runtime_container + stop_started_at = datetime.now(UTC) + stop_started = time.monotonic() + stop_error: BaseException | None = None + try: + # Do not call the weblog's explicit /flush endpoint here. The event must remain + # buffered until Docker sends SIGTERM and the language runtime runs its shutdown + # lifecycle. + self.weblog_infra.stop(flush=False, stop_timeout=DIRECT_EVP_STOP_TIMEOUT_SECONDS) + except BaseException as error: + stop_error = error + stop_duration = time.monotonic() - stop_started + stopped_at = datetime.now(UTC) + stopped_state = self._stopped_container_state(runtime_container) + shutdown_markers, shutdown_marker_errors = self._shutdown_markers(runtime_container) + + capture_observed = interface.wait_for(matcher, timeout=DIRECT_EVP_CAPTURE_WAIT_SECONDS) + interface.wait(DIRECT_EVP_CAPTURE_SETTLE_SECONDS) + matching_after = list(filter(matcher, interface.get_data())) + self._last_direct_evp_shutdown_evidence = { + "capture_files": [capture.get("log_filename") for capture in matching_after], + "capture_observed_after_stop": capture_observed, + "capture_request_started_at": [ + capture.get("request", {}).get("timestamp_start") for capture in matching_after + ], + "captures_after_settle": len(matching_after), + "captures_before_evaluation": len(matching_before), + "captures_before_stop": len(matching_before_stop), + "evaluation_finished_at": evaluation_finished_at.isoformat(), + "evaluation_started_at": evaluation_started_at.isoformat(), + "evaluation_status_code": response.status_code, + "explicit_flush": False, + "flag_key": evaluation.flag_key, + "flush_window_primed": True, + "priming_capture_files": [capture.get("log_filename") for capture in priming_captures], + "priming_capture_request_started_at": [ + capture.get("request", {}).get("timestamp_start") for capture in priming_captures + ], + "priming_evaluation_finished_at": priming_finished_at.isoformat(), + "priming_evaluation_started_at": priming_started_at.isoformat(), + "priming_evaluation_status_code": priming_response.status_code, + "priming_subject_id": priming_subject_id, + "shutdown_bound_seconds": DIRECT_EVP_SHUTDOWN_BOUND_SECONDS, + "shutdown_duration_seconds": stop_duration, + "shutdown_marker_errors": shutdown_marker_errors, + "shutdown_markers": shutdown_markers, + "signal_path": evaluation.signal_path, + "stop_error": None if stop_error is None else f"{type(stop_error).__name__}: {stop_error}", + "stop_started_at": stop_started_at.isoformat(), + "stop_timeout_seconds": DIRECT_EVP_STOP_TIMEOUT_SECONDS, + "stopped_at": stopped_at.isoformat(), + "stopped_container": stopped_state, + "subject_id": evaluation.subject_id, + } + self._direct_evp_shutdown_evidence_path.write_text( + json.dumps(self._last_direct_evp_shutdown_evidence, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + if stop_error is not None: + raise stop_error + + def _wait_for_expected_evp_captures(self, *, is_empty_test_run: bool) -> None: + """Keep the application alive until selected async EVP writers are observable.""" + if self.replay or self.exposure_egress is None or is_empty_test_run: + return + if not self._expected_evp_capture_paths and self._shutdown_evp_evaluation is None: + raise RuntimeError("A non-empty Feature Flags EVP run registered no expected capture paths") + + interface = interfaces.datadog_sidecar if self.exposure_egress == "sidecar" else interfaces.datadog_direct + missing_paths: list[str] = [] + for path in sorted(self._expected_evp_capture_paths): + + def captured(data: dict[str, Any], expected_path: str = path) -> bool: + return data.get("path") == expected_path + + logger.terminal.write_sep("-", f"Wait for {path} on {interface} before stopping the weblog") + logger.terminal.flush() + if not interface.wait_for(captured, timeout=DIRECT_EVP_CAPTURE_WAIT_SECONDS): + missing_paths.append(path) + + if missing_paths: + raise RuntimeError(f"Timed out waiting for Feature Flags EVP captures: {', '.join(missing_paths)}") + + # Preserve a bounded window in which an unsafe retry would appear as a duplicate capture. + interface.wait(DIRECT_EVP_CAPTURE_SETTLE_SECONDS) + + def _wait_and_stop_containers(self, *, is_empty_test_run: bool) -> None: + try: + self._wait_for_expected_evp_captures(is_empty_test_run=is_empty_test_run) + finally: + if self.replay and self.exposure_egress == "direct": + self._last_direct_evp_runtime_evidence = json.loads( + self._direct_evp_runtime_evidence_path.read_text(encoding="utf-8") + ) + if self._direct_evp_shutdown_evidence_path.exists(): + self._last_direct_evp_shutdown_evidence = json.loads( + self._direct_evp_shutdown_evidence_path.read_text(encoding="utf-8") + ) + super()._wait_and_stop_containers(is_empty_test_run=is_empty_test_run) + if self.exposure_egress is not None: + if self.replay: + self._load_telemetry_interfaces() + elif self.exposure_egress == "sidecar": + self.serverless_init_container.stop() - interfaces.datadog_sidecar.check_deserialization_errors() - interfaces.datadog_direct.check_deserialization_errors() + interfaces.datadog_sidecar.check_deserialization_errors() + interfaces.datadog_direct.check_deserialization_errors() @staticmethod def _load_telemetry_interfaces() -> None: diff --git a/utils/_context/_scenarios/endtoend.py b/utils/_context/_scenarios/endtoend.py index 1afdd08c190..0f15c413c67 100644 --- a/utils/_context/_scenarios/endtoend.py +++ b/utils/_context/_scenarios/endtoend.py @@ -495,7 +495,7 @@ def _wait_and_stop_containers(self, *, is_empty_test_run: bool): # An empty selection has no test-generated data to flush. An Agentless scenario also # has no Agent-backed writer target, so its flush endpoint can only time out. - self.weblog_infra.stop(flush=not is_empty_test_run and self.include_agent) + self._stop_weblog(is_empty_test_run=is_empty_test_run) interfaces.library.check_deserialization_errors() for container in self.buddies: @@ -518,6 +518,10 @@ def _wait_and_stop_containers(self, *, is_empty_test_run: bool): if self._mocked_backend_v2: interfaces.backend_v2.check_deserialization_errors() + def _stop_weblog(self, *, is_empty_test_run: bool) -> None: + """Stop the weblog after setup traffic has been generated.""" + self.weblog_infra.stop(flush=not is_empty_test_run and self.include_agent) + def _wait_interface(self, interface: ProxyBasedInterfaceValidator, timeout: int): logger.terminal.write_sep("-", f"Wait for {interface} ({timeout}s)") logger.terminal.flush() diff --git a/utils/_context/containers.py b/utils/_context/containers.py index d50b4806127..f97408e88f3 100644 --- a/utils/_context/containers.py +++ b/utils/_context/containers.py @@ -422,7 +422,14 @@ def _fix_host_pwd_in_volumes(self): self.volumes = result - def stop(self): + @property + def runtime_container(self) -> Container: + """Return the live Docker object backing this configured container.""" + if self._container is None: + raise RuntimeError(f"Container {self.name} has not been started") + return self._container + + def stop(self, *, timeout: int | None = None) -> None: self._starting_thread = None logger.debug(f"Stopping container {self.name}") @@ -434,7 +441,10 @@ def stop(self): pytest.exit(f"Container {self.name} is not running ({self._container.status}), please check logs", 1) try: - self._container.stop() + if timeout is None: + self._container.stop() + else: + self._container.stop(timeout=timeout) except requests.exceptions.Timeout as e: pytest.exit( f"Container {self.name} failed to stop: the docker client timed out waiting for a response " diff --git a/utils/_context/weblog_infrastructure.py b/utils/_context/weblog_infrastructure.py index 43758fbd505..6d21cd1cf19 100644 --- a/utils/_context/weblog_infrastructure.py +++ b/utils/_context/weblog_infrastructure.py @@ -35,7 +35,7 @@ def configure(self, config: pytest.Config) -> None: """Perform any configuration. Executed only if the weblog will be used""" @abstractmethod - def stop(self, *, flush: bool = True) -> None: + def stop(self, *, flush: bool = True, stop_timeout: int | None = None) -> None: """Stop the tested infra""" @@ -214,18 +214,18 @@ def get_containers(self) -> tuple[TestedContainer, ...]: ) return (self.http_container, *self._other_containers) - def stop(self, *, flush: bool = True) -> None: + def stop(self, *, flush: bool = True, stop_timeout: int | None = None) -> None: if self._is_proxy_weblog: if self._proxy_runtime_container: - self._proxy_runtime_container.stop() + self._proxy_runtime_container.stop(timeout=stop_timeout) if self._processor_container: - self._processor_container.stop() + self._processor_container.stop(timeout=stop_timeout) if self._dummy_server_container: - self._dummy_server_container.stop() + self._dummy_server_container.stop(timeout=stop_timeout) else: if flush: self.http_container.flush() - self.http_container.stop() + self.http_container.stop(timeout=stop_timeout) @property def library_name(self) -> str: From 7b8639c9862d7a625bf027c572f8a3b74e822ba3 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Fri, 11 Sep 2026 02:42:46 +0000 Subject: [PATCH 2/6] test(ffe): scope EVP teardown to enabled contracts Skip agentless EVP setup for manifest-declared expected failures so unsupported languages cannot register scenario-level capture obligations. Keep new direct contracts explicitly disabled until each language activation lands. Environment: Datadog workspace --- manifests/dotnet.yml | 1 + manifests/golang.yml | 3 +++ manifests/java.yml | 1 + manifests/nodejs.yml | 1 + manifests/php.yml | 1 + manifests/python.yml | 1 + manifests/ruby.yml | 1 + tests/ffe/test_exposure_egress.py | 5 ++++- tests/ffe/test_flag_eval_evp.py | 2 ++ tests/test_the_test/test_ffe_evp_contract.py | 20 ++++++++++++++++++++ 10 files changed, 35 insertions(+), 1 deletion(-) diff --git a/manifests/dotnet.yml b/manifests/dotnet.yml index cabb6513319..ea58d370235 100644 --- a/manifests/dotnet.yml +++ b/manifests/dotnet.yml @@ -780,6 +780,7 @@ manifest: tests/ffe/test_dynamic_evaluation.py::Test_FFE_Unknown_Operator_Tolerance: bug (FFL-2184) tests/ffe/test_exposure_egress.py: v3.36.0 tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct: missing_feature (Not yet implemented) + tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: missing_feature (Not yet implemented) tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Sidecar: missing_feature (Not yet implemented) tests/ffe/test_exposures_datadog_agent.py: v3.36.0 tests/ffe/test_exposures_datadog_agent.py::Test_FFE_Exposure_Caching_Serial_Id_Appears: missing_feature (EX-3416) diff --git a/manifests/golang.yml b/manifests/golang.yml index 8dc6af08cc7..1e0bb53e2d7 100644 --- a/manifests/golang.yml +++ b/manifests/golang.yml @@ -1344,6 +1344,7 @@ manifest: excluded_weblog: [echo, gin, chi, net-http-orchestrion, net-http, uds-echo] tests/ffe/test_exposure_egress.py: v2.6.0-dev # Easy win for chi, echo, gin, net-http, net-http-orchestrion, uds-echo and version 2.5.0 tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct: missing_feature (Not yet implemented) + tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: missing_feature (Not yet implemented) tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Sidecar: missing_feature (Not yet implemented) tests/ffe/test_exposures_datadog_agent.py: v2.6.0-dev # Easy win for chi, echo, gin, net-http, net-http-orchestrion, uds-echo and version 2.5.0 tests/ffe/test_exposures_datadog_agent.py::Test_FFE_Exposure_Caching_Serial_Id_Appears: missing_feature (EX-3414) @@ -1351,6 +1352,8 @@ manifest: tests/ffe/test_exposures_datadog_agent.py::Test_FFE_Exposure_Caching_Serial_Id_Disappears: missing_feature (EX-3414) tests/ffe/test_exposures_datadog_agent.py::Test_FFE_Exposure_Serial_Id: missing_feature (EX-3414) tests/ffe/test_flag_eval_evp.py: v2.10.0-dev + tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_Egress_Agentless_Direct: missing_feature (Not yet implemented) + tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_Egress_Agentless_Sidecar: missing_feature (Not yet implemented) tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_ObserveFullData_Absent_Hashed: v2.11.0-dev tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_ObserveFullData_False_Hashed: v2.11.0-dev tests/ffe/test_flag_eval_evp.py::Test_FFE_EVP_Flagevaluation_ObserveFullData_True_Unhashed: v2.11.0-dev # Easy win for chi, echo, gin, net-http, net-http-orchestrion, uds-echo and version 2.10.0 diff --git a/manifests/java.yml b/manifests/java.yml index 2423ede9155..9435fefbed6 100644 --- a/manifests/java.yml +++ b/manifests/java.yml @@ -3352,6 +3352,7 @@ manifest: "*": irrelevant spring-boot: v1.56.0 tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct: missing_feature (Not yet implemented) + tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: missing_feature (Not yet implemented) tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Sidecar: missing_feature (Not yet implemented) tests/ffe/test_exposures_datadog_agent.py: - weblog_declaration: diff --git a/manifests/nodejs.yml b/manifests/nodejs.yml index b4181747edb..03446f3187b 100644 --- a/manifests/nodejs.yml +++ b/manifests/nodejs.yml @@ -1837,6 +1837,7 @@ manifest: "*": incomplete_test_app express4: *ref_5_77_0 tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct: missing_feature (dd-trace-js#9527) + tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: missing_feature (dd-trace-js#9527) tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Sidecar: missing_feature (dd-trace-js#9527) tests/ffe/test_exposures_datadog_agent.py: - weblog_declaration: diff --git a/manifests/php.yml b/manifests/php.yml index f1dea045518..ef207d12f2a 100644 --- a/manifests/php.yml +++ b/manifests/php.yml @@ -1242,6 +1242,7 @@ manifest: laravel11x: incomplete_test_app symfony7x: incomplete_test_app tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct: missing_feature (Not yet implemented) + tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: missing_feature (Not yet implemented) tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Sidecar: missing_feature (Not yet implemented) tests/ffe/test_exposures_datadog_agent.py: - weblog_declaration: diff --git a/manifests/python.yml b/manifests/python.yml index 062f956d69a..b3cb4106c23 100644 --- a/manifests/python.yml +++ b/manifests/python.yml @@ -1463,6 +1463,7 @@ manifest: tests/ffe/test_dynamic_evaluation.py::Test_FFE_RC_Unavailable: flaky (FFL-1622) tests/ffe/test_exposure_egress.py: v4.2.0-dev tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct: missing_feature (Not yet implemented) + tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: missing_feature (Not yet implemented) tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Sidecar: missing_feature (Not yet implemented) tests/ffe/test_exposures_datadog_agent.py: v4.2.0-dev tests/ffe/test_exposures_datadog_agent.py::Test_FFE_Exposure_Caching_Serial_Id_Appears: missing_feature (EX-3413) diff --git a/manifests/ruby.yml b/manifests/ruby.yml index ea879a6cafa..4a3f96dd649 100644 --- a/manifests/ruby.yml +++ b/manifests/ruby.yml @@ -2036,6 +2036,7 @@ manifest: "*": irrelevant rails72: v2.23.0-dev tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct: missing_feature (Not yet implemented) + tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: missing_feature (Not yet implemented) tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Sidecar: missing_feature (Not yet implemented) tests/ffe/test_exposures_datadog_agent.py: - weblog_declaration: diff --git a/tests/ffe/test_exposure_egress.py b/tests/ffe/test_exposure_egress.py index f2968df841a..e98c1020f81 100644 --- a/tests/ffe/test_exposure_egress.py +++ b/tests/ffe/test_exposure_egress.py @@ -14,7 +14,7 @@ exposure_events_from_data, ) from tests.ffe.utils.fixtures import make_ufc_fixture -from utils import context, features, remote_config as rc, scenarios, weblog +from utils import context, features, remote_config as rc, scenario_crash, scenarios, weblog from utils._context._scenarios.agentless_endtoend import FeatureFlaggingAgentlessEndToEndScenario RC_PATH = "datadog/2/FFE_FLAGS" @@ -87,12 +87,14 @@ class Test_FFE_Exposure_Egress_Datadog_Agent(ExposureEgressContract): pass +@scenario_crash @scenarios.feature_flagging_and_experimentation_agentless_direct @features.feature_flags_exposures class Test_FFE_Exposure_Egress_Agentless_Direct(ExposureEgressContract): pass +@scenario_crash @scenarios.feature_flagging_and_experimentation_agentless_direct @features.feature_flags_exposures class Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: @@ -161,6 +163,7 @@ def test_exposure_egress_shutdown(self) -> None: ) +@scenario_crash @scenarios.feature_flagging_and_experimentation_agentless_serverless @features.feature_flags_exposures class Test_FFE_Exposure_Egress_Agentless_Sidecar(ExposureEgressContract): diff --git a/tests/ffe/test_flag_eval_evp.py b/tests/ffe/test_flag_eval_evp.py index 13b951eb7f9..28519e2e18f 100644 --- a/tests/ffe/test_flag_eval_evp.py +++ b/tests/ffe/test_flag_eval_evp.py @@ -340,12 +340,14 @@ class Test_FFE_EVP_Flagevaluation_Egress_Datadog_Agent(FlagevaluationEgressContr pass +@scenario_crash @scenarios.feature_flagging_and_experimentation_agentless_direct @features.feature_flags_evp_flagevaluation class Test_FFE_EVP_Flagevaluation_Egress_Agentless_Direct(FlagevaluationEgressContract): pass +@scenario_crash @scenarios.feature_flagging_and_experimentation_agentless_serverless @features.feature_flags_evp_flagevaluation class Test_FFE_EVP_Flagevaluation_Egress_Agentless_Sidecar(FlagevaluationEgressContract): diff --git a/tests/test_the_test/test_ffe_evp_contract.py b/tests/test_the_test/test_ffe_evp_contract.py index e8c7187baff..8f61d957931 100644 --- a/tests/test_the_test/test_ffe_evp_contract.py +++ b/tests/test_the_test/test_ffe_evp_contract.py @@ -5,6 +5,8 @@ import pytest +from tests.ffe import test_exposure_egress as exposure_egress_tests +from tests.ffe import test_flag_eval_evp as flag_eval_evp_tests from tests.ffe.utils.evp import ( EVP_ORIGINS, FeatureFlaggingEVPEgress, @@ -24,6 +26,15 @@ ) +AGENTLESS_EVP_CAPTURE_CONTRACTS = ( + exposure_egress_tests.Test_FFE_Exposure_Egress_Agentless_Direct, + exposure_egress_tests.Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown, + exposure_egress_tests.Test_FFE_Exposure_Egress_Agentless_Sidecar, + flag_eval_evp_tests.Test_FFE_EVP_Flagevaluation_Egress_Agentless_Direct, + flag_eval_evp_tests.Test_FFE_EVP_Flagevaluation_Egress_Agentless_Sidecar, +) + + def _direct_runtime_evidence(library_name: str = "nodejs") -> dict[str, Any]: executable = { "dotnet": "dotnet", @@ -166,6 +177,15 @@ def _direct_capture(*, path: str = "/api/v2/exposures", headers: list[list[str]] } +@pytest.mark.parametrize("contract", AGENTLESS_EVP_CAPTURE_CONTRACTS) +@scenarios.test_the_test +@features.not_reported +def test_agentless_evp_capture_contracts_skip_manifest_xfails_before_setup(contract: type[Any]) -> None: + marker_names = {marker.name for marker in getattr(contract, "pytestmark", ())} + + assert "skip_if_xfail" in marker_names + + @pytest.mark.parametrize(("library_name", "origin"), sorted(EVP_ORIGINS.items())) @scenarios.test_the_test @features.not_reported From ba1aa62fb95a03b379f0b0e3dbaff78b819cb94c Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Fri, 11 Sep 2026 04:39:39 +0000 Subject: [PATCH 3/6] fix(ffe): handle empty direct EVP replays Keep manifest-deactivated runs from requiring capture registration or evidence artifacts, and match shutdown flag evaluations by raw or hashed targeting key. Environment: Datadog workspace --- .../test_mock_ffe_agentless_backend.py | 80 ++++++++++++++++++- .../_context/_scenarios/agentless_endtoend.py | 47 ++++++++--- 2 files changed, 113 insertions(+), 14 deletions(-) diff --git a/tests/test_the_test/test_mock_ffe_agentless_backend.py b/tests/test_the_test/test_mock_ffe_agentless_backend.py index 3c0a538dbe2..71516b9d4d4 100644 --- a/tests/test_the_test/test_mock_ffe_agentless_backend.py +++ b/tests/test_the_test/test_mock_ffe_agentless_backend.py @@ -1,6 +1,7 @@ """Unit coverage for the mock FFE agentless backend test fixture.""" from collections.abc import Callable +import hashlib from pathlib import Path from typing import Any, Literal from unittest.mock import MagicMock @@ -189,15 +190,16 @@ def test_agentless_exposure_scenario_has_no_agent_and_two_capture_routes( @scenarios.test_the_test @features.not_reported -def test_agentless_evp_capture_registry_rejects_missing_and_unknown_paths() -> None: +def test_agentless_evp_capture_registry_allows_unregistered_and_rejects_unknown_paths() -> None: scenario = FeatureFlaggingAgentlessEndToEndScenario( "MOCK_FFE_AGENTLESS_DIRECT_CAPTURE_REGISTRY", doc="test", exposure_egress="direct", ) - with pytest.raises(RuntimeError, match="registered no expected capture paths"): - scenario._wait_for_expected_evp_captures(is_empty_test_run=False) # noqa: SLF001 + # Manifest-deactivated items are still collected without --skip-empty-scenario, but their + # setup methods do not run and therefore register no capture expectations. + scenario._wait_for_expected_evp_captures(is_empty_test_run=False) # noqa: SLF001 # Empty selections and replay runs have no live setup phase and therefore need no registration. scenario._wait_for_expected_evp_captures(is_empty_test_run=True) # noqa: SLF001 @@ -208,6 +210,42 @@ def test_agentless_evp_capture_registry_rejects_missing_and_unknown_paths() -> N scenario.register_expected_evp_capture("/api/v2/not-a-signal") +@pytest.mark.parametrize( + "captured_targeting_key", + [ + "shutdown-user", + f"sha256_{hashlib.sha256(b'shutdown-user').hexdigest()}", + ], +) +@scenarios.test_the_test +@features.not_reported +def test_direct_evp_shutdown_matcher_supports_flagevaluation_targeting_keys( + captured_targeting_key: str, +) -> None: + evaluation = agentless_endtoend_scenarios.DirectEVPShutdownEvaluation( + signal_path="/api/v2/flagevaluation", + request_path="/ffe", + body={}, + flag_key="shutdown-flag", + subject_id="shutdown-user", + ) + capture = { + "path": "/api/v2/flagevaluation", + "request": { + "content": { + "flagEvaluations": [ + { + "flag": {"key": "shutdown-flag"}, + "targeting_key": captured_targeting_key, + } + ] + } + }, + } + + assert FeatureFlaggingAgentlessEndToEndScenario._capture_has_shutdown_evaluation(capture, evaluation) # noqa: SLF001 + + @scenarios.test_the_test @features.not_reported def test_agentless_evp_capture_registry_waits_for_each_path_before_settling( @@ -266,6 +304,42 @@ def stop_containers(self: object, *, is_empty_test_run: bool) -> None: assert lifecycle == ["capture", "stop"] +@scenarios.test_the_test +@features.not_reported +@pytest.mark.parametrize("run_selection", ["manifest-deactivated", "empty"]) +def test_direct_evp_replay_skips_missing_runtime_evidence( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + run_selection: Literal["manifest-deactivated", "empty"], +) -> None: + monkeypatch.chdir(tmp_path) + scenario = FeatureFlaggingAgentlessEndToEndScenario( + "MOCK_FFE_AGENTLESS_DIRECT_EMPTY_REPLAY", + doc="test", + exposure_egress="direct", + ) + Path(scenario.host_log_folder).mkdir() + scenario.replay = True + is_empty_test_run = run_selection == "empty" + + stop_containers = MagicMock() + load_interfaces = MagicMock() + sidecar_errors = MagicMock() + direct_errors = MagicMock() + monkeypatch.setattr(endtoend_scenarios.DdTraceEndToEndScenario, "_wait_and_stop_containers", stop_containers) + monkeypatch.setattr(scenario, "_load_telemetry_interfaces", load_interfaces) + monkeypatch.setattr(interfaces.datadog_sidecar, "check_deserialization_errors", sidecar_errors) + monkeypatch.setattr(interfaces.datadog_direct, "check_deserialization_errors", direct_errors) + + scenario._wait_and_stop_containers(is_empty_test_run=is_empty_test_run) # noqa: SLF001 + + assert scenario._last_direct_evp_runtime_evidence is None # noqa: SLF001 + stop_containers.assert_called_once_with(is_empty_test_run=is_empty_test_run) + load_interfaces.assert_called_once_with() + sidecar_errors.assert_called_once_with() + direct_errors.assert_called_once_with() + + @scenarios.test_the_test @features.not_reported def test_direct_evp_shutdown_probe_uses_sigterm_without_explicit_flush( diff --git a/utils/_context/_scenarios/agentless_endtoend.py b/utils/_context/_scenarios/agentless_endtoend.py index 7a02cd32987..a3de541df40 100644 --- a/utils/_context/_scenarios/agentless_endtoend.py +++ b/utils/_context/_scenarios/agentless_endtoend.py @@ -1,5 +1,6 @@ from dataclasses import dataclass from datetime import UTC, datetime +import hashlib import json from pathlib import Path import time @@ -30,7 +31,9 @@ DIRECT_EVP_CA_BUNDLE_SOURCE = "./utils/build/docker/agent/ca-certificates.crt" DIRECT_EVP_CA_BUNDLE_CONTAINER_PATH = "/etc/ssl/certs/ca-certificates.crt" -DIRECT_EVP_SIGNAL_PATHS = frozenset({"/api/v2/exposures", "/api/v2/flagevaluation"}) +DIRECT_EVP_EXPOSURES_PATH = "/api/v2/exposures" +DIRECT_EVP_FLAGEVALUATIONS_PATH = "/api/v2/flagevaluation" +DIRECT_EVP_SIGNAL_PATHS = frozenset({DIRECT_EVP_EXPOSURES_PATH, DIRECT_EVP_FLAGEVALUATIONS_PATH}) DIRECT_EVP_CAPTURE_WAIT_SECONDS = 30 DIRECT_EVP_CAPTURE_SETTLE_SECONDS = 3 DIRECT_EVP_RUNTIME_EVIDENCE_FILENAME = "direct_evp_runtime.json" @@ -434,19 +437,34 @@ def _capture_has_shutdown_evaluation(data: dict[str, Any], evaluation: DirectEVP content = data.get("request", {}).get("content") if not isinstance(content, dict): return False - events = content.get("exposures") - if not isinstance(events, list): + + if evaluation.signal_path == DIRECT_EVP_EXPOSURES_PATH: + events = content.get("exposures") + elif evaluation.signal_path == DIRECT_EVP_FLAGEVALUATIONS_PATH: events = content.get("flagEvaluations") + else: + return False if not isinstance(events, list): return False - return any( - isinstance(event, dict) + + matching_events = [ + event + for event in events + if isinstance(event, dict) and isinstance(event.get("flag"), dict) and event["flag"].get("key") == evaluation.flag_key - and isinstance(event.get("subject"), dict) - and event["subject"].get("id") == evaluation.subject_id - for event in events - ) + ] + if evaluation.signal_path == DIRECT_EVP_EXPOSURES_PATH: + return any( + isinstance(event.get("subject"), dict) and event["subject"].get("id") == evaluation.subject_id + for event in matching_events + ) + + expected_targeting_keys = { + evaluation.subject_id, + f"sha256_{hashlib.sha256(evaluation.subject_id.encode()).hexdigest()}", + } + return any(event.get("targeting_key") in expected_targeting_keys for event in matching_events) @staticmethod def _stopped_container_state(container: Container) -> dict[str, Any]: @@ -612,7 +630,9 @@ def _wait_for_expected_evp_captures(self, *, is_empty_test_run: bool) -> None: if self.replay or self.exposure_egress is None or is_empty_test_run: return if not self._expected_evp_capture_paths and self._shutdown_evp_evaluation is None: - raise RuntimeError("A non-empty Feature Flags EVP run registered no expected capture paths") + # Manifest-deactivated tests remain collected when --skip-empty-scenario is disabled, + # but their setup methods never run and therefore register no capture expectations. + return interface = interfaces.datadog_sidecar if self.exposure_egress == "sidecar" else interfaces.datadog_direct missing_paths: list[str] = [] @@ -636,7 +656,12 @@ def _wait_and_stop_containers(self, *, is_empty_test_run: bool) -> None: try: self._wait_for_expected_evp_captures(is_empty_test_run=is_empty_test_run) finally: - if self.replay and self.exposure_egress == "direct": + if ( + self.replay + and self.exposure_egress == "direct" + and not is_empty_test_run + and self._direct_evp_runtime_evidence_path.exists() + ): self._last_direct_evp_runtime_evidence = json.loads( self._direct_evp_runtime_evidence_path.read_text(encoding="utf-8") ) From 80d3f90e3880f1dd48b9b380337569fe69b3772c Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Sat, 12 Sep 2026 04:58:39 +0000 Subject: [PATCH 4/6] test(dotnet): enable agentless EVP exposure egress --- manifests/dotnet.yml | 15 +++++-- .../test_dotnet_direct_evp_activation.py | 44 +++++++++++++++++++ utils/build/docker/dotnet/weblog/Program.cs | 18 +++++++- utils/build/docker/dotnet/weblog/app.sh | 4 ++ 4 files changed, 77 insertions(+), 4 deletions(-) create mode 100644 tests/test_the_test/test_dotnet_direct_evp_activation.py diff --git a/manifests/dotnet.yml b/manifests/dotnet.yml index ea58d370235..3b1d3cc6dc6 100644 --- a/manifests/dotnet.yml +++ b/manifests/dotnet.yml @@ -779,9 +779,18 @@ manifest: tests/ffe/test_dynamic_evaluation.py::Test_FFE_Flag_Parse_Error_Isolation: bug (FFL-2184) tests/ffe/test_dynamic_evaluation.py::Test_FFE_Unknown_Operator_Tolerance: bug (FFL-2184) tests/ffe/test_exposure_egress.py: v3.36.0 - tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct: missing_feature (Not yet implemented) - tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: missing_feature (Not yet implemented) - tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Sidecar: missing_feature (Not yet implemented) + tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct: + - weblog_declaration: + "*": irrelevant + poc: v3.54.0 + tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown: + - weblog_declaration: + "*": irrelevant + poc: v3.54.0 + tests/ffe/test_exposure_egress.py::Test_FFE_Exposure_Egress_Agentless_Sidecar: + - weblog_declaration: + "*": irrelevant + poc: v3.54.0 tests/ffe/test_exposures_datadog_agent.py: v3.36.0 tests/ffe/test_exposures_datadog_agent.py::Test_FFE_Exposure_Caching_Serial_Id_Appears: missing_feature (EX-3416) tests/ffe/test_exposures_datadog_agent.py::Test_FFE_Exposure_Caching_Serial_Id_Cycle: missing_feature (EX-3416) diff --git a/tests/test_the_test/test_dotnet_direct_evp_activation.py b/tests/test_the_test/test_dotnet_direct_evp_activation.py new file mode 100644 index 00000000000..183337b1782 --- /dev/null +++ b/tests/test_the_test/test_dotnet_direct_evp_activation.py @@ -0,0 +1,44 @@ +"""Guard the .NET-only activation and lifecycle for direct-EVP validation.""" + +from pathlib import Path + +from utils import features, scenarios + + +@scenarios.test_the_test +@features.not_reported +def test_dotnet_direct_evp_activation_is_exposure_only_and_poc_scoped() -> None: + manifest = Path("manifests/dotnet.yml").read_text(encoding="utf-8") + + for test_class in ( + "Test_FFE_Exposure_Egress_Agentless_Direct", + "Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown", + "Test_FFE_Exposure_Egress_Agentless_Sidecar", + ): + declaration = ( + f"tests/ffe/test_exposure_egress.py::{test_class}:\n" + ' - weblog_declaration:\n "*": irrelevant\n poc: v3.54.0' + ) + assert declaration in manifest + + assert "tests/ffe/test_flag_eval_evp.py: missing_feature (FFL-2446)" in manifest + + +@scenarios.test_the_test +@features.not_reported +def test_dotnet_direct_evp_shutdown_execs_runtime_and_marks_server_closed() -> None: + app_script = Path("utils/build/docker/dotnet/weblog/app.sh").read_text(encoding="utf-8") + program = Path("utils/build/docker/dotnet/weblog/Program.cs").read_text(encoding="utf-8") + + shutdown_switch = 'SYSTEM_TESTS_FFE_SHUTDOWN_FLUSH_ENABLED:-}" = "true"' + assert shutdown_switch in app_script + assert "exec dotnet app.dll" in app_script + assert app_script.index(shutdown_switch) < app_script.index("exec dotnet app.dll") + assert app_script.index("exec dotnet app.dll") < app_script.index("if ( ! dotnet app.dll)") + + assert 'GetEnvironmentVariable("SYSTEM_TESTS_FFE_SHUTDOWN_FLUSH_ENABLED") == "true"' in program + assert "GetRequiredService().ApplicationStopped.Register" in program + assert 'event = "system_tests.ffe.shutdown.server_closed"' in program + assert 'timestamp = DateTimeOffset.UtcNow.ToString("O")' in program + assert "Console.Out.Flush();" in program + assert program.index("ApplicationStopped.Register") < program.index("host.Run();") diff --git a/utils/build/docker/dotnet/weblog/Program.cs b/utils/build/docker/dotnet/weblog/Program.cs index 684ffb681ab..d773b53cd1a 100644 --- a/utils/build/docker/dotnet/weblog/Program.cs +++ b/utils/build/docker/dotnet/weblog/Program.cs @@ -1,6 +1,8 @@ using System; +using System.Text.Json; using System.Threading; using Microsoft.AspNetCore.Hosting; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Http; using Datadog.Trace; @@ -37,7 +39,21 @@ public static void Main(string[] args) settings.LogsInjectionEnabled = true; Tracer.Configure(settings); } - CreateHostBuilder(args).Build().Run(); + var host = CreateHostBuilder(args).Build(); + if (Environment.GetEnvironmentVariable("SYSTEM_TESTS_FFE_SHUTDOWN_FLUSH_ENABLED") == "true") + { + host.Services.GetRequiredService().ApplicationStopped.Register(() => + { + Console.WriteLine(JsonSerializer.Serialize(new + { + @event = "system_tests.ffe.shutdown.server_closed", + timestamp = DateTimeOffset.UtcNow.ToString("O") + })); + Console.Out.Flush(); + }); + } + + host.Run(); } public static IHostBuilder CreateHostBuilder(string[] args) { diff --git a/utils/build/docker/dotnet/weblog/app.sh b/utils/build/docker/dotnet/weblog/app.sh index e1e317f764b..182fac2b980 100755 --- a/utils/build/docker/dotnet/weblog/app.sh +++ b/utils/build/docker/dotnet/weblog/app.sh @@ -8,6 +8,10 @@ if [ "${UDS_WEBLOG:-0}" = "1" ]; then ./set-uds-transport.sh fi +if [ "${SYSTEM_TESTS_FFE_SHUTDOWN_FLUSH_ENABLED:-}" = "true" ]; then + exec dotnet app.dll +fi + if ( ! dotnet app.dll); then echo recovering dump to /var/log/system-tests/dumps mkdir -p /var/log/system-tests/dumps From 9c753fbcbdcc6122eed15fd5af827671fbc0771c Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Sat, 12 Sep 2026 05:12:51 +0000 Subject: [PATCH 5/6] test(ffe): use identity-compatible serverless-init Pin serverless-init 1.10.4 so sidecar scenarios advertise forwarding support for both EVP origin headers required by hardened SDKs. Environment: Datadog workspace --- mirror_images.lock.yaml | 8 ++++---- mirror_images.yaml | 2 +- tests/test_the_test/test_mock_ffe_agentless_backend.py | 1 + utils/_context/containers.py | 2 +- 4 files changed, 7 insertions(+), 6 deletions(-) diff --git a/mirror_images.lock.yaml b/mirror_images.lock.yaml index fdc59f0a430..00584a88f1e 100644 --- a/mirror_images.lock.yaml +++ b/mirror_images.lock.yaml @@ -26,10 +26,10 @@ images: digest: sha256:0199e251729b25276f97049cbcd39dd1997e98413a07b10a3c7e9e755d505a42 target: registry.ddbuild.io/system-tests/mirror/datadog/dd-appsec-php-ci:php-8.2-release tag: php-8.2-release - datadog/serverless-init:1.10.2: - digest: sha256:9936ccb50daa747d71a8b1148e74720a6ae60238ebdcc1793ea1834c01493bd3 - target: registry.ddbuild.io/system-tests/mirror/datadog/serverless-init:1.10.2 - tag: 1.10.2 + datadog/serverless-init:1.10.4: + digest: sha256:f4943347f155505156ba5f594317ea64b59678b3a6522e8c0993581af20956ff + target: registry.ddbuild.io/system-tests/mirror/datadog/serverless-init:1.10.4 + tag: 1.10.4 datadog/system-tests:apache-mod-7.0-zts.base-v1: digest: sha256:bafc072734341cafb86f45673df2de4851a04b951d0717c88f7958cac1815b20 target: registry.ddbuild.io/system-tests/mirror/datadog/system-tests:apache-mod-7.0-zts.base-v1 diff --git a/mirror_images.yaml b/mirror_images.yaml index 64ff2c125aa..e3502ddc2dd 100644 --- a/mirror_images.yaml +++ b/mirror_images.yaml @@ -14,7 +14,7 @@ - "datadog/agent:latest" - "datadog/dd-appsec-php-ci:php-8.0-release" - "datadog/dd-appsec-php-ci:php-8.2-release" -- "datadog/serverless-init:1.10.2" +- "datadog/serverless-init:1.10.4" - "datadog/system-tests:apache-mod-7.0-zts.base-v1" - "datadog/system-tests:apache-mod-7.0.base-v1" - "datadog/system-tests:apache-mod-7.1-zts.base-v1" diff --git a/tests/test_the_test/test_mock_ffe_agentless_backend.py b/tests/test_the_test/test_mock_ffe_agentless_backend.py index 71516b9d4d4..717a48a71d1 100644 --- a/tests/test_the_test/test_mock_ffe_agentless_backend.py +++ b/tests/test_the_test/test_mock_ffe_agentless_backend.py @@ -180,6 +180,7 @@ def test_agentless_exposure_scenario_has_no_agent_and_two_capture_routes( serverless_init = scenario.serverless_init_container assert serverless_init_containers == (serverless_init,) assert isinstance(serverless_init, ServerlessInitContainer) + assert serverless_init.image.original_name == "datadog/serverless-init:1.10.4" assert environment["DD_TRACE_AGENT_PORT"] == str(serverless_init.apm_receiver_port) assert environment["DD_TRACE_AGENT_URL"] == f"http://ffe-serverless-init:{serverless_init.apm_receiver_port}" assert serverless_init.healthcheck is not None diff --git a/utils/_context/containers.py b/utils/_context/containers.py index f97408e88f3..6242b2e41d6 100644 --- a/utils/_context/containers.py +++ b/utils/_context/containers.py @@ -879,7 +879,7 @@ def __init__(self) -> None: apm_receiver_port_hex = f"{self.apm_receiver_port:04X}" super().__init__( name="ffe-serverless-init", - image_name="datadog/serverless-init:1.10.2", + image_name="datadog/serverless-init:1.10.4", environment={ "DD_API_KEY": _FAKE_DD_API_KEY, "DD_SITE": "mock-intake.invalid", From 701c9a30ab7c8f5d9b1cf5b2884c74442f9ff3ba Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Wed, 23 Sep 2026 16:51:53 +0000 Subject: [PATCH 6/6] test(dotnet-evp): keep enablement focused on manifests and server wiring Remove redundant activation and source-text self-tests. Preserve existing end-to-end contracts and required server/proxy adjustments. Environment: Datadog workspace --- .../test_dotnet_direct_evp_activation.py | 63 ------------------- 1 file changed, 63 deletions(-) delete mode 100644 tests/test_the_test/test_dotnet_direct_evp_activation.py diff --git a/tests/test_the_test/test_dotnet_direct_evp_activation.py b/tests/test_the_test/test_dotnet_direct_evp_activation.py deleted file mode 100644 index 316f6425096..00000000000 --- a/tests/test_the_test/test_dotnet_direct_evp_activation.py +++ /dev/null @@ -1,63 +0,0 @@ -"""Guard the .NET-only activation and lifecycle for direct-EVP validation.""" - -from pathlib import Path - -import pytest - -from utils import features, scenarios -from utils._context.component_version import Version -from utils.manifest import Manifest - - -@scenarios.test_the_test -@features.not_reported -def test_dotnet_direct_evp_activation_is_exposure_only_and_poc_scoped() -> None: - manifest = Path("manifests/dotnet.yml").read_text(encoding="utf-8") - - for test_class in ( - "Test_FFE_Exposure_Egress_Agentless_Direct", - "Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown", - "Test_FFE_Exposure_Egress_Agentless_Sidecar", - ): - declaration = ( - f"tests/ffe/test_exposure_egress.py::{test_class}:\n" - ' - weblog_declaration:\n "*": irrelevant\n poc: v3.55.0' - ) - assert declaration in manifest - - assert "tests/ffe/test_flag_eval_evp.py: missing_feature (FFL-2446)" in manifest - - -@pytest.mark.parametrize("version", ["3.54.0", "3.55.0"]) -@pytest.mark.parametrize("weblog", ["poc", "aspnet", "iis"]) -@scenarios.test_the_test -@features.not_reported -def test_dotnet_evp_activation_excludes_released_configuration_only_sdk(version: str, weblog: str) -> None: - manifest = Manifest({"dotnet": Version(version)}, weblog) - for test_class in ( - "Test_FFE_Exposure_Egress_Agentless_Direct", - "Test_FFE_Exposure_Egress_Agentless_Direct_Shutdown", - "Test_FFE_Exposure_Egress_Agentless_Sidecar", - ): - declarations = manifest.get_declarations(f"tests/ffe/test_exposure_egress.py::{test_class}") - assert (not declarations) == (version == "3.55.0" and weblog == "poc"), declarations - - -@scenarios.test_the_test -@features.not_reported -def test_dotnet_direct_evp_shutdown_execs_runtime_and_marks_server_closed() -> None: - app_script = Path("utils/build/docker/dotnet/weblog/app.sh").read_text(encoding="utf-8") - program = Path("utils/build/docker/dotnet/weblog/Program.cs").read_text(encoding="utf-8") - - shutdown_switch = 'SYSTEM_TESTS_FFE_SHUTDOWN_FLUSH_ENABLED:-}" = "true"' - assert shutdown_switch in app_script - assert "exec dotnet app.dll" in app_script - assert app_script.index(shutdown_switch) < app_script.index("exec dotnet app.dll") - assert app_script.index("exec dotnet app.dll") < app_script.index("if ( ! dotnet app.dll)") - - assert 'GetEnvironmentVariable("SYSTEM_TESTS_FFE_SHUTDOWN_FLUSH_ENABLED") == "true"' in program - assert "GetRequiredService().ApplicationStopped.Register" in program - assert 'event = "system_tests.ffe.shutdown.server_closed"' in program - assert 'timestamp = DateTimeOffset.UtcNow.ToString("O")' in program - assert "Console.Out.Flush();" in program - assert program.index("ApplicationStopped.Register") < program.index("host.Run();")