LukeParke/connexion
1
0
Fork 0
mirror of https://github.com/LukeHagar/connexion.git synced 2025-12-10 04:19:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Drop aiohttp support (#1491)

Browse Source
This commit is contained in:
Robbe Sneyders
2022-03-22 22:26:41 +01:00
committed by GitHub
parent f236c6883d
commit ca70b16ff5
43 changed files with 8 additions and 3190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
ARCHITECTURE.rst
View File

@@ -11,8 +11,7 @@ This document describes the high-level architecture of Connexion.
Apps
----
A Connexion ``App`` or application wraps a specific framework application (currently Flask or
AioHttp) and exposes a standardized interface for users to create and configure their Connexion
A Connexion ``App`` or application wraps a specific framework application (currently Flask) and exposes a standardized interface for users to create and configure their Connexion
application.
While a Connexion app implements the WSGI interface, it only acts ass a pass-through and doesn't

8
connexion/__init__.py
View File

@@ -38,13 +38,5 @@ except ImportError as e: # pragma: no cover
App = FlaskApp
Api = FlaskApi
try:
from .apis.aiohttp_api import AioHttpApi
from .apps.aiohttp_app import AioHttpApp
except ImportError as e: # pragma: no cover
_aiohttp_not_installed_error = not_installed_error(e)
AioHttpApi = _aiohttp_not_installed_error
AioHttpApp = _aiohttp_not_installed_error
# This version is replaced during release process.
__version__ = '2020.0.dev1'

36
connexion/apis/abstract.py
View File

