Avoid async-generator finalization side effects

Author: p1c2u

openapi_core/contrib/falcon/requests.py

@@ -15,6 +15,8 @@
 from openapi_core.contrib.falcon.util import unpack_params
 from openapi_core.datatypes import RequestParameters
 
+_BODY_NOT_SET = object()
+
 
 class FalconOpenAPIRequest:
     def __init__(
@@ -28,7 +30,7 @@ def __init__(
         if default_when_empty is None:
             default_when_empty = {}
         self.default_when_empty = default_when_empty
-        self._body: Optional[bytes] = None
+        self._body: Any = _BODY_NOT_SET
 
         # Path gets deduced by path finder against spec
         self.parameters = RequestParameters(
@@ -54,21 +56,23 @@ def method(self) -> str:
 
     @property
     def body(self) -> Optional[bytes]:
-        if self._body is not None:
-            return self._body
+        if self._body is not _BODY_NOT_SET:
+            return cast(Optional[bytes], self._body)
 
         # Falcon doesn't store raw request stream.
         # That's why we need to revert deserialized data
 
         # Support falcon-jsonify.
         request_json = getattr(cast(Any, self.request), "json", None)
         if request_json is not None:
-            return dumps(request_json).encode("utf-8")
+            self._body = dumps(request_json).encode("utf-8")
+            return cast(Optional[bytes], self._body)
 
         media = self.request.get_media(
             default_when_empty=self.default_when_empty,
         )
-        return serialize_body(self.request, media, self.content_type)
+        self._body = serialize_body(self.request, media, self.content_type)
+        return cast(Optional[bytes], self._body)
 
     @property
     def content_type(self) -> str:

openapi_core/contrib/falcon/responses.py

@@ -4,7 +4,6 @@
 from io import BytesIO
 from itertools import tee
 from typing import Any
-from typing import AsyncIterator
 from typing import Iterable
 from typing import List
 
@@ -78,7 +77,7 @@ async def _get_asgi_response_data(cls, response: Any) -> bytes:
             assert isinstance(data, bytes)
             return data
 
-        charset = response_any.charset or "utf-8"
+        charset = getattr(response_any, "charset", None) or "utf-8"
         chunks: List[bytes] = []
         stream_any = stream
 
@@ -108,14 +107,26 @@ async def _get_asgi_response_data(cls, response: Any) -> bytes:
                 chunks.append(chunk)
             return b"".join(chunks)
 
-        response_any.stream = cls._iter_chunks(chunks)
+        response_any.stream = _AsyncChunksIterator(chunks)
         return b"".join(chunks)
 
-    @staticmethod
-    async def _iter_chunks(chunks: Iterable[bytes]) -> AsyncIterator[bytes]:
-        for chunk in chunks:
-            yield chunk
-
     @property
     def data(self) -> bytes:
         return self._data
+
+
+class _AsyncChunksIterator:
+    def __init__(self, chunks: List[bytes]):
+        self._chunks = chunks
+        self._index = 0
+
+    def __aiter__(self) -> "_AsyncChunksIterator":
+        return self
+
+    async def __anext__(self) -> bytes:
+        if self._index >= len(self._chunks):
+            raise StopAsyncIteration
+
+        chunk = self._chunks[self._index]
+        self._index += 1
+        return chunk

tests/integration/contrib/falcon/test_falcon_asgi_middleware.py

@@ -1,16 +1,22 @@
 from json import dumps
 from pathlib import Path
+from typing import Any
+from typing import cast
 
 import pytest
 import yaml
 from falcon import status_codes
 from falcon.asgi import App
+from falcon.asgi import Response
 from falcon.constants import MEDIA_JSON
 from falcon.testing import TestClient
 from jsonschema_path import SchemaPath
 
 from openapi_core.contrib.falcon.middlewares import FalconASGIOpenAPIMiddleware
 from openapi_core.contrib.falcon.middlewares import FalconOpenAPIMiddleware
+from openapi_core.contrib.falcon.requests import FalconAsgiOpenAPIRequest
+from openapi_core.contrib.falcon.responses import FalconAsgiOpenAPIResponse
+from openapi_core.contrib.falcon.util import serialize_body
 
 
 @pytest.fixture
@@ -52,6 +58,23 @@ async def on_get(self, req, resp):
         resp.set_header("X-Rate-Limit", "12")
 
 
+class _AsyncStream:
+    def __init__(self, chunks):
+        self._chunks = chunks
+        self._index = 0
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if self._index >= len(self._chunks):
+            raise StopAsyncIteration
+
+        chunk = self._chunks[self._index]
+        self._index += 1
+        return chunk
+
+
 def test_dual_mode_sync_middleware_works_with_asgi_app(spec):
     middleware = FalconOpenAPIMiddleware.from_spec(spec)
     app = App(middleware=[middleware])
@@ -121,3 +144,68 @@ def test_explicit_asgi_middleware_validates_response(spec):
 
     assert response.status_code == 400
     assert "errors" in response.json
+
+
+@pytest.mark.asyncio
+async def test_asgi_response_adapter_handles_stream_without_charset():
+    chunks = [
+        b'{"data": [',
+        b'{"id": 12, "name": "Cat", "ears": {"healthy": true}}',
+        b"]}",
+    ]
+    response = Response()
+    response.content_type = MEDIA_JSON
+    response.stream = _AsyncStream(chunks)
+
+    openapi_response = await FalconAsgiOpenAPIResponse.from_response(response)
+
+    assert openapi_response.data == b"".join(chunks)
+    assert response.stream is not None
+
+    replayed_chunks = []
+    async for chunk in response.stream:
+        replayed_chunks.append(chunk)
+    assert b"".join(replayed_chunks) == b"".join(chunks)
+
+
+def test_asgi_request_body_cached_none_skips_media_deserialization():
+    class _DummyRequest:
+        def get_media(self, *args, **kwargs):
+            raise AssertionError("get_media should not be called")
+
+    openapi_request = object.__new__(FalconAsgiOpenAPIRequest)
+    openapi_request.request = cast(Any, _DummyRequest())
+    openapi_request._body = None
+
+    assert openapi_request.body is None
+
+
+def test_multipart_unsupported_serialization_warns_and_returns_none():
+    content_type = "multipart/form-data; boundary=test"
+
+    class _DummyHandler:
+        def serialize(self, media, content_type):
+            raise NotImplementedError(
+                "multipart form serialization unsupported"
+            )
+
+    class _DummyMediaHandlers:
+        def _resolve(self, content_type, default_media_type):
+            return (_DummyHandler(), content_type, None)
+
+    class _DummyOptions:
+        media_handlers = _DummyMediaHandlers()
+        default_media_type = MEDIA_JSON
+
+    class _DummyRequest:
+        options = _DummyOptions()
+
+    with pytest.warns(
+        UserWarning,
+        match="body serialization for multipart/form-data",
+    ):
+        body = serialize_body(
+            cast(Any, _DummyRequest()), {"name": "Cat"}, content_type
+        )
+
+    assert body is None