Requests chunk encoded request handling

Author: p1c2u

openapi_core/contrib/requests/requests.py

@@ -1,4 +1,5 @@
 """OpenAPI core contrib requests requests module"""
+from typing import Mapping
 from typing import Optional
 from typing import Union
 from urllib.parse import parse_qs
@@ -7,6 +8,7 @@
 from requests import PreparedRequest
 from requests import Request
 from requests.cookies import RequestsCookieJar
+from requests.utils import rewind_body
 from werkzeug.datastructures import Headers
 from werkzeug.datastructures import ImmutableMultiDict
 
@@ -67,6 +69,17 @@ def method(self) -> str:
     def body(self) -> Optional[str]:
         if self.request.body is None:
             return None
+        is_stream = all(
+            [
+                hasattr(self.request.body, "__iter__"),
+                not isinstance(self.request.body, (str, list, tuple, Mapping)),
+            ]
+        )
+        if is_stream:
+            chunks = list(self.request.body)
+            body = "".join(chunks)
+            self.request.body = (x for x in chunks)
+            return body
         if isinstance(self.request.body, bytes):
             return self.request.body.decode("utf-8")
         assert isinstance(self.request.body, str)

tests/integration/contrib/requests/test_requests_validation.py

@@ -1,3 +1,5 @@
+from types import GeneratorType
+
 import pytest
 import requests
 import responses
@@ -72,6 +74,22 @@ def test_request_validator_path_pattern(self, request_unmarshaller):
         result = request_unmarshaller.unmarshal(openapi_request)
         assert not result.errors
 
+    def test_request_validator_encoded_chunks(self, request_unmarshaller):
+        def gen():
+            yield '{"param1": 1}'
+
+        request = requests.Request(
+            "POST",
+            "http://localhost/browse/12/",
+            params={"q": "string"},
+            headers={"content-type": "application/json"},
+            data=gen(),
+        )
+        openapi_request = RequestsOpenAPIRequest(request)
+        result = request_unmarshaller.unmarshal(openapi_request)
+        assert not result.errors
+        assert request.data is GeneratorType
+
     def test_request_validator_prepared_request(self, request_unmarshaller):
         request = requests.Request(
             "POST",