@@ -7,7 +7,6 @@ import logging
import pathlib
import sys
import typing as t
import warnings
from enum import Enum
from ..decorators.produces import NoContent
@@ -19,7 +18,6 @@ from ..operations import make_operation
from ..options import ConnexionOptions
from ..resolver import Resolver
from ..spec import Specification
from ..utils import is_json_mimetype
MODULE_PATH = pathlib.Path(__file__).absolute().parent.parent
SWAGGER_UI_URL = 'ui'
@@ -256,7 +254,6 @@ class AbstractAPI(metaclass=AbstractAPIMeta):
"""
This method converts a handler response to a framework response.
This method should just retrieve response from handler then call `cls._get_response`.
It is mainly here to handle AioHttp async handler.
:param response: A response to cast (tuple, framework response, etc).
:param mimetype: The response mimetype.
:type mimetype: Union[None, str]
@@ -348,18 +345,7 @@ class AbstractAPI(metaclass=AbstractAPIMeta):
def get_connexion_response(cls, response, mimetype=None):
""" Cast framework dependent response to ConnexionResponse used for schema validation """
if isinstance(response, ConnexionResponse):
# If body in ConnexionResponse is not byte, it may not pass schema validation.
# In this case, rebuild response with aiohttp to have consistency
if response.body is None or isinstance(response.body, bytes):
return response
else:
response = cls._build_response(
data=response.body,
mimetype=mimetype,
content_type=response.content_type,
headers=response.headers,
status_code=response.status_code
)
if not cls._is_framework_response(response):
response = cls._response_from_handler(response, mimetype)
@@ -430,27 +416,9 @@ class AbstractAPI(metaclass=AbstractAPIMeta):
return body, status_code, mimetype
@classmethod
@abc.abstractmethod
def _serialize_data(cls, data, mimetype):
# TODO: Harmonize with flask_api. Currently this is the backwards compatible with aiohttp_api._cast_body.
if not isinstance(data, bytes):
if isinstance(mimetype, str) and is_json_mimetype(mimetype):
body = cls.jsonifier.dumps(data)
elif isinstance(data, str):
body = data
else:
warnings.warn(
"Implicit (aiohttp) serialization with str() will change in the next major version. "
"This is triggered because a non-JSON response body is being stringified. "
"This will be replaced by something that is mimetype-specific and may "
"serialize some things as JSON or throw an error instead of silently "
"stringifying unknown response bodies. "
"Please make sure to specify media/mime types in your specs.",
FutureWarning # a Deprecation targeted at application users.
)
body = str(data)
else:
body = data
return body, mimetype
pass
def json_loads(self, data):
return self.jsonifier.loads(data)

447
connexion/apis/aiohttp_api.py
View File

@@ -1,447 +0,0 @@
"""
This module defines an AioHttp Connexion API which implements translations between AioHttp and
Connexion requests / responses.
"""
import asyncio
import logging
import re
import traceback
from contextlib import suppress
from http import HTTPStatus
from urllib.parse import parse_qs
import aiohttp_jinja2
import jinja2
from aiohttp import web
from aiohttp.web_exceptions import HTTPNotFound, HTTPPermanentRedirect
from aiohttp.web_middlewares import normalize_path_middleware
from werkzeug.exceptions import HTTPException as werkzeug_HTTPException
from connexion.apis.abstract import AbstractAPI
from connexion.exceptions import ProblemException
from connexion.handlers import AuthErrorHandler
from connexion.jsonifier import JSONEncoder, Jsonifier
from connexion.lifecycle import ConnexionRequest, ConnexionResponse
from connexion.problem import problem
from connexion.security import AioHttpSecurityHandlerFactory
from connexion.utils import yamldumper
logger = logging.getLogger('connexion.apis.aiohttp_api')
def _generic_problem(http_status: HTTPStatus, exc: Exception = None):
extra = None
if exc is not None:
loop = asyncio.get_event_loop()
if loop.get_debug():
tb = None
with suppress(Exception):
tb = traceback.format_exc()
if tb:
extra = {"traceback": tb}
return problem(
status=http_status.value,
title=http_status.phrase,
detail=http_status.description,
ext=extra,
)
@web.middleware
async def problems_middleware(request, handler):
try:
response = await handler(request)
except ProblemException as exc:
response = problem(status=exc.status, detail=exc.detail, title=exc.title,
type=exc.type, instance=exc.instance, headers=exc.headers, ext=exc.ext)
except (werkzeug_HTTPException, _HttpNotFoundError) as exc:
response = problem(status=exc.code, title=exc.name, detail=exc.description)
except web.HTTPError as exc:
if exc.text == f"{exc.status}: {exc.reason}":
detail = HTTPStatus(exc.status).description
else:
detail = exc.text
response = problem(status=exc.status, title=exc.reason, detail=detail)
except (
web.HTTPException, # eg raised HTTPRedirection or HTTPSuccessful
asyncio.CancelledError, # skipped in default web_protocol
):
# leave this to default handling in aiohttp.web_protocol.RequestHandler.start()
raise
except asyncio.TimeoutError as exc:
# overrides 504 from aiohttp.web_protocol.RequestHandler.start()
logger.debug('Request handler timed out.', exc_info=exc)
response = _generic_problem(HTTPStatus.GATEWAY_TIMEOUT, exc)
except Exception as exc:
# overrides 500 from aiohttp.web_protocol.RequestHandler.start()
logger.exception('Error handling request', exc_info=exc)
response = _generic_problem(HTTPStatus.INTERNAL_SERVER_ERROR, exc)
if isinstance(response, ConnexionResponse):
response = await AioHttpApi.get_response(response)
return response
class AioHttpApi(AbstractAPI):
def __init__(self, *args, **kwargs):
# NOTE we use HTTPPermanentRedirect (308) because
# clients sometimes turn POST requests into GET requests
# on 301, 302, or 303
# see https://tools.ietf.org/html/rfc7538
trailing_slash_redirect = normalize_path_middleware(
append_slash=True,
redirect_class=HTTPPermanentRedirect
)
self.subapp = web.Application(
middlewares=[
problems_middleware,
trailing_slash_redirect
]
)
AbstractAPI.__init__(self, *args, **kwargs)
aiohttp_jinja2.setup(
self.subapp,
loader=jinja2.FileSystemLoader(
str(self.options.openapi_console_ui_from_dir)
)
)
middlewares = self.options.as_dict().get('middlewares', [])
self.subapp.middlewares.extend(middlewares)
@staticmethod
def make_security_handler_factory(pass_context_arg_name):
""" Create default SecurityHandlerFactory to create all security check handlers """
return AioHttpSecurityHandlerFactory(pass_context_arg_name)
def _set_base_path(self, base_path):
AbstractAPI._set_base_path(self, base_path)
self._api_name = AioHttpApi.normalize_string(self.base_path)
@staticmethod
def normalize_string(string):
return re.sub(r'[^a-zA-Z0-9]', '_', string.strip('/'))
def _base_path_for_prefix(self, request):
"""
returns a modified basePath which includes the incoming request's
path prefix.
"""
base_path = self.base_path
if not request.path.startswith(self.base_path):
prefix = request.path.split(self.base_path)[0]
base_path = prefix + base_path
return base_path
def _spec_for_prefix(self, request):
"""
returns a spec with a modified basePath / servers block
which corresponds to the incoming request path.
This is needed when behind a path-altering reverse proxy.
"""
base_path = self._base_path_for_prefix(request)
return self.specification.with_base_path(base_path).raw
def add_openapi_json(self):
"""
Adds openapi json to {base_path}/openapi.json
(or {base_path}/swagger.json for swagger2)
"""
logger.debug('Adding spec json: %s/%s', self.base_path,
self.options.openapi_spec_path)
self.subapp.router.add_route(
'GET',
self.options.openapi_spec_path,
self._get_openapi_json
)
def add_openapi_yaml(self):
"""
Adds openapi json to {base_path}/openapi.json
(or {base_path}/swagger.json for swagger2)
"""
if not self.options.openapi_spec_path.endswith("json"):
return
openapi_spec_path_yaml = \
self.options.openapi_spec_path[:-len("json")] + "yaml"
logger.debug('Adding spec yaml: %s/%s', self.base_path,
openapi_spec_path_yaml)
self.subapp.router.add_route(
'GET',
openapi_spec_path_yaml,
self._get_openapi_yaml
)
async def _get_openapi_json(self, request):
return web.Response(
status=200,
content_type='application/json',
body=self.jsonifier.dumps(self._spec_for_prefix(request))
)
async def _get_openapi_yaml(self, request):
return web.Response(
status=200,
content_type='text/yaml',
body=yamldumper(self._spec_for_prefix(request))
)
def add_swagger_ui(self):
"""
Adds swagger ui to {base_path}/ui/
"""
console_ui_path = self.options.openapi_console_ui_path.strip().rstrip('/')
logger.debug('Adding swagger-ui: %s%s/',
self.base_path,
console_ui_path)
for path in (
console_ui_path + '/',
console_ui_path + '/index.html',
):
self.subapp.router.add_route(
'GET',
path,
self._get_swagger_ui_home
)
if self.options.openapi_console_ui_config is not None:
self.subapp.router.add_route(
'GET',
console_ui_path + '/swagger-ui-config.json',
self._get_swagger_ui_config
)
# we have to add an explicit redirect instead of relying on the
# normalize_path_middleware because we also serve static files
# from this dir (below)
async def redirect(request):
raise web.HTTPMovedPermanently(
location=self.base_path + console_ui_path + '/'
)
self.subapp.router.add_route(
'GET',
console_ui_path,
redirect
)
# this route will match and get a permission error when trying to
# serve index.html, so we add the redirect above.
self.subapp.router.add_static(
console_ui_path,
path=str(self.options.openapi_console_ui_from_dir),
name='swagger_ui_static'
)
@aiohttp_jinja2.template('index.j2')
async def _get_swagger_ui_home(self, req):
base_path = self._base_path_for_prefix(req)
template_variables = {
'openapi_spec_url': (base_path + self.options.openapi_spec_path),
**self.options.openapi_console_ui_index_template_variables,
}
if self.options.openapi_console_ui_config is not None:
template_variables['configUrl'] = 'swagger-ui-config.json'
return template_variables
async def _get_swagger_ui_config(self, req):
return web.Response(
status=200,
content_type='text/json',
body=self.jsonifier.dumps(self.options.openapi_console_ui_config)
)
def add_auth_on_not_found(self, security, security_definitions):
"""
Adds a 404 error handler to authenticate and only expose the 404 status if the security validation pass.
"""
logger.debug('Adding path not found authentication')
not_found_error = AuthErrorHandler(
self, _HttpNotFoundError(),
security=security,
security_definitions=security_definitions
)
endpoint_name = f"{self._api_name}_not_found"
self.subapp.router.add_route(
'*',
'/{not_found_path}',
not_found_error.function,
name=endpoint_name
)
def _add_operation_internal(self, method, path, operation):
method = method.upper()
operation_id = operation.operation_id or path
logger.debug('... Adding %s -> %s', method, operation_id,
extra=vars(operation))
handler = operation.function
endpoint_name = '{}_{}_{}'.format(
self._api_name,
AioHttpApi.normalize_string(path),
method.lower()
)
self.subapp.router.add_route(
method, path, handler, name=endpoint_name
)
if not path.endswith('/'):
self.subapp.router.add_route(
method, path + '/', handler, name=endpoint_name + '_'
)
@classmethod
async def get_request(cls, req):
"""Convert aiohttp request to connexion
:param req: instance of aiohttp.web.Request
:return: connexion request instance
:rtype: ConnexionRequest
"""
url = str(req.url)
logger.debug(
'Getting data and status code',
extra={
# has_body | can_read_body report if
# body has been read or not
# body_exists refers to underlying stream of data
'body_exists': req.body_exists,
'can_read_body': req.can_read_body,
'content_type': req.content_type,
'url': url,
},
)
query = parse_qs(req.rel_url.query_string)
headers = req.headers
body = None
# Note: if request is not 'application/x-www-form-urlencoded' nor 'multipart/form-data',
# then `post_data` will be left an empty dict and the stream will not be consumed.
post_data = await req.post()
files = {}
form = {}
if post_data:
logger.debug('Reading multipart data from request')
for k, v in post_data.items():
if isinstance(v, web.FileField):
if k in files:
# if multiple files arrive under the same name in the
# request, downstream requires that we put them all into
# a list under the same key in the files dict.
if isinstance(files[k], list):
files[k].append(v)
else:
files[k] = [files[k], v]
else:
files[k] = v
else:
# put normal fields as an array, that's how werkzeug does that for Flask
# and that's what Connexion expects in its processing functions
form[k] = [v]
body = b''
else:
logger.debug('Reading data from request')
body = await req.read()
return ConnexionRequest(url=url,
method=req.method.lower(),
path_params=dict(req.match_info),
query=query,
headers=headers,
body=body,
json_getter=lambda: cls.jsonifier.loads(body),
form=form,
files=files,
context=req,
cookies=req.cookies)
@classmethod
async def get_response(cls, response, mimetype=None, request=None):
"""Get response.
This method is used in the lifecycle decorators
:type response: aiohttp.web.StreamResponse | (Any,) | (Any, int) | (Any, dict) | (Any, int, dict)
:rtype: aiohttp.web.Response
"""
while asyncio.iscoroutine(response):
response = await response
url = str(request.url) if request else ''
return cls._get_response(response, mimetype=mimetype, extra_context={"url": url})
@classmethod
def _is_framework_response(cls, response):
""" Return True if `response` is a framework response class """
return isinstance(response, web.StreamResponse)
@classmethod
def _framework_to_connexion_response(cls, response, mimetype):
""" Cast framework response class to ConnexionResponse used for schema validation """
body = None
if hasattr(response, "body"): # StreamResponse and FileResponse don't have body
body = response.body
return ConnexionResponse(
status_code=response.status,
mimetype=mimetype,
content_type=response.content_type,
headers=response.headers,
body=body
)
@classmethod
def _connexion_to_framework_response(cls, response, mimetype, extra_context=None):
""" Cast ConnexionResponse to framework response class """
return cls._build_response(
mimetype=response.mimetype or mimetype,
status_code=response.status_code,
content_type=response.content_type,
headers=response.headers,
data=response.body,
extra_context=extra_context,
)
@classmethod
def _build_response(cls, data, mimetype, content_type=None, headers=None, status_code=None, extra_context=None):
if cls._is_framework_response(data):
raise TypeError("Cannot return web.StreamResponse in tuple. Only raw data can be returned in tuple.")
data, status_code, serialized_mimetype = cls._prepare_body_and_status_code(data=data, mimetype=mimetype, status_code=status_code, extra_context=extra_context)
if isinstance(data, str):
text = data
body = None
else:
text = None
body = data
content_type = content_type or mimetype or serialized_mimetype
return web.Response(body=body, text=text, headers=headers, status=status_code, content_type=content_type)
@classmethod
def _set_jsonifier(cls):
cls.jsonifier = Jsonifier(cls=JSONEncoder)
class _HttpNotFoundError(HTTPNotFound):
def __init__(self):
self.name = 'Not Found'
self.description = (
'The requested URL was not found on the server. '
'If you entered the URL manually please check your spelling and '
'try again.'
)
self.code = type(self).status_code
self.empty_body = True
HTTPNotFound.__init__(self, reason=self.name)

2
connexion/apis/flask_api.py
View File

@@ -199,8 +199,6 @@ class FlaskApi(AbstractAPI):
@classmethod
def _serialize_data(cls, data, mimetype):
# TODO: harmonize flask and aiohttp serialization when mimetype=None or mimetype is not JSON
# (cases where it might not make sense to jsonify the data)
if (isinstance(mimetype, str) and is_json_mimetype(mimetype)):
body = cls.jsonifier.dumps(data)
elif not (isinstance(data, bytes) or isinstance(data, str)):

99
connexion/apps/aiohttp_app.py
View File

@@ -1,99 +0,0 @@
"""
This module defines an AioHttpApp, a Connexion application to wrap an AioHttp application.
"""
import logging
import pathlib
import pkgutil
import sys
from aiohttp import web
from ..apis.aiohttp_api import AioHttpApi
from ..exceptions import ConnexionException
from .abstract import AbstractApp
logger = logging.getLogger('connexion.aiohttp_app')
class AioHttpApp(AbstractApp):
def __init__(self, import_name, only_one_api=False, **kwargs):
super().__init__(import_name, AioHttpApi, server='aiohttp', **kwargs)
self._only_one_api = only_one_api
self._api_added = False
def create_app(self):
return web.Application(**self.server_args)
def get_root_path(self):
mod = sys.modules.get(self.import_name)
if mod is not None and hasattr(mod, '__file__'):
return pathlib.Path(mod.__file__).resolve().parent
loader = pkgutil.get_loader(self.import_name)
filepath = None
if hasattr(loader, 'get_filename'):
filepath = loader.get_filename(self.import_name)
if filepath is None:
raise RuntimeError(f"Invalid import name '{self.import_name}'")
return pathlib.Path(filepath).resolve().parent
def set_errors_handlers(self):
pass
def add_api(self, specification, **kwargs):
if self._only_one_api:
if self._api_added:
raise ConnexionException(
"an api was already added, "
"create a new app with 'only_one_api=False' "
"to add more than one api"
)
else:
self.app = self._get_api(specification, kwargs).subapp
self._api_added = True
return self.app
api = self._get_api(specification, kwargs)
try:
self.app.add_subapp(api.base_path, api.subapp)
except ValueError:
raise ConnexionException(
"aiohttp doesn't allow to set empty base_path ('/'), "
"use non-empty instead, e.g /api"
)
return api
def _get_api(self, specification, kwargs):
return super().add_api(specification, **kwargs)
def run(self, port=None, server=None, debug=None, host=None, **options):
if port is not None:
self.port = port
elif self.port is None:
self.port = 5000
self.server = server or self.server
self.host = host or self.host or '0.0.0.0'
if debug is not None:
self.debug = debug
logger.debug('Starting %s HTTP server..', self.server, extra=vars(self))
if self.server == 'aiohttp':
logger.info('Listening on %s:%s..', self.host, self.port)
access_log = options.pop('access_log', None)
if options.pop('use_default_access_log', None):
access_log = logger
web.run_app(self.app, port=self.port, host=self.host, access_log=access_log, **options)
else:
raise Exception(f'Server {self.server} not recognized')

10
connexion/cli.py
View File

@@ -16,20 +16,16 @@ from connexion.mock import MockResolver
logger = logging.getLogger('connexion.cli')
CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
FLASK_APP = 'flask'
AIOHTTP_APP = 'aiohttp'
AVAILABLE_SERVERS = {
'flask': [FLASK_APP],
'gevent': [FLASK_APP],
'tornado': [FLASK_APP],
'aiohttp': [AIOHTTP_APP]
}
AVAILABLE_APPS = {
FLASK_APP: 'connexion.apps.flask_app.FlaskApp',
AIOHTTP_APP: 'connexion.apps.aiohttp_app.AioHttpApp'
}
DEFAULT_SERVERS = {
FLASK_APP: FLASK_APP,
AIOHTTP_APP: AIOHTTP_APP
}
@@ -153,12 +149,6 @@ def run(spec_file,
)
raise click.UsageError(message)
if app_framework == AIOHTTP_APP:
try:
import aiohttp # NOQA
except Exception:
fatal_error('aiohttp library is not installed')
logging_level = logging.WARN
if verbose > 0:
logging_level = logging.INFO

5
connexion/security/__init__.py
View File

@@ -16,8 +16,3 @@ try:
from .flask_security_handler_factory import FlaskSecurityHandlerFactory
except ImportError as err: # pragma: no cover
FlaskSecurityHandlerFactory = not_installed_error(err)
try:
from .aiohttp_security_handler_factory import AioHttpSecurityHandlerFactory
except ImportError as err: # pragma: no cover
AioHttpSecurityHandlerFactory = not_installed_error(err)

39
connexion/security/aiohttp_security_handler_factory.py
View File

@@ -1,39 +0,0 @@
"""
This module defines an aiohttp-specific SecurityHandlerFactory.
"""
import logging
import aiohttp
from .async_security_handler_factory import AbstractAsyncSecurityHandlerFactory
logger = logging.getLogger('connexion.api.security')
class AioHttpSecurityHandlerFactory(AbstractAsyncSecurityHandlerFactory):
def __init__(self, pass_context_arg_name):
super().__init__(pass_context_arg_name=pass_context_arg_name)
self.client_session = None
def get_token_info_remote(self, token_info_url):
"""
Return a function which will call `token_info_url` to retrieve token info.
Returned function must accept oauth token in parameter.
It must return a token_info dict in case of success, None otherwise.
:param token_info_url: Url to get information about the token
:type token_info_url: str
:rtype: types.FunctionType
"""
async def wrapper(token):
if not self.client_session:
# Must be created in a coroutine
self.client_session = aiohttp.ClientSession()
headers = {'Authorization': f'Bearer {token}'}
token_request = await self.client_session.get(token_info_url, headers=headers, timeout=5)
if token_request.status != 200:
return None
return token_request.json()
return wrapper

9
docs/quickstart.rst
View File

@@ -96,15 +96,6 @@ to ``tornado`` or ``gevent``:
app = connexion.FlaskApp(__name__, port = 8080, specification_dir='openapi/', server='tornado')
Connexion has the ``aiohttp`` framework as server backend too:
.. code-block:: python
import connexion
app = connexion.AioHttpApp(__name__, port = 8080, specification_dir='openapi/')
.. _Jinja2: http://jinja.pocoo.org/
.. _Tornado: http://www.tornadoweb.org/en/stable/
.. _gevent: http://www.gevent.org/

16
examples/openapi3/enforcedefaults_aiohttp/README.rst
View File

@@ -1,16 +0,0 @@
========================
Custom Validator Example
========================
In this example we fill-in non-provided properties with their defaults.
Validator code is based on example from `python-jsonschema docs`_.
Running:
.. code-block:: bash
$ ./enforcedefaults.py
Now open your browser and go to http://localhost:8080/v1/ui/ to see the Swagger
UI. If you send a ``POST`` request with empty body ``{}``, you should receive
echo with defaults filled-in.

39
examples/openapi3/enforcedefaults_aiohttp/enforcedefaults-api.yaml
View File

@@ -1,39 +0,0 @@
openapi: '3.0.0'
info:
version: '1'
title: Custom Validator Example
servers:
- url: http://localhost:8080/{basePath}
variables:
basePath:
default: api
paths:
/echo:
post:
description: Echo passed data
operationId: enforcedefaults.echo
requestBody:
required: true
content:
application/json:
schema:
$ref: '#/components/schemas/Data'
responses:
'200':
description: Data with defaults filled in by validator
default:
description: Unexpected error
content:
application/json:
schema:
$ref: '#/components/schemas/Error'
components:
schemas:
Data:
type: object
properties:
foo:
type: string
default: foo
Error:
type: string

56
examples/openapi3/enforcedefaults_aiohttp/enforcedefaults.py
View File

@@ -1,56 +0,0 @@
#!/usr/bin/env python3
import connexion
import jsonschema
import six
from connexion.decorators.validation import RequestBodyValidator
from connexion.json_schema import Draft4RequestValidator
async def echo(body):
return body
# via https://python-jsonschema.readthedocs.io/
def extend_with_set_default(validator_class):
validate_properties = validator_class.VALIDATORS['properties']
def set_defaults(validator, properties, instance, schema):
for property, subschema in six.iteritems(properties):
if 'default' in subschema:
instance.setdefault(property, subschema['default'])
for error in validate_properties(
validator, properties, instance, schema):
yield error
return jsonschema.validators.extend(
validator_class, {'properties': set_defaults})
DefaultsEnforcingDraft4Validator = extend_with_set_default(Draft4RequestValidator)
class DefaultsEnforcingRequestBodyValidator(RequestBodyValidator):
def __init__(self, *args, **kwargs):
super(DefaultsEnforcingRequestBodyValidator, self).__init__(
*args, validator=DefaultsEnforcingDraft4Validator, **kwargs)
validator_map = {
'body': DefaultsEnforcingRequestBodyValidator
}
if __name__ == '__main__':
app = connexion.AioHttpApp(
__name__,
port=8080,
specification_dir='.',
options={'swagger_ui': True}
)
app.add_api(
'enforcedefaults-api.yaml',
arguments={'title': 'Hello World Example'},
validator_map=validator_map,
)
app.run()

11
examples/openapi3/helloworld_aiohttp/README.rst
View File

@@ -1,11 +0,0 @@
===================
Hello World Example
===================
Running:
.. code-block:: bash
$ ./hello.py
Now open your browser and go to http://localhost:9090/v1.0/ui/ to see the Swagger UI.

14
examples/openapi3/helloworld_aiohttp/hello.py
View File

@@ -1,14 +0,0 @@
#!/usr/bin/env python3
import connexion
from aiohttp import web
async def post_greeting(name):
return web.Response(text=f'Hello {name}')
if __name__ == '__main__':
app = connexion.AioHttpApp(__name__, port=9090, specification_dir='openapi/')
app.add_api('helloworld-api.yaml', arguments={'title': 'Hello World Example'})
app.run()

30
examples/openapi3/helloworld_aiohttp/openapi/helloworld-api.yaml
View File

@@ -1,30 +0,0 @@
openapi: "3.0.0"
info:
title: Hello World
version: "1.0"
servers:
- url: http://localhost:9090/v1.0
paths:
/greeting/{name}:
post:
summary: Generate greeting
description: Generates a greeting message.
operationId: hello.post_greeting
responses:
200:
description: greeting response
content:
text/plain:
schema:
type: string
example: "hello dave!"
parameters:
- name: name
in: path
description: Name of the person to greet.
required: true
schema:
type: string
example: "dave"

58
examples/openapi3/reverseproxy_aiohttp/README.rst
View File

@@ -1,58 +0,0 @@
=====================
Reverse Proxy Example
=====================
This example demonstrates how to run a connexion application behind a path-altering reverse proxy.
You can either set the path in your app, or set the ``X-Forwarded-Path`` header.
Running:
.. code-block:: bash
$ sudo pip3 install --upgrade connexion[swagger-ui] aiohttp-remotes
$ ./app.py
Now open your browser and go to http://localhost:8080/reverse_proxied/ui/ to see the Swagger UI.
You can also use the ``X-Forwarded-Path`` header to modify the reverse proxy path.
For example:
.. code-block:: bash
curl -H "X-Forwarded-Path: /banana/" http://localhost:8080/openapi.json
{
"servers" : [
{
"url" : "banana"
}
],
"paths" : {
"/hello" : {
"get" : {
"responses" : {
"200" : {
"description" : "hello",
"content" : {
"text/plain" : {
"schema" : {
"type" : "string"
}
}
}
}
},
"operationId" : "app.hello",
"summary" : "say hi"
}
}
},
"openapi" : "3.0.0",
"info" : {
"version" : "1.0",
"title" : "Path-Altering Reverse Proxy Example"
}
}

82
examples/openapi3/reverseproxy_aiohttp/app.py
View File

@@ -1,82 +0,0 @@
#!/usr/bin/env python3
'''
example of aiohttp connexion running behind a path-altering reverse-proxy
'''
import json
import logging
import connexion
from aiohttp import web
from aiohttp_remotes.exceptions import RemoteError, TooManyHeaders
from aiohttp_remotes.x_forwarded import XForwardedBase
from yarl import URL
X_FORWARDED_PATH = "X-Forwarded-Path"
class XPathForwarded(XForwardedBase):
def __init__(self, num=1):
self._num = num
def get_forwarded_path(self, headers):
forwarded_host = headers.getall(X_FORWARDED_PATH, [])
if len(forwarded_host) > 1:
raise TooManyHeaders(X_FORWARDED_PATH)
return forwarded_host[0] if forwarded_host else None
@web.middleware
async def middleware(self, request, handler):
logging.warning(
"this demo is not secure by default!! "
"You'll want to make sure these headers are coming from your proxy, "
"and not directly from users on the web!"
)
try:
overrides = {}
headers = request.headers
forwarded_for = self.get_forwarded_for(headers)
if forwarded_for:
overrides['remote'] = str(forwarded_for[-self._num])
proto = self.get_forwarded_proto(headers)
if proto:
overrides['scheme'] = proto[-self._num]
host = self.get_forwarded_host(headers)
if host is not None:
overrides['host'] = host
prefix = self.get_forwarded_path(headers)
if prefix is not None:
prefix = '/' + prefix.strip('/') + '/'
request_path = URL(request.path.lstrip('/'))
overrides['rel_url'] = URL(prefix).join(request_path)
request = request.clone(**overrides)
return await handler(request)
except RemoteError as exc:
exc.log(request)
await self.raise_error(request)
def hello(request):
ret = {
"host": request.host,
"scheme": request.scheme,
"path": request.path,
"_href": str(request.url)
}
return web.Response(text=json.dumps(ret), status=200)
if __name__ == '__main__':
app = connexion.AioHttpApp(__name__)
app.add_api('openapi.yaml', pass_context_arg_name='request')
aio = app.app
reverse_proxied = XPathForwarded()
aio.middlewares.append(reverse_proxied.middleware)
app.run(port=8080)

38
examples/openapi3/reverseproxy_aiohttp/nginx.conf
View File

@@ -1,38 +0,0 @@
worker_processes 1;
error_log stderr;
daemon off;
pid nginx.pid;
events {
worker_connections 1024;
}
http {
include /etc/nginx/mime.types;
default_type application/octet-stream;
sendfile on;
keepalive_timeout 65;
ssl_protocols TLSv1 TLSv1.1 TLSv1.2;
ssl_prefer_server_ciphers on;
access_log access.log;
server {
listen localhost:9000;
location /reverse_proxied/ {
# Define the location of the proxy server to send the request to
proxy_pass http://localhost:8080/;
# Add prefix header
proxy_set_header X-Forwarded-Path /reverse_proxied/;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Host $host:$server_port;
proxy_set_header X-Forwarded-Server $host;
proxy_set_header X-Forwarded-Port $server_port;
proxy_set_header X-Forwarded-Proto http;
}
}
}

18
examples/openapi3/reverseproxy_aiohttp/openapi.yaml
View File

@@ -1,18 +0,0 @@
openapi: 3.0.0
info:
title: Path-Altering Reverse Proxy Example
version: '1.0'
servers:
- url: /api
paths:
/hello:
get:
summary: say hi
operationId: app.hello
responses:
'200':
description: hello
content:
text/plain:
schema:
type: string

10
setup.py
View File

@@ -36,11 +36,6 @@ flask_require = [
'flask>=1.0.4,<3',
'itsdangerous>=0.24',
]
aiohttp_require = [
'aiohttp>=2.3.10,<4',
'aiohttp-jinja2>=0.14.0,<2',
'MarkupSafe>=0.23',
]
tests_require = [
'decorator>=5,<6',
@@ -51,10 +46,6 @@ tests_require = [
swagger_ui_require
]
tests_require.extend(aiohttp_require)
tests_require.append('pytest-aiohttp')
tests_require.append('aiohttp-remotes')
docs_require = [
'sphinx-autoapi==1.8.1'
]
@@ -108,7 +99,6 @@ setup(
'tests': tests_require,
'flask': flask_require,
'swagger-ui': swagger_ui_require,
'aiohttp': aiohttp_require,
'docs': docs_require
},
cmdclass={'test': PyTest},

161
tests/aiohttp/test_aiohttp_api_secure.py
View File

@@ -1,161 +0,0 @@
import base64
from unittest.mock import MagicMock
import pytest
from connexion import AioHttpApp
class FakeAioHttpClientResponse:
def __init__(self, status_code, data):
"""
:type status_code: int
:type data: dict
"""
self.status = status_code
self.data = data
self.ok = status_code == 200
async def json(self):
return self.data
@pytest.fixture
def oauth_aiohttp_client(monkeypatch):
async def fake_get(url, params=None, headers=None, timeout=None):
"""
:type url: str
:type params: dict| None
"""
headers = headers or {}
assert url == "https://oauth.example/token_info"
token = headers.get('Authorization', 'invalid').split()[-1]
if token in ["100", "has_myscope"]:
return FakeAioHttpClientResponse(200, {"uid": "test-user", "scope": ["myscope"]})
elif token in ["200", "has_wrongscope"]:
return FakeAioHttpClientResponse(200, {"uid": "test-user", "scope": ["wrongscope"]})
elif token == "has_myscope_otherscope":
return FakeAioHttpClientResponse(200, {"uid": "test-user", "scope": ["myscope", "otherscope"]})
elif token in ["300", "is_not_invalid"]:
return FakeAioHttpClientResponse(404, {})
elif token == "has_scopes_in_scopes_with_s":
return FakeAioHttpClientResponse(200, {"uid": "test-user", "scopes": ["myscope", "otherscope"]})
else:
raise AssertionError('Not supported test token ' + token)
client_instance = MagicMock()
client_instance.get = fake_get
monkeypatch.setattr('aiohttp.ClientSession', MagicMock(return_value=client_instance))
async def test_auth_all_paths(oauth_aiohttp_client, aiohttp_api_spec_dir, aiohttp_client):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True, auth_all_paths=True)
app.add_api('swagger_secure.yaml')
app_client = await aiohttp_client(app.app)
get_inexistent_endpoint = await app_client.get(
'/v1.0/does-not-exist-valid-token',
headers={'Authorization': 'Bearer 100'}
)
assert get_inexistent_endpoint.status == 404
assert get_inexistent_endpoint.content_type == 'application/problem+json'
get_inexistent_endpoint = await app_client.get(
'/v1.0/does-not-exist-no-token'
)
assert get_inexistent_endpoint.status == 401
assert get_inexistent_endpoint.content_type == 'application/problem+json'
@pytest.mark.parametrize('spec', ['swagger_secure.yaml', 'openapi_secure.yaml'])
async def test_secure_app(oauth_aiohttp_client, aiohttp_api_spec_dir, aiohttp_client, spec):
"""
Test common authentication method between Swagger 2 and OpenApi 3
"""
app = AioHttpApp(__name__, port=5001, specification_dir=aiohttp_api_spec_dir, debug=True)
app.add_api(spec)
app_client = await aiohttp_client(app.app)
response = await app_client.get('/v1.0/all_auth')
assert response.status == 401
assert response.content_type == 'application/problem+json'
response = await app_client.get('/v1.0/all_auth', headers={'Authorization': 'Bearer 100'})
assert response.status == 200
assert (await response.json()) == {"scope": ['myscope'], "uid": 'test-user'}
response = await app_client.get('/v1.0/all_auth', headers={'authorization': 'Bearer 100'})
assert response.status == 200, "Authorization header in lower case should be accepted"
assert (await response.json()) == {"scope": ['myscope'], "uid": 'test-user'}
response = await app_client.get('/v1.0/all_auth', headers={'AUTHORIZATION': 'Bearer 100'})
assert response.status == 200, "Authorization header in upper case should be accepted"
assert (await response.json()) == {"scope": ['myscope'], "uid": 'test-user'}
basic_header = 'Basic ' + base64.b64encode(b'username:username').decode('ascii')
response = await app_client.get('/v1.0/all_auth', headers={'Authorization': basic_header})
assert response.status == 200
assert (await response.json()) == {"uid": 'username'}
basic_header = 'Basic ' + base64.b64encode(b'username:wrong').decode('ascii')
response = await app_client.get('/v1.0/all_auth', headers={'Authorization': basic_header})
assert response.status == 401, "Wrong password should trigger unauthorized"
assert response.content_type == 'application/problem+json'
response = await app_client.get('/v1.0/all_auth', headers={'X-API-Key': '{"foo": "bar"}'})
assert response.status == 200
assert (await response.json()) == {"foo": "bar"}
async def test_bearer_secure(aiohttp_api_spec_dir, aiohttp_client):
"""
Test authentication method specific to OpenApi 3
"""
app = AioHttpApp(__name__, port=5001, specification_dir=aiohttp_api_spec_dir, debug=True)
app.add_api('openapi_secure.yaml')
app_client = await aiohttp_client(app.app)
bearer_header = 'Bearer {"scope": ["myscope"], "uid": "test-user"}'
response = await app_client.get('/v1.0/bearer_auth', headers={'Authorization': bearer_header})
assert response.status == 200
assert (await response.json()) == {"scope": ['myscope'], "uid": 'test-user'}
async def test_async_secure(aiohttp_api_spec_dir, aiohttp_client):
app = AioHttpApp(__name__, port=5001, specification_dir=aiohttp_api_spec_dir, debug=True)
app.add_api('openapi_secure.yaml', pass_context_arg_name='request')
app_client = await aiohttp_client(app.app)
response = await app_client.get('/v1.0/async_auth')
assert response.status == 401
assert response.content_type == 'application/problem+json'
bearer_header = 'Bearer {"scope": ["myscope"], "uid": "test-user"}'
response = await app_client.get('/v1.0/async_auth', headers={'Authorization': bearer_header})
assert response.status == 200
assert (await response.json()) == {"scope": ['myscope'], "uid": 'test-user'}
bearer_header = 'Bearer {"scope": ["myscope", "other_scope"], "uid": "test-user"}'
response = await app_client.get('/v1.0/async_auth', headers={'Authorization': bearer_header})
assert response.status == 403, "async_scope_validation should deny access if scopes are not strictly the same"
basic_header = 'Basic ' + base64.b64encode(b'username:username').decode('ascii')
response = await app_client.get('/v1.0/async_auth', headers={'Authorization': basic_header})
assert response.status == 200
assert (await response.json()) == {"uid": 'username'}
basic_header = 'Basic ' + base64.b64encode(b'username:wrong').decode('ascii')
response = await app_client.get('/v1.0/async_auth', headers={'Authorization': basic_header})
assert response.status == 401, "Wrong password should trigger unauthorized"
assert response.content_type == 'application/problem+json'
response = await app_client.get('/v1.0/all_auth', headers={'X-API-Key': '{"foo": "bar"}'})
assert response.status == 200
assert (await response.json()) == {"foo": "bar"}
bearer_header = 'Bearer {"scope": ["myscope"], "uid": "test-user"}'
response = await app_client.get('/v1.0/async_bearer_auth', headers={'Authorization': bearer_header})
assert response.status == 200
assert (await response.json()) == {"scope": ['myscope'], "uid": 'test-user'}

160
tests/aiohttp/test_aiohttp_app.py
View File

@@ -1,160 +0,0 @@
import logging
import os
import pathlib
from unittest import mock
import pytest
from connexion import AioHttpApp
from connexion.exceptions import ConnexionException
from conftest import TEST_FOLDER
@pytest.fixture
def web_run_app_mock(monkeypatch):
mock_ = mock.MagicMock()
monkeypatch.setattr('connexion.apps.aiohttp_app.web.run_app', mock_)
return mock_
@pytest.fixture
def sys_modules_mock(monkeypatch):
monkeypatch.setattr('connexion.apps.aiohttp_app.sys.modules', {})
def test_app_run(web_run_app_mock, aiohttp_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.run(use_default_access_log=True)
logger = logging.getLogger('connexion.aiohttp_app')
assert web_run_app_mock.call_args_list == [
mock.call(app.app, port=5001, host='0.0.0.0', access_log=logger)
]
def test_app_run_new_port(web_run_app_mock, aiohttp_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.run(port=5002)
assert web_run_app_mock.call_args_list == [
mock.call(app.app, port=5002, host='0.0.0.0', access_log=None)
]
def test_app_run_default_port(web_run_app_mock, aiohttp_api_spec_dir):
app = AioHttpApp(__name__,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.run()
assert web_run_app_mock.call_args_list == [
mock.call(app.app, port=5000, host='0.0.0.0', access_log=None)
]
def test_app_run_debug(web_run_app_mock, aiohttp_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir)
app.add_api('swagger_simple.yaml')
app.run(debug=True)
assert web_run_app_mock.call_args_list == [
mock.call(app.app, port=5001, host='0.0.0.0', access_log=None)
]
def test_app_run_access_log(web_run_app_mock, aiohttp_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
logger = logging.getLogger('connexion.aiohttp_app')
app.run(access_log=logger)
assert web_run_app_mock.call_args_list == [
mock.call(app.app, port=5001, host='0.0.0.0', access_log=logger)
]
def test_app_run_server_error(web_run_app_mock, aiohttp_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir)
with pytest.raises(Exception) as exc_info:
app.run(server='other')
assert exc_info.value.args == ('Server other not recognized',)
def test_app_get_root_path_return_Path(aiohttp_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir)
assert isinstance(app.get_root_path(), pathlib.Path) == True
def test_app_get_root_path_exists(aiohttp_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir)
assert app.get_root_path().exists() == True
def test_app_get_root_path(aiohttp_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir)
root_path = app.get_root_path()
assert str(root_path).endswith(os.path.join('tests', 'aiohttp')) == True
def test_app_get_root_path_not_in_sys_modules(sys_modules_mock, aiohttp_api_spec_dir):
app = AioHttpApp('connexion', port=5001,
specification_dir=aiohttp_api_spec_dir)
root_path = app.get_root_path()
assert str(root_path).endswith(os.sep + 'connexion') == True
def test_app_get_root_path_invalid(sys_modules_mock, aiohttp_api_spec_dir):
with pytest.raises(RuntimeError) as exc_info:
AioHttpApp('error__', port=5001,
specification_dir=aiohttp_api_spec_dir)
assert exc_info.value.args == ("Invalid import name 'error__'",)
def test_app_with_empty_base_path_error(aiohttp_api_spec_dir):
spec_dir = '..' / aiohttp_api_spec_dir.relative_to(TEST_FOLDER)
app = AioHttpApp(__name__, port=5001,
specification_dir=spec_dir,
debug=True)
with pytest.raises(ConnexionException) as exc_info:
app.add_api('swagger_empty_base_path.yaml')
assert exc_info.value.args == (
"aiohttp doesn't allow to set empty base_path ('/'), "
"use non-empty instead, e.g /api",
)
def test_app_with_empty_base_path_and_only_one_api(aiohttp_api_spec_dir):
spec_dir = '..' / aiohttp_api_spec_dir.relative_to(TEST_FOLDER)
app = AioHttpApp(__name__, port=5001,
specification_dir=spec_dir,
debug=True,
only_one_api=True)
api = app.add_api('swagger_empty_base_path.yaml')
assert api is app.app
def test_app_add_two_apis_error_with_only_one_api(aiohttp_api_spec_dir):
spec_dir = '..' / aiohttp_api_spec_dir.relative_to(TEST_FOLDER)
app = AioHttpApp(__name__, port=5001,
specification_dir=spec_dir,
debug=True,
only_one_api=True)
app.add_api('swagger_empty_base_path.yaml')
with pytest.raises(ConnexionException) as exc_info:
app.add_api('swagger_empty_base_path.yaml')
assert exc_info.value.args == (
"an api was already added, "
"create a new app with 'only_one_api=False' "
"to add more than one api",
)

49
tests/aiohttp/test_aiohttp_datetime.py
View File

@@ -1,49 +0,0 @@
from connexion import AioHttpApp
try:
import ujson as json
except ImportError:
import json
async def test_swagger_json(aiohttp_api_spec_dir, aiohttp_client):
""" Verify the swagger.json file is returned for default setting passed to app. """
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.add_api('datetime_support.yaml')
app_client = await aiohttp_client(app.app)
swagger_json = await app_client.get('/v1.0/openapi.json')
spec_data = await swagger_json.json()
def get_value(data, path):
for part in path.split('.'):
data = data.get(part)
assert data, f"No data in part '{part}' of '{path}'"
return data
example = get_value(spec_data, 'paths./datetime.get.responses.200.content.application/json.schema.example.value')
assert example in [
'2000-01-23T04:56:07.000008+00:00', # PyYAML 5.3
'2000-01-23T04:56:07.000008Z'
]
example = get_value(spec_data, 'paths./date.get.responses.200.content.application/json.schema.example.value')
assert example == '2000-01-23'
example = get_value(spec_data, 'paths./uuid.get.responses.200.content.application/json.schema.example.value')
assert example == 'a7b8869c-5f24-4ce0-a5d1-3e44c3663aa9'
resp = await app_client.get('/v1.0/datetime')
assert resp.status == 200
json_data = await resp.json()
assert json_data == {'value': '2000-01-02T03:04:05.000006Z'}
resp = await app_client.get('/v1.0/date')
assert resp.status == 200
json_data = await resp.json()
assert json_data == {'value': '2000-01-02'}
resp = await app_client.get('/v1.0/uuid')
assert resp.status == 200
json_data = await resp.json()
assert json_data == {'value': 'e7ff66d0-3ec2-4c4e-bed0-6e4723c24c51'}

137
tests/aiohttp/test_aiohttp_errors.py
View File

@@ -1,137 +0,0 @@
import asyncio
import pytest
from connexion import AioHttpApp
from connexion.apis.aiohttp_api import HTTPStatus
import aiohttp.test_utils
def is_valid_problem_json(json_body):
return all(key in json_body for key in ["type", "title", "detail", "status"])
@pytest.fixture
def aiohttp_app(problem_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=problem_api_spec_dir,
debug=True)
options = {"validate_responses": True}
app.add_api('openapi.yaml', validate_responses=True, pass_context_arg_name='request_ctx', options=options)
return app
async def test_aiohttp_problems_404(aiohttp_app, aiohttp_client):
# TODO: This is a based on test_errors.test_errors(). That should be refactored
# so that it is parameterized for all web frameworks.
app_client = await aiohttp_client(aiohttp_app.app) # type: aiohttp.test_utils.TestClient
greeting404 = await app_client.get('/v1.0/greeting') # type: aiohttp.ClientResponse
assert greeting404.content_type == 'application/problem+json'
assert greeting404.status == 404
error404 = await greeting404.json()
assert is_valid_problem_json(error404)
assert error404['type'] == 'about:blank'
assert error404['title'] == 'Not Found'
assert error404['detail'] == HTTPStatus(404).description
assert error404['status'] == 404
assert 'instance' not in error404
async def test_aiohttp_problems_405(aiohttp_app, aiohttp_client):
# TODO: This is a based on test_errors.test_errors(). That should be refactored
# so that it is parameterized for all web frameworks.
app_client = await aiohttp_client(aiohttp_app.app) # type: aiohttp.test_utils.TestClient
get_greeting = await app_client.get('/v1.0/greeting/jsantos') # type: aiohttp.ClientResponse
assert get_greeting.content_type == 'application/problem+json'
assert get_greeting.status == 405
error405 = await get_greeting.json()
assert is_valid_problem_json(error405)
assert error405['type'] == 'about:blank'
assert error405['title'] == 'Method Not Allowed'
assert error405['detail'] == HTTPStatus(405).description
assert error405['status'] == 405
assert 'instance' not in error405
async def test_aiohttp_problems_500(aiohttp_app, aiohttp_client):
# TODO: This is a based on test_errors.test_errors(). That should be refactored
# so that it is parameterized for all web frameworks.
app_client = await aiohttp_client(aiohttp_app.app) # type: aiohttp.test_utils.TestClient
get500 = await app_client.get('/v1.0/except') # type: aiohttp.ClientResponse
assert get500.content_type == 'application/problem+json'
assert get500.status == 500
error500 = await get500.json()
assert is_valid_problem_json(error500)
assert error500['type'] == 'about:blank'
assert error500['title'] == 'Internal Server Error'
assert error500['detail'] == HTTPStatus(500).description
assert error500['status'] == 500
assert 'instance' not in error500
async def test_aiohttp_problems_418(aiohttp_app, aiohttp_client):
# TODO: This is a based on test_errors.test_errors(). That should be refactored
# so that it is parameterized for all web frameworks.
app_client = await aiohttp_client(aiohttp_app.app) # type: aiohttp.test_utils.TestClient
get_problem = await app_client.get('/v1.0/problem') # type: aiohttp.ClientResponse
assert get_problem.content_type == 'application/problem+json'
assert get_problem.status == 418
assert get_problem.headers['x-Test-Header'] == 'In Test'
error_problem = await get_problem.json()
assert is_valid_problem_json(error_problem)
assert error_problem['type'] == 'http://www.example.com/error'
assert error_problem['title'] == 'Some Error'
assert error_problem['detail'] == 'Something went wrong somewhere'
assert error_problem['status'] == 418
assert error_problem['instance'] == 'instance1'
async def test_aiohttp_problems_misc(aiohttp_app, aiohttp_client):
# TODO: This is a based on test_errors.test_errors(). That should be refactored
# so that it is parameterized for all web frameworks.
app_client = await aiohttp_client(aiohttp_app.app) # type: aiohttp.test_utils.TestClient
problematic_json = await app_client.get(
'/v1.0/json_response_with_undefined_value_to_serialize') # type: aiohttp.ClientResponse
assert problematic_json.content_type == 'application/problem+json'
assert problematic_json.status == 500
problematic_json_body = await problematic_json.json()
assert is_valid_problem_json(problematic_json_body)
custom_problem = await app_client.get('/v1.0/customized_problem_response') # type: aiohttp.ClientResponse
assert custom_problem.content_type == 'application/problem+json'
assert custom_problem.status == 403
problem_body = await custom_problem.json()
assert is_valid_problem_json(problem_body)
assert 'amount' in problem_body
problem_as_exception = await app_client.get('/v1.0/problem_exception_with_extra_args') # type: aiohttp.ClientResponse
assert problem_as_exception.content_type == "application/problem+json"
assert problem_as_exception.status == 400
problem_as_exception_body = await problem_as_exception.json()
assert is_valid_problem_json(problem_as_exception_body)
assert 'age' in problem_as_exception_body
assert problem_as_exception_body['age'] == 30
@pytest.mark.skip(reason="aiohttp_api.get_connexion_response uses _cast_body "
"to stringify the dict directly instead of using json.dumps. "
"This differs from flask usage, where there is no _cast_body.")
async def test_aiohttp_problem_with_text_content_type(aiohttp_app, aiohttp_client):
app_client = await aiohttp_client(aiohttp_app.app) # type: aiohttp.test_utils.TestClient
get_problem2 = await app_client.get('/v1.0/other_problem') # type: aiohttp.ClientResponse
assert get_problem2.content_type == 'application/problem+json'
assert get_problem2.status == 418
error_problem2 = await get_problem2.json()
assert is_valid_problem_json(error_problem2)
assert error_problem2['type'] == 'about:blank'
assert error_problem2['title'] == 'Some Error'
assert error_problem2['detail'] == 'Something went wrong somewhere'
assert error_problem2['status'] == 418
assert error_problem2['instance'] == 'instance1'

118
tests/aiohttp/test_aiohttp_multipart.py
View File

@@ -1,118 +0,0 @@
import os
from pathlib import Path
import pytest
from connexion import AioHttpApp
import aiohttp
try:
import ujson as json
except ImportError:
import json
@pytest.fixture
def aiohttp_app(aiohttp_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.add_api(
'openapi_multipart.yaml',
validate_responses=True,
strict_validation=True,
pythonic_params=True,
pass_context_arg_name='request_ctx',
)
return app
async def test_single_file_upload(aiohttp_app, aiohttp_client):
app_client = await aiohttp_client(aiohttp_app.app)
resp = await app_client.post(
'/v1.0/upload_file',
data=aiohttp.FormData(fields=[('myfile', open(__file__, 'rb'))])(),
)
data = await resp.json()
assert resp.status == 200
assert data['fileName'] == f'{__name__}.py'
assert data['myfile_content'] == Path(__file__).read_bytes().decode('utf8')
async def test_many_files_upload(aiohttp_app, aiohttp_client):
app_client = await aiohttp_client(aiohttp_app.app)
dir_name = os.path.dirname(__file__)
files_field = [
('myfiles', open(f'{dir_name}/{file_name}', 'rb')) \
for file_name in sorted(os.listdir(dir_name)) if file_name.endswith('.py')
]
form_data = aiohttp.FormData(fields=files_field)
resp = await app_client.post(
'/v1.0/upload_files',
data=form_data(),
)
data = await resp.json()
assert resp.status == 200
assert data['files_count'] == len(files_field)
assert data['myfiles_content'] == [
Path(f'{dir_name}/{file_name}').read_bytes().decode('utf8') \
for file_name in sorted(os.listdir(dir_name)) if file_name.endswith('.py')
]
async def test_mixed_multipart_single_file(aiohttp_app, aiohttp_client):
app_client = await aiohttp_client(aiohttp_app.app)
form_data = aiohttp.FormData()
form_data.add_field('dir_name', os.path.dirname(__file__))
form_data.add_field('myfile', open(__file__, 'rb'))
resp = await app_client.post(
'/v1.0/mixed_single_file',
data=form_data(),
)
data = await resp.json()
assert resp.status == 200
assert data['dir_name'] == os.path.dirname(__file__)
assert data['fileName'] == f'{__name__}.py'
assert data['myfile_content'] == Path(__file__).read_bytes().decode('utf8')
async def test_mixed_multipart_many_files(aiohttp_app, aiohttp_client):
app_client = await aiohttp_client(aiohttp_app.app)
dir_name = os.path.dirname(__file__)
files_field = [
('myfiles', open(f'{dir_name}/{file_name}', 'rb')) \
for file_name in sorted(os.listdir(dir_name)) if file_name.endswith('.py')
]
form_data = aiohttp.FormData(fields=files_field)
form_data.add_field('dir_name', os.path.dirname(__file__))
form_data.add_field('test_count', str(len(files_field)))
resp = await app_client.post(
'/v1.0/mixed_many_files',
data=form_data(),
)
data = await resp.json()
assert resp.status == 200
assert data['dir_name'] == os.path.dirname(__file__)
assert data['test_count'] == len(files_field)
assert data['files_count'] == len(files_field)
assert data['myfiles_content'] == [
Path(f'{dir_name}/{file_name}').read_bytes().decode('utf8') \
for file_name in sorted(os.listdir(dir_name)) if file_name.endswith('.py')
]

129
tests/aiohttp/test_aiohttp_reverse_proxy.py
View File

@@ -1,129 +0,0 @@
import asyncio
from aiohttp_remotes.exceptions import RemoteError, TooManyHeaders
from aiohttp_remotes.x_forwarded import XForwardedBase
from connexion import AioHttpApp
from yarl import URL
from aiohttp import web
X_FORWARDED_PATH = "X-Forwarded-Path"
class XPathForwarded(XForwardedBase):
def __init__(self, num=1):
self._num = num
def get_forwarded_path(self, headers):
forwarded_host = headers.getall(X_FORWARDED_PATH, [])
if len(forwarded_host) > 1:
raise TooManyHeaders(X_FORWARDED_PATH)
return forwarded_host[0] if forwarded_host else None
@web.middleware
async def middleware(self, request, handler):
try:
overrides = {}
headers = request.headers
forwarded_for = self.get_forwarded_for(headers)
if forwarded_for:
overrides['remote'] = str(forwarded_for[-self._num])
proto = self.get_forwarded_proto(headers)
if proto:
overrides['scheme'] = proto[-self._num]
host = self.get_forwarded_host(headers)
if host is not None:
overrides['host'] = host
prefix = self.get_forwarded_path(headers)
if prefix is not None:
prefix = '/' + prefix.strip('/') + '/'
request_path = URL(request.path.lstrip('/'))
overrides['rel_url'] = URL(prefix).join(request_path)
request = request.clone(**overrides)
return await handler(request)
except RemoteError as exc:
exc.log(request)
await self.raise_error(request)
async def test_swagger_json_behind_proxy(simple_api_spec_dir, aiohttp_client):
""" Verify the swagger.json file is returned with base_path updated
according to X-Forwarded-Path header. """
app = AioHttpApp(__name__, port=5001,
specification_dir=simple_api_spec_dir,
debug=True)
api = app.add_api('swagger.yaml')
aio = app.app
reverse_proxied = XPathForwarded()
aio.middlewares.append(reverse_proxied.middleware)
app_client = await aiohttp_client(app.app)
headers = {'X-Forwarded-Path': '/behind/proxy'}
swagger_ui = await app_client.get('/v1.0/ui/', headers=headers)
assert swagger_ui.status == 200
assert b'url = "/behind/proxy/v1.0/swagger.json"' in (
await swagger_ui.read()
)
swagger_json = await app_client.get('/v1.0/swagger.json', headers=headers)
assert swagger_json.status == 200
assert swagger_json.headers.get('Content-Type') == 'application/json'
json_ = await swagger_json.json()
assert api.specification.raw['basePath'] == '/v1.0', \
"Original specifications should not have been changed"
assert json_.get('basePath') == '/behind/proxy/v1.0', \
"basePath should contains original URI"
json_['basePath'] = api.specification.raw['basePath']
assert api.specification.raw == json_, \
"Only basePath should have been updated"
async def test_openapi_json_behind_proxy(simple_api_spec_dir, aiohttp_client):
""" Verify the swagger.json file is returned with base_path updated
according to X-Forwarded-Path header. """
app = AioHttpApp(__name__, port=5001,
specification_dir=simple_api_spec_dir,
debug=True)
api = app.add_api('openapi.yaml')
aio = app.app
reverse_proxied = XPathForwarded()
aio.middlewares.append(reverse_proxied.middleware)
app_client = await aiohttp_client(app.app)
headers = {'X-Forwarded-Path': '/behind/proxy'}
swagger_ui = await app_client.get('/v1.0/ui/', headers=headers)
assert swagger_ui.status == 200
assert b'url: "/behind/proxy/v1.0/openapi.json"' in (
await swagger_ui.read()
)
swagger_json = await app_client.get('/v1.0/openapi.json', headers=headers)
assert swagger_json.status == 200
assert swagger_json.headers.get('Content-Type') == 'application/json'
json_ = await swagger_json.json()
assert json_.get('servers', [{}])[0].get('url') == '/behind/proxy/v1.0', \
"basePath should contains original URI"
url = api.specification.raw.get('servers', [{}])[0].get('url')
assert url != '/behind/proxy/v1.0', \
"Original specifications should not have been changed"
json_['servers'] = api.specification.raw.get('servers')
assert api.specification.raw == json_, \
"Only there servers block should have been updated"

377
tests/aiohttp/test_aiohttp_simple_api.py
View File

@@ -1,377 +0,0 @@
import sys
import pytest
import yaml
from connexion import AioHttpApp
from conftest import TEST_FOLDER
try:
import ujson as json
except ImportError:
import json
@pytest.fixture
def aiohttp_app(aiohttp_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
options = {"validate_responses": True}
app.add_api('swagger_simple.yaml', validate_responses=True, pass_context_arg_name='request_ctx', options=options)
return app
async def test_app(aiohttp_app, aiohttp_client):
# Create the app and run the test_app testcase below.
app_client = await aiohttp_client(aiohttp_app.app)
get_bye = await app_client.get('/v1.0/bye/jsantos')
assert get_bye.status == 200
assert (await get_bye.read()) == b'Goodbye jsantos'
async def test_app_with_relative_path(aiohttp_api_spec_dir, aiohttp_client):
# Create the app with a relative path and run the test_app testcase below.
app = AioHttpApp(__name__, port=5001,
specification_dir='..' /
aiohttp_api_spec_dir.relative_to(TEST_FOLDER),
debug=True)
app.add_api('swagger_simple.yaml')
app_client = await aiohttp_client(app.app)
get_bye = await app_client.get('/v1.0/bye/jsantos')
assert get_bye.status == 200
assert (await get_bye.read()) == b'Goodbye jsantos'
async def test_swagger_json(aiohttp_api_spec_dir, aiohttp_client):
""" Verify the swagger.json file is returned for default setting passed to app. """
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
api = app.add_api('swagger_simple.yaml')
app_client = await aiohttp_client(app.app)
swagger_json = await app_client.get('/v1.0/swagger.json')
assert swagger_json.status == 200
json_ = await swagger_json.json()
assert api.specification.raw == json_
async def test_swagger_yaml(aiohttp_api_spec_dir, aiohttp_client):
""" Verify the swagger.yaml file is returned for default setting passed to app. """
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
api = app.add_api('swagger_simple.yaml')
app_client = await aiohttp_client(app.app)
spec_response = await app_client.get('/v1.0/swagger.yaml')
data_ = await spec_response.read()
assert spec_response.status == 200
assert api.specification.raw == yaml.load(data_, yaml.FullLoader)
async def test_no_swagger_json(aiohttp_api_spec_dir, aiohttp_client):
""" Verify the swagger.json file is not returned when set to False when creating app. """
options = {"swagger_json": False}
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
options=options,
debug=True)
app.add_api('swagger_simple.yaml')
app_client = await aiohttp_client(app.app)
swagger_json = await app_client.get('/v1.0/swagger.json') # type: flask.Response
assert swagger_json.status == 404
async def test_no_swagger_yaml(aiohttp_api_spec_dir, aiohttp_client):
""" Verify the swagger.json file is not returned when set to False when creating app. """
options = {"swagger_json": False}
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
options=options,
debug=True)
app.add_api('swagger_simple.yaml')
app_client = await aiohttp_client(app.app)
spec_response = await app_client.get('/v1.0/swagger.yaml') # type: flask.Response
assert spec_response.status == 404
async def test_swagger_ui(aiohttp_api_spec_dir, aiohttp_client):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.add_api('swagger_simple.yaml')
app_client = await aiohttp_client(app.app)
swagger_ui = await app_client.get('/v1.0/ui')
assert swagger_ui.status == 200
assert swagger_ui.url.path == '/v1.0/ui/'
assert b'url = "/v1.0/swagger.json"' in (await swagger_ui.read())
swagger_ui = await app_client.get('/v1.0/ui/')
assert swagger_ui.status == 200
assert b'url = "/v1.0/swagger.json"' in (await swagger_ui.read())
async def test_swagger_ui_config_json(aiohttp_api_spec_dir, aiohttp_client):
""" Verify the swagger-ui-config.json file is returned for swagger_ui_config option passed to app. """
swagger_ui_config = {"displayOperationId": True}
options = {"swagger_ui_config": swagger_ui_config}
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
options=options,
debug=True)
api = app.add_api('swagger_simple.yaml')
app_client = await aiohttp_client(app.app)
swagger_ui_config_json = await app_client.get('/v1.0/ui/swagger-ui-config.json')
json_ = await swagger_ui_config_json.read()
assert swagger_ui_config_json.status == 200
assert swagger_ui_config == json.loads(json_)
async def test_no_swagger_ui_config_json(aiohttp_api_spec_dir, aiohttp_client):
""" Verify the swagger-ui-config.json file is not returned when the swagger_ui_config option not passed to app. """
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.add_api('swagger_simple.yaml')
app_client = await aiohttp_client(app.app)
swagger_ui_config_json = await app_client.get('/v1.0/ui/swagger-ui-config.json')
assert swagger_ui_config_json.status == 404
async def test_swagger_ui_index(aiohttp_api_spec_dir, aiohttp_client):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.add_api('openapi_secure.yaml')
app_client = await aiohttp_client(app.app)
swagger_ui = await app_client.get('/v1.0/ui/index.html')
assert swagger_ui.status == 200
assert b'url: "/v1.0/openapi.json"' in (await swagger_ui.read())
assert b'swagger-ui-config.json' not in (await swagger_ui.read())
async def test_swagger_ui_index_with_config(aiohttp_api_spec_dir, aiohttp_client):
swagger_ui_config = {"displayOperationId": True}
options = {"swagger_ui_config": swagger_ui_config}
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
options=options,
debug=True)
app.add_api('openapi_secure.yaml')
app_client = await aiohttp_client(app.app)
swagger_ui = await app_client.get('/v1.0/ui/index.html')
assert swagger_ui.status == 200
assert b'configUrl: "swagger-ui-config.json"' in (await swagger_ui.read())
async def test_pythonic_path_param(aiohttp_api_spec_dir, aiohttp_client):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.add_api('openapi_simple.yaml', pythonic_params=True)
app_client = await aiohttp_client(app.app)
pythonic = await app_client.get('/v1.0/pythonic/100')
assert pythonic.status == 200
j = await pythonic.json()
assert j['id_'] == 100
async def test_cookie_param(aiohttp_api_spec_dir, aiohttp_client):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.add_api('openapi_simple.yaml', pass_context_arg_name="request")
app_client = await aiohttp_client(app.app)
response = await app_client.get('/v1.0/test-cookie-param', headers={"Cookie": "test_cookie=hello"})
assert response.status == 200
j = await response.json()
assert j['cookie_value'] == "hello"
async def test_swagger_ui_static(aiohttp_api_spec_dir, aiohttp_client):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.add_api('swagger_simple.yaml')
app_client = await aiohttp_client(app.app)
swagger_ui = await app_client.get('/v1.0/ui/lib/swagger-oauth.js')
assert swagger_ui.status == 200
app_client = await aiohttp_client(app.app)
swagger_ui = await app_client.get('/v1.0/ui/swagger-ui.min.js')
assert swagger_ui.status == 200
async def test_no_swagger_ui(aiohttp_api_spec_dir, aiohttp_client):
options = {"swagger_ui": False}
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
options=options, debug=True)
app.add_api('swagger_simple.yaml')
app_client = await aiohttp_client(app.app)
swagger_ui = await app_client.get('/v1.0/ui/')
assert swagger_ui.status == 404
app2 = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
options = {"swagger_ui": False}
app2.add_api('swagger_simple.yaml', options=options)
app2_client = await aiohttp_client(app.app)
swagger_ui2 = await app2_client.get('/v1.0/ui/')
assert swagger_ui2.status == 404
async def test_middlewares(aiohttp_api_spec_dir, aiohttp_client):
async def middleware(app, handler):
async def middleware_handler(request):
response = (await handler(request))
response.body += b' middleware'
return response
return middleware_handler
options = {"middlewares": [middleware]}
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True, options=options)
app.add_api('swagger_simple.yaml')
app_client = await aiohttp_client(app.app)
get_bye = await app_client.get('/v1.0/bye/jsantos')
assert get_bye.status == 200
assert (await get_bye.read()) == b'Goodbye jsantos middleware'
async def test_response_with_str_body(aiohttp_app, aiohttp_client):
# Create the app and run the test_app testcase below.
app_client = await aiohttp_client(aiohttp_app.app)
get_bye = await app_client.get('/v1.0/aiohttp_str_response')
assert get_bye.status == 200
assert (await get_bye.read()) == b'str response'
async def test_response_with_non_str_and_non_json_body(aiohttp_app, aiohttp_client):
app_client = await aiohttp_client(aiohttp_app.app)
get_bye = await app_client.get(
'/v1.0/aiohttp_non_str_non_json_response'
)
assert get_bye.status == 200
assert (await get_bye.read()) == b'1234'
async def test_response_with_bytes_body(aiohttp_app, aiohttp_client):
# Create the app and run the test_app testcase below.
app_client = await aiohttp_client(aiohttp_app.app)
get_bye = await app_client.get('/v1.0/aiohttp_bytes_response')
assert get_bye.status == 200
assert (await get_bye.read()) == b'bytes response'
async def test_validate_responses(aiohttp_app, aiohttp_client):
app_client = await aiohttp_client(aiohttp_app.app)
get_bye = await app_client.get('/v1.0/aiohttp_validate_responses')
assert get_bye.status == 200
assert (await get_bye.json()) == {"validate": True}
async def test_get_users(aiohttp_client, aiohttp_app):
app_client = await aiohttp_client(aiohttp_app.app)
resp = await app_client.get('/v1.0/users')
assert resp.url.path == '/v1.0/users/' # followed redirect
assert resp.status == 200
json_data = await resp.json()
assert json_data == \
[{'name': 'John Doe', 'id': 1}, {'name': 'Nick Carlson', 'id': 2}]
async def test_create_user(aiohttp_client, aiohttp_app):
app_client = await aiohttp_client(aiohttp_app.app)
user = {'name': 'Maksim'}
resp = await app_client.post('/v1.0/users', json=user, headers={'Content-type': 'application/json'})
assert resp.status == 201
async def test_access_request_context(aiohttp_client, aiohttp_app):
app_client = await aiohttp_client(aiohttp_app.app)
resp = await app_client.post('/v1.0/aiohttp_access_request_context/')
assert resp.status == 204
async def test_query_parsing_simple(aiohttp_client, aiohttp_app):
expected_query = 'query'
app_client = await aiohttp_client(aiohttp_app.app)
resp = await app_client.get(
'/v1.0/aiohttp_query_parsing_str',
params={
'query': expected_query,
},
)
assert resp.status == 200
json_data = await resp.json()
assert json_data == {'query': expected_query}
async def test_query_parsing_array(aiohttp_client, aiohttp_app):
expected_query = ['queryA', 'queryB']
app_client = await aiohttp_client(aiohttp_app.app)
resp = await app_client.get(
'/v1.0/aiohttp_query_parsing_array',
params={
'query': ','.join(expected_query),
},
)
assert resp.status == 200
json_data = await resp.json()
assert json_data == {'query': expected_query}
async def test_query_parsing_array_multi(aiohttp_client, aiohttp_app):
expected_query = ['queryA', 'queryB', 'queryC']
query_str = '&'.join(['query=%s' % q for q in expected_query])
app_client = await aiohttp_client(aiohttp_app.app)
resp = await app_client.get(
'/v1.0/aiohttp_query_parsing_array_multi?%s' % query_str,
)
assert resp.status == 200
json_data = await resp.json()
assert json_data == {'query': expected_query}
if sys.version_info[0:2] >= (3, 5):
@pytest.fixture
def aiohttp_app_async_def(aiohttp_api_spec_dir):
app = AioHttpApp(__name__, port=5001,
specification_dir=aiohttp_api_spec_dir,
debug=True)
app.add_api('swagger_simple_async_def.yaml', validate_responses=True)
return app
async def test_validate_responses_async_def(aiohttp_app_async_def, aiohttp_client):
app_client = await aiohttp_client(aiohttp_app_async_def.app)
get_bye = await app_client.get('/v1.0/aiohttp_validate_responses')
assert get_bye.status == 200
assert (await get_bye.read()) == b'{"validate": true}'

172
tests/aiohttp/test_get_response.py
View File

@@ -1,172 +0,0 @@
import json
import pytest
from connexion.apis.aiohttp_api import AioHttpApi
from connexion.lifecycle import ConnexionResponse
from aiohttp import web
@pytest.fixture(scope='module')
def api(aiohttp_api_spec_dir):
yield AioHttpApi(specification=aiohttp_api_spec_dir / 'swagger_secure.yaml')
async def test_get_response_from_aiohttp_response(api):
response = await api.get_response(web.Response(text='foo', status=201, headers={'X-header': 'value'}))
assert isinstance(response, web.Response)
assert response.status == 201
assert response.body == b'foo'
assert response.content_type == 'text/plain'
assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'}
async def test_get_response_from_aiohttp_stream_response(api):
response = await api.get_response(web.StreamResponse(status=201, headers={'X-header': 'value'}))
assert isinstance(response, web.StreamResponse)
assert response.status == 201
assert response.content_type == 'application/octet-stream'
assert dict(response.headers) == {'X-header': 'value'}
async def test_get_response_from_connexion_response(api):
response = await api.get_response(ConnexionResponse(status_code=201, mimetype='text/plain', body='foo', headers={'X-header': 'value'}))
assert isinstance(response, web.Response)
assert response.status == 201
assert response.body == b'foo'
assert response.content_type == 'text/plain'
assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'}
async def test_get_response_from_string(api):
response = await api.get_response('foo')
assert isinstance(response, web.Response)
assert response.status == 200
assert response.body == b'foo'
assert response.content_type == 'text/plain'
assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8'}
async def test_get_response_from_string_tuple(api):
response = await api.get_response(('foo',))
assert isinstance(response, web.Response)
assert response.status == 200
assert response.body == b'foo'
assert response.content_type == 'text/plain'
assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8'}
async def test_get_response_from_string_status(api):
response = await api.get_response(('foo', 201))
assert isinstance(response, web.Response)
assert response.status == 201
assert response.body == b'foo'
assert response.content_type == 'text/plain'
assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8'}
async def test_get_response_from_string_headers(api):
response = await api.get_response(('foo', {'X-header': 'value'}))
assert isinstance(response, web.Response)
assert response.status == 200
assert response.body == b'foo'
assert response.content_type == 'text/plain'
assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'}
async def test_get_response_from_string_status_headers(api):
response = await api.get_response(('foo', 201, {'X-header': 'value'}))
assert isinstance(response, web.Response)
assert response.status == 201
assert response.body == b'foo'
assert response.content_type == 'text/plain'
assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'}
async def test_get_response_from_tuple_error(api):
with pytest.raises(TypeError) as e:
await api.get_response((web.Response(text='foo', status=201, headers={'X-header': 'value'}), 200))
assert str(e.value) == "Cannot return web.StreamResponse in tuple. Only raw data can be returned in tuple."
async def test_get_response_from_dict(api):
response = await api.get_response({'foo': 'bar'})
assert isinstance(response, web.Response)
assert response.status == 200
# odd, yes. but backwards compatible. see test_response_with_non_str_and_non_json_body in tests/aiohttp/test_aiohttp_simple_api.py
# TODO: This should be made into JSON when aiohttp and flask serialization can be harmonized.
assert response.body == b"{'foo': 'bar'}"
assert response.content_type == 'text/plain'
assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8'}
async def test_get_response_from_dict_json(api):
response = await api.get_response({'foo': 'bar'}, mimetype='application/json')
assert isinstance(response, web.Response)
assert response.status == 200
assert json.loads(response.body.decode()) == {"foo": "bar"}
assert response.content_type == 'application/json'
assert dict(response.headers) == {'Content-Type': 'application/json; charset=utf-8'}
async def test_get_response_no_data(api):
response = await api.get_response(None, mimetype='application/json')
assert isinstance(response, web.Response)
assert response.status == 204
assert response.body is None
assert response.content_type == 'application/json'
assert dict(response.headers) == {'Content-Type': 'application/json'}
async def test_get_response_binary_json(api):
response = await api.get_response(b'{"foo":"bar"}', mimetype='application/json')
assert isinstance(response, web.Response)
assert response.status == 200
assert json.loads(response.body.decode()) == {"foo": "bar"}
assert response.content_type == 'application/json'
assert dict(response.headers) == {'Content-Type': 'application/json'}
async def test_get_response_binary_no_mimetype(api):
response = await api.get_response(b'{"foo":"bar"}')
assert isinstance(response, web.Response)
assert response.status == 200
assert response.body == b'{"foo":"bar"}'
assert response.content_type == 'application/octet-stream'
assert dict(response.headers) == {}
async def test_get_connexion_response_from_aiohttp_response(api):
response = api.get_connexion_response(web.Response(text='foo', status=201, headers={'X-header': 'value'}))
assert isinstance(response, ConnexionResponse)
assert response.status_code == 201
assert response.body == b'foo'
assert response.content_type == 'text/plain'
assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'}
async def test_get_connexion_response_from_connexion_response(api):
response = api.get_connexion_response(ConnexionResponse(status_code=201, content_type='text/plain', body='foo', headers={'X-header': 'value'}))
assert isinstance(response, ConnexionResponse)
assert response.status_code == 201
assert response.body == b'foo'
assert response.content_type == 'text/plain'
assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'}
async def test_get_connexion_response_from_tuple(api):
response = api.get_connexion_response(('foo', 201, {'X-header': 'value'}))
assert isinstance(response, ConnexionResponse)
assert response.status_code == 201
assert response.body == b'foo'
assert response.content_type == 'text/plain'
assert dict(response.headers) == {'Content-Type': 'text/plain; charset=utf-8', 'X-header': 'value'}
async def test_get_connexion_response_from_aiohttp_stream_response(api):
response = api.get_connexion_response(web.StreamResponse(status=201, headers={'X-header': 'value'}))
assert isinstance(response, ConnexionResponse)
assert response.status_code == 201
assert response.body == None
assert response.content_type == 'application/octet-stream'
assert dict(response.headers) == {'X-header': 'value'}

10
tests/conftest.py
View File

@@ -78,11 +78,6 @@ def simple_api_spec_dir():
return FIXTURES_FOLDER / 'simple'
@pytest.fixture(scope='session')
def aiohttp_api_spec_dir():
return FIXTURES_FOLDER / 'aiohttp'
@pytest.fixture
def problem_api_spec_dir():
return FIXTURES_FOLDER / 'problem'
@@ -227,8 +222,3 @@ def bad_operations_app(request):
return build_app_from_fixture('bad_operations', request.param,
resolver_error=501)
if sys.version_info < (3, 5, 3) and sys.version_info[0] == 3:
@pytest.fixture
def aiohttp_client(test_client):
return test_client

153
tests/fakeapi/aiohttp_handlers.py
View File

@@ -1,153 +0,0 @@
#!/usr/bin/env python3
import datetime
import uuid
from connexion.lifecycle import ConnexionResponse
import aiohttp
from aiohttp.web import Request
from aiohttp.web import Response as AioHttpResponse
async def get_bye(name):
return AioHttpResponse(text=f'Goodbye {name}')
async def aiohttp_str_response():
return 'str response'
async def aiohttp_non_str_non_json_response():
return 1234
async def aiohttp_bytes_response():
return b'bytes response'
async def aiohttp_validate_responses():
return {"validate": True}
async def aiohttp_post_greeting(name, **kwargs):
data = {'greeting': f'Hello {name}'}
return data
async def aiohttp_echo(**kwargs):
return aiohttp.web.json_response(data=kwargs, status=200)
async def aiohttp_access_request_context(request_ctx):
assert request_ctx is not None
assert isinstance(request_ctx, aiohttp.web.Request)
return None
async def aiohttp_query_parsing_str(query):
return {'query': query}
async def aiohttp_query_parsing_array(query):
return {'query': query}
async def aiohttp_query_parsing_array_multi(query):
return {'query': query}
USERS = [
{"id": 1, "name": "John Doe"},
{"id": 2, "name": "Nick Carlson"}
]
async def aiohttp_users_get(*args):
return aiohttp.web.json_response(data=USERS, status=200)
async def aiohttp_users_post(user):
if "name" not in user:
return ConnexionResponse(body={"error": "name is undefined"},
status_code=400,
content_type='application/json')
user['id'] = len(USERS) + 1
USERS.append(user)
return aiohttp.web.json_response(data=USERS[-1], status=201)
async def aiohttp_token_info(token_info):
return aiohttp.web.json_response(data=token_info)
async def aiohttp_all_auth(token_info):
return await aiohttp_token_info(token_info)
async def aiohttp_async_auth(token_info):
return await aiohttp_token_info(token_info)
async def aiohttp_bearer_auth(token_info):
return await aiohttp_token_info(token_info)
async def aiohttp_async_bearer_auth(token_info):
return await aiohttp_token_info(token_info)
async def get_datetime():
return ConnexionResponse(body={'value': datetime.datetime(2000, 1, 2, 3, 4, 5, 6)})
async def get_date():
return ConnexionResponse(body={'value': datetime.date(2000, 1, 2)})
async def get_uuid():
return ConnexionResponse(body={'value': uuid.UUID(hex='e7ff66d0-3ec2-4c4e-bed0-6e4723c24c51')})
async def aiohttp_multipart_single_file(myfile):
return aiohttp.web.json_response(
data={
'fileName': myfile.filename,
'myfile_content': myfile.file.read().decode('utf8')
},
)
async def aiohttp_multipart_many_files(myfiles):
return aiohttp.web.json_response(
data={
'files_count': len(myfiles),
'myfiles_content': [ f.file.read().decode('utf8') for f in myfiles ]
},
)
async def aiohttp_multipart_mixed_single_file(myfile, body):
dir_name = body['dir_name']
return aiohttp.web.json_response(
data={
'dir_name': dir_name,
'fileName': myfile.filename,
'myfile_content': myfile.file.read().decode('utf8'),
},
)
async def aiohttp_multipart_mixed_many_files(myfiles, body):
dir_name = body['dir_name']
test_count = body['test_count']
return aiohttp.web.json_response(
data={
'files_count': len(myfiles),
'dir_name': dir_name,
'test_count': test_count,
'myfiles_content': [ f.file.read().decode('utf8') for f in myfiles ]
},
)
async def test_cookie_param(request):
return {"cookie_value": request.cookies["test_cookie"]}

5
tests/fakeapi/aiohttp_handlers_async_def.py
View File

@@ -1,5 +0,0 @@
from connexion.lifecycle import ConnexionResponse
async def aiohttp_validate_responses():
return ConnexionResponse(body=b'{"validate": true}')

60
tests/fixtures/aiohttp/datetime_support.yaml vendored
View File

@@ -1,60 +0,0 @@
openapi: "3.0.1"
info:
title: "{{title}}"
version: "1.0"
servers:
- url: http://localhost:8080/v1.0
paths:
/datetime:
get:
summary: Generate data with date time
operationId: fakeapi.aiohttp_handlers.get_datetime
responses:
200:
description: date time example
content:
application/json:
schema:
type: object
properties:
value:
type: string
format: date-time
example:
value: 2000-01-23T04:56:07.000008+00:00
/date:
get:
summary: Generate data with date
operationId: fakeapi.aiohttp_handlers.get_date
responses:
200:
description: date example
content:
application/json:
schema:
type: object
properties:
value:
type: string
format: date
example:
value: 2000-01-23
/uuid:
get:
summary: Generate data with uuid
operationId: fakeapi.aiohttp_handlers.get_uuid
responses:
200:
description: uuid handler
content:
application/json:
schema:
type: object
properties:
value:
type: string
format: uuid
example:
value: 'a7b8869c-5f24-4ce0-a5d1-3e44c3663aa9'

28
tests/fixtures/aiohttp/openapi_empty_base_path.yaml vendored
View File

@@ -1,28 +0,0 @@
openapi: 3.0.0
servers:
- url: /
info:
title: '{{title}}'
version: '1.0'
paths:
'/bye/{name}':
get:
summary: Generate goodbye
description: Generates a goodbye message.
operationId: fakeapi.aiohttp_handlers.get_bye
responses:
'200':
description: goodbye response
content:
text/plain:
schema:
type: string
default:
description: unexpected error
parameters:
- name: name
in: path
description: Name of the person to say bye.
required: true
schema:
type: string

132
tests/fixtures/aiohttp/openapi_multipart.yaml vendored
View File

@@ -1,132 +0,0 @@
---
openapi: 3.0.0
servers:
- url: /v1.0
info:
title: "{{title}}"
version: "1.0"
paths:
"/upload_file":
post:
summary: Uploads single file
description: Handles multipart file upload.
operationId: fakeapi.aiohttp_handlers.aiohttp_multipart_single_file
responses:
"200":
description: OK response
content:
'application/json':
schema:
type: object
properties:
fileName:
type: string
default:
description: unexpected error
requestBody:
required: true
content:
multipart/form-data:
schema:
type: object
properties:
myfile:
type: string
format: binary
"/upload_files":
post:
summary: Uploads many files
description: Handles multipart file upload.
operationId: fakeapi.aiohttp_handlers.aiohttp_multipart_many_files
responses:
"200":
description: OK response
content:
'application/json':
schema:
type: object
properties:
files_count:
type: number
default:
description: unexpected error
requestBody:
required: true
content:
multipart/form-data:
schema:
type: object
properties:
myfiles:
type: array
items:
type: string
format: binary
"/mixed_single_file":
post:
summary: Reads multipart data
description: Handles multipart data reading
operationId: fakeapi.aiohttp_handlers.aiohttp_multipart_mixed_single_file
responses:
"200":
description: OK response
content:
'application/json':
schema:
type: object
properties:
dir_name:
type: string
fileName:
type: string
default:
description: unexpected error
requestBody:
required: true
content:
multipart/form-data:
schema:
type: object
properties:
dir_name:
type: string
myfile:
type: string
format: binary
"/mixed_many_files":
post:
summary: Reads multipart data
description: Handles multipart data reading
operationId: fakeapi.aiohttp_handlers.aiohttp_multipart_mixed_many_files
responses:
"200":
description: OK response
content:
'application/json':
schema:
type: object
properties:
dir_name:
type: string
test_count:
type: number
files_count:
type: number
default:
description: unexpected error
requestBody:
required: true
content:
multipart/form-data:
schema:
type: object
properties:
dir_name:
type: string
test_count:
type: number
myfiles:
type: array
items:
type: string
format: binary

99
tests/fixtures/aiohttp/openapi_secure.yaml vendored
View File

@@ -1,99 +0,0 @@
openapi: 3.0.0
servers:
- url: /v1.0
info:
title: '{{title}}'
version: '1.0'
paths:
'/all_auth':
get:
summary: Test basic and oauth auth
operationId: fakeapi.aiohttp_handlers.aiohttp_all_auth
security:
- oauth:
- myscope
- basic: []
- api_key: []
responses:
'200':
$ref: "#/components/responses/Success"
'/async_auth':
get:
summary: Test async auth
operationId: fakeapi.aiohttp_handlers.aiohttp_async_auth
security:
- async_oauth:
- myscope
- async_basic: []
- async_api_key: []
responses:
'200':
$ref: "#/components/responses/Success"
'/bearer_auth':
get:
summary: Test api key auth
operationId: fakeapi.aiohttp_handlers.aiohttp_bearer_auth
security:
- bearer: []
responses:
'200':
$ref: "#/components/responses/Success"
'/async_bearer_auth':
get:
summary: Test api key auth
operationId: fakeapi.aiohttp_handlers.aiohttp_async_bearer_auth
security:
- async_bearer: []
responses:
'200':
$ref: "#/components/responses/Success"
components:
responses:
Success:
description: "Operation succeed"
content:
application/json:
schema:
type: object
securitySchemes:
oauth:
type: oauth2
x-tokenInfoUrl: 'https://oauth.example/token_info'
flows:
password:
tokenUrl: 'https://oauth.example/token'
scopes:
myscope: can do stuff
basic:
type: http
scheme: basic
x-basicInfoFunc: fakeapi.auth.fake_basic_auth
api_key:
type: apiKey
in: header
name: X-API-Key
x-apikeyInfoFunc: fakeapi.auth.fake_json_auth
bearer:
type: http
scheme: bearer
x-bearerInfoFunc: fakeapi.auth.fake_json_auth
async_oauth:
type: oauth2
flows: {}
x-tokenInfoFunc: fakeapi.auth.async_json_auth
x-scopeValidateFunc: fakeapi.auth.async_scope_validation
async_basic:
type: http
scheme: basic
x-basicInfoFunc: fakeapi.auth.async_basic_auth
async_api_key:
type: apiKey
in: cookie
name: X-API-Key
x-apikeyInfoFunc: fakeapi.auth.async_json_auth
async_bearer:
type: http
scheme: bearer
x-bearerInfoFunc: fakeapi.auth.async_json_auth

35
tests/fixtures/aiohttp/openapi_simple.yaml vendored
View File

@@ -1,35 +0,0 @@
openapi: 3.0.0
servers:
- url: /v1.0
info:
title: '{{title}}'
version: '1.0'
paths:
'/pythonic/{id}':
get:
description: test overloading pythonic snake-case and builtins
operationId: fakeapi.aiohttp_handlers.aiohttp_echo
parameters:
- name: id
description: id field
in: path
required: true
schema:
type: integer
responses:
'200':
description: ok
security: []
/test-cookie-param:
get:
summary: Test cookie parameter support.
operationId: fakeapi.aiohttp_handlers.test_cookie_param
parameters:
- name: test_cookie
in: cookie
required: true
schema:
type: string
responses:
'200':
description: OK

29
tests/fixtures/aiohttp/swagger_empty_base_path.yaml vendored
View File

@@ -1,29 +0,0 @@
swagger: "2.0"
info:
title: "{{title}}"
version: "1.0"
basePath: /
paths:
/bye/{name}:
get:
summary: Generate goodbye
description: Generates a goodbye message.
operationId: fakeapi.aiohttp_handlers.get_bye
produces:
- text/plain
responses:
'200':
description: goodbye response
schema:
type: string
default:
description: "unexpected error"
parameters:
- name: name
in: path
description: Name of the person to say bye.
required: true
type: string

40
tests/fixtures/aiohttp/swagger_secure.yaml vendored
View File

@@ -1,40 +0,0 @@
swagger: "2.0"
info:
title: "{{title}}"
version: "1.0"
basePath: /v1.0
securityDefinitions:
oauth:
type: oauth2
flow: password
tokenUrl: https://oauth.example/token
x-tokenInfoUrl: https://oauth.example/token_info
scopes:
myscope: can do stuff
basic:
type: basic
x-basicInfoFunc: fakeapi.auth.fake_basic_auth
api_key:
type: apiKey
in: header
name: X-API-Key
x-apikeyInfoFunc: fakeapi.auth.fake_json_auth
security:
- oauth:
- myscope
- basic: []
- api_key: []
paths:
/all_auth:
get:
summary: Test different authentication
operationId: fakeapi.aiohttp_handlers.aiohttp_token_info
responses:
200:
description: greeting response
schema:
type: object

203
tests/fixtures/aiohttp/swagger_simple.yaml vendored
View File

@@ -1,203 +0,0 @@
swagger: "2.0"
info:
title: "{{title}}"
version: "1.0"
basePath: /v1.0
paths:
/bye/{name}:
get:
summary: Generate goodbye
description: Generates a goodbye message.
operationId: fakeapi.aiohttp_handlers.get_bye
produces:
- text/plain
responses:
'200':
description: goodbye response
schema:
type: string
default:
description: "unexpected error"
parameters:
- name: name
in: path
description: Name of the person to say bye.
required: true
type: string
/aiohttp_str_response:
get:
summary: Return a str response
description: Test returning a str response
operationId: fakeapi.aiohttp_handlers.aiohttp_str_response
produces:
- text/plain
responses:
200:
description: json response
schema:
type: string
/aiohttp_non_str_non_json_response:
get:
summary: Return a non str and non json response
description: Test returning a non str and non json response
operationId: fakeapi.aiohttp_handlers.aiohttp_non_str_non_json_response
produces:
- text/plain
responses:
200:
description: non str non json response
/aiohttp_bytes_response:
get:
summary: Return a bytes response
description: Test returning a bytes response
operationId: fakeapi.aiohttp_handlers.aiohttp_bytes_response
produces:
- text/plain
responses:
200:
description: bytes response
/aiohttp_validate_responses:
get:
summary: Return a bytes response
description: Test returning a bytes response
operationId: fakeapi.aiohttp_handlers.aiohttp_validate_responses
produces:
- application/json
responses:
200:
description: json response
schema:
type: object
/aiohttp_access_request_context:
post:
summary: Test request context access
description: Test request context access in handlers.
operationId: fakeapi.aiohttp_handlers.aiohttp_access_request_context
responses:
204:
description: success no content.
/users/:
get:
summary: Test get users
description: Get test users list
operationId: fakeapi.aiohttp_handlers.aiohttp_users_get
produces:
- application/json
responses:
200:
description: Return users
schema:
type: array
items:
$ref: '#/definitions/User'
post:
summary: Create a new user
description: Add new user to a list of users
operationId: fakeapi.aiohttp_handlers.aiohttp_users_post
consumes:
- application/json
parameters:
- in: body
name: user
description: The user to create
schema:
$ref: '#/definitions/User'
responses:
201:
description: json response
schema:
$ref: '#/definitions/User'
/aiohttp_query_parsing_str:
get:
summary: Test proper parsing of query parameters
description: Tests proper parsing
operationId: fakeapi.aiohttp_handlers.aiohttp_query_parsing_str
parameters:
- in: query
name: query
description: Simple query param
type: string
required: true
responses:
200:
description: Query parsing result
schema:
$ref: '#/definitions/SimpleQuery'
/aiohttp_query_parsing_array:
get:
summary: Test proper parsing of query parameters
description: Tests proper parsing
operationId: fakeapi.aiohttp_handlers.aiohttp_query_parsing_array
parameters:
- in: query
name: query
description: Array like query param
type: array
items:
type: string
required: true
responses:
200:
description: Query parsing result
schema:
$ref: '#/definitions/MultiQuery'
/aiohttp_query_parsing_array_multi:
get:
summary: Test proper parsing of query parameters
description: Tests proper parsing
operationId: fakeapi.aiohttp_handlers.aiohttp_query_parsing_array_multi
parameters:
- in: query
name: query
description: Array like query param
type: array
items:
type: string
collectionFormat: multi
required: true
responses:
200:
description: Query parsing result
schema:
$ref: '#/definitions/MultiQuery'
definitions:
SimpleQuery:
type: object
required:
- query
properties:
query:
type: string
MultiQuery:
type: object
required:
- query
properties:
query:
type: array
items:
type: string
User:
type: object
required:
- name
properties:
id:
type: number
name:
type: string

21
tests/fixtures/aiohttp/swagger_simple_async_def.yaml vendored
View File

@@ -1,21 +0,0 @@
swagger: "2.0"
info:
title: "{{title}}"
version: "1.0"
basePath: /v1.0
paths:
/aiohttp_validate_responses:
get:
summary: Return a bytes response
description: Test returning a bytes response
operationId: fakeapi.aiohttp_handlers_async_def.aiohttp_validate_responses
produces:
- application/json
responses:
200:
description: json response
schema:
type: object

40
tests/test_cli.py
View File

@@ -257,23 +257,6 @@ def test_run_with_wsgi_containers(mock_app_run, spec_file):
assert result.exit_code == 0
def test_run_with_aiohttp_not_installed(mock_app_run, spec_file):
import sys
aiohttp_bkp = sys.modules.pop('aiohttp', None)
sys.modules['aiohttp'] = None
runner = CliRunner()
# missing aiohttp
result = runner.invoke(main,
['run', spec_file, '-f', 'aiohttp'],
catch_exceptions=False)
sys.modules['aiohttp'] = aiohttp_bkp
assert 'aiohttp library is not installed' in result.output
assert result.exit_code == 1
def test_run_with_wsgi_server_and_server_opts(mock_app_run, spec_file):
runner = CliRunner()
@@ -284,26 +267,3 @@ def test_run_with_wsgi_server_and_server_opts(mock_app_run, spec_file):
catch_exceptions=False)
assert "these options are mutually exclusive" in result.output
assert result.exit_code == 2
def test_run_with_incompatible_server_and_default_framework(mock_app_run, spec_file):
runner = CliRunner()
result = runner.invoke(main,
['run', spec_file,
'-s', 'aiohttp'],
catch_exceptions=False)
assert "Invalid server 'aiohttp' for app-framework 'flask'" in result.output
assert result.exit_code == 2
def test_run_with_incompatible_server_and_framework(mock_app_run, spec_file):
runner = CliRunner()
result = runner.invoke(main,
['run', spec_file,
'-s', 'flask',
'-f', 'aiohttp'],
catch_exceptions=False)
assert "Invalid server 'flask' for app-framework 'aiohttp'" in result.output
assert result.exit_code == 2

8
tox.ini
View File

@@ -27,11 +27,11 @@ setenv=PYTHONPATH = {toxinidir}:{toxinidir}
deps=pytest
commands=
pip install Requirements-Builder
min: requirements-builder --level=min --extras aiohttp -o {toxworkdir}/requirements-min.txt setup.py
min: requirements-builder --level=min -o {toxworkdir}/requirements-min.txt setup.py
min: pip install --upgrade -r {toxworkdir}/requirements-min.txt
pypi: requirements-builder --level=pypi --extras aiohttp -o {toxworkdir}/requirements-pypi.txt setup.py
pypi: requirements-builder --level=pypi -o {toxworkdir}/requirements-pypi.txt setup.py
pypi: pip install --upgrade -r {toxworkdir}/requirements-pypi.txt
dev: requirements-builder --level=dev --extras aiohttp --req=requirements-devel.txt -o {toxworkdir}/requirements-dev.txt setup.py
dev: requirements-builder --level=dev --req=requirements-devel.txt -o {toxworkdir}/requirements-dev.txt setup.py
dev: pip install --upgrade -r {toxworkdir}/requirements-dev.txt
python setup.py test
@@ -57,7 +57,7 @@ commands=isort --thirdparty connexion --check-only --diff .
basepython=python3
deps=isort==5.9.1
changedir={toxinidir}/tests
commands=isort --thirdparty aiohttp,connexion --check-only --diff .
commands=isort --thirdparty connexion --check-only --diff .
[testenv:mypy]
deps=