Fixes double serialization bug in python-experimental (#12561)

* Partial template update, simplifies ref6570 serialization

* Updated serialization

* Typo fixes

* Fixes test_query_style_form_serialization

* Fixes test_cookie_style_form_serialization

* FIxes test_style_simple_in_path_serialization

* Fixes test_style_simple_in_header_serialization

* Fixes many parameter tests

* Deletes duplicate method

* Fixes TestParameter.test_path_params_no_style

* Fixes test_style_pipe_delimited_serialization

* Fixes test_header_params_no_style

* Fixes test_style_space_delimited_serialization

* Fixes test_query_style_form_serialization

* Fixes test_cookie_style_form_serialization

* Fixes test_query_params_no_style

* Fixes test_cookie_params_no_style

* Fixes test_delete_endpoint_without_request_body

* Fixes all tests

* Reverts version file

* Adds test_get_user_by_name to test path param url encoding

* Adds test_body_with_query_params

* Samples regenerated
This commit is contained in:
Justin Black 2022-06-13 07:45:15 -07:00 committed by GitHub
parent 3190084134
commit a00ae4631d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
66 changed files with 1374 additions and 1077 deletions

View File

@ -49,9 +49,12 @@ class RequestField(RequestFieldBase):
class JSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, (str, int, float)):
# instances based on primitive classes
return obj
if isinstance(obj, str):
return str(obj)
elif isinstance(obj, float):
return float(obj)
elif isinstance(obj, int):
return int(obj)
elif isinstance(obj, Decimal):
if obj.as_tuple().exponent >= 0:
return int(obj)
@ -84,180 +87,175 @@ class ParameterStyle(enum.Enum):
DEEP_OBJECT = 'deepObject'
class PrefixSeparatorIterator:
# A class to store prefixes and separators for rfc6570 expansions
def __init__(self, prefix: str, separator: str):
self.prefix = prefix
self.separator = separator
self.first = True
if separator in {'.', '|', '%20'}:
item_separator = separator
else:
item_separator = ','
self.item_separator = item_separator
def __iter__(self):
return self
def __next__(self):
if self.first:
self.first = False
return self.prefix
return self.separator
class ParameterSerializerBase:
@staticmethod
def __serialize_number(
in_data: typing.Union[int, float], name: str, prefix=''
) -> typing.Tuple[typing.Tuple[str, str]]:
return tuple([(name, prefix + str(in_data))])
@classmethod
def get_default_explode(cls, style: ParameterStyle) -> bool:
return False
@staticmethod
def __serialize_str(
in_data: str, name: str, prefix=''
) -> typing.Tuple[typing.Tuple[str, str]]:
return tuple([(name, prefix + quote(in_data))])
@staticmethod
def __serialize_bool(in_data: bool, name: str, prefix='') -> typing.Tuple[typing.Tuple[str, str]]:
if in_data:
return tuple([(name, prefix + 'true')])
return tuple([(name, prefix + 'false')])
@staticmethod
def __urlencode(in_data: typing.Any) -> str:
return quote(str(in_data))
def __serialize_list(
self,
in_data: typing.List[typing.Any],
style: ParameterStyle,
name: str,
explode: bool,
empty_val: typing.Union[typing.Tuple[str, str], typing.Tuple] = tuple(),
prefix: str = '',
separator: str = ',',
) -> typing.Tuple[typing.Union[typing.Tuple[str, str], typing.Tuple], ...]:
if not in_data:
return empty_val
if explode and style in {
ParameterStyle.FORM,
ParameterStyle.MATRIX,
ParameterStyle.SPACE_DELIMITED,
ParameterStyle.PIPE_DELIMITED
}:
if style is ParameterStyle.FORM:
return tuple((name, prefix + self.__urlencode(val)) for val in in_data)
else:
joined_vals = prefix + separator.join(name + '=' + self.__urlencode(val) for val in in_data)
else:
joined_vals = prefix + separator.join(map(self.__urlencode, in_data))
return tuple([(name, joined_vals)])
def __form_item_representation(self, in_data: typing.Any) -> typing.Optional[str]:
if isinstance(in_data, none_type):
def __ref6570_item_value(in_data: typing.Any, percent_encode: bool):
"""
Get representation if str/float/int/None/items in list/ values in dict
None is returned if an item is undefined, use cases are value=
- None
- []
- {}
- [None, None None]
- {'a': None, 'b': None}
"""
if type(in_data) in {str, float, int}:
if percent_encode:
return quote(str(in_data))
return str(in_data)
elif isinstance(in_data, none_type):
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return None
elif isinstance(in_data, list):
if not in_data:
return None
raise ApiValueError('Unable to generate a form representation of {}'.format(in_data))
elif isinstance(in_data, dict):
if not in_data:
return None
raise ApiValueError('Unable to generate a form representation of {}'.format(in_data))
elif isinstance(in_data, (bool, bytes)):
raise ApiValueError('Unable to generate a form representation of {}'.format(in_data))
# str, float, int
return self.__urlencode(in_data)
elif isinstance(in_data, list) and not in_data:
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return None
elif isinstance(in_data, dict) and not in_data:
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return None
raise ApiValueError('Unable to generate a ref6570 item representation of {}'.format(in_data))
def __serialize_dict(
self,
in_data: typing.Dict[str, typing.Any],
style: ParameterStyle,
name: str,
@staticmethod
def to_dict(name: str, value: str):
return {name: value}
@classmethod
def ref6570_expansion(
cls,
variable_name: str,
in_data: typing.Any,
explode: bool,
empty_val: typing.Union[typing.Tuple[str, str], typing.Tuple] = tuple(),
prefix: str = '',
separator: str = ',',
) -> typing.Tuple[typing.Tuple[str, str]]:
if not in_data:
return empty_val
if all(val is None for val in in_data.values()):
return empty_val
form_items = {}
if style is ParameterStyle.FORM:
for key, val in in_data.items():
new_val = self.__form_item_representation(val)
if new_val is None:
continue
form_items[key] = new_val
if explode:
if style is ParameterStyle.FORM:
return tuple((key, prefix + val) for key, val in form_items.items())
elif style in {
ParameterStyle.SIMPLE,
ParameterStyle.LABEL,
ParameterStyle.MATRIX,
ParameterStyle.SPACE_DELIMITED,
ParameterStyle.PIPE_DELIMITED
}:
joined_vals = prefix + separator.join(key + '=' + self.__urlencode(val) for key, val in in_data.items())
else:
raise ApiValueError(f'Invalid style {style} for dict serialization with explode=True')
elif style is ParameterStyle.FORM:
joined_vals = prefix + separator.join(key + separator + val for key, val in form_items.items())
else:
joined_vals = prefix + separator.join(
key + separator + self.__urlencode(val) for key, val in in_data.items())
return tuple([(name, joined_vals)])
def _serialize_x(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
style: ParameterStyle,
name: str,
explode: bool,
empty_val: typing.Union[typing.Tuple[str, str], typing.Tuple] = (),
prefix: str = '',
separator: str = ',',
) -> typing.Tuple[typing.Tuple[str, str], ...]:
if isinstance(in_data, none_type):
return empty_val
elif isinstance(in_data, bool):
# must be before int check
return self.__serialize_bool(in_data, name=name, prefix=prefix)
elif isinstance(in_data, (int, float)):
return self.__serialize_number(in_data, name=name, prefix=prefix)
elif isinstance(in_data, str):
return self.__serialize_str(in_data, name=name, prefix=prefix)
percent_encode: bool,
prefix_separator_iterator: PrefixSeparatorIterator
) -> str:
"""
Separator is for separate variables like dict with explode true, not for array item separation
"""
named_parameter_expansion = prefix_separator_iterator.separator in {'&', ';'}
var_name_piece = variable_name if named_parameter_expansion else ''
if type(in_data) in {str, float, int}:
item_value = cls.__ref6570_item_value(in_data, percent_encode)
if item_value is None:
return next(prefix_separator_iterator) + var_name_piece
elif item_value == '' and prefix_separator_iterator.separator == ';':
return next(prefix_separator_iterator) + var_name_piece
value_pair_equals = '=' if named_parameter_expansion else ''
return next(prefix_separator_iterator) + var_name_piece + value_pair_equals + item_value
elif isinstance(in_data, none_type):
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return ""
elif isinstance(in_data, list):
return self.__serialize_list(
in_data,
style=style,
name=name,
explode=explode,
empty_val=empty_val,
prefix=prefix,
separator=separator
item_values = [cls.__ref6570_item_value(v, percent_encode) for v in in_data]
item_values = [v for v in item_values if v is not None]
if not item_values:
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return ""
value_pair_equals = '=' if named_parameter_expansion else ''
if not explode:
return (
next(prefix_separator_iterator) +
var_name_piece +
value_pair_equals +
prefix_separator_iterator.item_separator.join(item_values)
)
# exploded
return next(prefix_separator_iterator) + next(prefix_separator_iterator).join(
[var_name_piece + value_pair_equals + val for val in item_values]
)
elif isinstance(in_data, dict):
return self.__serialize_dict(
in_data,
style=style,
name=name,
explode=explode,
empty_val=empty_val,
prefix=prefix,
separator=separator
in_data_transformed = {key: cls.__ref6570_item_value(val, percent_encode) for key, val in in_data.items()}
in_data_transformed = {key: val for key, val in in_data_transformed.items() if val is not None}
if not in_data_transformed:
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return ""
value_pair_equals = '=' if named_parameter_expansion else ''
if not explode:
return (
next(prefix_separator_iterator) +
var_name_piece + value_pair_equals +
prefix_separator_iterator.item_separator.join(
prefix_separator_iterator.item_separator.join(
item_pair
) for item_pair in in_data_transformed.items()
)
)
# exploded
return next(prefix_separator_iterator) + next(prefix_separator_iterator).join(
[key + '=' + val for key, val in in_data_transformed.items()]
)
# bool, bytes, etc
raise ApiValueError('Unable to generate a ref6570 representation of {}'.format(in_data))
class StyleFormSerializer(ParameterSerializerBase):
@classmethod
def get_default_explode(cls, style: ParameterStyle) -> bool:
if style is ParameterStyle.FORM:
return True
return super().get_default_explode(style)
def _serialize_form(
def serialize_form(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
name: str,
explode: bool,
) -> typing.Tuple[typing.Tuple[str, str], ...]:
return self._serialize_x(in_data, style=ParameterStyle.FORM, name=name, explode=explode)
percent_encode: bool,
prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator] = None
) -> str:
if prefix_separator_iterator is None:
prefix_separator_iterator = PrefixSeparatorIterator('?', '&')
return self.ref6570_expansion(
variable_name=name,
in_data=in_data,
explode=explode,
percent_encode=percent_encode,
prefix_separator_iterator=prefix_separator_iterator
)
class StyleSimpleSerializer(ParameterSerializerBase):
def _serialize_simple_tuple(
def serialize_simple(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
name: str,
explode: bool,
in_type: ParameterInType,
) -> typing.Tuple[typing.Tuple[str, str], ...]:
if in_type is ParameterInType.HEADER:
empty_val = ()
else:
empty_val = ((name, ''),)
return self._serialize_x(in_data, style=ParameterStyle.SIMPLE, name=name, explode=explode, empty_val=empty_val)
percent_encode: bool
) -> str:
prefix_separator_iterator = PrefixSeparatorIterator('', ',')
return self.ref6570_expansion(
variable_name=name,
in_data=in_data,
explode=explode,
percent_encode=percent_encode,
prefix_separator_iterator=prefix_separator_iterator
)
@dataclass
@ -345,8 +343,8 @@ class ParameterBase:
def _serialize_json(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list]
) -> typing.Tuple[typing.Tuple[str, str]]:
return tuple([(self.name, json.dumps(in_data))])
) -> str:
return json.dumps(in_data)
class PathParameter(ParameterBase, StyleSimpleSerializer):
@ -372,56 +370,45 @@ class PathParameter(ParameterBase, StyleSimpleSerializer):
content=content
)
def __serialize_label(
def _serialize_label(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list]
) -> typing.Dict[str, str]:
empty_val = ((self.name, ''),)
prefix = '.'
separator = '.'
return self._remove_empty_and_cast(
self._serialize_x(
in_data,
style=ParameterStyle.LABEL,
name=self.name,
explode=self.explode,
empty_val=empty_val,
prefix=prefix,
separator=separator
)
prefix_separator_iterator = PrefixSeparatorIterator('.', '.')
value = self.ref6570_expansion(
variable_name=self.name,
in_data=in_data,
explode=self.explode,
percent_encode=True,
prefix_separator_iterator=prefix_separator_iterator
)
return self.to_dict(self.name, value)
def __serialize_matrix(
def _serialize_matrix(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list]
) -> typing.Dict[str, str]:
separator = ','
if in_data == '':
prefix = ';' + self.name
elif isinstance(in_data, (dict, list)) and self.explode:
prefix = ';'
separator = ';'
else:
prefix = ';' + self.name + '='
empty_val = ((self.name, ''),)
return self._remove_empty_and_cast(
self._serialize_x(
in_data,
style=ParameterStyle.MATRIX,
name=self.name,
explode=self.explode,
prefix=prefix,
empty_val=empty_val,
separator=separator
)
prefix_separator_iterator = PrefixSeparatorIterator(';', ';')
value = self.ref6570_expansion(
variable_name=self.name,
in_data=in_data,
explode=self.explode,
percent_encode=True,
prefix_separator_iterator=prefix_separator_iterator
)
return self.to_dict(self.name, value)
def _serialize_simple(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
) -> typing.Dict[str, str]:
tuple_data = self._serialize_simple_tuple(in_data, self.name, self.explode, self.in_type)
return self._remove_empty_and_cast(tuple_data)
value = self.serialize_simple(
in_data=in_data,
name=self.name,
explode=self.explode,
percent_encode=True
)
return self.to_dict(self.name, value)
def serialize(
self,
@ -444,16 +431,16 @@ class PathParameter(ParameterBase, StyleSimpleSerializer):
if self.style is ParameterStyle.SIMPLE:
return self._serialize_simple(cast_in_data)
elif self.style is ParameterStyle.LABEL:
return self.__serialize_label(cast_in_data)
return self._serialize_label(cast_in_data)
elif self.style is ParameterStyle.MATRIX:
return self.__serialize_matrix(cast_in_data)
return self._serialize_matrix(cast_in_data)
# self.content will be length one
for content_type, schema in self.content.items():
cast_in_data = schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
if content_type == self._json_content_type:
tuple_data = self._serialize_json(cast_in_data)
return self._remove_empty_and_cast(tuple_data)
value = self._serialize_json(cast_in_data)
return self.to_dict(self.name, value)
raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))
@ -464,17 +451,20 @@ class QueryParameter(ParameterBase, StyleFormSerializer):
name: str,
required: bool = False,
style: typing.Optional[ParameterStyle] = None,
explode: bool = False,
explode: typing.Optional[bool] = None,
allow_reserved: typing.Optional[bool] = None,
schema: typing.Optional[typing.Type[Schema]] = None,
content: typing.Optional[typing.Dict[str, typing.Type[Schema]]] = None
):
used_style = ParameterStyle.FORM if style is None and content is None and schema else style
used_explode = self.get_default_explode(used_style) if explode is None else explode
super().__init__(
name,
in_type=ParameterInType.QUERY,
required=required,
style=style,
explode=explode,
style=used_style,
explode=used_explode,
allow_reserved=allow_reserved,
schema=schema,
content=content
@ -482,39 +472,68 @@ class QueryParameter(ParameterBase, StyleFormSerializer):
def __serialize_space_delimited(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list]
) -> typing.Tuple[typing.Tuple[str, str], ...]:
separator = '%20'
empty_val = ()
return self._serialize_x(
in_data,
style=ParameterStyle.SPACE_DELIMITED,
name=self.name,
in_data: typing.Union[None, int, float, str, bool, dict, list],
prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator]
) -> typing.Dict[str, str]:
if prefix_separator_iterator is None:
prefix_separator_iterator = self.get_prefix_separator_iterator()
value = self.ref6570_expansion(
variable_name=self.name,
in_data=in_data,
explode=self.explode,
separator=separator,
empty_val=empty_val
percent_encode=True,
prefix_separator_iterator=prefix_separator_iterator
)
return self.to_dict(self.name, value)
def __serialize_pipe_delimited(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list]
) -> typing.Tuple[typing.Tuple[str, str], ...]:
separator = '|'
empty_val = ()
return self._serialize_x(
in_data: typing.Union[None, int, float, str, bool, dict, list],
prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator]
) -> typing.Dict[str, str]:
if prefix_separator_iterator is None:
prefix_separator_iterator = self.get_prefix_separator_iterator()
value = self.ref6570_expansion(
variable_name=self.name,
in_data=in_data,
explode=self.explode,
percent_encode=True,
prefix_separator_iterator=prefix_separator_iterator
)
return self.to_dict(self.name, value)
def __serialize_form(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator]
) -> typing.Dict[str, str]:
if prefix_separator_iterator is None:
prefix_separator_iterator = self.get_prefix_separator_iterator()
value = self.serialize_form(
in_data,
style=ParameterStyle.PIPE_DELIMITED,
name=self.name,
explode=self.explode,
separator=separator,
empty_val=empty_val
percent_encode=True,
prefix_separator_iterator=prefix_separator_iterator
)
return self.to_dict(self.name, value)
def get_prefix_separator_iterator(self) -> typing.Optional[PrefixSeparatorIterator]:
if not self.schema:
return None
if self.style is ParameterStyle.FORM:
return PrefixSeparatorIterator('?', '&')
elif self.style is ParameterStyle.SPACE_DELIMITED:
return PrefixSeparatorIterator('', '%20')
elif self.style is ParameterStyle.PIPE_DELIMITED:
return PrefixSeparatorIterator('', '|')
def serialize(
self,
in_data: typing.Union[
Schema, Decimal, int, float, str, date, datetime, None, bool, list, tuple, dict, frozendict]
) -> typing.Tuple[typing.Tuple[str, str]]:
Schema, Decimal, int, float, str, date, datetime, None, bool, list, tuple, dict, frozendict],
prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator] = None
) -> typing.Dict[str, str]:
if self.schema:
cast_in_data = self.schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
@ -534,17 +553,18 @@ class QueryParameter(ParameterBase, StyleFormSerializer):
if self.style:
# TODO update query ones to omit setting values when [] {} or None is input
if self.style is ParameterStyle.FORM:
return self._serialize_form(cast_in_data, explode=self.explode, name=self.name)
return self.__serialize_form(cast_in_data, prefix_separator_iterator)
elif self.style is ParameterStyle.SPACE_DELIMITED:
return self.__serialize_space_delimited(cast_in_data)
return self.__serialize_space_delimited(cast_in_data, prefix_separator_iterator)
elif self.style is ParameterStyle.PIPE_DELIMITED:
return self.__serialize_pipe_delimited(cast_in_data)
return self.__serialize_pipe_delimited(cast_in_data, prefix_separator_iterator)
# self.content will be length one
for content_type, schema in self.content.items():
cast_in_data = schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
if content_type == self._json_content_type:
return self._serialize_json(cast_in_data)
value = self._serialize_json(cast_in_data)
return self.to_dict(self.name, value)
raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))
@ -555,17 +575,20 @@ class CookieParameter(ParameterBase, StyleFormSerializer):
name: str,
required: bool = False,
style: typing.Optional[ParameterStyle] = None,
explode: bool = False,
explode: typing.Optional[bool] = None,
allow_reserved: typing.Optional[bool] = None,
schema: typing.Optional[typing.Type[Schema]] = None,
content: typing.Optional[typing.Dict[str, typing.Type[Schema]]] = None
):
used_style = ParameterStyle.FORM if style is None and content is None and schema else style
used_explode = self.get_default_explode(used_style) if explode is None else explode
super().__init__(
name,
in_type=ParameterInType.COOKIE,
required=required,
style=style,
explode=explode,
style=used_style,
explode=used_explode,
allow_reserved=allow_reserved,
schema=schema,
content=content
@ -575,7 +598,7 @@ class CookieParameter(ParameterBase, StyleFormSerializer):
self,
in_data: typing.Union[
Schema, Decimal, int, float, str, date, datetime, None, bool, list, tuple, dict, frozendict]
) -> typing.Tuple[typing.Tuple[str, str]]:
) -> typing.Dict[str, str]:
if self.schema:
cast_in_data = self.schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
@ -584,13 +607,25 @@ class CookieParameter(ParameterBase, StyleFormSerializer):
returns fields: tuple
"""
if self.style:
return self._serialize_form(cast_in_data, explode=self.explode, name=self.name)
"""
TODO add escaping of comma, space, equals
or turn encoding on
"""
value = self.serialize_form(
cast_in_data,
explode=self.explode,
name=self.name,
percent_encode=False,
prefix_separator_iterator=PrefixSeparatorIterator('', '&')
)
return self.to_dict(self.name, value)
# self.content will be length one
for content_type, schema in self.content.items():
cast_in_data = schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
if content_type == self._json_content_type:
return self._serialize_json(cast_in_data)
value = self._serialize_json(cast_in_data)
return self.to_dict(self.name, value)
raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))
@ -628,9 +663,8 @@ class HeaderParameter(ParameterBase, StyleSimpleSerializer):
def _serialize_simple(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
) -> HTTPHeaderDict[str, str]:
tuple_data = self._serialize_simple_tuple(in_data, self.name, self.explode, self.in_type)
return self.__to_headers(tuple_data)
) -> str:
return self.serialize_simple(in_data, self.name, self.explode, False)
def serialize(
self,
@ -646,14 +680,15 @@ class HeaderParameter(ParameterBase, StyleSimpleSerializer):
returns headers: dict
"""
if self.style:
return self._serialize_simple(cast_in_data)
value = self._serialize_simple(cast_in_data)
return self.__to_headers(((self.name, value),))
# self.content will be length one
for content_type, schema in self.content.items():
cast_in_data = schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
if content_type == self._json_content_type:
tuple_data = self._serialize_json(cast_in_data)
return self.__to_headers(tuple_data)
value = self._serialize_json(cast_in_data)
return self.__to_headers(((self.name, value),))
raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))
@ -934,8 +969,6 @@ class ApiClient:
self,
resource_path: str,
method: str,
path_params: typing.Optional[typing.Dict[str, typing.Any]] = None,
query_params: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
headers: typing.Optional[HTTPHeaderDict] = None,
body: typing.Optional[typing.Union[str, bytes]] = None,
fields: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
@ -951,17 +984,8 @@ class ApiClient:
if self.cookie:
headers['Cookie'] = self.cookie
# path parameters
if path_params:
for k, v in path_params.items():
# specified safe chars, encode everything
resource_path = resource_path.replace(
'{%s}' % k,
quote(str(v), safe=self.configuration.safe_chars_for_path_param)
)
# auth setting
self.update_params_for_auth(headers, query_params,
self.update_params_for_auth(headers,
auth_settings, resource_path, method, body)
# request url
@ -975,7 +999,6 @@ class ApiClient:
response = {{#if asyncio}}await {{/if}}{{#if tornado}}yield {{/if}}self.request(
method,
url,
query_params=query_params,
headers=headers,
fields=fields,
body=body,
@ -988,8 +1011,6 @@ class ApiClient:
self,
resource_path: str,
method: str,
path_params: typing.Optional[typing.Dict[str, typing.Any]] = None,
query_params: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
headers: typing.Optional[HTTPHeaderDict] = None,
body: typing.Optional[typing.Union[str, bytes]] = None,
fields: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
@ -1005,8 +1026,6 @@ class ApiClient:
:param resource_path: Path to method endpoint.
:param method: Method to call.
:param path_params: Path parameters in the url.
:param query_params: Query parameters in the url.
:param headers: Header parameters to be
placed in the request header.
:param body: Request body.
@ -1039,8 +1058,6 @@ class ApiClient:
return self.__call_api(
resource_path,
method,
path_params,
query_params,
headers,
body,
fields,
@ -1055,8 +1072,6 @@ class ApiClient:
(
resource_path,
method,
path_params,
query_params,
headers,
body,
json,
@ -1072,7 +1087,6 @@ class ApiClient:
self,
method: str,
url: str,
query_params: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
headers: typing.Optional[HTTPHeaderDict] = None,
fields: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
body: typing.Optional[typing.Union[str, bytes]] = None,
@ -1082,19 +1096,16 @@ class ApiClient:
"""Makes the HTTP request using RESTClient."""
if method == "GET":
return self.rest_client.GET(url,
query_params=query_params,
stream=stream,
timeout=timeout,
headers=headers)
elif method == "HEAD":
return self.rest_client.HEAD(url,
query_params=query_params,
stream=stream,
timeout=timeout,
headers=headers)
elif method == "OPTIONS":
return self.rest_client.OPTIONS(url,
query_params=query_params,
headers=headers,
fields=fields,
stream=stream,
@ -1102,7 +1113,6 @@ class ApiClient:
body=body)
elif method == "POST":
return self.rest_client.POST(url,
query_params=query_params,
headers=headers,
fields=fields,
stream=stream,
@ -1110,7 +1120,6 @@ class ApiClient:
body=body)
elif method == "PUT":
return self.rest_client.PUT(url,
query_params=query_params,
headers=headers,
fields=fields,
stream=stream,
@ -1118,7 +1127,6 @@ class ApiClient:
body=body)
elif method == "PATCH":
return self.rest_client.PATCH(url,
query_params=query_params,
headers=headers,
fields=fields,
stream=stream,
@ -1126,7 +1134,6 @@ class ApiClient:
body=body)
elif method == "DELETE":
return self.rest_client.DELETE(url,
query_params=query_params,
headers=headers,
stream=stream,
timeout=timeout,
@ -1137,12 +1144,11 @@ class ApiClient:
" `POST`, `PATCH`, `PUT` or `DELETE`."
)
def update_params_for_auth(self, headers, querys, auth_settings,
def update_params_for_auth(self, headers, auth_settings,
resource_path, method, body):
"""Updates header and query params based on authentication setting.
:param headers: Header parameters dict to be updated.
:param querys: Query parameters tuple list to be updated.
:param auth_settings: Authentication setting identifiers list.
:param resource_path: A string representation of the HTTP request resource path.
:param method: A string representation of the HTTP request method.
@ -1165,13 +1171,18 @@ class ApiClient:
# The HTTP signature scheme requires multiple HTTP headers
# that are calculated dynamically.
signing_info = self.configuration.signing_info
querys = tuple()
auth_headers = signing_info.get_http_signature_headers(
resource_path, method, headers, body, querys)
for key, value in auth_headers.items():
headers.add(key, value)
{{/if}}
elif auth_setting['in'] == 'query':
querys.append((auth_setting['key'], auth_setting['value']))
""" TODO implement auth in query
need to pass in prefix_separator_iterator
and need to output resource_path with query params added
"""
raise ApiValueError("Auth in query not yet implemented")
else:
raise ApiValueError(
'Authentication token must be in `query` or `header`'
@ -1374,15 +1385,16 @@ class RequestBody(StyleFormSerializer, JSONDetector):
def __serialize_application_x_www_form_data(
self, in_data: typing.Any
) -> typing.Dict[str, tuple[tuple[str, str], ...]]:
) -> SerializedRequestBody:
"""
POST submission of form data in body
"""
if not isinstance(in_data, frozendict):
raise ValueError(
f'Unable to serialize {in_data} to application/x-www-form-urlencoded because it is not a dict of data')
cast_in_data = self.__json_encoder.default(in_data)
fields = self._serialize_form(cast_in_data, explode=True, name='')
if not fields:
return {}
return {'fields': fields}
value = self.serialize_form(cast_in_data, name='', explode=True, percent_encode=False)
return dict(body=value)
def serialize(
self, in_data: typing.Any, content_type: str

View File

@ -436,6 +436,7 @@ class {{operationIdCamelCase}}(api_client.Api):
{{#if cookieParams}}
self._verify_typed_dict_inputs(RequestCookieParams, cookie_params)
{{/if}}
used_path = _path
{{#if pathParams}}
_path_params = {}
@ -449,10 +450,13 @@ class {{operationIdCamelCase}}(api_client.Api):
continue
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
{{/if}}
{{#if queryParams}}
_query_params = []
prefix_separator_iterator = None
for parameter in (
{{#each queryParams}}
request_query_{{paramName}},
@ -461,8 +465,11 @@ class {{operationIdCamelCase}}(api_client.Api):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
{{/if}}
{{#or headerParams bodyParam produces}}
@ -509,14 +516,8 @@ class {{operationIdCamelCase}}(api_client.Api):
{{/if}}
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
{{#if pathParams}}
path_params=_path_params,
{{/if}}
{{#if queryParams}}
query_params=tuple(_query_params),
{{/if}}
{{#or headerParams bodyParam produces}}
headers=_headers,
{{/or}}

View File

@ -83,7 +83,6 @@ class RESTClientObject(object):
self,
method: str,
url: str,
query_params: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
headers: typing.Optional[HTTPHeaderDict] = None,
fields: typing.Optional[typing.Tuple[typing.Tuple[str, typing.Any], ...]] = None,
body: typing.Optional[typing.Union[str, bytes]] = None,
@ -94,7 +93,6 @@ class RESTClientObject(object):
:param method: http request method
:param url: http request url
:param query_params: query parameters in the url
:param headers: http request headers
:param body: request body, for other types
:param fields: request parameters for
@ -130,13 +128,10 @@ class RESTClientObject(object):
try:
# For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:
if query_params:
url += '?' + urlencode(query_params)
if 'Content-Type' not in headers and body is None:
r = self.pool_manager.request(
method,
url,
fields=query_params,
preload_content=not stream,
timeout=timeout,
headers=headers
@ -181,7 +176,6 @@ class RESTClientObject(object):
# For `GET`, `HEAD`
else:
r = self.pool_manager.request(method, url,
fields=query_params,
preload_content=not stream,
timeout=timeout,
headers=headers)
@ -195,63 +189,58 @@ class RESTClientObject(object):
return r
def GET(self, url, headers=None, query_params=None, stream=False,
def GET(self, url, headers=None, stream=False,
timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("GET", url,
headers=headers,
stream=stream,
timeout=timeout,
query_params=query_params, fields=fields)
fields=fields)
def HEAD(self, url, headers=None, query_params=None, stream=False,
def HEAD(self, url, headers=None, stream=False,
timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("HEAD", url,
headers=headers,
stream=stream,
timeout=timeout,
query_params=query_params, fields=fields)
fields=fields)
def OPTIONS(self, url, headers=None, query_params=None,
def OPTIONS(self, url, headers=None,
body=None, stream=False, timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("OPTIONS", url,
headers=headers,
query_params=query_params,
stream=stream,
timeout=timeout,
body=body, fields=fields)
def DELETE(self, url, headers=None, query_params=None, body=None,
def DELETE(self, url, headers=None, body=None,
stream=False, timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("DELETE", url,
headers=headers,
query_params=query_params,
stream=stream,
timeout=timeout,
body=body, fields=fields)
def POST(self, url, headers=None, query_params=None,
def POST(self, url, headers=None,
body=None, stream=False, timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("POST", url,
headers=headers,
query_params=query_params,
stream=stream,
timeout=timeout,
body=body, fields=fields)
def PUT(self, url, headers=None, query_params=None,
def PUT(self, url, headers=None,
body=None, stream=False, timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("PUT", url,
headers=headers,
query_params=query_params,
stream=stream,
timeout=timeout,
body=body, fields=fields)
def PATCH(self, url, headers=None, query_params=None,
def PATCH(self, url, headers=None,
body=None, stream=False, timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("PATCH", url,
headers=headers,
query_params=query_params,
stream=stream,
timeout=timeout,
body=body, fields=fields)

View File

@ -126,6 +126,7 @@ class Call123TestSpecialTags(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -145,7 +146,7 @@ class Call123TestSpecialTags(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -137,6 +137,7 @@ class FooGet(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -145,7 +146,7 @@ class FooGet(api_client.Api):
_headers.add('Accept', accept_content_type)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
stream=stream,

View File

@ -125,6 +125,7 @@ class AdditionalPropertiesWithArrayOfEnums(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -142,7 +143,7 @@ class AdditionalPropertiesWithArrayOfEnums(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -124,6 +124,7 @@ class ArrayModel(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -141,7 +142,7 @@ class ArrayModel(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -125,6 +125,7 @@ class ArrayOfEnums(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -142,7 +143,7 @@ class ArrayOfEnums(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -114,6 +114,7 @@ class BodyWithFileSchema(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -130,7 +131,7 @@ class BodyWithFileSchema(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -143,16 +143,20 @@ class BodyWithQueryParams(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestQueryParams, query_params)
used_path = _path
_query_params = []
prefix_separator_iterator = None
for parameter in (
request_query_query,
):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -169,9 +173,8 @@ class BodyWithQueryParams(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
query_params=tuple(_query_params),
headers=_headers,
fields=_fields,
body=_body,

View File

@ -122,6 +122,7 @@ class Boolean(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -139,7 +140,7 @@ class Boolean(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -145,8 +145,9 @@ class CaseSensitiveParams(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestQueryParams, query_params)
used_path = _path
_query_params = []
prefix_separator_iterator = None
for parameter in (
request_query_some_var,
request_query_some_var2,
@ -155,14 +156,16 @@ class CaseSensitiveParams(api_client.Api):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
# TODO add cookie handling
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
query_params=tuple(_query_params),
stream=stream,
timeout=timeout,
)

View File

@ -126,6 +126,7 @@ class ClientModel(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -145,7 +146,7 @@ class ClientModel(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -124,6 +124,7 @@ class ComposedOneOfDifferentTypes(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -141,7 +142,7 @@ class ComposedOneOfDifferentTypes(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -141,6 +141,7 @@ class DeleteCoffee(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestPathParams, path_params)
used_path = _path
_path_params = {}
for parameter in (
@ -151,12 +152,14 @@ class DeleteCoffee(api_client.Api):
continue
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
# TODO add cookie handling
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
stream=stream,
timeout=timeout,
)

View File

@ -257,6 +257,7 @@ class EndpointParameters(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -271,7 +272,7 @@ class EndpointParameters(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -427,8 +427,9 @@ class EnumParameters(api_client.Api):
"""
self._verify_typed_dict_inputs(RequestQueryParams, query_params)
self._verify_typed_dict_inputs(RequestHeaderParams, header_params)
used_path = _path
_query_params = []
prefix_separator_iterator = None
for parameter in (
request_query_enum_query_string_array,
request_query_enum_query_string,
@ -438,8 +439,11 @@ class EnumParameters(api_client.Api):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
_headers = HTTPHeaderDict()
for parameter in (
@ -463,9 +467,8 @@ class EnumParameters(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
query_params=tuple(_query_params),
headers=_headers,
fields=_fields,
body=_body,

View File

@ -113,6 +113,7 @@ class FakeHealthGet(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -121,7 +122,7 @@ class FakeHealthGet(api_client.Api):
_headers.add('Accept', accept_content_type)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
stream=stream,

View File

@ -191,8 +191,9 @@ class GroupParameters(api_client.Api):
"""
self._verify_typed_dict_inputs(RequestQueryParams, query_params)
self._verify_typed_dict_inputs(RequestHeaderParams, header_params)
used_path = _path
_query_params = []
prefix_separator_iterator = None
for parameter in (
request_query_required_string_group,
request_query_required_int64_group,
@ -202,8 +203,11 @@ class GroupParameters(api_client.Api):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
_headers = HTTPHeaderDict()
for parameter in (
@ -218,9 +222,8 @@ class GroupParameters(api_client.Api):
# TODO add cookie handling
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
query_params=tuple(_query_params),
headers=_headers,
auth_settings=_auth,
stream=stream,

View File

@ -132,6 +132,7 @@ class InlineAdditionalProperties(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -148,7 +149,7 @@ class InlineAdditionalProperties(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -524,8 +524,9 @@ class InlineComposition(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestQueryParams, query_params)
used_path = _path
_query_params = []
prefix_separator_iterator = None
for parameter in (
request_query_composition_at_root,
request_query_composition_in_property,
@ -533,8 +534,11 @@ class InlineComposition(api_client.Api):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -552,9 +556,8 @@ class InlineComposition(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
query_params=tuple(_query_params),
headers=_headers,
fields=_fields,
body=_body,

View File

@ -134,6 +134,7 @@ class JsonFormData(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -148,7 +149,7 @@ class JsonFormData(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -123,6 +123,7 @@ class JsonWithCharset(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -140,7 +141,7 @@ class JsonWithCharset(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -125,6 +125,7 @@ class Mammal(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -144,7 +145,7 @@ class Mammal(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -124,6 +124,7 @@ class NumberWithValidations(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -141,7 +142,7 @@ class NumberWithValidations(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -148,22 +148,25 @@ class ObjectInQuery(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestQueryParams, query_params)
used_path = _path
_query_params = []
prefix_separator_iterator = None
for parameter in (
request_query_map_bean,
):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
# TODO add cookie handling
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
query_params=tuple(_query_params),
stream=stream,
timeout=timeout,
)

View File

@ -124,6 +124,7 @@ class ObjectModelWithRefProps(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -141,7 +142,7 @@ class ObjectModelWithRefProps(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -351,6 +351,7 @@ class ParameterCollisions(api_client.Api):
self._verify_typed_dict_inputs(RequestHeaderParams, header_params)
self._verify_typed_dict_inputs(RequestPathParams, path_params)
self._verify_typed_dict_inputs(RequestCookieParams, cookie_params)
used_path = _path
_path_params = {}
for parameter in (
@ -366,7 +367,10 @@ class ParameterCollisions(api_client.Api):
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
_query_params = []
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
prefix_separator_iterator = None
for parameter in (
request_query__1,
request_query_a_b,
@ -377,8 +381,11 @@ class ParameterCollisions(api_client.Api):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
_headers = HTTPHeaderDict()
for parameter in (
@ -407,10 +414,8 @@ class ParameterCollisions(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
query_params=tuple(_query_params),
headers=_headers,
fields=_fields,
body=_body,

View File

@ -196,8 +196,9 @@ class QueryParameterCollectionFormat(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestQueryParams, query_params)
used_path = _path
_query_params = []
prefix_separator_iterator = None
for parameter in (
request_query_pipe,
request_query_ioutil,
@ -209,14 +210,16 @@ class QueryParameterCollectionFormat(api_client.Api):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
# TODO add cookie handling
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
query_params=tuple(_query_params),
stream=stream,
timeout=timeout,
)

View File

@ -129,22 +129,25 @@ class RefObjectInQuery(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestQueryParams, query_params)
used_path = _path
_query_params = []
prefix_separator_iterator = None
for parameter in (
request_query_map_bean,
):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
# TODO add cookie handling
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
query_params=tuple(_query_params),
stream=stream,
timeout=timeout,
)

View File

@ -112,6 +112,7 @@ class ResponseWithoutSchema(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -120,7 +121,7 @@ class ResponseWithoutSchema(api_client.Api):
_headers.add('Accept', accept_content_type)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
stream=stream,

View File

@ -122,6 +122,7 @@ class String(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -139,7 +140,7 @@ class String(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -124,6 +124,7 @@ class StringEnum(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -141,7 +142,7 @@ class StringEnum(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -124,6 +124,7 @@ class UploadDownloadFile(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -143,7 +144,7 @@ class UploadDownloadFile(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -149,6 +149,7 @@ class UploadFile(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -166,7 +167,7 @@ class UploadFile(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -151,6 +151,7 @@ class UploadFiles(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -168,7 +169,7 @@ class UploadFiles(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -129,6 +129,7 @@ class Classname(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -148,7 +149,7 @@ class Classname(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -146,6 +146,7 @@ class AddPet(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -164,7 +165,7 @@ class AddPet(api_client.Api):
host = self.get_host('add_pet', _servers, host_index)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -157,6 +157,7 @@ class DeletePet(api_client.Api):
"""
self._verify_typed_dict_inputs(RequestHeaderParams, header_params)
self._verify_typed_dict_inputs(RequestPathParams, path_params)
used_path = _path
_path_params = {}
for parameter in (
@ -168,6 +169,9 @@ class DeletePet(api_client.Api):
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
_headers = HTTPHeaderDict()
for parameter in (
request_header_api_key,
@ -180,9 +184,8 @@ class DeletePet(api_client.Api):
# TODO add cookie handling
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
headers=_headers,
auth_settings=_auth,
stream=stream,

View File

@ -212,16 +212,20 @@ class FindPetsByStatus(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestQueryParams, query_params)
used_path = _path
_query_params = []
prefix_separator_iterator = None
for parameter in (
request_query_status,
):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -230,9 +234,8 @@ class FindPetsByStatus(api_client.Api):
_headers.add('Accept', accept_content_type)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
query_params=tuple(_query_params),
headers=_headers,
auth_settings=_auth,
stream=stream,

View File

@ -186,16 +186,20 @@ class FindPetsByTags(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestQueryParams, query_params)
used_path = _path
_query_params = []
prefix_separator_iterator = None
for parameter in (
request_query_tags,
):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -204,9 +208,8 @@ class FindPetsByTags(api_client.Api):
_headers.add('Accept', accept_content_type)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
query_params=tuple(_query_params),
headers=_headers,
auth_settings=_auth,
stream=stream,

View File

@ -175,6 +175,7 @@ class GetPetById(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestPathParams, path_params)
used_path = _path
_path_params = {}
for parameter in (
@ -186,6 +187,9 @@ class GetPetById(api_client.Api):
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
_headers = HTTPHeaderDict()
# TODO add cookie handling
if accept_content_types:
@ -193,9 +197,8 @@ class GetPetById(api_client.Api):
_headers.add('Accept', accept_content_type)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
headers=_headers,
auth_settings=_auth,
stream=stream,

View File

@ -158,6 +158,7 @@ class UpdatePet(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -176,7 +177,7 @@ class UpdatePet(api_client.Api):
host = self.get_host('update_pet', _servers, host_index)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -166,6 +166,7 @@ class UpdatePetWithForm(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestPathParams, path_params)
used_path = _path
_path_params = {}
for parameter in (
@ -177,6 +178,9 @@ class UpdatePetWithForm(api_client.Api):
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -190,9 +194,8 @@ class UpdatePetWithForm(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
headers=_headers,
fields=_fields,
body=_body,

View File

@ -180,6 +180,7 @@ class UploadFileWithRequiredFile(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestPathParams, path_params)
used_path = _path
_path_params = {}
for parameter in (
@ -191,6 +192,9 @@ class UploadFileWithRequiredFile(api_client.Api):
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
_headers = HTTPHeaderDict()
# TODO add cookie handling
if accept_content_types:
@ -207,9 +211,8 @@ class UploadFileWithRequiredFile(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
headers=_headers,
fields=_fields,
body=_body,

View File

@ -178,6 +178,7 @@ class UploadImage(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestPathParams, path_params)
used_path = _path
_path_params = {}
for parameter in (
@ -189,6 +190,9 @@ class UploadImage(api_client.Api):
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
_headers = HTTPHeaderDict()
# TODO add cookie handling
if accept_content_types:
@ -205,9 +209,8 @@ class UploadImage(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
headers=_headers,
fields=_fields,
body=_body,

View File

@ -139,6 +139,7 @@ class DeleteOrder(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestPathParams, path_params)
used_path = _path
_path_params = {}
for parameter in (
@ -149,12 +150,14 @@ class DeleteOrder(api_client.Api):
continue
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
# TODO add cookie handling
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
stream=stream,
timeout=timeout,
)

View File

@ -133,6 +133,7 @@ class GetInventory(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -141,7 +142,7 @@ class GetInventory(api_client.Api):
_headers.add('Accept', accept_content_type)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
auth_settings=_auth,

View File

@ -181,6 +181,7 @@ class GetOrderById(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestPathParams, path_params)
used_path = _path
_path_params = {}
for parameter in (
@ -192,6 +193,9 @@ class GetOrderById(api_client.Api):
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
_headers = HTTPHeaderDict()
# TODO add cookie handling
if accept_content_types:
@ -199,9 +203,8 @@ class GetOrderById(api_client.Api):
_headers.add('Accept', accept_content_type)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
headers=_headers,
stream=stream,
timeout=timeout,

View File

@ -144,6 +144,7 @@ class PlaceOrder(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -163,7 +164,7 @@ class PlaceOrder(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -115,6 +115,7 @@ class CreateUser(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -131,7 +132,7 @@ class CreateUser(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -124,6 +124,7 @@ class CreateUsersWithArrayInput(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -140,7 +141,7 @@ class CreateUsersWithArrayInput(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -124,6 +124,7 @@ class CreateUsersWithListInput(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -140,7 +141,7 @@ class CreateUsersWithListInput(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
headers=_headers,
fields=_fields,

View File

@ -139,6 +139,7 @@ class DeleteUser(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestPathParams, path_params)
used_path = _path
_path_params = {}
for parameter in (
@ -149,12 +150,14 @@ class DeleteUser(api_client.Api):
continue
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
# TODO add cookie handling
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
stream=stream,
timeout=timeout,
)

View File

@ -172,6 +172,7 @@ class GetUserByName(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestPathParams, path_params)
used_path = _path
_path_params = {}
for parameter in (
@ -183,6 +184,9 @@ class GetUserByName(api_client.Api):
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
_headers = HTTPHeaderDict()
# TODO add cookie handling
if accept_content_types:
@ -190,9 +194,8 @@ class GetUserByName(api_client.Api):
_headers.add('Accept', accept_content_type)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
headers=_headers,
stream=stream,
timeout=timeout,

View File

@ -190,8 +190,9 @@ class LoginUser(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestQueryParams, query_params)
used_path = _path
_query_params = []
prefix_separator_iterator = None
for parameter in (
request_query_username,
request_query_password,
@ -199,8 +200,11 @@ class LoginUser(api_client.Api):
parameter_data = query_params.get(parameter.name, unset)
if parameter_data is unset:
continue
serialized_data = parameter.serialize(parameter_data)
_query_params.extend(serialized_data)
if prefix_separator_iterator is None:
prefix_separator_iterator = parameter.get_prefix_separator_iterator()
serialized_data = parameter.serialize(parameter_data, prefix_separator_iterator)
for serialized_value in serialized_data.values():
used_path += serialized_value
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -209,9 +213,8 @@ class LoginUser(api_client.Api):
_headers.add('Accept', accept_content_type)
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
query_params=tuple(_query_params),
headers=_headers,
stream=stream,
timeout=timeout,

View File

@ -99,10 +99,11 @@ class LogoutUser(api_client.Api):
api_response.body and api_response.headers will not be deserialized into schema
class instances
"""
used_path = _path
# TODO add cookie handling
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
stream=stream,
timeout=timeout,

View File

@ -155,6 +155,7 @@ class UpdateUser(api_client.Api):
class instances
"""
self._verify_typed_dict_inputs(RequestPathParams, path_params)
used_path = _path
_path_params = {}
for parameter in (
@ -166,6 +167,9 @@ class UpdateUser(api_client.Api):
serialized_data = parameter.serialize(parameter_data)
_path_params.update(serialized_data)
for k, v in _path_params.items():
used_path = used_path.replace('{%s}' % k, v)
_headers = HTTPHeaderDict()
# TODO add cookie handling
@ -181,9 +185,8 @@ class UpdateUser(api_client.Api):
elif 'body' in serialized_data:
_body = serialized_data['body']
response = self.api_client.call_api(
resource_path=_path,
resource_path=used_path,
method=_method,
path_params=_path_params,
headers=_headers,
fields=_fields,
body=_body,

View File

@ -53,9 +53,12 @@ class RequestField(RequestFieldBase):
class JSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, (str, int, float)):
# instances based on primitive classes
return obj
if isinstance(obj, str):
return str(obj)
elif isinstance(obj, float):
return float(obj)
elif isinstance(obj, int):
return int(obj)
elif isinstance(obj, Decimal):
if obj.as_tuple().exponent >= 0:
return int(obj)
@ -88,180 +91,175 @@ class ParameterStyle(enum.Enum):
DEEP_OBJECT = 'deepObject'
class PrefixSeparatorIterator:
# A class to store prefixes and separators for rfc6570 expansions
def __init__(self, prefix: str, separator: str):
self.prefix = prefix
self.separator = separator
self.first = True
if separator in {'.', '|', '%20'}:
item_separator = separator
else:
item_separator = ','
self.item_separator = item_separator
def __iter__(self):
return self
def __next__(self):
if self.first:
self.first = False
return self.prefix
return self.separator
class ParameterSerializerBase:
@staticmethod
def __serialize_number(
in_data: typing.Union[int, float], name: str, prefix=''
) -> typing.Tuple[typing.Tuple[str, str]]:
return tuple([(name, prefix + str(in_data))])
@classmethod
def get_default_explode(cls, style: ParameterStyle) -> bool:
return False
@staticmethod
def __serialize_str(
in_data: str, name: str, prefix=''
) -> typing.Tuple[typing.Tuple[str, str]]:
return tuple([(name, prefix + quote(in_data))])
@staticmethod
def __serialize_bool(in_data: bool, name: str, prefix='') -> typing.Tuple[typing.Tuple[str, str]]:
if in_data:
return tuple([(name, prefix + 'true')])
return tuple([(name, prefix + 'false')])
@staticmethod
def __urlencode(in_data: typing.Any) -> str:
return quote(str(in_data))
def __serialize_list(
self,
in_data: typing.List[typing.Any],
style: ParameterStyle,
name: str,
explode: bool,
empty_val: typing.Union[typing.Tuple[str, str], typing.Tuple] = tuple(),
prefix: str = '',
separator: str = ',',
) -> typing.Tuple[typing.Union[typing.Tuple[str, str], typing.Tuple], ...]:
if not in_data:
return empty_val
if explode and style in {
ParameterStyle.FORM,
ParameterStyle.MATRIX,
ParameterStyle.SPACE_DELIMITED,
ParameterStyle.PIPE_DELIMITED
}:
if style is ParameterStyle.FORM:
return tuple((name, prefix + self.__urlencode(val)) for val in in_data)
else:
joined_vals = prefix + separator.join(name + '=' + self.__urlencode(val) for val in in_data)
else:
joined_vals = prefix + separator.join(map(self.__urlencode, in_data))
return tuple([(name, joined_vals)])
def __form_item_representation(self, in_data: typing.Any) -> typing.Optional[str]:
if isinstance(in_data, none_type):
def __ref6570_item_value(in_data: typing.Any, percent_encode: bool):
"""
Get representation if str/float/int/None/items in list/ values in dict
None is returned if an item is undefined, use cases are value=
- None
- []
- {}
- [None, None None]
- {'a': None, 'b': None}
"""
if type(in_data) in {str, float, int}:
if percent_encode:
return quote(str(in_data))
return str(in_data)
elif isinstance(in_data, none_type):
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return None
elif isinstance(in_data, list):
if not in_data:
return None
raise ApiValueError('Unable to generate a form representation of {}'.format(in_data))
elif isinstance(in_data, dict):
if not in_data:
return None
raise ApiValueError('Unable to generate a form representation of {}'.format(in_data))
elif isinstance(in_data, (bool, bytes)):
raise ApiValueError('Unable to generate a form representation of {}'.format(in_data))
# str, float, int
return self.__urlencode(in_data)
elif isinstance(in_data, list) and not in_data:
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return None
elif isinstance(in_data, dict) and not in_data:
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return None
raise ApiValueError('Unable to generate a ref6570 item representation of {}'.format(in_data))
def __serialize_dict(
self,
in_data: typing.Dict[str, typing.Any],
style: ParameterStyle,
name: str,
@staticmethod
def to_dict(name: str, value: str):
return {name: value}
@classmethod
def ref6570_expansion(
cls,
variable_name: str,
in_data: typing.Any,
explode: bool,
empty_val: typing.Union[typing.Tuple[str, str], typing.Tuple] = tuple(),
prefix: str = '',
separator: str = ',',
) -> typing.Tuple[typing.Tuple[str, str]]:
if not in_data:
return empty_val
if all(val is None for val in in_data.values()):
return empty_val
form_items = {}
if style is ParameterStyle.FORM:
for key, val in in_data.items():
new_val = self.__form_item_representation(val)
if new_val is None:
continue
form_items[key] = new_val
if explode:
if style is ParameterStyle.FORM:
return tuple((key, prefix + val) for key, val in form_items.items())
elif style in {
ParameterStyle.SIMPLE,
ParameterStyle.LABEL,
ParameterStyle.MATRIX,
ParameterStyle.SPACE_DELIMITED,
ParameterStyle.PIPE_DELIMITED
}:
joined_vals = prefix + separator.join(key + '=' + self.__urlencode(val) for key, val in in_data.items())
else:
raise ApiValueError(f'Invalid style {style} for dict serialization with explode=True')
elif style is ParameterStyle.FORM:
joined_vals = prefix + separator.join(key + separator + val for key, val in form_items.items())
else:
joined_vals = prefix + separator.join(
key + separator + self.__urlencode(val) for key, val in in_data.items())
return tuple([(name, joined_vals)])
def _serialize_x(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
style: ParameterStyle,
name: str,
explode: bool,
empty_val: typing.Union[typing.Tuple[str, str], typing.Tuple] = (),
prefix: str = '',
separator: str = ',',
) -> typing.Tuple[typing.Tuple[str, str], ...]:
if isinstance(in_data, none_type):
return empty_val
elif isinstance(in_data, bool):
# must be before int check
return self.__serialize_bool(in_data, name=name, prefix=prefix)
elif isinstance(in_data, (int, float)):
return self.__serialize_number(in_data, name=name, prefix=prefix)
elif isinstance(in_data, str):
return self.__serialize_str(in_data, name=name, prefix=prefix)
percent_encode: bool,
prefix_separator_iterator: PrefixSeparatorIterator
) -> str:
"""
Separator is for separate variables like dict with explode true, not for array item separation
"""
named_parameter_expansion = prefix_separator_iterator.separator in {'&', ';'}
var_name_piece = variable_name if named_parameter_expansion else ''
if type(in_data) in {str, float, int}:
item_value = cls.__ref6570_item_value(in_data, percent_encode)
if item_value is None:
return next(prefix_separator_iterator) + var_name_piece
elif item_value == '' and prefix_separator_iterator.separator == ';':
return next(prefix_separator_iterator) + var_name_piece
value_pair_equals = '=' if named_parameter_expansion else ''
return next(prefix_separator_iterator) + var_name_piece + value_pair_equals + item_value
elif isinstance(in_data, none_type):
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return ""
elif isinstance(in_data, list):
return self.__serialize_list(
in_data,
style=style,
name=name,
explode=explode,
empty_val=empty_val,
prefix=prefix,
separator=separator
item_values = [cls.__ref6570_item_value(v, percent_encode) for v in in_data]
item_values = [v for v in item_values if v is not None]
if not item_values:
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return ""
value_pair_equals = '=' if named_parameter_expansion else ''
if not explode:
return (
next(prefix_separator_iterator) +
var_name_piece +
value_pair_equals +
prefix_separator_iterator.item_separator.join(item_values)
)
# exploded
return next(prefix_separator_iterator) + next(prefix_separator_iterator).join(
[var_name_piece + value_pair_equals + val for val in item_values]
)
elif isinstance(in_data, dict):
return self.__serialize_dict(
in_data,
style=style,
name=name,
explode=explode,
empty_val=empty_val,
prefix=prefix,
separator=separator
in_data_transformed = {key: cls.__ref6570_item_value(val, percent_encode) for key, val in in_data.items()}
in_data_transformed = {key: val for key, val in in_data_transformed.items() if val is not None}
if not in_data_transformed:
# ignored by the expansion process https://datatracker.ietf.org/doc/html/rfc6570#section-3.2.1
return ""
value_pair_equals = '=' if named_parameter_expansion else ''
if not explode:
return (
next(prefix_separator_iterator) +
var_name_piece + value_pair_equals +
prefix_separator_iterator.item_separator.join(
prefix_separator_iterator.item_separator.join(
item_pair
) for item_pair in in_data_transformed.items()
)
)
# exploded
return next(prefix_separator_iterator) + next(prefix_separator_iterator).join(
[key + '=' + val for key, val in in_data_transformed.items()]
)
# bool, bytes, etc
raise ApiValueError('Unable to generate a ref6570 representation of {}'.format(in_data))
class StyleFormSerializer(ParameterSerializerBase):
@classmethod
def get_default_explode(cls, style: ParameterStyle) -> bool:
if style is ParameterStyle.FORM:
return True
return super().get_default_explode(style)
def _serialize_form(
def serialize_form(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
name: str,
explode: bool,
) -> typing.Tuple[typing.Tuple[str, str], ...]:
return self._serialize_x(in_data, style=ParameterStyle.FORM, name=name, explode=explode)
percent_encode: bool,
prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator] = None
) -> str:
if prefix_separator_iterator is None:
prefix_separator_iterator = PrefixSeparatorIterator('?', '&')
return self.ref6570_expansion(
variable_name=name,
in_data=in_data,
explode=explode,
percent_encode=percent_encode,
prefix_separator_iterator=prefix_separator_iterator
)
class StyleSimpleSerializer(ParameterSerializerBase):
def _serialize_simple_tuple(
def serialize_simple(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
name: str,
explode: bool,
in_type: ParameterInType,
) -> typing.Tuple[typing.Tuple[str, str], ...]:
if in_type is ParameterInType.HEADER:
empty_val = ()
else:
empty_val = ((name, ''),)
return self._serialize_x(in_data, style=ParameterStyle.SIMPLE, name=name, explode=explode, empty_val=empty_val)
percent_encode: bool
) -> str:
prefix_separator_iterator = PrefixSeparatorIterator('', ',')
return self.ref6570_expansion(
variable_name=name,
in_data=in_data,
explode=explode,
percent_encode=percent_encode,
prefix_separator_iterator=prefix_separator_iterator
)
@dataclass
@ -349,8 +347,8 @@ class ParameterBase:
def _serialize_json(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list]
) -> typing.Tuple[typing.Tuple[str, str]]:
return tuple([(self.name, json.dumps(in_data))])
) -> str:
return json.dumps(in_data)
class PathParameter(ParameterBase, StyleSimpleSerializer):
@ -376,56 +374,45 @@ class PathParameter(ParameterBase, StyleSimpleSerializer):
content=content
)
def __serialize_label(
def _serialize_label(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list]
) -> typing.Dict[str, str]:
empty_val = ((self.name, ''),)
prefix = '.'
separator = '.'
return self._remove_empty_and_cast(
self._serialize_x(
in_data,
style=ParameterStyle.LABEL,
name=self.name,
explode=self.explode,
empty_val=empty_val,
prefix=prefix,
separator=separator
)
prefix_separator_iterator = PrefixSeparatorIterator('.', '.')
value = self.ref6570_expansion(
variable_name=self.name,
in_data=in_data,
explode=self.explode,
percent_encode=True,
prefix_separator_iterator=prefix_separator_iterator
)
return self.to_dict(self.name, value)
def __serialize_matrix(
def _serialize_matrix(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list]
) -> typing.Dict[str, str]:
separator = ','
if in_data == '':
prefix = ';' + self.name
elif isinstance(in_data, (dict, list)) and self.explode:
prefix = ';'
separator = ';'
else:
prefix = ';' + self.name + '='
empty_val = ((self.name, ''),)
return self._remove_empty_and_cast(
self._serialize_x(
in_data,
style=ParameterStyle.MATRIX,
name=self.name,
explode=self.explode,
prefix=prefix,
empty_val=empty_val,
separator=separator
)
prefix_separator_iterator = PrefixSeparatorIterator(';', ';')
value = self.ref6570_expansion(
variable_name=self.name,
in_data=in_data,
explode=self.explode,
percent_encode=True,
prefix_separator_iterator=prefix_separator_iterator
)
return self.to_dict(self.name, value)
def _serialize_simple(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
) -> typing.Dict[str, str]:
tuple_data = self._serialize_simple_tuple(in_data, self.name, self.explode, self.in_type)
return self._remove_empty_and_cast(tuple_data)
value = self.serialize_simple(
in_data=in_data,
name=self.name,
explode=self.explode,
percent_encode=True
)
return self.to_dict(self.name, value)
def serialize(
self,
@ -448,16 +435,16 @@ class PathParameter(ParameterBase, StyleSimpleSerializer):
if self.style is ParameterStyle.SIMPLE:
return self._serialize_simple(cast_in_data)
elif self.style is ParameterStyle.LABEL:
return self.__serialize_label(cast_in_data)
return self._serialize_label(cast_in_data)
elif self.style is ParameterStyle.MATRIX:
return self.__serialize_matrix(cast_in_data)
return self._serialize_matrix(cast_in_data)
# self.content will be length one
for content_type, schema in self.content.items():
cast_in_data = schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
if content_type == self._json_content_type:
tuple_data = self._serialize_json(cast_in_data)
return self._remove_empty_and_cast(tuple_data)
value = self._serialize_json(cast_in_data)
return self.to_dict(self.name, value)
raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))
@ -468,17 +455,20 @@ class QueryParameter(ParameterBase, StyleFormSerializer):
name: str,
required: bool = False,
style: typing.Optional[ParameterStyle] = None,
explode: bool = False,
explode: typing.Optional[bool] = None,
allow_reserved: typing.Optional[bool] = None,
schema: typing.Optional[typing.Type[Schema]] = None,
content: typing.Optional[typing.Dict[str, typing.Type[Schema]]] = None
):
used_style = ParameterStyle.FORM if style is None and content is None and schema else style
used_explode = self.get_default_explode(used_style) if explode is None else explode
super().__init__(
name,
in_type=ParameterInType.QUERY,
required=required,
style=style,
explode=explode,
style=used_style,
explode=used_explode,
allow_reserved=allow_reserved,
schema=schema,
content=content
@ -486,39 +476,68 @@ class QueryParameter(ParameterBase, StyleFormSerializer):
def __serialize_space_delimited(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list]
) -> typing.Tuple[typing.Tuple[str, str], ...]:
separator = '%20'
empty_val = ()
return self._serialize_x(
in_data,
style=ParameterStyle.SPACE_DELIMITED,
name=self.name,
in_data: typing.Union[None, int, float, str, bool, dict, list],
prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator]
) -> typing.Dict[str, str]:
if prefix_separator_iterator is None:
prefix_separator_iterator = self.get_prefix_separator_iterator()
value = self.ref6570_expansion(
variable_name=self.name,
in_data=in_data,
explode=self.explode,
separator=separator,
empty_val=empty_val
percent_encode=True,
prefix_separator_iterator=prefix_separator_iterator
)
return self.to_dict(self.name, value)
def __serialize_pipe_delimited(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list]
) -> typing.Tuple[typing.Tuple[str, str], ...]:
separator = '|'
empty_val = ()
return self._serialize_x(
in_data: typing.Union[None, int, float, str, bool, dict, list],
prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator]
) -> typing.Dict[str, str]:
if prefix_separator_iterator is None:
prefix_separator_iterator = self.get_prefix_separator_iterator()
value = self.ref6570_expansion(
variable_name=self.name,
in_data=in_data,
explode=self.explode,
percent_encode=True,
prefix_separator_iterator=prefix_separator_iterator
)
return self.to_dict(self.name, value)
def __serialize_form(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator]
) -> typing.Dict[str, str]:
if prefix_separator_iterator is None:
prefix_separator_iterator = self.get_prefix_separator_iterator()
value = self.serialize_form(
in_data,
style=ParameterStyle.PIPE_DELIMITED,
name=self.name,
explode=self.explode,
separator=separator,
empty_val=empty_val
percent_encode=True,
prefix_separator_iterator=prefix_separator_iterator
)
return self.to_dict(self.name, value)
def get_prefix_separator_iterator(self) -> typing.Optional[PrefixSeparatorIterator]:
if not self.schema:
return None
if self.style is ParameterStyle.FORM:
return PrefixSeparatorIterator('?', '&')
elif self.style is ParameterStyle.SPACE_DELIMITED:
return PrefixSeparatorIterator('', '%20')
elif self.style is ParameterStyle.PIPE_DELIMITED:
return PrefixSeparatorIterator('', '|')
def serialize(
self,
in_data: typing.Union[
Schema, Decimal, int, float, str, date, datetime, None, bool, list, tuple, dict, frozendict]
) -> typing.Tuple[typing.Tuple[str, str]]:
Schema, Decimal, int, float, str, date, datetime, None, bool, list, tuple, dict, frozendict],
prefix_separator_iterator: typing.Optional[PrefixSeparatorIterator] = None
) -> typing.Dict[str, str]:
if self.schema:
cast_in_data = self.schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
@ -538,17 +557,18 @@ class QueryParameter(ParameterBase, StyleFormSerializer):
if self.style:
# TODO update query ones to omit setting values when [] {} or None is input
if self.style is ParameterStyle.FORM:
return self._serialize_form(cast_in_data, explode=self.explode, name=self.name)
return self.__serialize_form(cast_in_data, prefix_separator_iterator)
elif self.style is ParameterStyle.SPACE_DELIMITED:
return self.__serialize_space_delimited(cast_in_data)
return self.__serialize_space_delimited(cast_in_data, prefix_separator_iterator)
elif self.style is ParameterStyle.PIPE_DELIMITED:
return self.__serialize_pipe_delimited(cast_in_data)
return self.__serialize_pipe_delimited(cast_in_data, prefix_separator_iterator)
# self.content will be length one
for content_type, schema in self.content.items():
cast_in_data = schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
if content_type == self._json_content_type:
return self._serialize_json(cast_in_data)
value = self._serialize_json(cast_in_data)
return self.to_dict(self.name, value)
raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))
@ -559,17 +579,20 @@ class CookieParameter(ParameterBase, StyleFormSerializer):
name: str,
required: bool = False,
style: typing.Optional[ParameterStyle] = None,
explode: bool = False,
explode: typing.Optional[bool] = None,
allow_reserved: typing.Optional[bool] = None,
schema: typing.Optional[typing.Type[Schema]] = None,
content: typing.Optional[typing.Dict[str, typing.Type[Schema]]] = None
):
used_style = ParameterStyle.FORM if style is None and content is None and schema else style
used_explode = self.get_default_explode(used_style) if explode is None else explode
super().__init__(
name,
in_type=ParameterInType.COOKIE,
required=required,
style=style,
explode=explode,
style=used_style,
explode=used_explode,
allow_reserved=allow_reserved,
schema=schema,
content=content
@ -579,7 +602,7 @@ class CookieParameter(ParameterBase, StyleFormSerializer):
self,
in_data: typing.Union[
Schema, Decimal, int, float, str, date, datetime, None, bool, list, tuple, dict, frozendict]
) -> typing.Tuple[typing.Tuple[str, str]]:
) -> typing.Dict[str, str]:
if self.schema:
cast_in_data = self.schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
@ -588,13 +611,25 @@ class CookieParameter(ParameterBase, StyleFormSerializer):
returns fields: tuple
"""
if self.style:
return self._serialize_form(cast_in_data, explode=self.explode, name=self.name)
"""
TODO add escaping of comma, space, equals
or turn encoding on
"""
value = self.serialize_form(
cast_in_data,
explode=self.explode,
name=self.name,
percent_encode=False,
prefix_separator_iterator=PrefixSeparatorIterator('', '&')
)
return self.to_dict(self.name, value)
# self.content will be length one
for content_type, schema in self.content.items():
cast_in_data = schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
if content_type == self._json_content_type:
return self._serialize_json(cast_in_data)
value = self._serialize_json(cast_in_data)
return self.to_dict(self.name, value)
raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))
@ -632,9 +667,8 @@ class HeaderParameter(ParameterBase, StyleSimpleSerializer):
def _serialize_simple(
self,
in_data: typing.Union[None, int, float, str, bool, dict, list],
) -> HTTPHeaderDict[str, str]:
tuple_data = self._serialize_simple_tuple(in_data, self.name, self.explode, self.in_type)
return self.__to_headers(tuple_data)
) -> str:
return self.serialize_simple(in_data, self.name, self.explode, False)
def serialize(
self,
@ -650,14 +684,15 @@ class HeaderParameter(ParameterBase, StyleSimpleSerializer):
returns headers: dict
"""
if self.style:
return self._serialize_simple(cast_in_data)
value = self._serialize_simple(cast_in_data)
return self.__to_headers(((self.name, value),))
# self.content will be length one
for content_type, schema in self.content.items():
cast_in_data = schema(in_data)
cast_in_data = self._json_encoder.default(cast_in_data)
if content_type == self._json_content_type:
tuple_data = self._serialize_json(cast_in_data)
return self.__to_headers(tuple_data)
value = self._serialize_json(cast_in_data)
return self.__to_headers(((self.name, value),))
raise NotImplementedError('Serialization of {} has not yet been implemented'.format(content_type))
@ -935,8 +970,6 @@ class ApiClient:
self,
resource_path: str,
method: str,
path_params: typing.Optional[typing.Dict[str, typing.Any]] = None,
query_params: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
headers: typing.Optional[HTTPHeaderDict] = None,
body: typing.Optional[typing.Union[str, bytes]] = None,
fields: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
@ -952,17 +985,8 @@ class ApiClient:
if self.cookie:
headers['Cookie'] = self.cookie
# path parameters
if path_params:
for k, v in path_params.items():
# specified safe chars, encode everything
resource_path = resource_path.replace(
'{%s}' % k,
quote(str(v), safe=self.configuration.safe_chars_for_path_param)
)
# auth setting
self.update_params_for_auth(headers, query_params,
self.update_params_for_auth(headers,
auth_settings, resource_path, method, body)
# request url
@ -976,7 +1000,6 @@ class ApiClient:
response = self.request(
method,
url,
query_params=query_params,
headers=headers,
fields=fields,
body=body,
@ -989,8 +1012,6 @@ class ApiClient:
self,
resource_path: str,
method: str,
path_params: typing.Optional[typing.Dict[str, typing.Any]] = None,
query_params: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
headers: typing.Optional[HTTPHeaderDict] = None,
body: typing.Optional[typing.Union[str, bytes]] = None,
fields: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
@ -1006,8 +1027,6 @@ class ApiClient:
:param resource_path: Path to method endpoint.
:param method: Method to call.
:param path_params: Path parameters in the url.
:param query_params: Query parameters in the url.
:param headers: Header parameters to be
placed in the request header.
:param body: Request body.
@ -1040,8 +1059,6 @@ class ApiClient:
return self.__call_api(
resource_path,
method,
path_params,
query_params,
headers,
body,
fields,
@ -1056,8 +1073,6 @@ class ApiClient:
(
resource_path,
method,
path_params,
query_params,
headers,
body,
json,
@ -1073,7 +1088,6 @@ class ApiClient:
self,
method: str,
url: str,
query_params: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
headers: typing.Optional[HTTPHeaderDict] = None,
fields: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
body: typing.Optional[typing.Union[str, bytes]] = None,
@ -1083,19 +1097,16 @@ class ApiClient:
"""Makes the HTTP request using RESTClient."""
if method == "GET":
return self.rest_client.GET(url,
query_params=query_params,
stream=stream,
timeout=timeout,
headers=headers)
elif method == "HEAD":
return self.rest_client.HEAD(url,
query_params=query_params,
stream=stream,
timeout=timeout,
headers=headers)
elif method == "OPTIONS":
return self.rest_client.OPTIONS(url,
query_params=query_params,
headers=headers,
fields=fields,
stream=stream,
@ -1103,7 +1114,6 @@ class ApiClient:
body=body)
elif method == "POST":
return self.rest_client.POST(url,
query_params=query_params,
headers=headers,
fields=fields,
stream=stream,
@ -1111,7 +1121,6 @@ class ApiClient:
body=body)
elif method == "PUT":
return self.rest_client.PUT(url,
query_params=query_params,
headers=headers,
fields=fields,
stream=stream,
@ -1119,7 +1128,6 @@ class ApiClient:
body=body)
elif method == "PATCH":
return self.rest_client.PATCH(url,
query_params=query_params,
headers=headers,
fields=fields,
stream=stream,
@ -1127,7 +1135,6 @@ class ApiClient:
body=body)
elif method == "DELETE":
return self.rest_client.DELETE(url,
query_params=query_params,
headers=headers,
stream=stream,
timeout=timeout,
@ -1138,12 +1145,11 @@ class ApiClient:
" `POST`, `PATCH`, `PUT` or `DELETE`."
)
def update_params_for_auth(self, headers, querys, auth_settings,
def update_params_for_auth(self, headers, auth_settings,
resource_path, method, body):
"""Updates header and query params based on authentication setting.
:param headers: Header parameters dict to be updated.
:param querys: Query parameters tuple list to be updated.
:param auth_settings: Authentication setting identifiers list.
:param resource_path: A string representation of the HTTP request resource path.
:param method: A string representation of the HTTP request method.
@ -1165,12 +1171,17 @@ class ApiClient:
# The HTTP signature scheme requires multiple HTTP headers
# that are calculated dynamically.
signing_info = self.configuration.signing_info
querys = tuple()
auth_headers = signing_info.get_http_signature_headers(
resource_path, method, headers, body, querys)
for key, value in auth_headers.items():
headers.add(key, value)
elif auth_setting['in'] == 'query':
querys.append((auth_setting['key'], auth_setting['value']))
""" TODO implement auth in query
need to pass in prefix_separator_iterator
and need to output resource_path with query params added
"""
raise ApiValueError("Auth in query not yet implemented")
else:
raise ApiValueError(
'Authentication token must be in `query` or `header`'
@ -1373,15 +1384,16 @@ class RequestBody(StyleFormSerializer, JSONDetector):
def __serialize_application_x_www_form_data(
self, in_data: typing.Any
) -> typing.Dict[str, tuple[tuple[str, str], ...]]:
) -> SerializedRequestBody:
"""
POST submission of form data in body
"""
if not isinstance(in_data, frozendict):
raise ValueError(
f'Unable to serialize {in_data} to application/x-www-form-urlencoded because it is not a dict of data')
cast_in_data = self.__json_encoder.default(in_data)
fields = self._serialize_form(cast_in_data, explode=True, name='')
if not fields:
return {}
return {'fields': fields}
value = self.serialize_form(cast_in_data, name='', explode=True, percent_encode=False)
return dict(body=value)
def serialize(
self, in_data: typing.Any, content_type: str

View File

@ -90,7 +90,6 @@ class RESTClientObject(object):
self,
method: str,
url: str,
query_params: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
headers: typing.Optional[HTTPHeaderDict] = None,
fields: typing.Optional[typing.Tuple[typing.Tuple[str, typing.Any], ...]] = None,
body: typing.Optional[typing.Union[str, bytes]] = None,
@ -101,7 +100,6 @@ class RESTClientObject(object):
:param method: http request method
:param url: http request url
:param query_params: query parameters in the url
:param headers: http request headers
:param body: request body, for other types
:param fields: request parameters for
@ -137,13 +135,10 @@ class RESTClientObject(object):
try:
# For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:
if query_params:
url += '?' + urlencode(query_params)
if 'Content-Type' not in headers and body is None:
r = self.pool_manager.request(
method,
url,
fields=query_params,
preload_content=not stream,
timeout=timeout,
headers=headers
@ -188,7 +183,6 @@ class RESTClientObject(object):
# For `GET`, `HEAD`
else:
r = self.pool_manager.request(method, url,
fields=query_params,
preload_content=not stream,
timeout=timeout,
headers=headers)
@ -202,63 +196,58 @@ class RESTClientObject(object):
return r
def GET(self, url, headers=None, query_params=None, stream=False,
def GET(self, url, headers=None, stream=False,
timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("GET", url,
headers=headers,
stream=stream,
timeout=timeout,
query_params=query_params, fields=fields)
fields=fields)
def HEAD(self, url, headers=None, query_params=None, stream=False,
def HEAD(self, url, headers=None, stream=False,
timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("HEAD", url,
headers=headers,
stream=stream,
timeout=timeout,
query_params=query_params, fields=fields)
fields=fields)
def OPTIONS(self, url, headers=None, query_params=None,
def OPTIONS(self, url, headers=None,
body=None, stream=False, timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("OPTIONS", url,
headers=headers,
query_params=query_params,
stream=stream,
timeout=timeout,
body=body, fields=fields)
def DELETE(self, url, headers=None, query_params=None, body=None,
def DELETE(self, url, headers=None, body=None,
stream=False, timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("DELETE", url,
headers=headers,
query_params=query_params,
stream=stream,
timeout=timeout,
body=body, fields=fields)
def POST(self, url, headers=None, query_params=None,
def POST(self, url, headers=None,
body=None, stream=False, timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("POST", url,
headers=headers,
query_params=query_params,
stream=stream,
timeout=timeout,
body=body, fields=fields)
def PUT(self, url, headers=None, query_params=None,
def PUT(self, url, headers=None,
body=None, stream=False, timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("PUT", url,
headers=headers,
query_params=query_params,
stream=stream,
timeout=timeout,
body=body, fields=fields)
def PATCH(self, url, headers=None, query_params=None,
def PATCH(self, url, headers=None,
body=None, stream=False, timeout=None, fields=None) -> urllib3.HTTPResponse:
return self.request("PATCH", url,
headers=headers,
query_params=query_params,
stream=stream,
timeout=timeout,
body=body, fields=fields)

View File

@ -0,0 +1,104 @@
import json
import typing
import unittest
import urllib3
from urllib3._collections import HTTPHeaderDict
from petstore_api import api_client
class ApiTestMixin(unittest.TestCase):
json_content_type = 'application/json'
user_agent = 'OpenAPI-Generator/1.0.0/python'
@classmethod
def assert_request_called_with(
cls,
mock_request,
url: str,
method: str = 'POST',
body: typing.Optional[bytes] = None,
content_type: typing.Optional[str] = 'application/json',
fields: typing.Optional[tuple[api_client.RequestField, ...]] = None,
accept_content_type: typing.Optional[str] = 'application/json',
stream: bool = False,
):
headers = {
'User-Agent': cls.user_agent
}
if accept_content_type:
headers['Accept'] = accept_content_type
if content_type:
headers['Content-Type'] = content_type
kwargs = dict(
headers=HTTPHeaderDict(headers),
fields=fields,
stream=stream,
timeout=None,
)
if method != 'GET':
kwargs['body'] = body
mock_request.assert_called_with(
method,
url,
**kwargs
)
@classmethod
def assert_pool_manager_request_called_with(
cls,
mock_request,
url: str,
method: str = 'POST',
body: typing.Optional[bytes] = None,
content_type: typing.Optional[str] = 'application/json',
accept_content_type: typing.Optional[str] = 'application/json',
stream: bool = False,
):
headers = {
'User-Agent': cls.user_agent
}
if accept_content_type:
headers['Accept'] = accept_content_type
if content_type:
headers['Content-Type'] = content_type
kwargs = dict(
headers=HTTPHeaderDict(headers),
preload_content=not stream,
timeout=None,
)
if content_type and method != 'GET':
kwargs['body'] = body
mock_request.assert_called_with(
method,
url,
**kwargs
)
@staticmethod
def headers_for_content_type(content_type: str) -> dict[str, str]:
return {'content-type': content_type}
@classmethod
def response(
cls,
body: typing.Union[str, bytes],
status: int = 200,
content_type: str = json_content_type,
headers: typing.Optional[dict[str, str]] = None,
preload_content: bool = True
) -> urllib3.HTTPResponse:
if headers is None:
headers = {}
headers.update(cls.headers_for_content_type(content_type))
return urllib3.HTTPResponse(
body,
headers=headers,
status=status,
preload_content=preload_content
)
@staticmethod
def json_bytes(in_data: typing.Any) -> bytes:
return json.dumps(in_data, separators=(",", ":"), ensure_ascii=False).encode('utf-8')

View File

@ -43,7 +43,6 @@ class ConfigurationTests(unittest.TestCase):
mock_request.assert_called_with(
'POST',
'http://path-server-test.petstore.local/v2/pet',
query_params=None,
headers=HTTPHeaderDict({
'Content-Type': 'application/json',
'User-Agent': 'OpenAPI-Generator/1.0.0/python'
@ -60,7 +59,6 @@ class ConfigurationTests(unittest.TestCase):
mock_request.assert_called_with(
'DELETE',
'https://localhost:8080/v1/pet/123456789',
query_params=None,
headers={'User-Agent': 'OpenAPI-Generator/1.0.0/python'},
fields=None,
body=None,

View File

@ -13,18 +13,17 @@ from email.mime import nonmultipart
import io
import sys
import unittest
import json
import typing
from unittest.mock import patch
import urllib3
from urllib3._collections import HTTPHeaderDict
import petstore_api
from petstore_api import api_client, schemas, exceptions
from petstore_api.api.fake_api import FakeApi # noqa: E501
from petstore_api.rest import RESTClientObject
from . import ApiTestMixin
class MIMEFormdata(nonmultipart.MIMENonMultipart):
def __init__(self, keyname, *args, **kwargs):
@ -33,108 +32,10 @@ class MIMEFormdata(nonmultipart.MIMENonMultipart):
"Content-Disposition", "form-data; name=\"%s\"" % keyname)
class TestFakeApi(unittest.TestCase):
class TestFakeApi(ApiTestMixin):
"""FakeApi unit test stubs"""
json_content_type = 'application/json'
configuration = petstore_api.Configuration()
api = FakeApi(api_client=api_client.ApiClient(configuration=configuration))
user_agent = 'OpenAPI-Generator/1.0.0/python'
@staticmethod
def headers_for_content_type(content_type: str) -> dict[str, str]:
return {'content-type': content_type}
@classmethod
def __response(
cls,
body: typing.Union[str, bytes],
status: int = 200,
content_type: str = json_content_type,
headers: typing.Optional[dict[str, str]] = None,
preload_content: bool = True
) -> urllib3.HTTPResponse:
if headers is None:
headers = {}
headers.update(cls.headers_for_content_type(content_type))
return urllib3.HTTPResponse(
body,
headers=headers,
status=status,
preload_content=preload_content
)
@staticmethod
def __json_bytes(in_data: typing.Any) -> bytes:
return json.dumps(in_data, separators=(",", ":"), ensure_ascii=False).encode('utf-8')
@classmethod
def __assert_request_called_with(
cls,
mock_request,
url: str,
method: str = 'POST',
body: typing.Optional[bytes] = None,
content_type: typing.Optional[str] = 'application/json',
fields: typing.Optional[tuple[api_client.RequestField, ...]] = None,
accept_content_type: str = 'application/json',
stream: bool = False,
query_params: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None
):
headers = {
'Accept': accept_content_type,
'User-Agent': cls.user_agent
}
if accept_content_type:
headers['Accept'] = accept_content_type
if content_type:
headers['Content-Type'] = content_type
kwargs = dict(
headers=HTTPHeaderDict(headers),
query_params=query_params,
fields=fields,
stream=stream,
timeout=None,
)
if method != 'GET':
kwargs['body'] = body
mock_request.assert_called_with(
method,
url,
**kwargs
)
@classmethod
def __assert_pool_manager_request_called_with(
cls,
mock_request,
url: str,
method: str = 'POST',
body: typing.Optional[bytes] = None,
content_type: typing.Optional[str] = 'application/json',
accept_content_type: typing.Optional[str] = 'application/json',
stream: bool = False,
query_params: typing.Optional[typing.Tuple[typing.Tuple[str, str], ...]] = None,
):
headers = {
'User-Agent': cls.user_agent
}
if accept_content_type:
headers['Accept'] = accept_content_type
if content_type:
headers['Content-Type'] = content_type
kwargs = dict(
headers=HTTPHeaderDict(headers),
fields=query_params,
preload_content=not stream,
timeout=None,
)
if content_type and method != 'GET':
kwargs['body'] = body
mock_request.assert_called_with(
method,
url,
**kwargs
)
def test_array_model(self):
from petstore_api.model import animal_farm, animal
@ -142,17 +43,17 @@ class TestFakeApi(unittest.TestCase):
# serialization + deserialization works
with patch.object(RESTClientObject, 'request') as mock_request:
json_data = [{"className": "Cat", "color": "black"}]
mock_request.return_value = self.__response(
self.__json_bytes(json_data)
mock_request.return_value = self.response(
self.json_bytes(json_data)
)
cat = animal.Animal(className="Cat", color="black")
body = animal_farm.AnimalFarm([cat])
api_response = self.api.array_model(body=body)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/refs/arraymodel',
body=self.__json_bytes(json_data)
body=self.json_bytes(json_data)
)
assert isinstance(api_response.body, animal_farm.AnimalFarm)
@ -172,15 +73,15 @@ class TestFakeApi(unittest.TestCase):
value = [string_enum.StringEnum("placed")]
body = array_of_enums.ArrayOfEnums(value)
value_simple = ["placed"]
mock_request.return_value = self.__response(
self.__json_bytes(value_simple)
mock_request.return_value = self.response(
self.json_bytes(value_simple)
)
api_response = self.api.array_of_enums(body=body)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/refs/array-of-enums',
body=self.__json_bytes(value_simple)
body=self.json_bytes(value_simple)
)
assert isinstance(api_response.body, array_of_enums.ArrayOfEnums)
@ -193,15 +94,15 @@ class TestFakeApi(unittest.TestCase):
with patch.object(RESTClientObject, 'request') as mock_request:
value = 10.0
body = number_with_validations.NumberWithValidations(value)
mock_request.return_value = self.__response(
self.__json_bytes(value)
mock_request.return_value = self.response(
self.json_bytes(value)
)
api_response = self.api.number_with_validations(body=body)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/refs/number',
body=self.__json_bytes(value)
body=self.json_bytes(value)
)
assert isinstance(api_response.body, number_with_validations.NumberWithValidations)
@ -225,15 +126,15 @@ class TestFakeApi(unittest.TestCase):
]
for (body, value_simple) in cast_to_simple_value:
with patch.object(RESTClientObject, 'request') as mock_request:
mock_request.return_value = self.__response(
self.__json_bytes(value_simple)
mock_request.return_value = self.response(
self.json_bytes(value_simple)
)
api_response = self.api.composed_one_of_different_types(body=body)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/refs/composed_one_of_number_with_validations',
body=self.__json_bytes(value_simple)
body=self.json_bytes(value_simple)
)
assert isinstance(api_response.body, composed_one_of_different_types.ComposedOneOfDifferentTypes)
@ -242,15 +143,15 @@ class TestFakeApi(unittest.TestCase):
# inputting the uncast values into the endpoint also works
for (body, value_simple) in cast_to_simple_value:
with patch.object(RESTClientObject, 'request') as mock_request:
mock_request.return_value = self.__response(
self.__json_bytes(value_simple)
mock_request.return_value = self.response(
self.json_bytes(value_simple)
)
api_response = self.api.composed_one_of_different_types(body=value_simple)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/refs/composed_one_of_number_with_validations',
body=self.__json_bytes(value_simple)
body=self.json_bytes(value_simple)
)
assert isinstance(api_response.body, composed_one_of_different_types.ComposedOneOfDifferentTypes)
@ -261,15 +162,15 @@ class TestFakeApi(unittest.TestCase):
with patch.object(RESTClientObject, 'request') as mock_request:
body = "blah"
value_simple = body
mock_request.return_value = self.__response(
self.__json_bytes(value_simple)
mock_request.return_value = self.response(
self.json_bytes(value_simple)
)
api_response = self.api.string(body=body)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/refs/string',
body=self.__json_bytes(value_simple)
body=self.json_bytes(value_simple)
)
assert isinstance(api_response.body, str)
@ -281,15 +182,15 @@ class TestFakeApi(unittest.TestCase):
with patch.object(RESTClientObject, 'request') as mock_request:
value = "placed"
body = string_enum.StringEnum(value)
mock_request.return_value = self.__response(
self.__json_bytes(value)
mock_request.return_value = self.response(
self.json_bytes(value)
)
api_response = self.api.string_enum(body=body)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/refs/enum',
body=self.__json_bytes(value)
body=self.json_bytes(value)
)
assert isinstance(api_response.body, string_enum.StringEnum)
@ -301,15 +202,15 @@ class TestFakeApi(unittest.TestCase):
with patch.object(RESTClientObject, 'request') as mock_request:
body = Mammal(className="BasquePig")
value_simple = dict(className='BasquePig')
mock_request.return_value = self.__response(
self.__json_bytes(value_simple)
mock_request.return_value = self.response(
self.json_bytes(value_simple)
)
api_response = self.api.mammal(body=body)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/refs/mammal',
body=self.__json_bytes(value_simple)
body=self.json_bytes(value_simple)
)
assert isinstance(api_response.body, Mammal)
@ -333,6 +234,36 @@ class TestFakeApi(unittest.TestCase):
with self.assertRaises(petstore_api.ApiValueError):
self.api.body_with_query_params(body=schemas.unset, query_params=dict(query=schemas.unset))
def test_body_with_query_params(self):
from petstore_api.model import user
with patch.object(RESTClientObject, 'request') as mock_request:
value_simple = dict(
id=1,
username='first last',
firstName='first',
lastName='last'
)
body = user.User(**value_simple)
mock_request.return_value = self.response(
b''
)
api_response = self.api.body_with_query_params(
body=body,
query_params=dict(query='hi there')
)
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/body-with-query-params?query=hi%20there',
method='PUT',
body=self.json_bytes(value_simple),
accept_content_type=None
)
assert isinstance(api_response.body, api_client.Unset)
assert api_response.response.status == 200
def test_upload_download_file_tx_bytes_and_file(self):
"""Test case for upload_download_file
uploads a file and downloads a file using application/octet-stream # noqa: E501
@ -346,7 +277,7 @@ class TestFakeApi(unittest.TestCase):
with open(file_path1, "rb") as some_file:
file_bytes = some_file.read()
file1 = open(file_path1, "rb")
mock_response = self.__response(
mock_response = self.response(
file_bytes,
content_type='application/octet-stream'
)
@ -354,7 +285,7 @@ class TestFakeApi(unittest.TestCase):
with patch.object(RESTClientObject, 'request') as mock_request:
mock_request.return_value = mock_response
api_response = self.api.upload_download_file(body=file1)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/uploadDownloadFile',
body=file_bytes,
@ -374,7 +305,7 @@ class TestFakeApi(unittest.TestCase):
with patch.object(RESTClientObject, 'request') as mock_request:
mock_request.return_value = mock_response
api_response = self.api.upload_download_file(body=file_bytes)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/uploadDownloadFile',
body=file_bytes,
@ -411,7 +342,7 @@ class TestFakeApi(unittest.TestCase):
streamable_body = StreamableBody(file1)
mock_response = self.__response(
mock_response = self.response(
streamable_body,
content_type='application/octet-stream',
preload_content=False
@ -419,7 +350,7 @@ class TestFakeApi(unittest.TestCase):
with patch.object(RESTClientObject, 'request') as mock_request:
mock_request.return_value = mock_response
api_response = self.api.upload_download_file(body=file_bytes, stream=True)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/uploadDownloadFile',
body=file_bytes,
@ -443,7 +374,7 @@ class TestFakeApi(unittest.TestCase):
when streaming is used and the response contains the content disposition header with a filename
that filename is used when saving the file locally
"""
mock_response = self.__response(
mock_response = self.response(
streamable_body,
content_type='application/octet-stream',
headers={'content-disposition': f'attachment; filename="{saved_file_name}"'},
@ -452,7 +383,7 @@ class TestFakeApi(unittest.TestCase):
with patch.object(RESTClientObject, 'request') as mock_request:
mock_request.return_value = mock_response
api_response = self.api.upload_download_file(body=file_bytes, stream=True)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/uploadDownloadFile',
body=file_bytes,
@ -489,11 +420,11 @@ class TestFakeApi(unittest.TestCase):
}
try:
with patch.object(RESTClientObject, 'request') as mock_request:
mock_request.return_value = self.__response(
self.__json_bytes(response_json)
mock_request.return_value = self.response(
self.json_bytes(response_json)
)
api_response = self.api.upload_file(body={'file': file1})
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/uploadFile',
fields=(
@ -514,11 +445,11 @@ class TestFakeApi(unittest.TestCase):
# sending just bytes works also
with patch.object(RESTClientObject, 'request') as mock_request:
mock_request.return_value = self.__response(
self.__json_bytes(response_json)
mock_request.return_value = self.response(
self.json_bytes(response_json)
)
api_response = self.api.upload_file(body={'file': file_bytes})
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/uploadFile',
fields=(
@ -567,11 +498,11 @@ class TestFakeApi(unittest.TestCase):
}
try:
with patch.object(RESTClientObject, 'request') as mock_request:
mock_request.return_value = self.__response(
self.__json_bytes(response_json)
mock_request.return_value = self.response(
self.json_bytes(response_json)
)
api_response = self.api.upload_files(body={'files': [file1, file1]})
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/uploadFiles',
fields=(
@ -598,11 +529,11 @@ class TestFakeApi(unittest.TestCase):
# sending just bytes works also
with patch.object(RESTClientObject, 'request') as mock_request:
mock_request.return_value = self.__response(
self.__json_bytes(response_json)
mock_request.return_value = self.response(
self.json_bytes(response_json)
)
api_response = self.api.upload_files(body={'files': [file_bytes, file_bytes]})
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/uploadFiles',
fields=(
@ -640,11 +571,11 @@ class TestFakeApi(unittest.TestCase):
testing composed schemas at inline locations # noqa: E501
"""
single_char_str = 'a'
json_bytes = self.__json_bytes(single_char_str)
json_bytes = self.json_bytes(single_char_str)
# tx and rx json with composition at root level of schema for request + response body
content_type = 'application/json'
mock_request.return_value = self.__response(
mock_request.return_value = self.response(
json_bytes
)
api_response = self.api.inline_composition(
@ -655,12 +586,11 @@ class TestFakeApi(unittest.TestCase):
},
accept_content_types=(content_type,)
)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/inlineComposition/',
'http://petstore.swagger.io:80/v2/fake/inlineComposition/?compositionAtRoot=a&someProp=a',
accept_content_type=content_type,
content_type=content_type,
query_params=(('compositionAtRoot', 'a'), ('someProp', 'a')),
body=json_bytes
)
self.assertEqual(api_response.body, single_char_str)
@ -669,7 +599,7 @@ class TestFakeApi(unittest.TestCase):
# tx and rx json with composition at property level of schema for request + response body
content_type = 'multipart/form-data'
multipart_response = self.__encode_multipart_formdata(fields={'someProp': single_char_str})
mock_request.return_value = self.__response(
mock_request.return_value = self.response(
bytes(multipart_response),
content_type=multipart_response.get_content_type()
)
@ -682,12 +612,11 @@ class TestFakeApi(unittest.TestCase):
content_type=content_type,
accept_content_types=(content_type,)
)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/inlineComposition/',
'http://petstore.swagger.io:80/v2/fake/inlineComposition/?compositionAtRoot=a&someProp=a',
accept_content_type=content_type,
content_type=content_type,
query_params=(('compositionAtRoot', 'a'), ('someProp', 'a')),
fields=(
api_client.RequestField(
name='someProp',
@ -707,7 +636,7 @@ class TestFakeApi(unittest.TestCase):
values[invalid_index] = invalid_value
with self.assertRaises(exceptions.ApiValueError):
multipart_response = self.__encode_multipart_formdata(fields={'someProp': values[0]})
mock_request.return_value = self.__response(
mock_request.return_value = self.response(
bytes(multipart_response),
content_type=multipart_response.get_content_type()
)
@ -726,16 +655,16 @@ class TestFakeApi(unittest.TestCase):
with patch.object(RESTClientObject, 'request') as mock_request:
body = None
content_type_with_charset = 'application/json; charset=utf-8'
mock_request.return_value = self.__response(
self.__json_bytes(body),
mock_request.return_value = self.response(
self.json_bytes(body),
content_type=content_type_with_charset
)
api_response = self.api.json_with_charset(body=body)
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/jsonWithCharset',
body=self.__json_bytes(body),
body=self.json_bytes(body),
content_type=content_type_with_charset,
accept_content_type=content_type_with_charset
)
@ -749,12 +678,12 @@ class TestFakeApi(unittest.TestCase):
with patch.object(RESTClientObject, 'request') as mock_request:
body = None
content_type = 'application/json'
mock_request.return_value = self.__response(
self.__json_bytes(body),
mock_request.return_value = self.response(
self.json_bytes(body),
)
api_response = self.api.response_without_schema()
self.__assert_request_called_with(
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/responseWithoutSchema',
method='GET',
@ -766,7 +695,7 @@ class TestFakeApi(unittest.TestCase):
with patch.object(RESTClientObject, 'request') as mock_request:
mock_request.return_value = self.__response(
mock_request.return_value = self.response(
'blah',
content_type='text/plain'
)
@ -779,12 +708,12 @@ class TestFakeApi(unittest.TestCase):
with patch.object(urllib3.PoolManager, 'request') as mock_request:
body = None
mock_request.return_value = self.__response(
self.__json_bytes(body),
mock_request.return_value = self.response(
self.json_bytes(body),
)
api_response = self.api.delete_coffee(path_params=dict(id='1'))
self.__assert_pool_manager_request_called_with(
self.assert_pool_manager_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/fake/deleteCoffee/1',
method='DELETE',

View File

@ -11,8 +11,7 @@
import unittest
import collections
from petstore_api import api_client
from petstore_api import schemas
from petstore_api import api_client, exceptions, schemas
ParamTestCase = collections.namedtuple('ParamTestCase', 'payload expected_serialization explode', defaults=[False])
@ -25,6 +24,11 @@ class TestParameter(unittest.TestCase):
api_client.ParameterInType.HEADER: api_client.HeaderParameter,
}
invalid_inputs = (
True,
False
)
def test_throws_exception_when_schema_and_content_omitted(self):
with self.assertRaises(ValueError):
api_client.QueryParameter(
@ -95,68 +99,52 @@ class TestParameter(unittest.TestCase):
test_cases = (
ParamTestCase(
None,
()
dict(color='')
),
ParamTestCase(
1,
(('color', '1'),)
dict(color='?color=1')
),
ParamTestCase(
3.14,
(('color', '3.14'),)
dict(color='?color=3.14')
),
ParamTestCase(
'blue',
(('color', 'blue'),)
dict(color='?color=blue')
),
ParamTestCase(
'hello world',
(('color', 'hello%20world'),)
dict(color='?color=hello%20world')
),
ParamTestCase(
'',
(('color', ''),)
),
ParamTestCase(
True,
(('color', 'true'),)
),
ParamTestCase(
False,
(('color', 'false'),)
dict(color='?color=')
),
ParamTestCase(
[],
()
dict(color='')
),
ParamTestCase(
['blue', 'black', 'brown'],
(('color', 'blue,black,brown'),)
dict(color='?color=blue,black,brown')
),
ParamTestCase(
['blue', 'black', 'brown'],
(
('color', 'blue'),
('color', 'black'),
('color', 'brown'),
),
dict(color='?color=blue&color=black&color=brown'),
explode=True
),
ParamTestCase(
{},
()
dict(color='')
),
ParamTestCase(
dict(R=100, G=200, B=150),
(('color', 'R,100,G,200,B,150'),)
dict(color='?color=R,100,G,200,B,150')
),
ParamTestCase(
dict(R=100, G=200, B=150),
(
('R', '100'),
('G', '200'),
('B', '150'),
),
dict(color='?R=100&G=200&B=150'),
explode=True
),
)
@ -170,73 +158,68 @@ class TestParameter(unittest.TestCase):
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
for explode in (True, False):
parameter = api_client.QueryParameter(
name=name,
style=api_client.ParameterStyle.FORM,
schema=schemas.AnyTypeSchema,
explode=explode,
)
parameter.serialize(invalid_input)
def test_cookie_style_form_serialization(self):
name = 'color'
test_cases = (
ParamTestCase(
None,
()
dict(color='')
),
ParamTestCase(
1,
(('color', '1'),)
dict(color='color=1')
),
ParamTestCase(
3.14,
(('color', '3.14'),)
dict(color='color=3.14')
),
ParamTestCase(
'blue',
(('color', 'blue'),)
dict(color='color=blue')
),
ParamTestCase(
'hello world',
(('color', 'hello%20world'),)
dict(color='color=hello world')
),
ParamTestCase(
'',
(('color', ''),)
),
ParamTestCase(
True,
(('color', 'true'),)
),
ParamTestCase(
False,
(('color', 'false'),)
dict(color='color=')
),
ParamTestCase(
[],
()
dict(color='')
),
ParamTestCase(
['blue', 'black', 'brown'],
(('color', 'blue,black,brown'),)
dict(color='color=blue,black,brown')
),
ParamTestCase(
['blue', 'black', 'brown'],
(
('color', 'blue'),
('color', 'black'),
('color', 'brown'),
),
dict(color='color=blue&color=black&color=brown'),
explode=True
),
ParamTestCase(
{},
()
dict(color='')
),
ParamTestCase(
dict(R=100, G=200, B=150),
(('color', 'R,100,G,200,B,150'),)
dict(color='color=R,100,G,200,B,150')
),
ParamTestCase(
dict(R=100, G=200, B=150),
(
('R', '100'),
('G', '200'),
('B', '150'),
),
dict(color='R=100&G=200&B=150'),
explode=True
),
)
@ -250,6 +233,17 @@ class TestParameter(unittest.TestCase):
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
for explode in (True, False):
parameter = api_client.CookieParameter(
name=name,
style=api_client.ParameterStyle.FORM,
schema=schemas.AnyTypeSchema,
explode=explode,
)
parameter.serialize(invalid_input)
def test_style_simple_in_path_serialization(self):
name = 'color'
test_cases = (
@ -277,14 +271,6 @@ class TestParameter(unittest.TestCase):
'',
dict(color='')
),
ParamTestCase(
True,
dict(color='true')
),
ParamTestCase(
False,
dict(color='false')
),
ParamTestCase(
[],
dict(color='')
@ -322,12 +308,23 @@ class TestParameter(unittest.TestCase):
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
for explode in (True, False):
parameter = api_client.PathParameter(
name=name,
style=api_client.ParameterStyle.SIMPLE,
schema=schemas.AnyTypeSchema,
explode=explode,
)
parameter.serialize(invalid_input)
def test_style_simple_in_header_serialization(self):
name = 'color'
test_cases = (
ParamTestCase(
None,
{}
dict(color='')
),
ParamTestCase(
1,
@ -343,23 +340,15 @@ class TestParameter(unittest.TestCase):
),
ParamTestCase(
'hello world',
dict(color='hello%20world')
dict(color='hello world')
),
ParamTestCase(
'',
dict(color='')
),
ParamTestCase(
True,
dict(color='true')
),
ParamTestCase(
False,
dict(color='false')
),
ParamTestCase(
[],
{}
dict(color='')
),
ParamTestCase(
['blue', 'black', 'brown'],
@ -372,7 +361,7 @@ class TestParameter(unittest.TestCase):
),
ParamTestCase(
{},
{}
dict(color='')
),
ParamTestCase(
dict(R=100, G=200, B=150),
@ -385,7 +374,6 @@ class TestParameter(unittest.TestCase):
),
)
for test_case in test_cases:
print(test_case.payload)
parameter = api_client.HeaderParameter(
name=name,
style=api_client.ParameterStyle.SIMPLE,
@ -395,6 +383,17 @@ class TestParameter(unittest.TestCase):
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
for explode in (True, False):
parameter = api_client.HeaderParameter(
name=name,
style=api_client.ParameterStyle.SIMPLE,
schema=schemas.AnyTypeSchema,
explode=explode,
)
parameter.serialize(invalid_input)
def test_style_label_in_path_serialization(self):
name = 'color'
test_cases = (
@ -422,14 +421,6 @@ class TestParameter(unittest.TestCase):
'',
dict(color='.')
),
ParamTestCase(
True,
dict(color='.true')
),
ParamTestCase(
False,
dict(color='.false')
),
ParamTestCase(
[],
dict(color='')
@ -467,6 +458,17 @@ class TestParameter(unittest.TestCase):
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
for explode in (True, False):
parameter = api_client.PathParameter(
name=name,
style=api_client.ParameterStyle.LABEL,
schema=schemas.AnyTypeSchema,
explode=explode,
)
parameter.serialize(invalid_input)
def test_style_matrix_in_path_serialization(self):
name = 'color'
test_cases = (
@ -494,14 +496,6 @@ class TestParameter(unittest.TestCase):
'',
dict(color=';color')
),
ParamTestCase(
True,
dict(color=';color=true')
),
ParamTestCase(
False,
dict(color=';color=false')
),
ParamTestCase(
[],
dict(color='')
@ -539,66 +533,59 @@ class TestParameter(unittest.TestCase):
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
for explode in (True, False):
parameter = api_client.PathParameter(
name=name,
style=api_client.ParameterStyle.MATRIX,
schema=schemas.AnyTypeSchema,
explode=explode,
)
parameter.serialize(invalid_input)
def test_style_space_delimited_serialization(self):
name = 'color'
test_cases = (
ParamTestCase(
None,
()
dict(color='')
),
ParamTestCase(
1,
(('color', '1'),)
dict(color='1')
),
ParamTestCase(
3.14,
(('color', '3.14'),)
dict(color='3.14')
),
ParamTestCase(
'blue',
(('color', 'blue'),)
dict(color='blue')
),
ParamTestCase(
'hello world',
(('color', 'hello%20world'),)
dict(color='hello%20world')
),
ParamTestCase(
'',
(('color', ''),)
),
ParamTestCase(
True,
(('color', 'true'),)
),
ParamTestCase(
False,
(('color', 'false'),)
dict(color='')
),
ParamTestCase(
[],
()
dict(color='')
),
ParamTestCase(
['blue', 'black', 'brown'],
(('color', 'blue%20black%20brown'),)
),
ParamTestCase(
['blue', 'black', 'brown'],
(('color', 'color=blue%20color=black%20color=brown'),),
explode=True
dict(color='blue%20black%20brown')
),
ParamTestCase(
{},
()
dict(color='')
),
ParamTestCase(
dict(R=100, G=200, B=150),
(('color', 'R%20100%20G%20200%20B%20150'),)
),
ParamTestCase(
dict(R=100, G=200, B=150),
(('color', 'R=100%20G=200%20B=150'),),
explode=True
dict(color='R%20100%20G%20200%20B%20150')
),
)
for test_case in test_cases:
@ -611,66 +598,59 @@ class TestParameter(unittest.TestCase):
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
for explode in (True, False):
parameter = api_client.QueryParameter(
name=name,
style=api_client.ParameterStyle.SPACE_DELIMITED,
schema=schemas.AnyTypeSchema,
explode=explode,
)
parameter.serialize(invalid_input)
def test_style_pipe_delimited_serialization(self):
name = 'color'
test_cases = (
ParamTestCase(
None,
()
dict(color='')
),
ParamTestCase(
1,
(('color', '1'),)
dict(color='1')
),
ParamTestCase(
3.14,
(('color', '3.14'),)
dict(color='3.14')
),
ParamTestCase(
'blue',
(('color', 'blue'),)
dict(color='blue')
),
ParamTestCase(
'hello world',
(('color', 'hello%20world'),)
dict(color='hello%20world')
),
ParamTestCase(
'',
(('color', ''),)
),
ParamTestCase(
True,
(('color', 'true'),)
),
ParamTestCase(
False,
(('color', 'false'),)
dict(color='')
),
ParamTestCase(
[],
()
dict(color='')
),
ParamTestCase(
['blue', 'black', 'brown'],
(('color', 'blue|black|brown'),)
),
ParamTestCase(
['blue', 'black', 'brown'],
(('color', 'color=blue|color=black|color=brown'),),
explode=True
dict(color='blue|black|brown')
),
ParamTestCase(
{},
()
dict(color='')
),
ParamTestCase(
dict(R=100, G=200, B=150),
(('color', 'R|100|G|200|B|150'),)
),
ParamTestCase(
dict(R=100, G=200, B=150),
(('color', 'R=100|G=200|B=150'),),
explode=True
dict(color='R|100|G|200|B|150')
),
)
for test_case in test_cases:
@ -683,6 +663,17 @@ class TestParameter(unittest.TestCase):
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
for explode in (True, False):
parameter = api_client.QueryParameter(
name=name,
style=api_client.ParameterStyle.PIPE_DELIMITED,
schema=schemas.AnyTypeSchema,
explode=explode,
)
parameter.serialize(invalid_input)
def test_path_params_no_style(self):
name = 'color'
test_cases = (
@ -710,14 +701,6 @@ class TestParameter(unittest.TestCase):
'',
dict(color='')
),
ParamTestCase(
True,
dict(color='true')
),
ParamTestCase(
False,
dict(color='false')
),
ParamTestCase(
[],
dict(color='')
@ -743,12 +726,20 @@ class TestParameter(unittest.TestCase):
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
parameter = api_client.PathParameter(
name=name,
schema=schemas.AnyTypeSchema,
)
parameter.serialize(invalid_input)
def test_header_params_no_style(self):
name = 'color'
test_cases = (
ParamTestCase(
None,
{}
dict(color='')
),
ParamTestCase(
1,
@ -764,23 +755,15 @@ class TestParameter(unittest.TestCase):
),
ParamTestCase(
'hello world',
dict(color='hello%20world')
dict(color='hello world')
),
ParamTestCase(
'',
dict(color='')
),
ParamTestCase(
True,
dict(color='true')
),
ParamTestCase(
False,
dict(color='false')
),
ParamTestCase(
[],
{}
dict(color='')
),
ParamTestCase(
['blue', 'black', 'brown'],
@ -788,7 +771,7 @@ class TestParameter(unittest.TestCase):
),
ParamTestCase(
{},
{}
dict(color='')
),
ParamTestCase(
dict(R=100, G=200, B=150),
@ -803,67 +786,136 @@ class TestParameter(unittest.TestCase):
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
def test_query_or_cookie_params_no_style(self):
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
parameter = api_client.HeaderParameter(
name=name,
schema=schemas.AnyTypeSchema,
)
parameter.serialize(invalid_input)
def test_query_params_no_style(self):
name = 'color'
test_cases = (
ParamTestCase(
None,
()
dict(color='')
),
ParamTestCase(
1,
(('color', '1'),)
dict(color='?color=1')
),
ParamTestCase(
3.14,
(('color', '3.14'),)
dict(color='?color=3.14')
),
ParamTestCase(
'blue',
(('color', 'blue'),)
dict(color='?color=blue')
),
ParamTestCase(
'hello world',
(('color', 'hello%20world'),)
dict(color='?color=hello%20world')
),
ParamTestCase(
'',
(('color', ''),)
),
ParamTestCase(
True,
(('color', 'true'),)
),
ParamTestCase(
False,
(('color', 'false'),)
dict(color='?color=')
),
ParamTestCase(
[],
()
dict(color='')
),
ParamTestCase(
['blue', 'black', 'brown'],
(('color', 'blue,black,brown'),)
dict(color='?color=blue&color=black&color=brown')
),
ParamTestCase(
{},
()
dict(color='')
),
ParamTestCase(
dict(R=100, G=200, B=150),
(('color', 'R,100,G,200,B,150'),)
dict(color='?R=100&G=200&B=150')
),
)
for in_type in {api_client.ParameterInType.QUERY, api_client.ParameterInType.COOKIE}:
for test_case in test_cases:
parameter_cls = self.in_type_to_parameter_cls[in_type]
parameter = parameter_cls(
for test_case in test_cases:
parameter = api_client.QueryParameter(
name=name,
schema=schemas.AnyTypeSchema,
)
print(parameter.explode)
print(test_case.payload)
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
parameter = api_client.QueryParameter(
name=name,
schema=schemas.AnyTypeSchema,
)
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
parameter.serialize(invalid_input)
def test_cookie_params_no_style(self):
name = 'color'
test_cases = (
ParamTestCase(
None,
dict(color='')
),
ParamTestCase(
1,
dict(color='color=1')
),
ParamTestCase(
3.14,
dict(color='color=3.14')
),
ParamTestCase(
'blue',
dict(color='color=blue')
),
ParamTestCase(
'hello world',
dict(color='color=hello world')
),
ParamTestCase(
'',
dict(color='color=')
),
ParamTestCase(
[],
dict(color='')
),
ParamTestCase(
['blue', 'black', 'brown'],
dict(color='color=blue&color=black&color=brown')
),
ParamTestCase(
{},
dict(color='')
),
ParamTestCase(
dict(R=100, G=200, B=150),
dict(color='R=100&G=200&B=150')
),
)
for test_case in test_cases:
parameter = api_client.CookieParameter(
name=name,
schema=schemas.AnyTypeSchema,
)
print(test_case.payload)
serialization = parameter.serialize(test_case.payload)
self.assertEqual(serialization, test_case.expected_serialization)
with self.assertRaises(exceptions.ApiValueError):
for invalid_input in self.invalid_inputs:
parameter = api_client.CookieParameter(
name=name,
schema=schemas.AnyTypeSchema,
)
parameter.serialize(invalid_input)
def test_checks_content_lengths(self):
with self.assertRaises(ValueError):

View File

@ -123,13 +123,13 @@ class TestParameter(unittest.TestCase):
serialization = request_body.serialize(payload, content_type)
self.assertEqual(
serialization,
{'fields': (('some_str', 'a'), ('some_int', '1'), ('some_float', '3.14'))}
dict(body='?some_str=a&some_int=1&some_float=3.14')
)
serialization = request_body.serialize({}, content_type)
self.assertEqual(
serialization,
{}
dict(body='')
)
invalid_payloads = [

View File

@ -0,0 +1,108 @@
# coding: utf-8
"""
OpenAPI Petstore
This spec is mainly for testing Petstore server and contains fake endpoints, models. Please do not use this for any other purpose. Special characters: \" \\ # noqa: E501
The version of the OpenAPI document: 1.0.0
Generated by: https://openapi-generator.tech
"""
import unittest
from unittest.mock import patch
import petstore_api
from petstore_api import api_client
from petstore_api.api.user_api import UserApi
from petstore_api.rest import RESTClientObject
from . import ApiTestMixin
class TestUserApi(ApiTestMixin):
"""UserApi unit test stubs"""
configuration = petstore_api.Configuration()
api = UserApi(api_client=api_client.ApiClient(configuration=configuration))
def test_create_user(self):
"""Test case for create_user
Create user # noqa: E501
"""
pass
def test_create_users_with_array_input(self):
"""Test case for create_users_with_array_input
Creates list of users with given input array # noqa: E501
"""
pass
def test_create_users_with_list_input(self):
"""Test case for create_users_with_list_input
Creates list of users with given input array # noqa: E501
"""
pass
def test_delete_user(self):
"""Test case for delete_user
Delete user # noqa: E501
"""
pass
def test_get_user_by_name(self):
from petstore_api.model import user
# serialization + deserialization works
with patch.object(RESTClientObject, 'request') as mock_request:
value_simple = dict(
id=1,
username='first last',
firstName='first',
lastName='last'
)
body = user.User(**value_simple)
mock_request.return_value = self.response(
self.json_bytes(value_simple)
)
api_response = self.api.get_user_by_name(
path_params=dict(username='first last')
)
self.assert_request_called_with(
mock_request,
'http://petstore.swagger.io:80/v2/user/first%20last',
method='GET',
accept_content_type='application/xml, application/json',
content_type=None
)
assert isinstance(api_response.body, user.User)
assert api_response.body == body
def test_login_user(self):
"""Test case for login_user
Logs user into the system # noqa: E501
"""
pass
def test_logout_user(self):
"""Test case for logout_user
Logs out current logged in user session # noqa: E501
"""
pass
def test_update_user(self):
"""Test case for update_user
Updated user # noqa: E501
"""
pass
if __name__ == '__main__':
unittest.main()