🐛 Log ResponseValidationError from streaming endpoint serializers

fastapi/routing.py

@@ -58,6 +58,7 @@ from fastapi.exceptions import (
     ResponseValidationError,
     WebSocketRequestValidationError,
 )
+from fastapi.logger import logger
 from fastapi.sse import (
     _PING_INTERVAL,
     KEEPALIVE_COMMENT,
@@ -556,7 +557,14 @@ def get_request_handler(
                     async def _producer() -> None:
                         async with send_stream:
                             async for raw_item in sse_aiter:
-                                await send_stream.send(_serialize_sse_item(raw_item))
+                                try:
+                                    chunk = _serialize_sse_item(raw_item)
+                                except ResponseValidationError:
+                                    logger.exception(
+                                        "ResponseValidationError while serializing SSE stream item"
+                                    )
+                                    raise
+                                await send_stream.send(chunk)
 
                     send_keepalive, receive_keepalive = (
                         anyio.create_memory_object_stream[bytes](max_buffer_size=1)
@@ -628,7 +636,14 @@ def get_request_handler(
 
                     async def _async_stream_jsonl() -> AsyncIterator[bytes]:
                         async for item in gen:
-                            yield _serialize_item(item)
+                            try:
+                                chunk = _serialize_item(item)
+                            except ResponseValidationError:
+                                logger.exception(
+                                    "ResponseValidationError while serializing JSONL stream item"
+                                )
+                                raise
+                            yield chunk
                             # To allow for cancellation to trigger
                             # Ref: https://github.com/fastapi/fastapi/issues/14680
                             await anyio.sleep(0)
@@ -640,7 +655,14 @@ def get_request_handler(
 
                     def _sync_stream_jsonl() -> Iterator[bytes]:
                         for item in gen:  # ty: ignore[not-iterable]
-                            yield _serialize_item(item)
+                            try:
+                                chunk = _serialize_item(item)
+                            except ResponseValidationError:
+                                logger.exception(
+                                    "ResponseValidationError while serializing JSONL stream item"
+                                )
+                                raise
+                            yield chunk
 
                     jsonl_stream_content = _sync_stream_jsonl()
 

tests/test_stream_response_validation_error.py

@@ -0,0 +1,244 @@
+"""Tests for logging ResponseValidationError from streaming endpoint serializers.
+
+Regression tests for the bug where ResponseValidationError raised during
+serialization of a yielded item was silently swallowed with no log output.
+
+Covers:
+- SSE async generator
+- SSE sync generator
+- JSONL async generator
+- JSONL sync generator
+
+For each variant, verifies:
+1. Valid streams are unaffected (no spurious ERROR logs, correct body).
+2. Invalid items trigger an ERROR-level log entry with exc_info attached.
+3. The exception still propagates (is not swallowed).
+"""
+
+import logging
+from collections.abc import AsyncIterable, Iterable
+
+import pytest
+from fastapi import FastAPI
+from fastapi.exceptions import ResponseValidationError
+from fastapi.responses import EventSourceResponse
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+
+
+class Item(BaseModel):
+    id: int
+    name: str
+
+
+# ── App setup ────────────────────────────────────────────────────────────────
+
+app = FastAPI()
+
+
+@app.get("/sse/valid", response_class=EventSourceResponse)
+async def sse_valid() -> AsyncIterable[Item]:
+    yield Item(id=1, name="ok")
+    yield Item(id=2, name="also-ok")
+
+
+@app.get("/sse/valid-sync", response_class=EventSourceResponse)
+def sse_valid_sync() -> Iterable[Item]:
+    yield Item(id=1, name="ok")
+    yield Item(id=2, name="also-ok")
+
+
+@app.get("/sse/invalid", response_class=EventSourceResponse)
+async def sse_invalid() -> AsyncIterable[Item]:
+    # "id" must be an int; passing a non-coercible string causes
+    # ResponseValidationError during serialization.
+    yield {"id": "NOT_AN_INT", "name": "bad"}  # type: ignore[misc]
+
+
+@app.get("/sse/invalid-sync", response_class=EventSourceResponse)
+def sse_invalid_sync() -> Iterable[Item]:
+    yield {"id": "NOT_AN_INT", "name": "bad"}  # type: ignore[misc]
+
+
+@app.get("/jsonl/valid")
+async def jsonl_valid() -> AsyncIterable[Item]:
+    yield Item(id=1, name="ok")
+    yield Item(id=2, name="also-ok")
+
+
+@app.get("/jsonl/valid-sync")
+def jsonl_valid_sync() -> Iterable[Item]:
+    yield Item(id=1, name="ok")
+    yield Item(id=2, name="also-ok")
+
+
+@app.get("/jsonl/invalid")
+async def jsonl_invalid() -> AsyncIterable[Item]:
+    yield {"id": "NOT_AN_INT", "name": "bad"}  # type: ignore[misc]
+
+
+@app.get("/jsonl/invalid-sync")
+def jsonl_invalid_sync() -> Iterable[Item]:
+    yield {"id": "NOT_AN_INT", "name": "bad"}  # type: ignore[misc]
+
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture(name="client")
+def client_fixture():
+    with TestClient(app, raise_server_exceptions=False) as c:
+        yield c
+
+
+@pytest.fixture(name="raising_client")
+def raising_client_fixture():
+    """Client that lets ResponseValidationError propagate into the test."""
+    with TestClient(app, raise_server_exceptions=True) as c:
+        yield c
+
+
+# ── Valid stream tests (no regressions) ──────────────────────────────────────
+
+
+def test_sse_valid_stream_unaffected(
+    client: TestClient, caplog: pytest.LogCaptureFixture
+):
+    """Valid SSE streams must still produce correct output and zero ERROR logs."""
+    with caplog.at_level(logging.ERROR, logger="fastapi"):
+        response = client.get("/sse/valid")
+
+    assert response.status_code == 200
+    assert "text/event-stream" in response.headers["content-type"]
+    data_lines = [
+        line for line in response.text.splitlines() if line.startswith("data: ")
+    ]
+    assert len(data_lines) == 2
+    error_records = [r for r in caplog.records if r.levelno >= logging.ERROR]
+    assert error_records == [], (
+        f"Unexpected ERROR logs on valid stream: {error_records}"
+    )
+
+
+def test_jsonl_valid_stream_unaffected(
+    client: TestClient, caplog: pytest.LogCaptureFixture
+):
+    """Valid JSONL streams must still produce correct output and zero ERROR logs."""
+    with caplog.at_level(logging.ERROR, logger="fastapi"):
+        response = client.get("/jsonl/valid")
+
+    assert response.status_code == 200
+    assert "application/jsonl" in response.headers["content-type"]
+    lines = [line for line in response.text.splitlines() if line.strip()]
+    assert len(lines) == 2
+    error_records = [r for r in caplog.records if r.levelno >= logging.ERROR]
+    assert error_records == [], (
+        f"Unexpected ERROR logs on valid stream: {error_records}"
+    )
+
+
+def test_valid_sse_produces_no_error_logs(
+    client: TestClient, caplog: pytest.LogCaptureFixture
+):
+    """Sync valid SSE produces no spurious ERROR log."""
+    with caplog.at_level(logging.ERROR, logger="fastapi"):
+        response = client.get("/sse/valid-sync")
+    assert response.status_code == 200
+    error_records = [r for r in caplog.records if r.levelno >= logging.ERROR]
+    assert error_records == []
+
+
+def test_valid_jsonl_produces_no_error_logs(
+    client: TestClient, caplog: pytest.LogCaptureFixture
+):
+    """Sync valid JSONL produces no spurious ERROR log."""
+    with caplog.at_level(logging.ERROR, logger="fastapi"):
+        response = client.get("/jsonl/valid-sync")
+    assert response.status_code == 200
+    error_records = [r for r in caplog.records if r.levelno >= logging.ERROR]
+    assert error_records == []
+
+
+# ── Invalid item logging tests ────────────────────────────────────────────────
+
+
+def test_sse_invalid_item_is_logged(
+    client: TestClient, caplog: pytest.LogCaptureFixture
+):
+    """ResponseValidationError in async SSE serializer must emit an ERROR log with exc_info."""
+    with caplog.at_level(logging.ERROR, logger="fastapi"):
+        client.get("/sse/invalid")
+
+    error_records = [r for r in caplog.records if r.levelno >= logging.ERROR]
+    assert error_records, (
+        "Expected at least one ERROR log record for SSE validation failure"
+    )
+    record = error_records[0]
+    assert record.exc_info is not None, "ERROR log must carry exc_info (full traceback)"
+    assert "ResponseValidationError" in record.getMessage() or issubclass(
+        record.exc_info[0],
+        ResponseValidationError,  # type: ignore[index]
+    ), "ERROR log message or exc_info must reference ResponseValidationError"
+
+
+def test_jsonl_invalid_item_is_logged(
+    client: TestClient, caplog: pytest.LogCaptureFixture
+):
+    """ResponseValidationError in async JSONL serializer must emit an ERROR log with exc_info."""
+    with caplog.at_level(logging.ERROR, logger="fastapi"):
+        client.get("/jsonl/invalid")
+
+    error_records = [r for r in caplog.records if r.levelno >= logging.ERROR]
+    assert error_records, (
+        "Expected at least one ERROR log record for JSONL validation failure"
+    )
+    record = error_records[0]
+    assert record.exc_info is not None, "ERROR log must carry exc_info (full traceback)"
+    assert "ResponseValidationError" in record.getMessage() or issubclass(
+        record.exc_info[0],
+        ResponseValidationError,  # type: ignore[index]
+    ), "ERROR log message or exc_info must reference ResponseValidationError"
+
+
+def test_sse_invalid_item_logged_sync(
+    client: TestClient, caplog: pytest.LogCaptureFixture
+):
+    """ResponseValidationError in sync SSE serializer must emit an ERROR log with exc_info."""
+    with caplog.at_level(logging.ERROR, logger="fastapi"):
+        client.get("/sse/invalid-sync")
+
+    error_records = [r for r in caplog.records if r.levelno >= logging.ERROR]
+    assert error_records, "Expected ERROR log for sync SSE validation failure"
+    assert error_records[0].exc_info is not None
+
+
+def test_jsonl_invalid_item_logged_sync(
+    client: TestClient, caplog: pytest.LogCaptureFixture
+):
+    """ResponseValidationError in sync JSONL serializer must emit an ERROR log with exc_info."""
+    with caplog.at_level(logging.ERROR, logger="fastapi"):
+        client.get("/jsonl/invalid-sync")
+
+    error_records = [r for r in caplog.records if r.levelno >= logging.ERROR]
+    assert error_records, "Expected ERROR log for sync JSONL validation failure"
+    assert error_records[0].exc_info is not None
+
+
+# ── Exception propagation tests ───────────────────────────────────────────────
+
+
+def test_sse_invalid_item_propagates(raising_client: TestClient):
+    """ResponseValidationError from async SSE must still propagate (not be swallowed).
+
+    The SSE _producer task runs inside an anyio task group, so the error is
+    wrapped in an ExceptionGroup when it escapes the group boundary.
+    """
+    # anyio task group wraps the sub-task exception in an ExceptionGroup
+    with pytest.raises((ResponseValidationError, ExceptionGroup)):  # type: ignore[name-defined]  # noqa: F821
+        raising_client.get("/sse/invalid")
+
+
+def test_jsonl_invalid_item_propagates(raising_client: TestClient):
+    """ResponseValidationError from async JSONL must still propagate (not be swallowed)."""
+    with pytest.raises(ResponseValidationError):
+        raising_client.get("/jsonl/invalid")