test: validate opaque stream IDs and hide log-reader credentials

tests/e2e/llm_translation/test_responses_bridge_streaming_e2e.py

@@ -16,8 +16,10 @@ into a chat completion chunk. Two customer-visible contracts only hold on that p
 
 from __future__ import annotations
 
+from typing import Final, Literal
+
 import pytest
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from e2e_config import unique_marker
 from e2e_http import StreamingResponse
@@ -51,7 +53,8 @@ class _BridgeChoice(BaseModel):
 
 class _BridgeChunk(BaseModel):
     id: str
-    choices: list[_BridgeChoice] = []
+    object: Literal["chat.completion.chunk"]
+    choices: list[_BridgeChoice] = Field(default_factory=list)
 
 
 class _WeatherArgs(BaseModel):
@@ -103,16 +106,19 @@ class TestResponsesBridgeChatCompletionsStreaming:
             resources.key(),
             ChatBody(
                 model=bridged_model,
-                messages=[ChatMessage(role="user", content=f"Count from 1 to 5, one number per line. {unique_marker()}")],
+                messages=[
+                    ChatMessage(role="user", content=f"Count from 1 to 5, one number per line. {unique_marker()}")
+                ],
                 max_tokens=64,
                 stream=True,
             ),
         )
 
-        chunks = _bridge_chunks(result)
-        ids = {chunk.id for chunk in chunks}
+        chunks: Final = _bridge_chunks(result)
+        assert len(chunks) > 1, "the shared-id contract needs more than one streamed chunk"
+        ids: Final = frozenset(chunk.id for chunk in chunks)
         assert len(ids) == 1, f"bridged stream used {len(ids)} different chunk ids: {sorted(ids)[:5]}"
-        assert ids.pop().startswith("chatcmpl-"), f"bridged chunk id is not chat-completion shaped: {chunks[0].id}"
+        assert chunks[0].id.strip(), "bridged stream emitted an empty chunk id"
 
     @pytest.mark.covers(
         "llm.chat_completions.openai.basic.stream.bridge_streams_sse",
@@ -134,9 +140,9 @@ class TestResponsesBridgeChatCompletionsStreaming:
         chunks = _bridge_chunks(result)
         content = "".join(choice.delta.content or "" for chunk in chunks for choice in chunk.choices)
         assert content.strip(), f"bridged stream completed with no content deltas: {result.stream_events[:3]}"
-        assert any(
-            choice.finish_reason for chunk in chunks for choice in chunk.choices
-        ), f"bridged stream never emitted a finish_reason: {result.stream_events[-3:]}"
+        assert any(choice.finish_reason for chunk in chunks for choice in chunk.choices), (
+            f"bridged stream never emitted a finish_reason: {result.stream_events[-3:]}"
+        )
         assert result.stream_done, f"bridged stream did not terminate with [DONE]: {result.stream_events[-2:]}"
 
     @pytest.mark.covers(

tests/e2e/logging/datadog_reader.py

@@ -13,7 +13,7 @@ empty result. External reads go through ``e2e_http``.
 from __future__ import annotations
 
 import time
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 import pytest
 from pydantic import BaseModel, ConfigDict, Field
@@ -36,8 +36,8 @@ _RATE_LIMIT_RETRIES = 5
 
 
 class _DdAuthHeaders(Headers):
-    api_key: str = Field(serialization_alias="DD-API-KEY")
-    app_key: str = Field(serialization_alias="DD-APPLICATION-KEY")
+    api_key: str = Field(serialization_alias="DD-API-KEY", repr=False)
+    app_key: str = Field(serialization_alias="DD-APPLICATION-KEY", repr=False)
 
 
 class _SearchFilter(BaseModel):
@@ -88,8 +88,8 @@ class _SearchResponse(BaseModel):
 @dataclass(frozen=True, slots=True)
 class DdLogsReader:
     site: str
-    api_key: str
-    app_key: str
+    api_key: str = field(repr=False)
+    app_key: str = field(repr=False)
 
     def events_for_marker(self, marker: str) -> list[DdLogEvent]:
         """Every ingested event whose attributes carry the marker. DataDog

tests/e2e/logging/test_datadog_reader.py

@@ -0,0 +1,20 @@
+from typing import Final
+
+from datadog_reader import DdLogsReader
+from datadog_reader import _DdAuthHeaders  # pyright: ignore[reportPrivateUsage]  # verifies private auth-header serialization
+
+
+def test_failure_diagnostics_hide_credentials_without_changing_auth_headers() -> None:
+    api_key: Final = "test-datadog-api-secret"
+    app_key: Final = "test-datadog-app-secret"
+    reader: Final = DdLogsReader(site="datadoghq.com", api_key=api_key, app_key=app_key)
+    headers: Final = _DdAuthHeaders(api_key=api_key, app_key=app_key)
+
+    for value in (reader, headers):
+        assert api_key not in repr(value)
+        assert app_key not in repr(value)
+
+    assert headers.model_dump(by_alias=True) == {
+        "DD-API-KEY": api_key,
+        "DD-APPLICATION-KEY": app_key,
+    }