Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_bad_operation_id():
# Unresolvable operationIDs should result in a well-defined error that can
# be handled upstream.
with pytest.raises(ResolverError):
Resolver().resolve_function_from_operation_id('ohai.I.do.not.exist')
with pytest.raises(ResolverError):
RestyResolver('connexion').resolve_function_from_operation_id('ohai.I.do.not.exist')
def test_resty_resolve_x_router_controller_without_operation_id():
operation = Swagger2Operation(api=None,
method='GET',
path='/hello/{id}',
path_parameters=[],
operation={'x-swagger-router-controller': 'fakeapi.hello'},
app_produces=['application/json'],
app_consumes=['application/json'],
app_security=[],
security_definitions={},
definitions={},
parameter_definitions=PARAMETER_DEFINITIONS,
resolver=RestyResolver('fakeapi'))
assert operation.operation_id == 'fakeapi.hello.get'
def test_resty_resolve_with_default_module_name_can_resolve_api_root():
operation = Swagger2Operation(api=None,
method='GET',
path='/',
path_parameters=[],
operation={},
app_produces=['application/json'],
app_consumes=['application/json'],
app_security=[],
security_definitions={},
definitions={},
parameter_definitions=PARAMETER_DEFINITIONS,
resolver=RestyResolver('fakeapi'))
assert operation.operation_id == 'fakeapi.get'
def test_resty_resolve_with_default_module_name_will_translate_dashes_in_resource_name():
operation = Swagger2Operation(api=None,
method='GET',
path='/foo-bar',
path_parameters=[],
operation={},
app_produces=['application/json'],
app_consumes=['application/json'],
app_security=[],
security_definitions={},
definitions={},
parameter_definitions=PARAMETER_DEFINITIONS,
resolver=RestyResolver('fakeapi'))
assert operation.operation_id == 'fakeapi.foo_bar.search'
def test_resty_resolve_with_default_module_name_will_resolve_resource_root_as_configured():
operation = Swagger2Operation(api=None,
method='GET',
path='/hello',
path_parameters=[],
operation={},
app_produces=['application/json'],
app_consumes=['application/json'],
app_security=[],
security_definitions={},
definitions={},
parameter_definitions=PARAMETER_DEFINITIONS,
resolver=RestyResolver('fakeapi', 'api_list'))
assert operation.operation_id == 'fakeapi.hello.api_list'
__name__,
validator_map={
'body': DSSRequestBodyValidator,
'parameter': DSSParameterValidator,
},
)
# The Flask/Connection app's logger has its own multi-line formatter and configuration. Rather than suppressing
# it we let it do its thing, give it a special name and only enable it if DSS_DEBUG > 1. Most of the DSS web
# app's logging is done through the DSSChaliceApp.app logger not the Flask app's logger.
#
app.app.logger_name = 'dss.api'
debug = Config.debug_level() > 0
app.app.debug = debug
app.app.logger.info('Flask debug is %s.', 'enabled' if debug else 'disabled')
resolver = RestyResolver("dss.api", collection_endpoint_name="list")
app.add_api('../dss-api.yml', resolver=resolver, validate_responses=True, arguments=os.environ)
app.add_error_handler(DSSException, dss_exception_handler)
return app
:param path: The current path where is instantiated Microservice class:
```
Microservice(path=__file__)
^^^^--- This param
```
:return: Flask
"""
check_package_exists("connexion")
specification_dir = self.path
application_root = self._get_application_root(config)
if not os.path.isabs(self.path):
specification_dir = os.path.join(path, self.path)
app = connexion.App(__name__,
specification_dir=specification_dir,
resolver=RestyResolver(self.project_dir))
params = {
"specification": get_bundled_specs(
Path(os.path.join(specification_dir, self.file))) if prance else self.file,
"arguments": {'title': config.APP_NAME},
"base_path": application_root,
"options": {"swagger_url": self.url},
}
# Fix Connexion issue https://github.com/zalando/connexion/issues/1135
if application_root == "/":
del params["base_path"]
app.add_api(**params)
# Invert the objects, instead connexion with a Flask object, a Flask object with
application = app.app
import connexion
import logging
import os
from connexion.resolver import RestyResolver
from flask import redirect
logging.basicConfig(level=logging.INFO)
def go_to_ui():
return redirect('/api/v1/ui')
if __name__ == '__main__':
port = int(os.environ.get("PORT", 5000))
app = connexion.App(__name__)
app.app.add_url_rule('/', 'index', go_to_ui)
app.add_api('scaner_api.yaml', arguments={'title': 'Scaner\'s API'}, resolver=RestyResolver('scaner.controllers'))
from scaner import tasks
app.app.tasks = tasks
app.run(host='0.0.0.0',port=port)
return name
def get_function_name():
method = operation.method
is_collection_endpoint = \
method.lower() == 'get' \
and path_match.group('resource_name') \
and not path_match.group('extended_path')
return self.collection_endpoint_name if is_collection_endpoint else method.lower()
return '{}.{}'.format(get_controller_name(), get_function_name())
class MethodViewResolver(RestyResolver):
"""
Resolves endpoint functions based on Flask's MethodView semantics, e.g. ::
paths:
/foo_bar:
get:
# Implied function call: api.FooBarView().get
class FooBarView(MethodView):
def get(self):
return ...
def post(self):
return ...
"""
def resolve_operation_id(self, operation):
ElasticSearchFactory(
os.environ['ELASTICSEARCH_HOST'],
os.environ['ELASTICSEARCH_PORT'],
),
'rooms',
'room',
room_mapping
)
)
return binder
if __name__ == '__main__':
app = connexion.App(__name__, specification_dir='swagger/')
app.add_api('indexer.yaml', resolver=RestyResolver('api'))
FlaskInjector(app=app.app, modules=[configure])
app.run(port=9090)