request response binary format integration tests

openapi_core/contrib/starlette/requests.py

@@ -45,7 +45,7 @@ def body(self) -> Optional[str]:
 
     @property
     def mimetype(self) -> str:
-        content_type = self.request.headers["Content-Type"]
+        content_type = self.request.headers.get("Content-Type")
         if content_type:
             return content_type.partition(";")[0]
 

tests/integration/conftest.py

@@ -1,3 +1,4 @@
+from base64 import b64decode
 from os import path
 from urllib import request
 
@@ -25,6 +26,19 @@ def spec_from_url(base_uri):
     return Spec.from_dict(spec_dict, base_uri=base_uri)
 
 
+@pytest.fixture(scope="session")
+def data_gif():
+    return b64decode(
+        """
+R0lGODlhEAAQAMQAAO3t7eHh4srKyvz8/P5pDP9rENLS0v/28P/17tXV1dHEvPDw8M3Nzfn5+d3d
+3f5jA97Syvnv6MfLzcfHx/1mCPx4Kc/S1Pf189C+tP+xgv/k1N3OxfHy9NLV1/39/f///yH5BAAA
+AAAALAAAAAAQABAAAAVq4CeOZGme6KhlSDoexdO6H0IUR+otwUYRkMDCUwIYJhLFTyGZJACAwQcg
+EAQ4kVuEE2AIGAOPQQAQwXCfS8KQGAwMjIYIUSi03B7iJ+AcnmclHg4TAh0QDzIpCw4WGBUZeikD
+Fzk0lpcjIQA7
+"""
+    )
+
+
 class Factory(dict):
     __getattr__ = dict.__getitem__
     __setattr__ = dict.__setitem__

tests/integration/contrib/aiohttp/data/v3.0/aiohttpproject/__main__.py

@@ -0,0 +1,15 @@
+from aiohttp import web
+from aiohttpproject.pets.views import PetPhotoView
+
+routes = [
+    web.view("/v1/pets/{petId}/photo", PetPhotoView),
+]
+
+
+def get_app(loop=None):
+    app = web.Application(loop=loop)
+    app.add_routes(routes)
+    return app
+
+
+app = get_app()

tests/integration/contrib/aiohttp/data/v3.0/aiohttpproject/openapi.py

@@ -0,0 +1,9 @@
+from pathlib import Path
+
+import yaml
+
+from openapi_core import Spec
+
+openapi_spec_path = Path("tests/integration/data/v3.0/petstore.yaml")
+spec_dict = yaml.load(openapi_spec_path.read_text(), yaml.Loader)
+spec = Spec.from_dict(spec_dict)

tests/integration/contrib/aiohttp/data/v3.0/aiohttpproject/pets/views.py

@@ -0,0 +1,52 @@
+from base64 import b64decode
+from io import BytesIO
+
+from aiohttp import web
+from aiohttpproject.openapi import spec
+from multidict import MultiDict
+
+from openapi_core import unmarshal_request
+from openapi_core import unmarshal_response
+from openapi_core.contrib.aiohttp import AIOHTTPOpenAPIWebRequest
+from openapi_core.contrib.aiohttp import AIOHTTPOpenAPIWebResponse
+
+
+class PetPhotoView(web.View):
+    OPENID_LOGO = b64decode(
+        """
+R0lGODlhEAAQAMQAAO3t7eHh4srKyvz8/P5pDP9rENLS0v/28P/17tXV1dHEvPDw8M3Nzfn5+d3d
+3f5jA97Syvnv6MfLzcfHx/1mCPx4Kc/S1Pf189C+tP+xgv/k1N3OxfHy9NLV1/39/f///yH5BAAA
+AAAALAAAAAAQABAAAAVq4CeOZGme6KhlSDoexdO6H0IUR+otwUYRkMDCUwIYJhLFTyGZJACAwQcg
+EAQ4kVuEE2AIGAOPQQAQwXCfS8KQGAwMjIYIUSi03B7iJ+AcnmclHg4TAh0QDzIpCw4WGBUZeikD
+Fzk0lpcjIQA7
+"""
+    )
+
+    async def get(self):
+        request_body = await self.request.text()
+        openapi_request = AIOHTTPOpenAPIWebRequest(
+            self.request, body=request_body
+        )
+        request_unmarshalled = unmarshal_request(openapi_request, spec=spec)
+        response = web.Response(
+            body=self.OPENID_LOGO,
+            content_type="image/gif",
+        )
+        openapi_response = AIOHTTPOpenAPIWebResponse(response)
+        response_unmarshalled = unmarshal_response(
+            openapi_request, openapi_response, spec=spec
+        )
+        return response
+
+    async def post(self):
+        request_body = await self.request.read()
+        openapi_request = AIOHTTPOpenAPIWebRequest(
+            self.request, body=request_body
+        )
+        request_unmarshalled = unmarshal_request(openapi_request, spec=spec)
+        response = web.Response(status=201)
+        openapi_response = AIOHTTPOpenAPIWebResponse(response)
+        response_unmarshalled = unmarshal_response(
+            openapi_request, openapi_response, spec=spec
+        )
+        return response

