[python] fixes multipart/form-data bug (#13926)

* api_client template updated

* Samples regenerated

* Fixes tests

* Adds missing test file

* Adds test fix
This commit is contained in:
Justin Black 2022-11-05 17:10:24 +01:00 committed by GitHub
parent d5f896fe20
commit b35ea31e82
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 193 additions and 99 deletions

View File

@ -1399,24 +1399,24 @@ class RequestBody(StyleFormSerializer, JSONDetector):
def __multipart_json_item(self, key: str, value: Schema) -> RequestField: def __multipart_json_item(self, key: str, value: Schema) -> RequestField:
json_value = self.__json_encoder.default(value) json_value = self.__json_encoder.default(value)
return RequestField(name=key, data=json.dumps(json_value), headers={'Content-Type': 'application/json'}) request_field = RequestField(name=key, data=json.dumps(json_value))
request_field.make_multipart(content_type='application/json')
return request_field
def __multipart_form_item(self, key: str, value: Schema) -> RequestField: def __multipart_form_item(self, key: str, value: Schema) -> RequestField:
if isinstance(value, str): if isinstance(value, str):
return RequestField(name=key, data=str(value), headers={'Content-Type': 'text/plain'}) request_field = RequestField(name=key, data=str(value))
request_field.make_multipart(content_type='text/plain')
elif isinstance(value, bytes): elif isinstance(value, bytes):
return RequestField(name=key, data=value, headers={'Content-Type': 'application/octet-stream'}) request_field = RequestField(name=key, data=value)
request_field.make_multipart(content_type='application/octet-stream')
elif isinstance(value, FileIO): elif isinstance(value, FileIO):
request_field = RequestField( # TODO use content.encoding to limit allowed content types if they are present
name=key, request_field = RequestField.from_tuples(key, (os.path.basename(value.name), value.read()))
data=value.read(),
filename=os.path.basename(value.name),
headers={'Content-Type': 'application/octet-stream'}
)
value.close() value.close()
return request_field
else: else:
return self.__multipart_json_item(key=key, value=value) request_field = self.__multipart_json_item(key=key, value=value)
return request_field
def __serialize_multipart_form_data( def __serialize_multipart_form_data(
self, in_data: Schema self, in_data: Schema

View File

@ -1389,24 +1389,24 @@ class RequestBody(StyleFormSerializer, JSONDetector):
def __multipart_json_item(self, key: str, value: Schema) -> RequestField: def __multipart_json_item(self, key: str, value: Schema) -> RequestField:
json_value = self.__json_encoder.default(value) json_value = self.__json_encoder.default(value)
return RequestField(name=key, data=json.dumps(json_value), headers={'Content-Type': 'application/json'}) request_field = RequestField(name=key, data=json.dumps(json_value))
request_field.make_multipart(content_type='application/json')
return request_field
def __multipart_form_item(self, key: str, value: Schema) -> RequestField: def __multipart_form_item(self, key: str, value: Schema) -> RequestField:
if isinstance(value, str): if isinstance(value, str):
return RequestField(name=key, data=str(value), headers={'Content-Type': 'text/plain'}) request_field = RequestField(name=key, data=str(value))
request_field.make_multipart(content_type='text/plain')
elif isinstance(value, bytes): elif isinstance(value, bytes):
return RequestField(name=key, data=value, headers={'Content-Type': 'application/octet-stream'}) request_field = RequestField(name=key, data=value)
request_field.make_multipart(content_type='application/octet-stream')
elif isinstance(value, FileIO): elif isinstance(value, FileIO):
request_field = RequestField( # TODO use content.encoding to limit allowed content types if they are present
name=key, request_field = RequestField.from_tuples(key, (os.path.basename(value.name), value.read()))
data=value.read(),
filename=os.path.basename(value.name),
headers={'Content-Type': 'application/octet-stream'}
)
value.close() value.close()
return request_field
else: else:
return self.__multipart_json_item(key=key, value=value) request_field = self.__multipart_json_item(key=key, value=value)
return request_field
def __serialize_multipart_form_data( def __serialize_multipart_form_data(
self, in_data: Schema self, in_data: Schema

View File

@ -1389,24 +1389,24 @@ class RequestBody(StyleFormSerializer, JSONDetector):
def __multipart_json_item(self, key: str, value: Schema) -> RequestField: def __multipart_json_item(self, key: str, value: Schema) -> RequestField:
json_value = self.__json_encoder.default(value) json_value = self.__json_encoder.default(value)
return RequestField(name=key, data=json.dumps(json_value), headers={'Content-Type': 'application/json'}) request_field = RequestField(name=key, data=json.dumps(json_value))
request_field.make_multipart(content_type='application/json')
return request_field
def __multipart_form_item(self, key: str, value: Schema) -> RequestField: def __multipart_form_item(self, key: str, value: Schema) -> RequestField:
if isinstance(value, str): if isinstance(value, str):
return RequestField(name=key, data=str(value), headers={'Content-Type': 'text/plain'}) request_field = RequestField(name=key, data=str(value))
request_field.make_multipart(content_type='text/plain')
elif isinstance(value, bytes): elif isinstance(value, bytes):
return RequestField(name=key, data=value, headers={'Content-Type': 'application/octet-stream'}) request_field = RequestField(name=key, data=value)
request_field.make_multipart(content_type='application/octet-stream')
elif isinstance(value, FileIO): elif isinstance(value, FileIO):
request_field = RequestField( # TODO use content.encoding to limit allowed content types if they are present
name=key, request_field = RequestField.from_tuples(key, (os.path.basename(value.name), value.read()))
data=value.read(),
filename=os.path.basename(value.name),
headers={'Content-Type': 'application/octet-stream'}
)
value.close() value.close()
return request_field
else: else:
return self.__multipart_json_item(key=key, value=value) request_field = self.__multipart_json_item(key=key, value=value)
return request_field
def __serialize_multipart_form_data( def __serialize_multipart_form_data(
self, in_data: Schema self, in_data: Schema

View File

@ -1398,24 +1398,24 @@ class RequestBody(StyleFormSerializer, JSONDetector):
def __multipart_json_item(self, key: str, value: Schema) -> RequestField: def __multipart_json_item(self, key: str, value: Schema) -> RequestField:
json_value = self.__json_encoder.default(value) json_value = self.__json_encoder.default(value)
return RequestField(name=key, data=json.dumps(json_value), headers={'Content-Type': 'application/json'}) request_field = RequestField(name=key, data=json.dumps(json_value))
request_field.make_multipart(content_type='application/json')
return request_field
def __multipart_form_item(self, key: str, value: Schema) -> RequestField: def __multipart_form_item(self, key: str, value: Schema) -> RequestField:
if isinstance(value, str): if isinstance(value, str):
return RequestField(name=key, data=str(value), headers={'Content-Type': 'text/plain'}) request_field = RequestField(name=key, data=str(value))
request_field.make_multipart(content_type='text/plain')
elif isinstance(value, bytes): elif isinstance(value, bytes):
return RequestField(name=key, data=value, headers={'Content-Type': 'application/octet-stream'}) request_field = RequestField(name=key, data=value)
request_field.make_multipart(content_type='application/octet-stream')
elif isinstance(value, FileIO): elif isinstance(value, FileIO):
request_field = RequestField( # TODO use content.encoding to limit allowed content types if they are present
name=key, request_field = RequestField.from_tuples(key, (os.path.basename(value.name), value.read()))
data=value.read(),
filename=os.path.basename(value.name),
headers={'Content-Type': 'application/octet-stream'}
)
value.close() value.close()
return request_field
else: else:
return self.__multipart_json_item(key=key, value=value) request_field = self.__multipart_json_item(key=key, value=value)
return request_field
def __serialize_multipart_form_data( def __serialize_multipart_form_data(
self, in_data: Schema self, in_data: Schema

View File

@ -35,8 +35,7 @@ class MIMEFormdata(nonmultipart.MIMENonMultipart):
class TestFakeApi(ApiTestMixin): class TestFakeApi(ApiTestMixin):
"""FakeApi unit test stubs""" """FakeApi unit test stubs"""
configuration = petstore_api.Configuration() configuration = petstore_api.Configuration()
api_client = api_client.ApiClient(configuration=configuration) api = FakeApi(api_client=api_client.ApiClient(configuration=configuration))
api = FakeApi(api_client=api_client)
def test_array_model(self): def test_array_model(self):
from petstore_api.model import animal_farm, animal from petstore_api.model import animal_farm, animal
@ -469,7 +468,11 @@ class TestFakeApi(ApiTestMixin):
name='file', name='file',
data=file_bytes, data=file_bytes,
filename=file_name, filename=file_name,
headers={'Content-Type': 'application/octet-stream'} headers={
'Content-Location': None,
'Content-Type': 'image/png',
"Content-Disposition": "form-data; name=\"file\"; filename=\"1px_pic1.png\""
}
), ),
), ),
content_type='multipart/form-data' content_type='multipart/form-data'
@ -493,7 +496,11 @@ class TestFakeApi(ApiTestMixin):
api_client.RequestField( api_client.RequestField(
name='file', name='file',
data=file_bytes, data=file_bytes,
headers={'Content-Type': 'application/octet-stream'} headers={
'Content-Type': 'application/octet-stream',
"Content-Disposition": "form-data; name=\"file\"",
"Content-Location": None
}
), ),
), ),
content_type='multipart/form-data' content_type='multipart/form-data'
@ -548,13 +555,22 @@ class TestFakeApi(ApiTestMixin):
name='files', name='files',
data=file_bytes, data=file_bytes,
filename=file_name, filename=file_name,
headers={'Content-Type': 'application/octet-stream'} headers={
'Content-Type': 'image/png',
"Content-Disposition": "form-data; name=\"files\"; filename=\"1px_pic1.png\"",
"Content-Location": None
}
), ),
api_client.RequestField( api_client.RequestField(
name='files', name='files',
data=file_bytes, data=file_bytes,
filename=file_name, filename=file_name,
headers={'Content-Type': 'application/octet-stream'} headers={
'Content-Type': 'image/png',
"Content-Disposition": "form-data; name=\"files\"; filename=\"1px_pic1.png\"",
"Content-Location": None
}
), ),
), ),
content_type='multipart/form-data' content_type='multipart/form-data'
@ -579,12 +595,20 @@ class TestFakeApi(ApiTestMixin):
api_client.RequestField( api_client.RequestField(
name='files', name='files',
data=file_bytes, data=file_bytes,
headers={'Content-Type': 'application/octet-stream'} headers={
'Content-Type': 'application/octet-stream',
"Content-Disposition": "form-data; name=\"files\"",
"Content-Location": None
}
), ),
api_client.RequestField( api_client.RequestField(
name='files', name='files',
data=file_bytes, data=file_bytes,
headers={'Content-Type': 'application/octet-stream'} headers={
'Content-Type': 'application/octet-stream',
"Content-Disposition": "form-data; name=\"files\"",
"Content-Location": None
}
), ),
), ),
content_type='multipart/form-data' content_type='multipart/form-data'
@ -657,11 +681,15 @@ class TestFakeApi(ApiTestMixin):
accept_content_type=content_type, accept_content_type=content_type,
content_type=content_type, content_type=content_type,
fields=( fields=(
api_client.RequestField( api_client.RequestField(
name='someProp', name='someProp',
data=single_char_str, data=single_char_str,
headers={'Content-Type': 'text/plain'} headers={
), 'Content-Type': 'text/plain',
"Content-Disposition": "form-data; name=\"someProp\"",
"Content-Location": None
}
),
), ),
) )
self.assertEqual(api_response.body, {'someProp': single_char_str}) self.assertEqual(api_response.body, {'someProp': single_char_str})
@ -764,32 +792,6 @@ class TestFakeApi(ApiTestMixin):
assert isinstance(api_response.body, schemas.Unset) assert isinstance(api_response.body, schemas.Unset)
assert isinstance(api_response.headers, schemas.Unset) assert isinstance(api_response.headers, schemas.Unset)
def test_x_www_form_urlencoded(self):
with patch.object(urllib3.PoolManager, 'request') as mock_request:
from urllib3._collections import HTTPHeaderDict
from petstore_api.apis.tags import pet_api
pet_id = dict(petId=2345)
pet_values = dict(
name='mister furball award',
status='happy, fuzzy, and bouncy'
)
mock_request.return_value = self.response("")
api_instance = pet_api.PetApi(self.api_client)
api_instance.update_pet_with_form(path_params=pet_id, body=pet_values)
mock_request.assert_called_with(
'POST',
'http://petstore.swagger.io:80/v2/pet/2345',
body='name=mister%20furball%20award&status=happy%2C%20fuzzy%2C%20and%20bouncy',
fields={},
encode_multipart=False,
preload_content=True,
timeout=None,
headers=HTTPHeaderDict({'User-Agent': self.user_agent,
'Content-Type': 'application/x-www-form-urlencoded'})
)
def test_json_patch(self): def test_json_patch(self):
with patch.object(urllib3.PoolManager, 'request') as mock_request: with patch.object(urllib3.PoolManager, 'request') as mock_request:
from petstore_api.model import json_patch_request from petstore_api.model import json_patch_request

