From b83682d6f07a25343daa68fc707c80d5832883e8 Mon Sep 17 00:00:00 2001 From: Anik Bhattacharjee Date: Thu, 3 Sep 2026 12:51:42 -0400 Subject: [PATCH] LCORE-1822: Enable OpenTelemetry delivery E2E test with mock OTLP collector MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the Behave step definitions for the previously-skipped OpenTelemetry E2E scenario and turns the test on, so CI now verifies end-to-end that the Lightspeed Core Stack actually delivers telemetry to an OTLP backend. The scenario asserts that a `responses` request carrying a `safety_identifier` marker results in that marker reaching a collector. Two pieces were needed to make it real: 1. **A mock OTLP/HTTP collector** to receive and assert on exports. 2. **Instrumentation** so the marker (`safety_identifier`) is actually emitted on a span — previously it was only forwarded to the model provider, so the scenario could never pass and was tagged `@skip`. --- docker-compose-library.yaml | 35 +++- docker-compose.yaml | 35 +++- src/app/endpoints/responses.py | 20 +- src/utils/otel_tracing.py | 1 + tests/e2e/features/opentelemetry.feature | 4 +- tests/e2e/features/steps/opentelemetry.py | 87 +++++++++ tests/e2e/mock_otel_collector/Dockerfile | 5 + tests/e2e/mock_otel_collector/README.md | 53 ++++++ tests/e2e/mock_otel_collector/server.py | 174 ++++++++++++++++++ .../app/endpoints/responses_otel_helpers.py | 2 + .../unit/app/endpoints/test_responses_otel.py | 48 +++++ 11 files changed, 443 insertions(+), 21 deletions(-) create mode 100644 tests/e2e/features/steps/opentelemetry.py create mode 100644 tests/e2e/mock_otel_collector/Dockerfile create mode 100644 tests/e2e/mock_otel_collector/README.md create mode 100644 tests/e2e/mock_otel_collector/server.py diff --git a/docker-compose-library.yaml b/docker-compose-library.yaml index 0d750b071..ee7090e43 100755 --- a/docker-compose-library.yaml +++ b/docker-compose-library.yaml @@ -11,6 +11,8 @@ services: depends_on: mock-mcp: condition: service_healthy + mock-otel: + condition: service_healthy networks: - lightspeednet volumes: @@ -67,12 +69,15 @@ services: - PDF_KV_RAG_PATH=/tmp/e2e-rag-work/pdf_kv_store.db # Prevent HuggingFace Hub update checks (HTTP 429 rate-limiting in CI from parallel jobs). - HF_HUB_OFFLINE=1 - # OpenTelemetry configuration (tracing disabled by default) - - OTEL_EXPORTER_OTLP_ENDPOINT=${OTEL_EXPORTER_OTLP_ENDPOINT:-} - - OTEL_EXPORTER_OTLP_PROTOCOL=${OTEL_EXPORTER_OTLP_PROTOCOL:-} - - OTEL_SERVICE_NAME=${OTEL_SERVICE_NAME:-} + # OpenTelemetry configuration. Export is enabled by default and points at + # the mock OTLP/HTTP collector so the OpenTelemetry delivery E2E test works + # without per-scenario reconfiguration. Override any OTEL_* var (or set + # OTEL_SDK_DISABLED=true) to change or disable export. + - OTEL_EXPORTER_OTLP_ENDPOINT=${OTEL_EXPORTER_OTLP_ENDPOINT:-http://mock-otel:4318} + - OTEL_EXPORTER_OTLP_PROTOCOL=${OTEL_EXPORTER_OTLP_PROTOCOL:-http/protobuf} + - OTEL_SERVICE_NAME=${OTEL_SERVICE_NAME:-lightspeed-stack-e2e} - OTEL_ANONYMIZATION_SECRET=${OTEL_ANONYMIZATION_SECRET:-lightspeed-stack-otel-anonymization-dev-default} - - OTEL_SDK_DISABLED=${OTEL_SDK_DISABLED:-true} + - OTEL_SDK_DISABLED=${OTEL_SDK_DISABLED:-false} healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8080/liveness"] interval: 10s # how often to run the check @@ -113,6 +118,26 @@ services: retries: 3 start_period: 2s + # Mock OTLP/HTTP collector for OpenTelemetry E2E tests. + # lightspeed-stack exports to it by default (see OTEL_* above) and waits for it + # to be healthy, so telemetry is delivered from startup. The port is bound to + # loopback only so it is not exposed beyond the host running the tests. + mock-otel: + build: + context: ./tests/e2e/mock_otel_collector + dockerfile: Dockerfile + container_name: mock-otel + ports: + - "127.0.0.1:4318:4318" + networks: + - lightspeednet + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:4318/health')"] + interval: 5s + timeout: 3s + retries: 3 + start_period: 2s + networks: lightspeednet: diff --git a/docker-compose.yaml b/docker-compose.yaml index 6dff730b7..2ec0c0d74 100755 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -114,17 +114,22 @@ services: # Substituted in mounted lightspeed-stack.yaml (ogx.url); GitHub Actions sets E2E_LLAMA_HOSTNAME - E2E_OGX_HOSTNAME=${E2E_OGX_HOSTNAME:-${E2E_LLAMA_HOSTNAME:-ogx}} - E2E_OGX_PORT=${E2E_OGX_PORT:-${E2E_LLAMA_PORT:-8321}} - # OpenTelemetry configuration (tracing disabled by default) - - OTEL_EXPORTER_OTLP_ENDPOINT=${OTEL_EXPORTER_OTLP_ENDPOINT:-} - - OTEL_EXPORTER_OTLP_PROTOCOL=${OTEL_EXPORTER_OTLP_PROTOCOL:-} - - OTEL_SERVICE_NAME=${OTEL_SERVICE_NAME:-} + # OpenTelemetry configuration. Export is enabled by default and points at + # the mock OTLP/HTTP collector so the OpenTelemetry delivery E2E test works + # without per-scenario reconfiguration. Override any OTEL_* var (or set + # OTEL_SDK_DISABLED=true) to change or disable export. + - OTEL_EXPORTER_OTLP_ENDPOINT=${OTEL_EXPORTER_OTLP_ENDPOINT:-http://mock-otel:4318} + - OTEL_EXPORTER_OTLP_PROTOCOL=${OTEL_EXPORTER_OTLP_PROTOCOL:-http/protobuf} + - OTEL_SERVICE_NAME=${OTEL_SERVICE_NAME:-lightspeed-stack-e2e} - OTEL_ANONYMIZATION_SECRET=${OTEL_ANONYMIZATION_SECRET:-lightspeed-stack-otel-anonymization-dev-default} - - OTEL_SDK_DISABLED=${OTEL_SDK_DISABLED:-true} + - OTEL_SDK_DISABLED=${OTEL_SDK_DISABLED:-false} depends_on: ogx: condition: service_healthy mock-mcp: condition: service_healthy + mock-otel: + condition: service_healthy networks: - lightspeednet healthcheck: @@ -167,6 +172,26 @@ services: retries: 3 start_period: 2s + # Mock OTLP/HTTP collector for OpenTelemetry E2E tests. + # lightspeed-stack exports to it by default (see OTEL_* above) and waits for it + # to be healthy, so telemetry is delivered from startup. The port is bound to + # loopback only so it is not exposed beyond the host running the tests. + mock-otel: + build: + context: ./tests/e2e/mock_otel_collector + dockerfile: Dockerfile + container_name: mock-otel + ports: + - "127.0.0.1:4318:4318" + networks: + - lightspeednet + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:4318/health')"] + interval: 5s + timeout: 3s + retries: 3 + start_period: 2s + # Mock TLS inference server for TLS E2E tests mock-tls-inference: build: diff --git a/src/app/endpoints/responses.py b/src/app/endpoints/responses.py index 33dadc561..ef72faa97 100644 --- a/src/app/endpoints/responses.py +++ b/src/app/endpoints/responses.py @@ -579,14 +579,18 @@ async def handle_responses_with_tracing( # pylint: disable=too-many-locals ) attachments_count = _count_request_attachments(original_request.input) - set_span_attributes( - root_span, - { - SpanAttributes.USER_ID: anonymize_value(user_id), - SpanAttributes.INPUT: anonymize_value(input_text), - SpanAttributes.REQUEST_ATTACHMENTS_COUNT: attachments_count, - }, - ) + span_attributes: dict[str, Any] = { + SpanAttributes.USER_ID: anonymize_value(user_id), + SpanAttributes.INPUT: anonymize_value(input_text), + SpanAttributes.REQUEST_ATTACHMENTS_COUNT: attachments_count, + } + # safety_identifier is a caller-supplied, non-PII identifier, so it is + # recorded verbatim (not anonymized) when present. + if original_request.safety_identifier is not None: + span_attributes[SpanAttributes.SAFETY_IDENTIFIER] = ( + original_request.safety_identifier + ) + set_span_attributes(root_span, span_attributes) await check_mcp_auth(configuration, mcp_headers, token, request.headers) diff --git a/src/utils/otel_tracing.py b/src/utils/otel_tracing.py index 9b4d887c4..ddbd5948f 100644 --- a/src/utils/otel_tracing.py +++ b/src/utils/otel_tracing.py @@ -24,6 +24,7 @@ class SpanAttributes(StrEnum): SESSION_ID = "session.id" USER_ID = "user.id" # anonymized + SAFETY_IDENTIFIER = "request.safety_identifier" # caller-supplied identifier INPUT = "request.input" # anonymized OUTPUT = "response.output" # anonymized RESPONSE_ERROR = "response.error" diff --git a/tests/e2e/features/opentelemetry.feature b/tests/e2e/features/opentelemetry.feature index 8d8f79a59..aba242621 100644 --- a/tests/e2e/features/opentelemetry.feature +++ b/tests/e2e/features/opentelemetry.feature @@ -1,11 +1,10 @@ -@cfg_authorized @OTel @skip +@cfg_authorized @OTel @skip-in-prow Feature: OpenTelemetry observability tests Background: Given The service is started locally And The system is in default state And An OpenTelemetry service is running and listening for OTLP data - And The service is configured to export data to the OpenTelemetry service And I set the Authorization header to Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6Ikpva And REST API service prefix is /v1 And the Lightspeed stack configuration directory is "tests/e2e/configuration" @@ -25,5 +24,4 @@ Feature: OpenTelemetry observability tests } """ Then The status code of the response is 200 - And The service exported an OpenTelemetry event containing e2e-otel-delivery-marker And The OpenTelemetry service received data containing e2e-otel-delivery-marker diff --git a/tests/e2e/features/steps/opentelemetry.py b/tests/e2e/features/steps/opentelemetry.py new file mode 100644 index 000000000..4d9f25939 --- /dev/null +++ b/tests/e2e/features/steps/opentelemetry.py @@ -0,0 +1,87 @@ +"""Step definitions for the OpenTelemetry telemetry-delivery E2E scenario. + +The Lightspeed Core Stack exports spans/events to the mock OTLP/HTTP collector +(the ``mock-otel`` Docker Compose service) from startup: the ``OTEL_*`` +environment variables that enable ``opentelemetry-instrument`` and point the +exporter at the collector are baked into the Compose files, so no per-scenario +reconfiguration is needed. These steps only reset the collector's buffer at the +start of the scenario and assert that telemetry containing a scenario marker is +delivered. +""" + +import os +import time + +import requests +from behave import given, then # pyright: ignore[reportAttributeAccessIssue] +from behave.runner import Context + +from tests.e2e.utils.utils import wait_for_container_health + +# Compose service / container name for the mock collector (see docker-compose*.yaml). +MOCK_OTEL_SERVICE = "mock-otel" + +# Host-side control API of the mock collector (published port from docker-compose). +_MOCK_OTEL_HOST = os.getenv("E2E_OTEL_MOCK_HOST", "localhost") +_MOCK_OTEL_PORT = os.getenv("E2E_OTEL_MOCK_PORT", "4318") +MOCK_OTEL_CONTROL_BASE = f"http://{_MOCK_OTEL_HOST}:{_MOCK_OTEL_PORT}" + +# Delivery is asynchronous: the SDK batches spans before export. Poll generously. +_DELIVERY_TIMEOUT_S = float(os.getenv("E2E_OTEL_DELIVERY_TIMEOUT_S", "45")) +_DELIVERY_POLL_INTERVAL_S = 2.0 + + +def _reset_mock_collector() -> None: + """Clear any telemetry buffered by the mock collector from prior runs.""" + response = requests.post(f"{MOCK_OTEL_CONTROL_BASE}/reset", timeout=5) + assert ( + response.status_code == 200 + ), f"Failed to reset mock OTEL collector: HTTP {response.status_code}" + + +def _poll_collector_contains(marker: str) -> bool: + """Return True once the collector has buffered a payload containing ``marker``.""" + url = f"{MOCK_OTEL_CONTROL_BASE}/received" + deadline = time.monotonic() + _DELIVERY_TIMEOUT_S + while time.monotonic() < deadline: + try: + response = requests.get(url, params={"contains": marker}, timeout=5) + if response.status_code == 200 and response.json().get("found"): + return True + except requests.RequestException: + pass + time.sleep(_DELIVERY_POLL_INTERVAL_S) + return False + + +@given("An OpenTelemetry service is running and listening for OTLP data") +def otel_service_running(context: Context) -> None: + """Wait for the mock OTLP collector to be healthy and clear its buffer. + + The ``mock-otel`` Compose service starts with the rest of the stack and its + readiness is enforced by the Compose healthcheck, so this step waits for + that health status and resets any previously buffered telemetry so the + scenario starts from a clean slate. The Lightspeed Core Stack already + exports to the collector via the ``OTEL_*`` variables set in the Compose + files. + """ + wait_for_container_health(MOCK_OTEL_SERVICE) + _reset_mock_collector() + context.otel_collector_ready = True + + +@then("The OpenTelemetry service received data containing {marker}") +def collector_received_data(context: Context, marker: str) -> None: + """Assert the mock collector buffered telemetry containing ``marker``. + + Verifies delivery from the collector's perspective; polls to tolerate the + SDK's batched, asynchronous export. + """ + assert getattr( + context, "otel_collector_ready", False + ), "The OpenTelemetry service must be started before asserting on delivery" + marker = marker.strip() + assert _poll_collector_contains(marker), ( + f"Mock OTEL collector did not receive data containing {marker!r} " + f"within {_DELIVERY_TIMEOUT_S:.0f}s" + ) diff --git a/tests/e2e/mock_otel_collector/Dockerfile b/tests/e2e/mock_otel_collector/Dockerfile new file mode 100644 index 000000000..6e2195273 --- /dev/null +++ b/tests/e2e/mock_otel_collector/Dockerfile @@ -0,0 +1,5 @@ +FROM python:3.12-slim +WORKDIR /app +COPY server.py . +EXPOSE 4318 +CMD ["python", "server.py"] diff --git a/tests/e2e/mock_otel_collector/README.md b/tests/e2e/mock_otel_collector/README.md new file mode 100644 index 000000000..d208411af --- /dev/null +++ b/tests/e2e/mock_otel_collector/README.md @@ -0,0 +1,53 @@ +# Mock OTEL collector + +A minimal OTLP/HTTP collector used by the OpenTelemetry E2E scenario +(`tests/e2e/features/opentelemetry.feature`) to verify that the Lightspeed Core +Stack delivers spans/events to a telemetry backend. + +It is a stdlib-only `http.server` that buffers the raw OTLP export bodies in +memory and exposes a small control API so Behave steps can assert what was +received. See `server.py` for the full endpoint list. + +## Endpoints + +| Method & path | Purpose | +| -------------------- | ------------------------------------------------------------- | +| `POST /v1/*` | Receive an OTLP export (traces/logs/metrics); body buffered. | +| `GET /received` | Report the count of buffered exports. | +| `GET /received?contains=` | Report whether `` appears in any payload. | +| `POST /reset` | Clear the buffer (called at scenario start). | +| `GET /health` | Liveness probe (`{"status": "ok"}`). | + +Substring queries search the raw request bytes. OTLP protobuf encodes string +fields as UTF-8, so a plaintext marker embedded in a span attribute value is +found without decoding protobuf. + +## Running + +Locally: + +```bash +python server.py [port] # default port 4318 +``` + +In E2E it runs as the `mock-otel` Docker Compose service on the `lightspeednet` +network. It starts with the rest of the stack (`docker compose up -d`) and +`lightspeed-stack` lists it under `depends_on` (waiting for it to become +healthy), so telemetry is delivered from startup. The +`An OpenTelemetry service is running and listening for OTLP data` step only waits +for it to become healthy and resets its buffer. + +## Pointing the service at it + +The Lightspeed Core Stack exports via HTTP/protobuf when launched with the OTEL +SDK enabled. The Compose files set these by default so export is always on in +E2E (override or set `OTEL_SDK_DISABLED=true` to change or disable it): + +```bash +OTEL_SDK_DISABLED=false +OTEL_EXPORTER_OTLP_ENDPOINT=http://mock-otel:4318 +OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf +``` + +The `scripts/entrypoint.sh` gate launches the service under +`opentelemetry-instrument` whenever `OTEL_SDK_DISABLED=false`. diff --git a/tests/e2e/mock_otel_collector/server.py b/tests/e2e/mock_otel_collector/server.py new file mode 100644 index 000000000..1e8bff0a5 --- /dev/null +++ b/tests/e2e/mock_otel_collector/server.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +"""Minimal mock OpenTelemetry (OTLP/HTTP) collector for E2E tests. + +Accepts OTLP/HTTP exports from the Lightspeed Core Stack and buffers the raw +request bodies in memory so Behave steps can assert that telemetry was +delivered. Uses only the Python standard library. + +Endpoints +--------- +- ``POST /v1/*`` : Receive an OTLP export (traces/logs/metrics). The body is + buffered and the request is answered with an empty ``application/x-protobuf`` + 200, which the OTLP/HTTP exporter accepts as success. +- ``GET /received`` : Report how many exports have been buffered. With + ``?contains=`` it reports whether that substring appears in any buffered + payload (raw-byte search; OTLP protobuf stores string fields as UTF-8, so a + plaintext marker embedded in an attribute value is found). +- ``POST /reset`` : Clear the buffer (used at the start of a scenario). +- ``GET /health`` : Liveness probe returning ``{"status": "ok"}``. + +Run as ``python server.py [port]``; default port is 4318 (OTLP/HTTP). + +The exporter must be configured for HTTP/protobuf, e.g.:: + + OTEL_EXPORTER_OTLP_ENDPOINT=http://mock-otel:4318 + OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf +""" + +import json +import sys +import threading +from collections import deque +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any + +# Resource bounds so a reachable caller cannot exhaust process memory by sending +# large or repeated exports (the Compose service publishes port 4318). The +# per-body cap bounds each entry; the entry-count and total-byte caps bound the +# whole buffer (oldest entries are evicted first). With 1000 entries of up to +# 5 MiB each an unbounded buffer could reach ~5 GiB, so the total-byte cap is the +# effective ceiling. +_MAX_BODY_BYTES = 5 * 1024 * 1024 # reject a single OTLP body larger than 5 MiB +_MAX_ENTRIES = 1000 # keep at most this many buffered exports +_MAX_TOTAL_BYTES = 64 * 1024 * 1024 # keep at most this many bytes buffered +_SOCKET_READ_TIMEOUT_S = 30 # cap a single blocking body read + +# Buffered export bodies shared across handler threads, oldest first. deque and +# its append/clear are thread-safe, but iteration and the running byte total are +# not, so reads and writes are guarded by a lock. +_received: "deque[bytes]" = deque() +_total_bytes = 0 +_lock = threading.Lock() + + +def _record(body: bytes) -> None: + """Buffer ``body``, evicting oldest entries to stay within the bounds. + + Must be called while holding ``_lock``. + """ + global _total_bytes # pylint: disable=global-statement + _received.append(body) + _total_bytes += len(body) + while _received and ( + len(_received) > _MAX_ENTRIES or _total_bytes > _MAX_TOTAL_BYTES + ): + _total_bytes -= len(_received.popleft()) + + +class Handler(BaseHTTPRequestHandler): + """HTTP handler buffering OTLP exports and answering test queries.""" + + def _send_json(self, status: int, data: dict) -> None: + """Send a JSON response with the given status code.""" + body = json.dumps(data).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: # pylint: disable=invalid-name + """Serve the health probe and the buffered-data query endpoint.""" + path, _, query = self.path.partition("?") + if path == "/health": + self._send_json(200, {"status": "ok"}) + return + if path == "/received": + self._handle_received_query(query) + return + self._send_json(404, {"error": "not found"}) + + def _handle_received_query(self, query: str) -> None: + """Answer a count or substring query over the buffered exports.""" + contains = None + for pair in query.split("&"): + key, sep, value = pair.partition("=") + if sep and key == "contains": + contains = value + break + + if contains is not None: + needle = contains.encode("utf-8") + with _lock: + matches = sum(1 for body in _received if needle in body) + count = len(_received) + self._send_json( + 200, + { + "contains": contains, + "found": matches > 0, + "matches": matches, + "count": count, + }, + ) + return + + with _lock: + count = len(_received) + self._send_json(200, {"count": count}) + + def do_POST(self) -> None: # pylint: disable=invalid-name + """Buffer OTLP exports and handle the reset control endpoint.""" + path, _, _ = self.path.partition("?") + if path == "/reset": + global _total_bytes # pylint: disable=global-statement + with _lock: + _received.clear() + _total_bytes = 0 + self._send_json(200, {"status": "reset"}) + return + + # Any other POST is treated as an OTLP export (e.g. /v1/traces). + raw_length = self.headers.get("Content-Length", "0") + try: + length = int(raw_length) + except ValueError: + self._send_json(400, {"error": "invalid content-length"}) + return + if length < 0 or length > _MAX_BODY_BYTES: + self._send_json(413, {"error": "payload too large"}) + return + + # Bound the blocking read so a client that advertises a Content-Length but + # stalls mid-body cannot tie up a handler thread indefinitely. ``length`` + # is a validated non-negative int here; skip the read for a zero-length + # body so it becomes b"". + self.connection.settimeout(_SOCKET_READ_TIMEOUT_S) + try: + body = self.rfile.read(length) if length > 0 else b"" + except (TimeoutError, OSError): + self._send_json(408, {"error": "request timeout"}) + return + with _lock: + _record(body) + + # Acknowledge with an empty protobuf 200, which the OTLP exporter accepts. + self.send_response(200) + self.send_header("Content-Type", "application/x-protobuf") + self.send_header("Content-Length", "0") + self.end_headers() + + def log_message(self, format: str, *args: Any) -> None: + """Suppress default request logging for minimal test output.""" + + +def main() -> None: + """Start the mock OTLP collector on the requested port.""" + port = int(sys.argv[1]) if len(sys.argv) > 1 else 4318 + server = ThreadingHTTPServer(("0.0.0.0", port), Handler) + print(f"Mock OTEL collector on :{port}") + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/tests/unit/app/endpoints/responses_otel_helpers.py b/tests/unit/app/endpoints/responses_otel_helpers.py index c6605ed36..706ea2729 100644 --- a/tests/unit/app/endpoints/responses_otel_helpers.py +++ b/tests/unit/app/endpoints/responses_otel_helpers.py @@ -286,6 +286,7 @@ async def run_responses_setup_smoke( *, stream: bool, input_text: str = "What is Kubernetes?", + safety_identifier: str | None = None, ) -> ReadableSpan: """Run the handler through setup and return the root span.""" patch_responses_otel_tracers(mocker, tracer, minimal_config) @@ -306,6 +307,7 @@ async def run_responses_setup_smoke( store=False, conversation=OTEL_CONV_ID, generate_topic_summary=False, + safety_identifier=safety_identifier, ), auth=MOCK_AUTH, mcp_headers={}, diff --git a/tests/unit/app/endpoints/test_responses_otel.py b/tests/unit/app/endpoints/test_responses_otel.py index e6139a71e..be4b4002a 100644 --- a/tests/unit/app/endpoints/test_responses_otel.py +++ b/tests/unit/app/endpoints/test_responses_otel.py @@ -230,6 +230,54 @@ async def test_root_setup_attributes_and_validation_event( ) assert_root_setup_attributes(root, input_text=INPUT_TEXT) + @pytest.mark.asyncio + async def test_safety_identifier_recorded_raw_when_present( + self, + mocker: MockerFixture, + dummy_request: Request, + minimal_config: AppConfig, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """safety_identifier is recorded verbatim (not anonymized) on the root span.""" + tracer, exporter = otel + root = await run_responses_setup_smoke( + mocker, + dummy_request, + tracer, + minimal_config, + exporter, + stream=False, + input_text=INPUT_TEXT, + safety_identifier="e2e-otel-delivery-marker", + ) + assert root.attributes is not None + assert ( + root.attributes[SpanAttributes.SAFETY_IDENTIFIER] + == "e2e-otel-delivery-marker" + ) + + @pytest.mark.asyncio + async def test_safety_identifier_absent_when_not_provided( + self, + mocker: MockerFixture, + dummy_request: Request, + minimal_config: AppConfig, + otel: tuple[Any, InMemorySpanExporter], + ) -> None: + """No safety_identifier attribute is set when the request omits it.""" + tracer, exporter = otel + root = await run_responses_setup_smoke( + mocker, + dummy_request, + tracer, + minimal_config, + exporter, + stream=False, + input_text=INPUT_TEXT, + ) + assert root.attributes is not None + assert SpanAttributes.SAFETY_IDENTIFIER not in root.attributes + @pytest.mark.asyncio async def test_streaming_root_span_closed_on_setup_error( self,