tests/integration/contrib/aiohttp/test_aiohttp_project.py

@@ -0,0 +1,83 @@
+import os
+import sys
+from base64 import b64encode
+
+import pytest
+from starlette.testclient import TestClient
+
+
+@pytest.fixture(autouse=True, scope="session")
+def project_setup():
+    directory = os.path.abspath(os.path.dirname(__file__))
+    project_dir = os.path.join(directory, "data/v3.0")
+    sys.path.insert(0, project_dir)
+    yield
+    sys.path.remove(project_dir)
+
+
+@pytest.fixture
+def app(project_setup, loop):
+    from aiohttpproject.__main__ import get_app
+
+    return get_app(loop=loop)
+
+
+@pytest.fixture
+async def client(app, aiohttp_client):
+    return await aiohttp_client(app)
+
+
+class BaseTestPetstore:
+    api_key = "12345"
+
+    @property
+    def api_key_encoded(self):
+        api_key_bytes = self.api_key.encode("utf8")
+        api_key_bytes_enc = b64encode(api_key_bytes)
+        return str(api_key_bytes_enc, "utf8")
+
+
+class TestPetPhotoView(BaseTestPetstore):
+    @pytest.mark.xfail(
+        reason="response binary format not supported",
+        strict=True,
+    )
+    async def test_get_valid(self, client, data_gif):
+        headers = {
+            "Authorization": "Basic testuser",
+            "Api-Key": self.api_key_encoded,
+            "Host": "petstore.swagger.io",
+        }
+
+        cookies = {"user": "1"}
+        response = await client.get(
+            "/v1/pets/1/photo",
+            headers=headers,
+            cookies=cookies,
+        )
+
+        assert await response.content.read() == data_gif
+        assert response.status == 200
+
+    async def test_post_valid(self, client, data_gif):
+        content_type = "image/gif"
+        headers = {
+            "Authorization": "Basic testuser",
+            "Api-Key": self.api_key_encoded,
+            "Content-Type": content_type,
+            "Host": "petstore.swagger.io",
+        }
+        data = {
+            "file": data_gif,
+        }
+
+        cookies = {"user": "1"}
+        response = await client.post(
+            "/v1/pets/1/photo",
+            headers=headers,
+            data=data,
+            cookies=cookies,
+        )
+
+        assert not await response.text()
+        assert response.status == 201

tests/integration/contrib/django/data/v3.0/djangoproject/pets/views.py

@@ -1,3 +1,7 @@
+from base64 import b64decode
+
+from django.conf import settings
+from django.http import FileResponse
 from django.http import HttpResponse
 from django.http import JsonResponse
 from rest_framework.views import APIView
@@ -76,6 +80,43 @@ def get(self, request, petId):
         }
         django_response = JsonResponse(response_dict)
         django_response["X-Rate-Limit"] = "12"
+        return django_response
+
+    @staticmethod
+    def get_extra_actions():
+        return []
+
+
+class PetPhotoView(APIView):
+    OPENID_LOGO = b64decode(
+        """
+R0lGODlhEAAQAMQAAO3t7eHh4srKyvz8/P5pDP9rENLS0v/28P/17tXV1dHEvPDw8M3Nzfn5+d3d
+3f5jA97Syvnv6MfLzcfHx/1mCPx4Kc/S1Pf189C+tP+xgv/k1N3OxfHy9NLV1/39/f///yH5BAAA
+AAAALAAAAAAQABAAAAVq4CeOZGme6KhlSDoexdO6H0IUR+otwUYRkMDCUwIYJhLFTyGZJACAwQcg
+EAQ4kVuEE2AIGAOPQQAQwXCfS8KQGAwMjIYIUSi03B7iJ+AcnmclHg4TAh0QDzIpCw4WGBUZeikD
+Fzk0lpcjIQA7
+"""
+    )
+
+    def get(self, request, petId):
+        assert request.openapi
+        assert not request.openapi.errors
+        assert request.openapi.parameters.path == {
+            "petId": 12,
+        }
+        django_response = FileResponse(
+            [self.OPENID_LOGO],
+            content_type="image/gif",
+        )
+        return django_response
+
+    def post(self, request):
+        assert request.openapi
+        assert not request.openapi.errors
+
+        # implement file upload here
+
+        django_response = HttpResponse(status=201)
 
         return django_response
 