View File

@ -0,0 +1,60 @@
# coding: utf-8
"""
Generated by: https://openapi-generator.tech
"""
import unittest
from unittest.mock import patch
import urllib3
import petstore_api
from petstore_api.paths.pet_pet_id import post
from petstore_api import configuration, schemas, api_client
from ... import ApiTestMixin
class TestPetPetId(ApiTestMixin, unittest.TestCase):
"""
PetPetId unit test stubs
Updates a pet in the store with form data # noqa: E501
"""
def test_post(self):
used_api_client = api_client.ApiClient()
api = post.ApiForpost(api_client=used_api_client)
with patch.object(urllib3.PoolManager, 'request') as mock_request:
path_params = {'petId': 2345}
body = {
'name': 'mister furball award',
'status': 'happy, fuzzy, and bouncy'
}
mock_request.return_value = self.response("")
api_response = api.post(path_params=path_params, body=body)
mock_request.assert_called_with(
'POST',
'http://petstore.swagger.io:80/v2/pet/2345',
body='name=mister%20furball%20award&status=happy%2C%20fuzzy%2C%20and%20bouncy',
fields={},
encode_multipart=False,
preload_content=True,
timeout=None,
headers={
'User-Agent': self.user_agent,
'Content-Type': 'application/x-www-form-urlencoded'
}
)
assert isinstance(api_response.response, urllib3.HTTPResponse)
assert isinstance(api_response.body, schemas.Unset)
assert isinstance(api_response.headers, schemas.Unset)
assert api_response.response.status == 200
if __name__ == '__main__':
unittest.main()

