Commit 849e0aaa authored by Jérome Perrin's avatar Jérome Perrin

open_api: more request body support

This fixes python2 support of request body and adds support for
base64 format. Multipart request body are not supported at this point.
parent 831038c2
......@@ -25,9 +25,16 @@
#
##############################################################################
import binascii
import json
import typing
from six.moves.urllib.parse import unquote
import six
if six.PY2:
from base64 import decodestring as base64_decodebytes
else:
from base64 import decodebytes as base64_decodebytes
if typing.TYPE_CHECKING:
from typing import Any, Callable, Optional
from erp5.component.document.OpenAPITypeInformation import OpenAPIOperation, OpenAPIParameter
......@@ -258,10 +265,8 @@ class OpenAPIService(XMLObject):
parameter,
parameter.getJSONSchema(),
)
requestBody = self.validateParameter(
'request body',
requestBody = self.validateRequestBody(
operation.getRequestBodyValue(request),
{},
operation.getRequestBodyJSONSchema(request),
)
if requestBody:
......@@ -296,6 +301,31 @@ class OpenAPIService(XMLObject):
parameter_name=parameter_name, e=e.message), str(e))
return parameter_value
security.declareProtected(
Permissions.AccessContentsInformation, 'validateRequestBody')
def validateRequestBody(self, parameter_value, schema):
# type: (str, dict) -> Any
"""Validate the request body raising a ParameterValidationError
when the parameter is not valid according to the corresponding schema.
"""
if schema is not None:
if schema.get('type') == 'string':
if schema.get('format') == 'base64':
try:
return base64_decodebytes(parameter_value)
except binascii.Error as e:
raise ParameterValidationError(
'Error validating request body: {e}'.format(e=str(e)))
elif schema.get('format') == 'binary':
return parameter_value or b''
return self.validateParameter(
'request body',
parameter_value,
{},
schema,
)
def executeMethod(self, request):
# type: (HTTPRequest) -> Any
operation = self.getMatchingOperation(request)
......
......@@ -175,13 +175,15 @@ class OpenAPIOperation(dict):
# type: (HTTPRequest) -> Optional[dict]
"""Returns the schema for the request body, or None if no `requestBody` defined
"""
request_content_type = request.getHeader('content-type')
# TODO there might be $ref ?
request_body_definition = self.get(
'requestBody', {'content': {}})['content'].get(request_content_type)
if request_body_definition:
return SchemaWithComponents(
self._schema, request_body_definition.get('schema', {}))
exact_request_content_type = request.getHeader('content-type')
wildcard_request_content_type = '%s/*' % ((exact_request_content_type or '').split('/')[0])
for request_content_type in exact_request_content_type, wildcard_request_content_type, '*/*':
# TODO there might be $ref ?
request_body_definition = self.get(
'requestBody', {'content': {}})['content'].get(request_content_type)
if request_body_definition:
return SchemaWithComponents(
self._schema, request_body_definition.get('schema', {}))
class OpenAPIParameter(dict):
......
......@@ -25,6 +25,11 @@
#
##############################################################################
import six
if six.PY2:
from base64 import encodestring as base64_encodebytes
else:
from base64 import encodebytes as base64_encodebytes
import io
import json
import unittest
......@@ -1242,3 +1247,72 @@ class TestURLPathWithWebSiteAndVirtualHost(OpenAPIPetStoreTestCase):
self.connector.getRelativeUrl()
))
self.assertEqual(response.getBody(), b'"ok"')
class TestOpenAPIRequestBody(OpenAPITestCase):
_type_id = 'Test Open API Request Body'
_open_api_schema = json.dumps(
{
'openapi': '3.0.3',
'info': {
'title': 'TestOpenAPIRequestBody',
'version': '0.0.0'
},
'paths': {
'/post': {
'post': {
'operationId': 'testPostByContentType',
'requestBody': {
'content': {
'image/*': {
'schema': {
'type': 'string',
'format': 'binary',
}
},
'application/x-base64': {
'schema': {
'type': 'string',
'format': 'base64',
}
}
}
}
}
}
}
})
def test_request_body_content_encoding(self):
self.addPythonScript(
'TestOpenAPIRequestBody_testPostByContentType',
'body=None',
'container.REQUEST.RESPONSE.setHeader("Content-Type", "application/octet-stream")\n'
'return body',
)
response = self.publish(
self.connector.getPath() + '/post',
request_method='POST',
stdin=io.BytesIO(b'png file content'),
env={"CONTENT_TYPE": 'image/png'})
self.assertEqual(response.getBody(), b'png file content')
self.assertEqual(response.getStatus(), 200)
response = self.publish(
self.connector.getPath() + '/post',
request_method='POST',
stdin=io.BytesIO(base64_encodebytes(b'base64 file content')),
env={"CONTENT_TYPE": 'application/x-base64'})
self.assertEqual(response.getBody(), b'base64 file content')
self.assertEqual(response.getStatus(), 200)
response = self.publish(
self.connector.getPath() + '/post',
request_method='POST',
stdin=io.BytesIO(b'not base64'),
env={"CONTENT_TYPE": 'application/x-base64'})
self.assertEqual(response.getStatus(), 400)
body = json.loads(response.getBody())
self.assertEqual(body['type'], 'parameter-validation-error')
self.assertIn('Error validating request body:', body['title'])
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment