Align agentless mock with the UFC CDN response

.github/CODEOWNERS

@@ -14,10 +14,12 @@
 /utils/build/docker/vcr/cassettes/aiguard @DataDog/k9-ai-guard @DataDog/system-tests-core
 /utils/scripts/generate-ai-guard-cassettes.sh @DataDog/k9-ai-guard @DataDog/system-tests-core
 /utils/docker_fixtures/spec/llm_observability.py @DataDog/ml-observability @DataDog/system-tests-core
+/utils/docker_fixtures/_mock_ffe_agentless_backend.py @DataDog/feature-flagging-and-experimentation-sdk @DataDog/system-tests-core
 /utils/telemetry/intake/ @DataDog/apm-sdk-capabilities @DataDog/system-tests-core
 /utils/telemetry/intake/static/ @DataDog/apm-sdk
 
 /tests/ffe/ @DataDog/feature-flagging-and-experimentation-sdk @DataDog/system-tests-core
+/tests/test_the_test/test_mock_ffe_agentless_backend.py @DataDog/feature-flagging-and-experimentation-sdk @DataDog/system-tests-core
 /tests/parametric/ @DataDog/system-tests-core @DataDog/apm-sdk-capabilities
 /tests/parametric/test_config_consistency.py @DataDog/apm-sdk-capabilities
 /tests/parametric/test_ffe/ @DataDog/feature-flagging-and-experimentation-sdk @DataDog/system-tests-core

tests/test_library_conf.py

@@ -445,8 +445,11 @@ def retrieve_span_links(span: DataDogAgentSpan) -> list[dict] | None:
     if span_meta.get("_dd.span_links") is None:
         return None
 
-    # Convert span_links tags into msgpack v0.4 format
-    json_links = json.loads(span_meta["_dd.span_links"])
+    # Convert span_links tags into msgpack v0.4 format. The v1 trace proxy
+    # normalizes JSON meta values while v0.4 payloads retain the serialized
+    # string, so accept both representations here.
+    raw_links = span_meta["_dd.span_links"]
+    json_links = json.loads(raw_links) if isinstance(raw_links, (str, bytes, bytearray)) else raw_links
     links = []
     for json_link in json_links:
         link = {}

tests/test_the_test/test_mock_ffe_agentless_backend.py

@@ -7,19 +7,39 @@
 from utils.docker_fixtures._core import HOST_GATEWAY_EXTRA_HOSTS, extra_hosts_for_environment
 from utils.docker_fixtures._mock_ffe_agentless_backend import (
     CONFIG_PATH,
+    CONFIG_QUERY,
     EXPECTED_API_KEY,
+    EXPECTED_DD_ENV,
     MockFFEAgentlessBackendServer,
+    UFC_RESPONSE_TYPE,
 )
 
 
 @scenarios.test_the_test
 @features.not_reported
 def test_mock_ffe_agentless_backend_serves_fixture_and_tracks_metadata(worker_id: str) -> None:
-    server = MockFFEAgentlessBackendServer(worker_id)
+    server = MockFFEAgentlessBackendServer(worker_id, port=0)
     try:
-        response = requests.get(server.base_url + CONFIG_PATH, headers={"dd-api-key": EXPECTED_API_KEY}, timeout=5)
+        for invalid_query in ("", "?dd_env=", "?dd_env=wrong", f"?dd_env={EXPECTED_DD_ENV}&dd_env=wrong"):
+            response = requests.get(
+                server.base_url + CONFIG_PATH + invalid_query,
+                headers={"DD-API-KEY": EXPECTED_API_KEY},
+                timeout=5,
+            )
+            assert response.status_code == 404
+
+        response = requests.get(
+            f"{server.base_url}{CONFIG_PATH}?{CONFIG_QUERY}",
+            headers={"DD-API-KEY": EXPECTED_API_KEY},
+            timeout=5,
+        )
         response.raise_for_status()
 
+        payload = response.json()
+        assert payload["data"]["type"] == UFC_RESPONSE_TYPE
+        assert payload["data"]["attributes"]["environment"]["name"] == "Test"
+        assert "new-user-onboarding" in payload["data"]["attributes"]["flags"]
+
         status = server.status()
         assert status["requests_total"] == 1
         assert status["last_auth_present"] is True