View File

@ -71,21 +71,53 @@ class TestParameter(unittest.TestCase):
dict( dict(
fields=( fields=(
api_client.RequestField( api_client.RequestField(
name='some_null', data='null', headers={'Content-Type': 'application/json'}), name='some_null', data='null', headers={
'Content-Type': 'application/json',
"Content-Disposition": "form-data; name=\"some_null\"",
"Content-Location": None
}),
api_client.RequestField( api_client.RequestField(
name='some_bool', data='true', headers={'Content-Type': 'application/json'}), name='some_bool', data='true', headers={
'Content-Type': 'application/json',
"Content-Disposition": "form-data; name=\"some_bool\"",
"Content-Location": None
}),
api_client.RequestField( api_client.RequestField(
name='some_str', data='a', headers={'Content-Type': 'text/plain'}), name='some_str', data='a', headers={
'Content-Type': 'text/plain',
"Content-Disposition": "form-data; name=\"some_str\"",
"Content-Location": None
}),
api_client.RequestField( api_client.RequestField(
name='some_int', data='1', headers={'Content-Type': 'application/json'}), name='some_int', data='1', headers={
'Content-Type': 'application/json',
"Content-Disposition": "form-data; name=\"some_int\"",
"Content-Location": None
}),
api_client.RequestField( api_client.RequestField(
name='some_float', data='3.14', headers={'Content-Type': 'application/json'}), name='some_float', data='3.14', headers={
'Content-Type': 'application/json',
"Content-Disposition": "form-data; name=\"some_float\"",
"Content-Location": None
}),
api_client.RequestField( api_client.RequestField(
name='some_list', data='[]', headers={'Content-Type': 'application/json'}), name='some_list', data='[]', headers={
'Content-Type': 'application/json',
"Content-Disposition": "form-data; name=\"some_list\"",
"Content-Location": None
}),
api_client.RequestField( api_client.RequestField(
name='some_dict', data='{}', headers={'Content-Type': 'application/json'}), name='some_dict', data='{}', headers={
'Content-Type': 'application/json',
"Content-Disposition": "form-data; name=\"some_dict\"",
"Content-Location": None
}),
api_client.RequestField( api_client.RequestField(
name='some_bytes', data=b'abc', headers={'Content-Type': 'application/octet-stream'}) name='some_bytes', data=b'abc', headers={
'Content-Type': 'application/octet-stream',
"Content-Disposition": "form-data; name=\"some_bytes\"",
"Content-Location": None
})
) )
) )
) )
@ -110,7 +142,7 @@ class TestParameter(unittest.TestCase):
def test_application_x_www_form_urlencoded_serialization(self): def test_application_x_www_form_urlencoded_serialization(self):
payload = dict( payload = dict(
some_null=None, some_null=None,
some_str='hi, spacether!', some_str='hi there',
some_int=1, some_int=1,
some_float=3.14, some_float=3.14,
some_list=[], some_list=[],
@ -123,7 +155,7 @@ class TestParameter(unittest.TestCase):
serialization = request_body.serialize(payload, content_type) serialization = request_body.serialize(payload, content_type)
self.assertEqual( self.assertEqual(
serialization, serialization,
dict(body='some_str=hi%2C%20spacether%21&some_int=1&some_float=3.14') dict(body='some_str=hi%20there&some_int=1&some_float=3.14')
) )
serialization = request_body.serialize({}, content_type) serialization = request_body.serialize({}, content_type)