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_standard_resolve_x_router_controller():
operation = Swagger2Operation(api=None,
method='GET',
path='endpoint',
path_parameters=[],
operation={
'x-swagger-router-controller': 'fakeapi.hello',
'operationId': 'post_greeting',
},
app_produces=['application/json'],
app_consumes=['application/json'],
app_security=[],
security_definitions={},
definitions={},
parameter_definitions=PARAMETER_DEFINITIONS,
resolver=Resolver())
assert operation.operation_id == 'fakeapi.hello.post_greeting'
def test_default(api):
op_spec = make_operation(OPERATION4)
op_spec['parameters'][1]['default'] = 1
Swagger2Operation(
api=api, method='GET', path='endpoint', path_parameters=[],
operation=op_spec, app_produces=['application/json'],
app_consumes=['application/json'], app_security=[],
security_definitions={}, definitions=DEFINITIONS,
parameter_definitions=PARAMETER_DEFINITIONS, resolver=Resolver()
)
op_spec = make_operation(OPERATION6, parameters=False)
op_spec['parameters'][0]['default'] = {
'keep_stacks': 1,
'image_version': 'one',
'senza_yaml': 'senza.yaml',
'new_traffic': 100
}
Swagger2Operation(
api=api, method='POST', path='endpoint', path_parameters=[],
operation=op_spec, app_produces=['application/json'],
app_consumes=['application/json'], app_security=[],
security_definitions={}, definitions=DEFINITIONS,
parameter_definitions={}, resolver=Resolver()
)
verify_oauth = mock.MagicMock(return_value=dummy)
monkeypatch.setattr('connexion.operations.secure.verify_oauth', verify_oauth)
op_spec = make_operation(OPERATION8)
operation = Swagger2Operation(api=api,
method='GET',
path='endpoint',
path_parameters=[],
operation=op_spec,
app_produces=['application/json'],
app_consumes=['application/json'],
app_security=[],
security_definitions=SECURITY_DEFINITIONS_BOTH,
definitions=DEFINITIONS,
parameter_definitions=PARAMETER_DEFINITIONS,
resolver=Resolver())
assert isinstance(operation.function, types.FunctionType)
security_decorator = operation.security_decorator
assert security_decorator.func is verify_security
assert len(security_decorator.args[0]) == 1
assert security_decorator.args[0][0] is dummy
assert security_decorator.args[1] == ['uid']
call_args = verify_oauth.call_args[0]
assert call_args[0] is math.ceil
assert call_args[1] is validate_scope
assert operation.method == 'GET'
assert operation.produces == ['application/json']
assert operation.consumes == ['application/json']
assert operation.security == [{'oauth': ['uid']}]
expected_body_schema = op_spec["parameters"][0]["schema"]
def test_no_token_info(api):
op_spec = make_operation(OPERATION1)
operation = Swagger2Operation(api=api,
method='GET',
path='endpoint',
path_parameters=[],
operation=op_spec,
app_produces=['application/json'],
app_consumes=['application/json'],
app_security=SECURITY_DEFINITIONS_WO_INFO,
security_definitions=SECURITY_DEFINITIONS_WO_INFO,
definitions=DEFINITIONS,
parameter_definitions=PARAMETER_DEFINITIONS,
resolver=Resolver())
assert isinstance(operation.function, types.FunctionType)
security_decorator = operation.security_decorator
assert security_decorator.func is verify_security
assert len(security_decorator.args[0]) == 0
assert operation.method == 'GET'
assert operation.produces == ['application/json']
assert operation.consumes == ['application/json']
assert operation.security == [{'oauth': ['uid']}]
expected_body_schema = {'definitions': DEFINITIONS}
expected_body_schema.update(DEFINITIONS["new_stack"])
assert operation.body_schema == expected_body_schema
def test_multi_body(api):
with pytest.raises(InvalidSpecification) as exc_info: # type: py.code.ExceptionInfo
op_spec = make_operation(OPERATION2)
operation = Swagger2Operation(api=api,
method='GET',
path='endpoint',
path_parameters=[],
operation=op_spec,
app_produces=['application/json'],
app_consumes=['application/json'],
app_security=[],
security_definitions={},
definitions=DEFINITIONS,
parameter_definitions=PARAMETER_DEFINITIONS,
resolver=Resolver())
operation.body_schema
exception = exc_info.value
assert str(exception) == "GET endpoint There can be one 'body' parameter at most"
assert repr(exception) == """"""
import connexion
from connexion.resolver import Resolver
import connexion.utils as utils
import myapp
app = connexion.App(__name__, specification_dir='swagger/')
def rs(x):
return utils.get_function_from_name("cwl_runner_wes." + x)
app.add_api('proto/workflow_execution.swagger.json', resolver=Resolver(rs))
app.run(port=8080)
:param options: New style options dictionary.
:type options: dict | None
:param pass_context_arg_name: Name of argument in handler functions to pass request context to.
:type pass_context_arg_name: str | None
:param validator_map: map of validators
:type validator_map: dict
:rtype: AbstractAPI
"""
# Turn the resolver_error code into a handler object
self.resolver_error = resolver_error
resolver_error_handler = None
if self.resolver_error is not None:
resolver_error_handler = self._resolver_error_handler
resolver = resolver or self.resolver
resolver = Resolver(resolver) if hasattr(resolver, '__call__') else resolver
auth_all_paths = auth_all_paths if auth_all_paths is not None else self.auth_all_paths
# TODO test if base_path starts with an / (if not none)
arguments = arguments or dict()
arguments = dict(self.arguments, **arguments) # copy global arguments and update with api specfic
if isinstance(specification, dict):
specification = specification
else:
specification = self.specification_dir / specification
api_options = self.options.extend(options)
api = self.api_cls(specification,
base_path=base_path,
arguments=arguments,
def resolve_operation_id(self, operation):
"""
Resolves the operationId using REST semantics unless explicitly configured in the spec
:type operation: connexion.operations.AbstractOperation
"""
if operation.operation_id:
return Resolver.resolve_operation_id(self, operation)
return self.resolve_operation_id_using_rest_semantics(operation)
logging.info("Using config:")
for n in args.__dict__:
logging.info(" %s: %s", n, getattr(args, n))
app = connexion.App(__name__)
backend = utils.get_function_from_name(
args.backend + ".create_backend")(args.opt)
def rs(x):
return getattr(backend, x)
app.add_api(
'openapi/workflow_execution_service.swagger.yaml',
resolver=Resolver(rs))
return app
self.specification = Specification.load(specification, arguments=arguments)
logger.debug('Read specification', extra={'spec': self.specification})
self.options = ConnexionOptions(options, oas_version=self.specification.version)
logger.debug('Options Loaded',
extra={'swagger_ui': self.options.openapi_console_ui_available,
'swagger_path': self.options.openapi_console_ui_from_dir,
'swagger_url': self.options.openapi_console_ui_path})
self._set_base_path(base_path)
logger.debug('Security Definitions: %s', self.specification.security_definitions)
self.resolver = resolver or Resolver()
logger.debug('Validate Responses: %s', str(validate_responses))
self.validate_responses = validate_responses
logger.debug('Strict Request Validation: %s', str(strict_validation))
self.strict_validation = strict_validation
logger.debug('Pythonic params: %s', str(pythonic_params))
self.pythonic_params = pythonic_params
logger.debug('pass_context_arg_name: %s', pass_context_arg_name)
self.pass_context_arg_name = pass_context_arg_name
if self.options.openapi_spec_available:
self.add_openapi_json()
self.add_openapi_yaml()