tests/integration/contrib/django/data/v3.0/djangoproject/urls.py

@@ -18,6 +18,7 @@
 from django.urls import path
 from djangoproject.pets.views import PetDetailView
 from djangoproject.pets.views import PetListView
+from djangoproject.pets.views import PetPhotoView
 from djangoproject.tags.views import TagListView
 
 urlpatterns = [
@@ -36,6 +37,11 @@
         PetDetailView.as_view(),
         name="pet_detail_view",
     ),
+    path(
+        "v1/pets/<int:petId>/photo",
+        PetPhotoView.as_view(),
+        name="pet_photo_view",
+    ),
     path(
         "v1/tags",
         TagListView.as_view(),

tests/integration/contrib/django/test_django_project.py

@@ -395,3 +395,38 @@ def test_get_skip_response_validation(self, client):
 
         assert response.status_code == 200
         assert response.content == b"success"
+
+
+class TestPetPhotoView(BaseTestDjangoProject):
+    @pytest.mark.xfail(
+        reason="response binary format not supported",
+        strict=True,
+    )
+    def test_get_valid(self, client, data_gif):
+        headers = {
+            "HTTP_AUTHORIZATION": "Basic testuser",
+            "HTTP_HOST": "petstore.swagger.io",
+        }
+        response = client.get("/v1/pets/12/photo", **headers)
+
+        assert response.status_code == 200
+        assert b"".join(list(response.streaming_content)) == data_gif
+
+    @pytest.mark.xfail(
+        reason="request binary format not supported",
+        strict=True,
+    )
+    def test_post_valid(self, client, data_gif):
+        client.cookies.load({"user": 1})
+        content_type = "image/gif"
+        headers = {
+            "HTTP_AUTHORIZATION": "Basic testuser",
+            "HTTP_HOST": "petstore.swagger.io",
+            "HTTP_API_KEY": self.api_key_encoded,
+        }
+        response = client.post(
+            "/v1/pets/12/photo", data_gif, content_type, secure=True, **headers
+        )
+
+        assert response.status_code == 201
+        assert not response.content

tests/integration/contrib/falcon/data/v3.0/falconproject/__main__.py

@@ -2,11 +2,14 @@
 from falconproject.openapi import openapi_middleware
 from falconproject.pets.resources import PetDetailResource
 from falconproject.pets.resources import PetListResource
+from falconproject.pets.resources import PetPhotoResource
 
 app = App(middleware=[openapi_middleware])
 
 pet_list_resource = PetListResource()
 pet_detail_resource = PetDetailResource()
+pet_photo_resource = PetPhotoResource()
 
 app.add_route("/v1/pets", pet_list_resource)
 app.add_route("/v1/pets/{petId}", pet_detail_resource)
+app.add_route("/v1/pets/{petId}/photo", pet_photo_resource)

tests/integration/contrib/falcon/data/v3.0/falconproject/pets/resources.py

@@ -1,5 +1,7 @@
+from base64 import b64decode
 from json import dumps
 
+from falcon.constants import MEDIA_JPEG
 from falcon.constants import MEDIA_JSON
 from falcon.status_codes import HTTP_200
 from falcon.status_codes import HTTP_201
@@ -74,3 +76,23 @@ def on_get(self, request, response, petId=None):
         response.content_type = MEDIA_JSON
         response.text = dumps({"data": data})
         response.set_header("X-Rate-Limit", "12")
+
+
+class PetPhotoResource:
+    OPENID_LOGO = b64decode(
+        """
+R0lGODlhEAAQAMQAAO3t7eHh4srKyvz8/P5pDP9rENLS0v/28P/17tXV1dHEvPDw8M3Nzfn5+d3d
+3f5jA97Syvnv6MfLzcfHx/1mCPx4Kc/S1Pf189C+tP+xgv/k1N3OxfHy9NLV1/39/f///yH5BAAA
+AAAALAAAAAAQABAAAAVq4CeOZGme6KhlSDoexdO6H0IUR+otwUYRkMDCUwIYJhLFTyGZJACAwQcg
+EAQ4kVuEE2AIGAOPQQAQwXCfS8KQGAwMjIYIUSi03B7iJ+AcnmclHg4TAh0QDzIpCw4WGBUZeikD
+Fzk0lpcjIQA7
+"""
+    )
+
+    def on_get(self, request, response, petId=None):
+        response.content_type = MEDIA_JPEG
+        response.stream = [self.OPENID_LOGO]
+
+    def on_post(self, request, response, petId=None):
+        data = request.stream.read()
+        response.status = HTTP_201