@@ -37,8 +57,9 @@ def test_mock_ffe_agentless_backend_host_gateway_mapping(monkeypatch: pytest.Mon
     monkeypatch.delenv("SYSTEM_TESTS_MOCK_FFE_AGENTLESS_BACKEND_HOST", raising=False)
     monkeypatch.delenv("SYSTEM_TESTS_MOCK_AGENTLESS_BACKEND_HOST", raising=False)
 
-    server = MockFFEAgentlessBackendServer(worker_id)
+    server = MockFFEAgentlessBackendServer(worker_id, port=0)
     try:
+        assert server.library_config_url.endswith(f"{CONFIG_PATH}?{CONFIG_QUERY}")
         env = {"DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL": server.library_config_url}
         assert extra_hosts_for_environment(env) == HOST_GATEWAY_EXTRA_HOSTS
     finally:
@@ -48,7 +69,7 @@ def test_mock_ffe_agentless_backend_host_gateway_mapping(monkeypatch: pytest.Mon
 @scenarios.test_the_test
 @features.not_reported
 def test_mock_ffe_agentless_backend_status_is_metadata_only(worker_id: str) -> None:
-    server = MockFFEAgentlessBackendServer(worker_id)
+    server = MockFFEAgentlessBackendServer(worker_id, port=0)
     try:
         status = server.status()
         assert set(status) == {

utils/docker_fixtures/_mock_ffe_agentless_backend.py

@@ -17,7 +17,7 @@
 import threading
 import time
 from typing import TYPE_CHECKING, Any, TypedDict, cast
-from urllib.parse import urlparse
+from urllib.parse import parse_qs, urlencode, urlparse
 
 import pytest
 import requests
@@ -44,10 +44,13 @@
 DELAYED_RESPONSE_SECONDS = 0.5
 TIMEOUT_RESPONSE_SECONDS = 1.5
 MAX_CONTROL_BODY_BYTES = 512
-CONFIG_PATH = "/api/v2/feature-flagging/config/server-distribution"
+CONFIG_PATH = "/api/v2/feature-flagging/config/rules-based/server"
+EXPECTED_DD_ENV = "test"
+CONFIG_QUERY = urlencode({"dd_env": EXPECTED_DD_ENV})
 REPO_ROOT = Path(__file__).parents[2]
 UFC_FIXTURE_PATH = REPO_ROOT / "tests" / "parametric" / "test_ffe" / "flags-v1.json"
 MALFORMED_UFC_BYTES = b'{"flags": ['
+UFC_RESPONSE_TYPE = "universal-flag-configuration"
 
 
 class MockFFEAgentlessBackendStatus(TypedDict):
@@ -143,18 +146,20 @@ def __init__(self, server_address: tuple[str, int]) -> None:
 
 class MockFFEAgentlessBackendRequestHandler(BaseHTTPRequestHandler):
     # Endpoint contract:
-    # - GET /api/v2/feature-flagging/config/server-distribution
+    # - GET /api/v2/feature-flagging/config/rules-based/server?dd_env=test
     # - GET /status
     # - POST /control/responses
     # - POST /control/reset
     server: MockFFEAgentlessBackendHTTPServer
 
     def do_GET(self) -> None:
-        path = urlparse(self.path).path
-        if path == CONFIG_PATH:
+        parsed = urlparse(self.path)
+        if parsed.path == CONFIG_PATH and parse_qs(parsed.query, keep_blank_values=True) == {
+            "dd_env": [EXPECTED_DD_ENV]
+        }:
             self._handle_config()
             return
-        if path == "/status":
+        if parsed.path == "/status":
             self._write_json(HTTPStatus.OK, self.server.state.status())
             return
         self._write_json(HTTPStatus.NOT_FOUND, {"error": "not found"})
@@ -232,11 +237,20 @@ def _write_json(self, status_code: HTTPStatus, payload: dict[str, Any] | MockFFE
 
 def _has_auth(headers: Mapping[str, str]) -> bool:
     normalized = {key.lower(): value for key, value in headers.items()}
-    return any(normalized.get(header) == EXPECTED_API_KEY for header in ("dd-api-key", "x-datadog-api-key"))
+    return normalized.get("dd-api-key") == EXPECTED_API_KEY
 
 
 def _valid_ufc_bytes() -> bytes:
-    return UFC_FIXTURE_PATH.read_bytes()
+    attributes = json.loads(UFC_FIXTURE_PATH.read_text())
+    return json.dumps(
+        {
+            "data": {
+                "id": "1",
+                "type": UFC_RESPONSE_TYPE,
+                "attributes": attributes,
+            }
+        }
+    ).encode("utf-8")
 
 
 def validate_responses(responses: object) -> list[str]:
@@ -269,13 +283,19 @@ def _response_for_response(response: str, *, has_auth: bool) -> tuple[int, bytes
 
 
 def _strip_config_path(url: str) -> str:
-    return url.removesuffix(CONFIG_PATH)
+    parsed = urlparse(url)
+    if not parsed.path.endswith(CONFIG_PATH):
+        return url
+
+    base_path = parsed.path.removesuffix(CONFIG_PATH).rstrip("/")
+    return parsed._replace(path=base_path, params="", query="", fragment="").geturl().rstrip("/")
 
 
 class MockFFEAgentlessBackendServer:
-    def __init__(self, worker_id: str) -> None:
-        self.port = get_host_port(worker_id, 4900)
+    def __init__(self, worker_id: str, *, port: int | None = None) -> None:
+        self.port = get_host_port(worker_id, 4900) if port is None else port
         self._server = MockFFEAgentlessBackendHTTPServer(("0.0.0.0", self.port))  # noqa: S104 - test fixture must be container-reachable.
+        self.port = self._server.server_port
         self._thread = threading.Thread(
             target=self._server.serve_forever, name="mock-ffe-agentless-backend", daemon=True
         )
@@ -300,7 +320,7 @@ def library_base_url(self) -> str:
 
     @property
     def library_config_url(self) -> str:
-        return f"{self.library_base_url}{CONFIG_PATH}"
+        return f"{self.library_base_url}{CONFIG_PATH}?{CONFIG_QUERY}"
 
     def reset(self) -> None:
         response = requests.post(f"{self.base_url}/control/reset", timeout=5)