mirror of
https://github.com/LukeHagar/connexion.git
synced 2025-12-09 12:27:46 +00:00
edb0381af3
This PR adds an interface for the ConnexionMiddleware, similar to the
interface of the Connexion Apps.
The Connexion Apps are now a simple wrapper around the
ConnexionMiddleware and framework app, delegating the work to the
middleware. This enables a similar interface and behavior for users when
using either the middleware or apps.
The arguments are repeated everywhere there is a user interface, but are
parsed in a central place. Repeating the arguments is not DRY, but
needed to provide users with IDE autocomplete, typing, etc. They are
parsed in a single `_Options` class, which also provides a mechanism to
set default options on an App level, and override them on the more
granular API level.
This makes the long list of provided parameters a lot more manageable,
so I would like to use it for the `Jsonifier` as well, and re-add the
`debug` and `extra_files` arguments which I have dropped in previous
PRs. I'll submit a separate PR for this.
I renamed the `options` parameter to `swagger_ui_options` since it only
contains swagger UI options. This is a breaking change though, and we'll
need to highlight this upon release.
We still have quite a lot of `App`, `MiddlewareApp`, and abstract
classes. It would be great if we could find a way to reduce those
further, or at least find better naming to make it more clear what each
one does 🙂 .
Finally, I added examples on how the middleware can be used with third
party frameworks under `examples/frameworks`. Currently there's an
example for Starlette and Quart, but this should be easy to extend. They
also show how the `ASGIDecorator` and `StarletteDecorator` from my
previous PR can be used.
328 lines
12 KiB
Python
328 lines
12 KiB
Python
import json
|
|
from unittest import mock
|
|
|
|
import jinja2
|
|
import pytest
|
|
import yaml
|
|
from connexion import App
|
|
from connexion.exceptions import InvalidSpecification
|
|
from connexion.http_facts import METHODS
|
|
from connexion.json_schema import ExtendedSafeLoader
|
|
from connexion.middleware.abstract import AbstractRoutingAPI
|
|
|
|
from conftest import TEST_FOLDER, build_app_from_fixture
|
|
|
|
SPECS = ["swagger.yaml", "openapi.yaml"]
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_app_with_relative_path(simple_api_spec_dir, spec):
|
|
# Create the app with a relative path and run the test_app testcase below.
|
|
app = App(
|
|
__name__,
|
|
specification_dir=".." / simple_api_spec_dir.relative_to(TEST_FOLDER),
|
|
)
|
|
app.add_api(spec)
|
|
|
|
app_client = app.test_client()
|
|
get_bye = app_client.get("/v1.0/bye/jsantos") # type: flask.Response
|
|
assert get_bye.status_code == 200
|
|
assert get_bye.data == b"Goodbye jsantos"
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_app_with_resolver(simple_api_spec_dir, spec):
|
|
from connexion.resolver import Resolver
|
|
|
|
resolver = Resolver()
|
|
app = App(
|
|
__name__,
|
|
specification_dir=".." / simple_api_spec_dir.relative_to(TEST_FOLDER),
|
|
resolver=resolver,
|
|
)
|
|
api = app.add_api(spec)
|
|
assert api.resolver is resolver
|
|
|
|
|
|
def test_app_with_different_uri_parser(simple_api_spec_dir):
|
|
from connexion.uri_parsing import FirstValueURIParser
|
|
|
|
app = App(
|
|
__name__,
|
|
specification_dir=".." / simple_api_spec_dir.relative_to(TEST_FOLDER),
|
|
uri_parser_class=FirstValueURIParser,
|
|
)
|
|
app.add_api("swagger.yaml")
|
|
|
|
app_client = app.test_client()
|
|
resp = app_client.get(
|
|
"/v1.0/test_array_csv_query_param?items=a,b,c&items=d,e,f"
|
|
) # type: flask.Response
|
|
assert resp.status_code == 200
|
|
j = json.loads(resp.get_data(as_text=True))
|
|
assert j == ["a", "b", "c"]
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_swagger_ui(simple_api_spec_dir, spec):
|
|
app = App(__name__, specification_dir=simple_api_spec_dir)
|
|
app.add_api(spec)
|
|
app_client = app.test_client()
|
|
swagger_ui = app_client.get("/v1.0/ui/") # type: flask.Response
|
|
assert swagger_ui.status_code == 200
|
|
spec_json_filename = "/v1.0/{spec}".format(spec=spec.replace("yaml", "json"))
|
|
assert spec_json_filename.encode() in swagger_ui.data
|
|
if "openapi" in spec:
|
|
assert b"swagger-ui-config.json" not in swagger_ui.data
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_swagger_ui_with_config(simple_api_spec_dir, spec):
|
|
swagger_ui_config = {"displayOperationId": True}
|
|
swagger_ui_options = {"swagger_ui_config": swagger_ui_config}
|
|
app = App(
|
|
__name__,
|
|
specification_dir=simple_api_spec_dir,
|
|
swagger_ui_options=swagger_ui_options,
|
|
)
|
|
app.add_api(spec)
|
|
app_client = app.test_client()
|
|
swagger_ui = app_client.get("/v1.0/ui/") # type: flask.Response
|
|
assert swagger_ui.status_code == 200
|
|
if "openapi" in spec:
|
|
assert b'configUrl: "swagger-ui-config.json"' in swagger_ui.data
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_no_swagger_ui(simple_api_spec_dir, spec):
|
|
swagger_ui_options = {"swagger_ui": False}
|
|
app = App(
|
|
__name__,
|
|
specification_dir=simple_api_spec_dir,
|
|
swagger_ui_options=swagger_ui_options,
|
|
)
|
|
app.add_api(spec)
|
|
|
|
app_client = app.test_client()
|
|
swagger_ui = app_client.get("/v1.0/ui/") # type: flask.Response
|
|
assert swagger_ui.status_code == 404
|
|
|
|
app2 = App(__name__, specification_dir=simple_api_spec_dir)
|
|
app2.add_api(spec, swagger_ui_options={"swagger_ui": False})
|
|
app2_client = app2.test_client()
|
|
swagger_ui2 = app2_client.get("/v1.0/ui/") # type: flask.Response
|
|
assert swagger_ui2.status_code == 404
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_swagger_ui_config_json(simple_api_spec_dir, spec):
|
|
"""Verify the swagger-ui-config.json file is returned for swagger_ui_config option passed to app."""
|
|
swagger_ui_config = {"displayOperationId": True}
|
|
swagger_ui_options = {"swagger_ui_config": swagger_ui_config}
|
|
app = App(
|
|
__name__,
|
|
specification_dir=simple_api_spec_dir,
|
|
swagger_ui_options=swagger_ui_options,
|
|
)
|
|
app.add_api(spec)
|
|
app_client = app.test_client()
|
|
url = "/v1.0/ui/swagger-ui-config.json"
|
|
swagger_ui_config_json = app_client.get(url) # type: flask.Response
|
|
assert swagger_ui_config_json.status_code == 200
|
|
assert swagger_ui_config == json.loads(
|
|
swagger_ui_config_json.get_data(as_text=True)
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_no_swagger_ui_config_json(simple_api_spec_dir, spec):
|
|
"""Verify the swagger-ui-config.json file is not returned when the swagger_ui_config option not passed to app."""
|
|
app = App(__name__, specification_dir=simple_api_spec_dir)
|
|
app.add_api(spec)
|
|
app_client = app.test_client()
|
|
url = "/v1.0/ui/swagger-ui-config.json"
|
|
swagger_ui_config_json = app_client.get(url) # type: flask.Response
|
|
assert swagger_ui_config_json.status_code == 404
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_swagger_json_app(simple_api_spec_dir, spec):
|
|
"""Verify the spec json file is returned for default setting passed to app."""
|
|
app = App(__name__, specification_dir=simple_api_spec_dir)
|
|
app.add_api(spec)
|
|
app_client = app.test_client()
|
|
url = "/v1.0/{spec}"
|
|
url = url.format(spec=spec.replace("yaml", "json"))
|
|
spec_json = app_client.get(url) # type: flask.Response
|
|
assert spec_json.status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_swagger_yaml_app(simple_api_spec_dir, spec):
|
|
"""Verify the spec yaml file is returned for default setting passed to app."""
|
|
app = App(__name__, specification_dir=simple_api_spec_dir)
|
|
app.add_api(spec)
|
|
app_client = app.test_client()
|
|
url = "/v1.0/{spec}"
|
|
url = url.format(spec=spec)
|
|
spec_response = app_client.get(url) # type: flask.Response
|
|
assert spec_response.status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_no_swagger_json_app(simple_api_spec_dir, spec):
|
|
"""Verify the spec json file is not returned when set to False when creating app."""
|
|
swagger_ui_options = {"serve_spec": False}
|
|
app = App(
|
|
__name__,
|
|
specification_dir=simple_api_spec_dir,
|
|
swagger_ui_options=swagger_ui_options,
|
|
)
|
|
app.add_api(spec)
|
|
|
|
app_client = app.test_client()
|
|
url = "/v1.0/{spec}"
|
|
url = url.format(spec=spec.replace("yaml", "json"))
|
|
spec_json = app_client.get(url) # type: flask.Response
|
|
assert spec_json.status_code == 404
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_dict_as_yaml_path(simple_api_spec_dir, spec):
|
|
openapi_yaml_path = simple_api_spec_dir / spec
|
|
|
|
with openapi_yaml_path.open(mode="rb") as openapi_yaml:
|
|
contents = openapi_yaml.read()
|
|
try:
|
|
openapi_template = contents.decode()
|
|
except UnicodeDecodeError:
|
|
openapi_template = contents.decode("utf-8", "replace")
|
|
|
|
openapi_string = jinja2.Template(openapi_template).render({})
|
|
specification = yaml.load(openapi_string, ExtendedSafeLoader) # type: dict
|
|
|
|
app = App(__name__, specification_dir=simple_api_spec_dir)
|
|
app.add_api(specification)
|
|
|
|
app_client = app.test_client()
|
|
url = "/v1.0/{spec}".format(spec=spec.replace("yaml", "json"))
|
|
swagger_json = app_client.get(url) # type: flask.Response
|
|
assert swagger_json.status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_swagger_json_api(simple_api_spec_dir, spec):
|
|
"""Verify the spec json file is returned for default setting passed to api."""
|
|
app = App(__name__, specification_dir=simple_api_spec_dir)
|
|
app.add_api(spec)
|
|
|
|
app_client = app.test_client()
|
|
url = "/v1.0/{spec}".format(spec=spec.replace("yaml", "json"))
|
|
swagger_json = app_client.get(url) # type: flask.Response
|
|
assert swagger_json.status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_no_swagger_json_api(simple_api_spec_dir, spec):
|
|
"""Verify the spec json file is not returned when set to False when adding api."""
|
|
app = App(__name__, specification_dir=simple_api_spec_dir)
|
|
app.add_api(spec, swagger_ui_options={"serve_spec": False})
|
|
|
|
app_client = app.test_client()
|
|
url = "/v1.0/{spec}".format(spec=spec.replace("yaml", "json"))
|
|
swagger_json = app_client.get(url) # type: flask.Response
|
|
assert swagger_json.status_code == 404
|
|
|
|
|
|
def test_swagger_json_content_type(simple_app):
|
|
app_client = simple_app.test_client()
|
|
spec = simple_app._spec_file
|
|
url = "/v1.0/{spec}".format(spec=spec.replace("yaml", "json"))
|
|
response = app_client.get(url) # type: flask.Response
|
|
assert response.status_code == 200
|
|
assert response.content_type == "application/json"
|
|
|
|
|
|
def test_single_route():
|
|
app = App(__name__)
|
|
|
|
def route1():
|
|
return "single 1"
|
|
|
|
@app.route("/single2", methods=["POST"])
|
|
def route2():
|
|
return "single 2"
|
|
|
|
app_client = app.test_client()
|
|
|
|
app.add_url_rule("/single1", "single1", route1, methods=["GET"])
|
|
|
|
get_single1 = app_client.get("/single1") # type: flask.Response
|
|
assert get_single1.data == b"single 1"
|
|
|
|
post_single1 = app_client.post("/single1") # type: flask.Response
|
|
assert post_single1.status_code == 405
|
|
|
|
post_single2 = app_client.post("/single2") # type: flask.Response
|
|
assert post_single2.data == b"single 2"
|
|
|
|
get_single2 = app_client.get("/single2") # type: flask.Response
|
|
assert get_single2.status_code == 405
|
|
|
|
|
|
def test_resolve_method(simple_app):
|
|
app_client = simple_app.test_client()
|
|
resp = app_client.get("/v1.0/resolver-test/method") # type: flask.Response
|
|
assert resp.data == b'"DummyClass"\n'
|
|
|
|
|
|
def test_resolve_classmethod(simple_app):
|
|
app_client = simple_app.test_client()
|
|
resp = app_client.get("/v1.0/resolver-test/classmethod") # type: flask.Response
|
|
assert resp.data.decode("utf-8", "replace") == '"DummyClass"\n'
|
|
|
|
|
|
@pytest.mark.parametrize("spec", SPECS)
|
|
def test_add_api_with_function_resolver_function_is_wrapped(simple_api_spec_dir, spec):
|
|
app = App(__name__, specification_dir=simple_api_spec_dir)
|
|
api = app.add_api(spec, resolver=lambda oid: (lambda foo: "bar"))
|
|
assert api.resolver.resolve_function_from_operation_id("faux")("bah") == "bar"
|
|
|
|
|
|
def test_default_query_param_does_not_match_defined_type(default_param_error_spec_dir):
|
|
with pytest.raises(InvalidSpecification):
|
|
build_app_from_fixture(default_param_error_spec_dir, validate_responses=True)
|
|
|
|
|
|
def test_handle_add_operation_error(simple_api_spec_dir, monkeypatch):
|
|
app = App(__name__, specification_dir=simple_api_spec_dir)
|
|
monkeypatch.setattr(
|
|
AbstractRoutingAPI,
|
|
"add_operation",
|
|
mock.MagicMock(side_effect=Exception("operation error!")),
|
|
)
|
|
with pytest.raises(Exception):
|
|
app.add_api("swagger.yaml", resolver=lambda oid: (lambda foo: "bar"))
|
|
|
|
|
|
def test_using_all_fields_in_path_item(simple_api_spec_dir):
|
|
"""Test that connexion will try to add an endpoint only on http methods.
|
|
|
|
test also that each http methods has its own endpoint.
|
|
"""
|
|
app = App(__name__, specification_dir=simple_api_spec_dir)
|
|
app.add_api("openapi.yaml")
|
|
|
|
test_methods = set()
|
|
for rule in app.app.url_map.iter_rules():
|
|
if rule.rule != "/v1.0/add_operation_on_http_methods_only":
|
|
continue
|
|
test_methods.update({method.lower() for method in rule.methods})
|
|
assert set(test_methods) == METHODS
|
|
|
|
|
|
def test_async_route(simple_app):
|
|
app_client = simple_app.test_client()
|
|
resp = app_client.get("/v1.0/async-route")
|
|
assert resp.status_code == 200
|