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

Make tests framework agnostic

Browse Source
This commit is contained in:
Robbe Sneyders
2023-02-13 19:40:48 +01:00
parent 642a5f2dc3
commit 515e324c80
16 changed files with 478 additions and 470 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
connexion/apps/asynchronous.py
View File

@@ -8,6 +8,7 @@ import typing as t
from starlette.responses import Response as StarletteResponse
from starlette.routing import Router
from starlette.testclient import TestClient
from starlette.types import Receive, Scope, Send
from connexion.apps.abstract import AbstractApp
@@ -187,4 +188,4 @@ class AsyncApp(AbstractApp):
"""TODO: implement"""
def test_client(self, **kwargs):
"""TODO: implement"""
return TestClient(self)

9
connexion/apps/flask.py
View File

@@ -10,6 +10,7 @@ import flask
import werkzeug.exceptions
from flask import Response as FlaskResponse
from flask import signals
from flask.testing import FlaskClient
from starlette.types import Receive, Scope, Send
from connexion.apps.abstract import AbstractApp
@@ -253,4 +254,12 @@ class FlaskApp(AbstractApp):
def test_client(self, **kwargs):
self.app.wsgi_app = a2wsgi.ASGIMiddleware(self.middleware)
self.app.test_client_class = ConnexionTestClient
return self.app.test_client(**kwargs)
class ConnexionTestClient(FlaskClient):
def open(self, *args, **kwargs):
# Align with async test client
kwargs["query_string"] = kwargs.pop("params", None)
return super().open(*args, **kwargs)

102
tests/api/conftest.py
View File

@@ -7,17 +7,21 @@ from conftest import FIXTURES_FOLDER, OPENAPI3_SPEC, build_app_from_fixture
@pytest.fixture(scope="session")
def simple_app(spec):
return build_app_from_fixture("simple", validate_responses=True)
def simple_app(spec, app_class):
return build_app_from_fixture(
"simple", app_class=app_class, spec_file=spec, validate_responses=True
)
@pytest.fixture(scope="session")
def simple_openapi_app():
return build_app_from_fixture("simple", OPENAPI3_SPEC, validate_responses=True)
def simple_openapi_app(app_class):
return build_app_from_fixture(
"simple", app_class=app_class, spec_file=OPENAPI3_SPEC, validate_responses=True
)
@pytest.fixture(scope="session")
def reverse_proxied_app(spec):
def reverse_proxied_app(spec, app_class):
class ReverseProxied:
def __init__(self, app, root_path=None, scheme=None, server=None):
self.app = app
@@ -47,62 +51,90 @@ def reverse_proxied_app(spec):
return await self.app(scope, receive, send)
app = build_app_from_fixture("simple", spec, validate_responses=True)
app = build_app_from_fixture(
"simple", app_class=app_class, spec_file=spec, validate_responses=True
)
app.middleware = ReverseProxied(app.middleware, root_path="/reverse_proxied/")
return app
@pytest.fixture(scope="session")
def snake_case_app(spec):
def snake_case_app(spec, app_class):
return build_app_from_fixture(
"snake_case", spec, validate_responses=True, pythonic_params=True
"snake_case",
app_class=app_class,
spec_file=spec,
validate_responses=True,
pythonic_params=True,
)
@pytest.fixture(scope="session")
def invalid_resp_allowed_app(spec):
return build_app_from_fixture("simple", spec, validate_responses=False)
@pytest.fixture(scope="session")
def strict_app(spec):
def invalid_resp_allowed_app(spec, app_class):
return build_app_from_fixture(
"simple", spec, validate_responses=True, strict_validation=True
"simple", app_class=app_class, spec_file=spec, validate_responses=False
)
@pytest.fixture(scope="session")
def problem_app(spec):
return build_app_from_fixture("problem", spec, validate_responses=True)
@pytest.fixture(scope="session")
def schema_app(spec):
return build_app_from_fixture("different_schemas", spec, validate_responses=True)
@pytest.fixture(scope="session")
def secure_endpoint_app(spec):
def strict_app(spec, app_class):
return build_app_from_fixture(
"secure_endpoint",
spec,
"simple",
app_class=app_class,
spec_file=spec,
validate_responses=True,
strict_validation=True,
)
@pytest.fixture(scope="session")
def problem_app(spec, app_class):
return build_app_from_fixture(
"problem", app_class=app_class, spec_file=spec, validate_responses=True
)
@pytest.fixture(scope="session")
def schema_app(spec, app_class):
return build_app_from_fixture(
"different_schemas",
app_class=app_class,
spec_file=spec,
validate_responses=True,
)
@pytest.fixture(scope="session")
def secure_api_app(spec):
options = {"swagger_ui": False}
def secure_endpoint_app(spec, app_class):
return build_app_from_fixture(
"secure_api", spec, options=options, auth_all_paths=True
"secure_endpoint",
app_class=app_class,
spec_file=spec,
validate_responses=True,
)
@pytest.fixture(scope="session")
def unordered_definition_app(spec):
return build_app_from_fixture("unordered_definition", spec)
def secure_api_app(spec, app_class):
options = {"swagger_ui": False}
return build_app_from_fixture(
"secure_api",
app_class=app_class,
spec_file=spec,
options=options,
auth_all_paths=True,
)
@pytest.fixture(scope="session")
def bad_operations_app(spec):
return build_app_from_fixture("bad_operations", spec, resolver_error=501)
def unordered_definition_app(spec, app_class):
return build_app_from_fixture(
"unordered_definition", app_class=app_class, spec_file=spec
)
@pytest.fixture(scope="session")
def bad_operations_app(spec, app_class):
return build_app_from_fixture(
"bad_operations", app_class=app_class, spec_file=spec, resolver_error=501
)

61
tests/api/test_bootstrap.py
View File

@@ -22,7 +22,7 @@ def test_app_with_relative_path(simple_api_spec_dir, spec):
app.add_api(spec)
app_client = app.test_client()
get_bye = app_client.get("/v1.0/bye/jsantos") # type: flask.Response
get_bye = app_client.get("/v1.0/bye/jsantos")
assert get_bye.status_code == 200
assert get_bye.data == b"Goodbye jsantos"
@@ -51,9 +51,7 @@ def test_app_with_different_uri_parser(simple_api_spec_dir):
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
resp = app_client.get("/v1.0/test_array_csv_query_param?items=a,b,c&items=d,e,f")
assert resp.status_code == 200
j = json.loads(resp.get_data(as_text=True))
assert j == ["a", "b", "c"]
@@ -63,7 +61,7 @@ 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
swagger_ui = app_client.get("/v1.0/ui/")
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
@@ -81,7 +79,7 @@ def test_swagger_ui_with_config(simple_api_spec_dir, spec):
)
app.add_api(spec)
app_client = app.test_client()
swagger_ui = app_client.get("/v1.0/ui/") # type: flask.Response
swagger_ui = app_client.get("/v1.0/ui/")
assert swagger_ui.status_code == 200
if "openapi" in spec:
assert b'configUrl: "swagger-ui-config.json"' in swagger_ui.data
@@ -97,13 +95,13 @@ def test_no_swagger_ui(simple_api_spec_dir, spec):
app.add_api(spec)
app_client = app.test_client()
swagger_ui = app_client.get("/v1.0/ui/") # type: flask.Response
swagger_ui = app_client.get("/v1.0/ui/")
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
swagger_ui2 = app2_client.get("/v1.0/ui/")
assert swagger_ui2.status_code == 404
@@ -119,7 +117,7 @@ def test_swagger_ui_config_json(simple_api_spec_dir, spec):
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
swagger_ui_config_json = app_client.get(url)
assert swagger_ui_config_json.status_code == 200
assert swagger_ui_config == json.loads(
swagger_ui_config_json.get_data(as_text=True)
@@ -132,7 +130,7 @@ def test_no_swagger_ui_config_json(simple_api_spec_dir, spec):
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
swagger_ui_config_json = app_client.get(url)
assert swagger_ui_config_json.status_code == 404
@@ -143,7 +141,7 @@ def test_swagger_json_app(simple_api_spec_dir, 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
spec_json = app_client.get(url)
assert spec_json.status_code == 200
@@ -154,7 +152,7 @@ def test_swagger_yaml_app(simple_api_spec_dir, spec):
app_client = app.test_client()
url = "/v1.0/{spec}"
url = url.format(spec=spec)
spec_response = app_client.get(url) # type: flask.Response
spec_response = app_client.get(url)
assert spec_response.status_code == 200
@@ -171,7 +169,7 @@ def test_no_swagger_json_app(simple_api_spec_dir, 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
spec_json = app_client.get(url)
assert spec_json.status_code == 404
@@ -193,7 +191,7 @@ def test_dict_as_yaml_path(simple_api_spec_dir, 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
swagger_json = app_client.get(url)
assert swagger_json.status_code == 200
@@ -204,7 +202,7 @@ def test_swagger_json_api(simple_api_spec_dir, 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
swagger_json = app_client.get(url)
assert swagger_json.status_code == 200
@@ -215,7 +213,7 @@ def test_no_swagger_json_api(simple_api_spec_dir, 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
swagger_json = app_client.get(url)
assert swagger_json.status_code == 404
@@ -223,9 +221,9 @@ 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
response = app_client.get(url)
assert response.status_code == 200
assert response.content_type == "application/json"
assert response.headers.get("content-type") == "application/json"
def test_single_route():
@@ -242,29 +240,29 @@ def test_single_route():
app.add_url_rule("/single1", "single1", route1, methods=["GET"])
get_single1 = app_client.get("/single1") # type: flask.Response
get_single1 = app_client.get("/single1")
assert get_single1.data == b"single 1"
post_single1 = app_client.post("/single1") # type: flask.Response
post_single1 = app_client.post("/single1")
assert post_single1.status_code == 405
post_single2 = app_client.post("/single2") # type: flask.Response
post_single2 = app_client.post("/single2")
assert post_single2.data == b"single 2"
get_single2 = app_client.get("/single2") # type: flask.Response
get_single2 = app_client.get("/single2")
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'
resp = app_client.get("/v1.0/resolver-test/method")
assert resp.text == '"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'
resp = app_client.get("/v1.0/resolver-test/classmethod")
assert resp.text == '"DummyClass"\n'
def test_add_api_with_function_resolver_function_is_wrapped(simple_api_spec_dir, spec):
@@ -273,9 +271,16 @@ def test_add_api_with_function_resolver_function_is_wrapped(simple_api_spec_dir,
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):
def test_default_query_param_does_not_match_defined_type(
default_param_error_spec_dir, app_class, spec
):
with pytest.raises(InvalidSpecification):
build_app_from_fixture(default_param_error_spec_dir, validate_responses=True)
build_app_from_fixture(
default_param_error_spec_dir,
app_class=app_class,
spec_file=spec,
validate_responses=True,
)
def test_handle_add_operation_error(simple_api_spec_dir, monkeypatch):

40
tests/api/test_errors.py
View File

@@ -10,8 +10,8 @@ def fix_data(data):
def test_errors(problem_app):
app_client = problem_app.test_client()
greeting404 = app_client.get("/v1.0/greeting") # type: flask.Response
assert greeting404.content_type == "application/problem+json"
greeting404 = app_client.get("/v1.0/greeting")
assert greeting404.headers.get("content-type") == "application/problem+json"
assert greeting404.status_code == 404
error404 = flask.json.loads(fix_data(greeting404.data))
assert error404["type"] == "about:blank"
@@ -23,19 +23,19 @@ def test_errors(problem_app):
assert error404["status"] == 404
assert "instance" not in error404
get_greeting = app_client.get("/v1.0/greeting/jsantos") # type: flask.Response
assert get_greeting.content_type == "application/problem+json"
get_greeting = app_client.get("/v1.0/greeting/jsantos")
assert get_greeting.headers.get("content-type") == "application/problem+json"
assert get_greeting.status_code == 405
error405 = json.loads(get_greeting.data.decode("utf-8", "replace"))
error405 = json.loads(get_greeting.text)
assert error405["type"] == "about:blank"
assert error405["title"] == "Method Not Allowed"
assert error405["status"] == 405
assert "instance" not in error405
get500 = app_client.get("/v1.0/except") # type: flask.Response
assert get500.content_type == "application/problem+json"
get500 = app_client.get("/v1.0/except")
assert get500.headers.get("content-type") == "application/problem+json"
assert get500.status_code == 500
error500 = json.loads(get500.data.decode("utf-8", "replace"))
error500 = json.loads(get500.text)
assert error500["type"] == "about:blank"
assert error500["title"] == "Internal Server Error"
assert (
@@ -46,21 +46,21 @@ def test_errors(problem_app):
assert error500["status"] == 500
assert "instance" not in error500
get_problem = app_client.get("/v1.0/problem") # type: flask.Response
assert get_problem.content_type == "application/problem+json"
get_problem = app_client.get("/v1.0/problem")
assert get_problem.headers.get("content-type") == "application/problem+json"
assert get_problem.status_code == 402
assert get_problem.headers["x-Test-Header"] == "In Test"
error_problem = json.loads(get_problem.data.decode("utf-8", "replace"))
error_problem = json.loads(get_problem.text)
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"] == 402
assert error_problem["instance"] == "instance1"
get_problem2 = app_client.get("/v1.0/other_problem") # type: flask.Response
assert get_problem2.content_type == "application/problem+json"
get_problem2 = app_client.get("/v1.0/other_problem")
assert get_problem2.headers.get("content-type") == "application/problem+json"
assert get_problem2.status_code == 402
error_problem2 = json.loads(get_problem2.data.decode("utf-8", "replace"))
error_problem2 = json.loads(get_problem2.text)
assert error_problem2["type"] == "about:blank"
assert error_problem2["title"] == "Some Error"
assert error_problem2["detail"] == "Something went wrong somewhere"
@@ -69,20 +69,18 @@ def test_errors(problem_app):
problematic_json = app_client.get(
"/v1.0/json_response_with_undefined_value_to_serialize"
) # type: flask.Response
)
assert problematic_json.status_code == 500
custom_problem = app_client.get("/v1.0/customized_problem_response")
assert custom_problem.status_code == 403
problem_body = json.loads(custom_problem.data.decode("utf-8", "replace"))
problem_body = json.loads(custom_problem.text)
assert "amount" in problem_body
assert problem_body["amount"] == 23.0
problem_as_exception = app_client.get("/v1.0/problem_exception_with_extra_args")
assert problem_as_exception.status_code == 400
problem_as_exception_body = json.loads(
problem_as_exception.data.decode("utf-8", "replace")
)
problem_as_exception_body = json.loads(problem_as_exception.text)
assert "age" in problem_as_exception_body
assert problem_as_exception_body["age"] == 30
@@ -90,9 +88,7 @@ def test_errors(problem_app):
"/v1.0/post_wrong_content_type", data="<html></html>", content_type="text/html"
)
assert unsupported_media_type.status_code == 415
unsupported_media_type_body = json.loads(
unsupported_media_type.data.decode("utf-8", "replace")
)
unsupported_media_type_body = json.loads(unsupported_media_type.text)
assert unsupported_media_type_body["type"] == "about:blank"
assert unsupported_media_type_body["title"] == "Unsupported Media Type"
assert unsupported_media_type_body["detail"].startswith(

12
tests/api/test_headers.py
View File

@@ -4,7 +4,7 @@ import json
def test_headers_jsonifier(simple_app):
app_client = simple_app.test_client()
response = app_client.post("/v1.0/goodday/dan", data={}) # type: flask.Response
response = app_client.post("/v1.0/goodday/dan", data={})
assert response.status_code == 201
# Default Werkzeug behavior was changed in 2.1 (https://github.com/pallets/werkzeug/issues/2352)
assert response.headers["Location"] in ["http://localhost/my/uri", "/my/uri"]
@@ -13,7 +13,7 @@ def test_headers_jsonifier(simple_app):
def test_headers_produces(simple_app):
app_client = simple_app.test_client()
response = app_client.post("/v1.0/goodevening/dan", data={}) # type: flask.Response
response = app_client.post("/v1.0/goodevening/dan", data={})
assert response.status_code == 201
# Default Werkzeug behavior was changed in 2.1 (https://github.com/pallets/werkzeug/issues/2352)
assert response.headers["Location"] in ["http://localhost/my/uri", "/my/uri"]
@@ -22,14 +22,12 @@ def test_headers_produces(simple_app):
def test_header_not_returned(simple_openapi_app):
app_client = simple_openapi_app.test_client()
response = app_client.post(
"/v1.0/goodday/noheader", data={}
) # type: flask.Response
response = app_client.post("/v1.0/goodday/noheader", data={})
assert (
response.status_code == 500
) # view_func has not returned what was promised in spec
assert response.content_type == "application/problem+json"
data = json.loads(response.data.decode("utf-8", "replace"))
assert response.headers.get("content-type") == "application/problem+json"
data = json.loads(response.text)
assert data["type"] == "about:blank"
assert data["title"] == "Response headers do not conform to specification"
assert (

171
tests/api/test_parameters.py
View File

@@ -10,29 +10,21 @@ def test_parameter_validation(simple_app):
url = "/v1.0/test_parameter_validation"
response = app_client.get(
url, query_string={"date": "2015-08-26"}
) # type: flask.Response
response = app_client.get(url, params={"date": "2015-08-26"})
assert response.status_code == 200
for invalid_int in "", "foo", "0.1":
response = app_client.get(
url, query_string={"int": invalid_int}
) # type: flask.Response
response = app_client.get(url, params={"int": invalid_int})
assert response.status_code == 400
response = app_client.get(url, query_string={"int": "123"}) # type: flask.Response
response = app_client.get(url, params={"int": "123"})
assert response.status_code == 200
for invalid_bool in "", "foo", "yes":
response = app_client.get(
url, query_string={"bool": invalid_bool}
) # type: flask.Response
response = app_client.get(url, params={"bool": invalid_bool})
assert response.status_code == 400
response = app_client.get(
url, query_string={"bool": "true"}
) # type: flask.Response
response = app_client.get(url, params={"bool": "true"})
assert response.status_code == 200
@@ -43,7 +35,7 @@ def test_required_query_param(simple_app):
response = app_client.get(url)
assert response.status_code == 400
response = app_client.get(url, query_string={"n": "1.23"})
response = app_client.get(url, params={"n": "1.23"})
assert response.status_code == 200
@@ -52,38 +44,34 @@ def test_array_query_param(simple_app):
headers = {"Content-type": "application/json"}
url = "/v1.0/test_array_csv_query_param"
response = app_client.get(url, headers=headers)
array_response: List[str] = json.loads(response.data.decode("utf-8", "replace"))
array_response: List[str] = json.loads(response.text)
assert array_response == ["squash", "banana"]
url = "/v1.0/test_array_csv_query_param?items=one,two,three"
response = app_client.get(url, headers=headers)
array_response: List[str] = json.loads(response.data.decode("utf-8", "replace"))
array_response: List[str] = json.loads(response.text)
assert array_response == ["one", "two", "three"]
url = "/v1.0/test_array_pipes_query_param?items=1|2|3"
response = app_client.get(url, headers=headers)
array_response: List[int] = json.loads(response.data.decode("utf-8", "replace"))
array_response: List[int] = json.loads(response.text)
assert array_response == [1, 2, 3]
url = "/v1.0/test_array_unsupported_query_param?items=1;2;3"
response = app_client.get(url, headers=headers)
array_response: List[str] = json.loads(
response.data.decode("utf-8", "replace")
response.text
) # unsupported collectionFormat
assert array_response == ["1;2;3"]
url = "/v1.0/test_array_csv_query_param?items=A&items=B&items=C&items=D,E,F"
response = app_client.get(url, headers=headers)
array_response: List[str] = json.loads(
response.data.decode("utf-8", "replace")
) # multi array with csv format
array_response: List[str] = json.loads(response.text) # multi array with csv format
assert array_response == ["D", "E", "F"]
url = "/v1.0/test_array_multi_query_param?items=A&items=B&items=C&items=D,E,F"
response = app_client.get(url, headers=headers)
array_response: List[str] = json.loads(
response.data.decode("utf-8", "replace")
) # multi array with csv format
array_response: List[str] = json.loads(response.text) # multi array with csv format
assert array_response == ["A", "B", "C", "D", "E", "F"]
url = "/v1.0/test_array_pipes_query_param?items=4&items=5&items=6&items=7|8|9"
response = app_client.get(url, headers=headers)
array_response: List[int] = json.loads(
response.data.decode("utf-8", "replace")
response.text
) # multi array with pipes format
assert array_response == [7, 8, 9]
@@ -93,28 +81,26 @@ def test_array_form_param(simple_app):
headers = {"Content-type": "application/x-www-form-urlencoded"}
url = "/v1.0/test_array_csv_form_param"
response = app_client.post(url, headers=headers)
array_response: List[str] = json.loads(response.data.decode("utf-8", "replace"))
array_response: List[str] = json.loads(response.text)
assert array_response == ["squash", "banana"]
url = "/v1.0/test_array_csv_form_param"
response = app_client.post(url, headers=headers, data={"items": "one,two,three"})
array_response: List[str] = json.loads(response.data.decode("utf-8", "replace"))
array_response: List[str] = json.loads(response.text)
assert array_response == ["one", "two", "three"]
url = "/v1.0/test_array_pipes_form_param"
response = app_client.post(url, headers=headers, data={"items": "1|2|3"})
array_response: List[int] = json.loads(response.data.decode("utf-8", "replace"))
array_response: List[int] = json.loads(response.text)
assert array_response == [1, 2, 3]
url = "/v1.0/test_array_csv_form_param"
data = "items=A&items=B&items=C&items=D,E,F"
response = app_client.post(url, headers=headers, data=data)
array_response: List[str] = json.loads(
response.data.decode("utf-8", "replace")
) # multi array with csv format
array_response: List[str] = json.loads(response.text) # multi array with csv format
assert array_response == ["D", "E", "F"]
url = "/v1.0/test_array_pipes_form_param"
data = "items=4&items=5&items=6&items=7|8|9"
response = app_client.post(url, headers=headers, data=data)
array_response: List[int] = json.loads(
response.data.decode("utf-8", "replace")
response.text
) # multi array with pipes format
assert array_response == [7, 8, 9]
@@ -133,7 +119,7 @@ def test_strict_extra_query_param(strict_app):
url = "/v1.0/test_parameter_validation?extra_parameter=true"
resp = app_client.get(url, headers=headers)
assert resp.status_code == 400
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response["detail"] == "Extra query parameter(s) extra_parameter not in spec"
@@ -142,7 +128,7 @@ def test_strict_formdata_param(strict_app):
headers = {"Content-type": "application/x-www-form-urlencoded"}
url = "/v1.0/test_array_csv_form_param"
resp = app_client.post(url, headers=headers, data={"items": "mango"})
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response == ["mango"]
assert resp.status_code == 200
@@ -164,14 +150,14 @@ def test_strict_formdata_param(strict_app):
def test_path_parameter_someint(simple_app, arg, result):
assert isinstance(arg, str) # sanity check
app_client = simple_app.test_client()
resp = app_client.get(f"/v1.0/test-int-path/{arg}") # type: flask.Response
assert resp.data.decode("utf-8", "replace") == f'"{result}"\n'
resp = app_client.get(f"/v1.0/test-int-path/{arg}")
assert resp.text == f'"{result}"\n'
def test_path_parameter_someint__bad(simple_app):
# non-integer values will not match Flask route
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/test-int-path/foo") # type: flask.Response
resp = app_client.get("/v1.0/test-int-path/foo")
assert resp.status_code == 404, resp.text
@@ -197,14 +183,14 @@ def test_path_parameter_someint__bad(simple_app):
def test_path_parameter_somefloat(simple_app, arg, result):
assert isinstance(arg, str) # sanity check
app_client = simple_app.test_client()
resp = app_client.get(f"/v1.0/test-float-path/{arg}") # type: flask.Response
assert resp.data.decode("utf-8", "replace") == f'"{result}"\n'
resp = app_client.get(f"/v1.0/test-float-path/{arg}")
assert resp.text == f'"{result}"\n'
def test_path_parameter_somefloat__bad(simple_app):
# non-float values will not match Flask route
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/test-float-path/123,45") # type: flask.Response
resp = app_client.get("/v1.0/test-float-path/123,45")
assert resp.status_code == 404, resp.text
@@ -212,20 +198,20 @@ def test_default_param(strict_app):
app_client = strict_app.test_client()
resp = app_client.get("/v1.0/test-default-query-parameter")
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response["app_name"] == "connexion"
def test_falsy_param(simple_app):
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/test-falsy-param", query_string={"falsy": 0})
resp = app_client.get("/v1.0/test-falsy-param", params={"falsy": 0})
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response == 0
resp = app_client.get("/v1.0/test-falsy-param")
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response == 1
@@ -233,7 +219,7 @@ def test_formdata_param(simple_app):
app_client = simple_app.test_client()
resp = app_client.post("/v1.0/test-formData-param", data={"formData": "test"})
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response == "test"
@@ -241,7 +227,7 @@ def test_formdata_bad_request(simple_app):
app_client = simple_app.test_client()
resp = app_client.post("/v1.0/test-formData-param")
assert resp.status_code == 400
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response["detail"] in [
"Missing formdata parameter 'formData'",
"'formData' is a required property", # OAS3
@@ -270,7 +256,7 @@ def test_strict_formdata_extra_param(strict_app):
"/v1.0/test-formData-param", data={"formData": "test", "extra_formData": "test"}
)
assert resp.status_code == 400
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert (
response["detail"] == "Extra formData parameter(s) extra_formData not in spec"
)
@@ -281,9 +267,10 @@ def test_formdata_file_upload(simple_app):
resp = app_client.post(
"/v1.0/test-formData-file-upload",
data={"fileData": (BytesIO(b"file contents"), "filename.txt")},
headers={"content-type": "multipart/form-data"},
)
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response == {"filename.txt": "file contents"}
@@ -291,7 +278,7 @@ def test_formdata_file_upload_bad_request(simple_app):
app_client = simple_app.test_client()
resp = app_client.post("/v1.0/test-formData-file-upload")
assert resp.status_code == 400
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response["detail"] in [
"Missing formdata parameter 'fileData'",
"'fileData' is a required property", # OAS3
@@ -303,8 +290,9 @@ def test_formdata_file_upload_missing_param(simple_app):
resp = app_client.post(
"/v1.0/test-formData-file-upload-missing-param",
data={"missing_fileData": (BytesIO(b"file contents"), "example.txt")},
headers={"content-type": "multipart/form-data"},
)
assert resp.status_code == 200
assert resp.status_code == 200, resp.text
def test_body_not_allowed_additional_properties(simple_app):
@@ -317,7 +305,7 @@ def test_body_not_allowed_additional_properties(simple_app):
)
assert resp.status_code == 400
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert "Additional properties are not allowed" in response["detail"]
@@ -326,22 +314,22 @@ def test_bool_as_default_param(simple_app):
resp = app_client.get("/v1.0/test-bool-param")
assert resp.status_code == 200
resp = app_client.get("/v1.0/test-bool-param", query_string={"thruthiness": True})
resp = app_client.get("/v1.0/test-bool-param", params={"thruthiness": True})
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response is True
def test_bool_param(simple_app):
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/test-bool-param", query_string={"thruthiness": True})
resp = app_client.get("/v1.0/test-bool-param", params={"thruthiness": True})
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response is True
resp = app_client.get("/v1.0/test-bool-param", query_string={"thruthiness": False})
resp = app_client.get("/v1.0/test-bool-param", params={"thruthiness": False})
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response is False
@@ -349,13 +337,13 @@ def test_bool_array_param(simple_app):
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/test-bool-array-param?thruthiness=true,true,true")
assert resp.status_code == 200, resp.text
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response is True
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/test-bool-array-param?thruthiness=true,true,false")
assert resp.status_code == 200, resp.text
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response is False
app_client = simple_app.test_client()
@@ -369,7 +357,7 @@ def test_required_param_miss_config(simple_app):
resp = app_client.get("/v1.0/test-required-param")
assert resp.status_code == 400
resp = app_client.get("/v1.0/test-required-param", query_string={"simple": "test"})
resp = app_client.get("/v1.0/test-required-param", params={"simple": "test"})
assert resp.status_code == 200
resp = app_client.get("/v1.0/test-required-param")
@@ -380,7 +368,7 @@ def test_parameters_defined_in_path_level(simple_app):
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/parameters-in-root-path?title=nice-get")
assert resp.status_code == 200
assert json.loads(resp.data.decode("utf-8", "replace")) == ["nice-get"]
assert json.loads(resp.text) == ["nice-get"]
resp = app_client.get("/v1.0/parameters-in-root-path")
assert resp.status_code == 400
@@ -389,10 +377,10 @@ def test_parameters_defined_in_path_level(simple_app):
def test_array_in_path(simple_app):
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/test-array-in-path/one_item")
assert json.loads(resp.data.decode("utf-8", "replace")) == ["one_item"]
assert json.loads(resp.text) == ["one_item"]
resp = app_client.get("/v1.0/test-array-in-path/one_item,another_item")
assert json.loads(resp.data.decode("utf-8", "replace")) == [
assert json.loads(resp.text) == [
"one_item",
"another_item",
]
@@ -401,43 +389,43 @@ def test_array_in_path(simple_app):
def test_nullable_parameter(simple_app):
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/nullable-parameters?time_start=null")
assert json.loads(resp.data.decode("utf-8", "replace")) == "it was None"
assert json.loads(resp.text) == "it was None"
resp = app_client.get("/v1.0/nullable-parameters?time_start=None")
assert json.loads(resp.data.decode("utf-8", "replace")) == "it was None"
assert json.loads(resp.text) == "it was None"
time_start = 1010
resp = app_client.get(f"/v1.0/nullable-parameters?time_start={time_start}")
assert json.loads(resp.data.decode("utf-8", "replace")) == time_start
assert json.loads(resp.text) == time_start
resp = app_client.post("/v1.0/nullable-parameters", data={"post_param": "None"})
assert json.loads(resp.data.decode("utf-8", "replace")) == "it was None"
assert json.loads(resp.text) == "it was None"
resp = app_client.post("/v1.0/nullable-parameters", data={"post_param": "null"})
assert json.loads(resp.data.decode("utf-8", "replace")) == "it was None"
assert json.loads(resp.text) == "it was None"
headers = {"Content-Type": "application/json"}
resp = app_client.put("/v1.0/nullable-parameters", data="null", headers=headers)
assert json.loads(resp.data.decode("utf-8", "replace")) == "it was None"
assert json.loads(resp.text) == "it was None"
resp = app_client.put("/v1.0/nullable-parameters", data="None", headers=headers)
assert json.loads(resp.data.decode("utf-8", "replace")) == "it was None"
assert json.loads(resp.text) == "it was None"
resp = app_client.put(
"/v1.0/nullable-parameters-noargs", data="None", headers=headers
)
assert json.loads(resp.data.decode("utf-8", "replace")) == "hello"
assert json.loads(resp.text) == "hello"
def test_args_kwargs(simple_app):
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/query-params-as-kwargs")
assert resp.status_code == 200
assert json.loads(resp.data.decode("utf-8", "replace")) == {}
assert json.loads(resp.text) == {}
resp = app_client.get("/v1.0/query-params-as-kwargs?foo=a&bar=b")
assert resp.status_code == 200
assert json.loads(resp.data.decode("utf-8", "replace")) == {"foo": "a"}
assert json.loads(resp.text) == {"foo": "a"}
if simple_app._spec_file == "openapi.yaml":
body = {"foo": "a", "bar": "b"}
@@ -448,7 +436,7 @@ def test_args_kwargs(simple_app):
)
assert resp.status_code == 200
# having only kwargs, the handler would have been passed 'body'
assert json.loads(resp.data.decode("utf-8", "replace")) == {
assert json.loads(resp.text) == {
"body": {"foo": "a", "bar": "b"},
}
@@ -457,13 +445,13 @@ def test_param_sanitization(simple_app):
app_client = simple_app.test_client()
resp = app_client.post("/v1.0/param-sanitization")
assert resp.status_code == 200
assert json.loads(resp.data.decode("utf-8", "replace")) == {}
assert json.loads(resp.text) == {}
resp = app_client.post(
"/v1.0/param-sanitization?$query=queryString", data={"$form": "formString"}
)
assert resp.status_code == 200
assert json.loads(resp.data.decode("utf-8", "replace")) == {
assert json.loads(resp.text) == {
"query": "queryString",
"form": "formString",
}
@@ -475,7 +463,7 @@ def test_param_sanitization(simple_app):
headers={"Content-Type": "application/json"},
)
assert resp.status_code == 200
assert json.loads(resp.data.decode("utf-8", "replace")) == body
assert json.loads(resp.text) == body
body = {"body1": "bodyString", "body2": 12, "body3": {"a": "otherString"}}
resp = app_client.post(
@@ -484,7 +472,7 @@ def test_param_sanitization(simple_app):
headers={"Content-Type": "application/json"},
)
assert resp.status_code == 200
assert json.loads(resp.data.decode("utf-8", "replace")) == body
assert json.loads(resp.text) == body
body = {
"body1": "bodyString",
@@ -497,7 +485,7 @@ def test_param_sanitization(simple_app):
headers={"Content-Type": "application/json"},
)
assert resp.status_code == 200
assert json.loads(resp.data.decode("utf-8", "replace")) == body
assert json.loads(resp.text) == body
def test_no_sanitization_in_request_body(simple_app):
@@ -514,7 +502,7 @@ def test_no_sanitization_in_request_body(simple_app):
response = app_client.post("/v1.0/forward", json=data)
assert response.status_code == 200
assert response.json == data
assert json.loads(response.text) == data
def test_parameters_snake_case(snake_case_app):
@@ -558,19 +546,25 @@ def test_parameters_snake_case(snake_case_app):
"/v1.0/test-get-camel-case-version?truthiness=true&orderBy=asc"
)
assert resp.status_code == 200, resp.text
assert resp.get_json() == {"truthiness": True, "order_by": "asc"}
assert json.loads(resp.text) == {"truthiness": True, "order_by": "asc"}
resp = app_client.get("/v1.0/test-get-camel-case-version?truthiness=5")
assert resp.status_code == 400
assert resp.get_json()["detail"].startswith("'5' is not of type 'boolean'")
assert json.loads(resp.text)["detail"].startswith("'5' is not of type 'boolean'")
# Incorrectly cased params should be ignored
resp = app_client.get(
"/v1.0/test-get-camel-case-version?Truthiness=true&order_by=asc"
)
assert resp.status_code == 200
assert resp.get_json() == {"truthiness": False, "order_by": None} # default values
assert json.loads(resp.text) == {
"truthiness": False,
"order_by": None,
} # default values
resp = app_client.get("/v1.0/test-get-camel-case-version?Truthiness=5&order_by=4")
assert resp.status_code == 200
assert resp.get_json() == {"truthiness": False, "order_by": None} # default values
assert json.loads(resp.text) == {
"truthiness": False,
"order_by": None,
} # default values
# TODO: Add tests for body parameters
@@ -579,12 +573,15 @@ def test_get_unicode_request(simple_app):
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/get_unicode_request?price=%C2%A319.99") # £19.99
assert resp.status_code == 200
assert json.loads(resp.data.decode("utf-8"))["price"] == "£19.99"
assert json.loads(resp.text)["price"] == "£19.99"
def test_cookie_param(simple_app):
app_client = simple_app.test_client()
app_client.set_cookie("localhost", "test_cookie", "hello")
try:
app_client = simple_app.test_client(cookies={"test_cookie": "hello"})
except TypeError:
app_client = simple_app.test_client()
app_client.set_cookie("localhost", "test_cookie", "hello")
response = app_client.get("/v1.0/test-cookie-param")
assert response.status_code == 200
assert response.json == {"cookie_value": "hello"}

171
tests/api/test_responses.py
View File

@@ -9,43 +9,39 @@ def test_app(simple_app):
app_client = simple_app.test_client()
# by default the Swagger UI is enabled
swagger_ui = app_client.get("/v1.0/ui/") # type: flask.Response
swagger_ui = app_client.get("/v1.0/ui/")
assert swagger_ui.status_code == 200
assert b"Swagger UI" in swagger_ui.data
assert "Swagger UI" in swagger_ui.text
# test return Swagger UI static files
swagger_icon = app_client.get("/v1.0/ui/swagger-ui.js") # type: flask.Response
swagger_icon = app_client.get("/v1.0/ui/swagger-ui.js")
assert swagger_icon.status_code == 200
post_greeting_url = app_client.post(
"/v1.0/greeting/jsantos/the/third/of/his/name", data={}
) # type: flask.Response
)
assert post_greeting_url.status_code == 200
assert post_greeting_url.content_type == "application/json"
greeting_response_url = json.loads(post_greeting_url.data.decode("utf-8"))
assert post_greeting_url.headers.get("content-type") == "application/json"
greeting_response_url = json.loads(post_greeting_url.text)
assert (
greeting_response_url["greeting"]
== "Hello jsantos thanks for the/third/of/his/name"
)
post_greeting = app_client.post(
"/v1.0/greeting/jsantos", data={}
) # type: flask.Response
post_greeting = app_client.post("/v1.0/greeting/jsantos", data={})
assert post_greeting.status_code == 200
assert post_greeting.content_type == "application/json"
greeting_response = json.loads(post_greeting.data.decode("utf-8"))
assert post_greeting.headers.get("content-type") == "application/json"
greeting_response = json.loads(post_greeting.text)
assert greeting_response["greeting"] == "Hello jsantos"
get_bye = app_client.get("/v1.0/bye/jsantos") # type: flask.Response
get_bye = app_client.get("/v1.0/bye/jsantos")
assert get_bye.status_code == 200
assert get_bye.data == b"Goodbye jsantos"
assert get_bye.text == "Goodbye jsantos"
post_greeting = app_client.post(
"/v1.0/greeting/jsantos", data={}
) # type: flask.Response
post_greeting = app_client.post("/v1.0/greeting/jsantos", data={})
assert post_greeting.status_code == 200
assert post_greeting.content_type == "application/json"
greeting_response = json.loads(post_greeting.data.decode("utf-8"))
assert post_greeting.headers.get("content-type") == "application/json"
greeting_response = json.loads(post_greeting.text)
assert greeting_response["greeting"] == "Hello jsantos"
@@ -65,15 +61,15 @@ def test_openapi_yaml_behind_proxy(reverse_proxied_app):
)
assert openapi_yaml.status_code == 200
assert openapi_yaml.headers.get("Content-Type").startswith("text/yaml")
spec = yaml.load(openapi_yaml.data.decode("utf-8"), Loader=yaml.BaseLoader)
spec = yaml.load(openapi_yaml.text, Loader=yaml.BaseLoader)
if reverse_proxied_app._spec_file == "swagger.yaml":
assert b'url: "/behind/proxy/v1.0/swagger.json"' in swagger_ui.data
assert 'url: "/behind/proxy/v1.0/swagger.json"' in swagger_ui.text
assert (
spec.get("basePath") == "/behind/proxy/v1.0"
), "basePath should contains original URI"
else:
assert b'url: "/behind/proxy/v1.0/openapi.json"' in swagger_ui.data
assert 'url: "/behind/proxy/v1.0/openapi.json"' in swagger_ui.text
url = spec.get("servers", [{}])[0].get("url")
assert url == "/behind/proxy/v1.0", "basePath should contains original URI"
@@ -81,47 +77,42 @@ def test_openapi_yaml_behind_proxy(reverse_proxied_app):
def test_produce_decorator(simple_app):
app_client = simple_app.test_client()
get_bye = app_client.get("/v1.0/bye/jsantos") # type: flask.Response
assert get_bye.content_type == "text/plain; charset=utf-8"
get_bye = app_client.get("/v1.0/bye/jsantos")
assert get_bye.headers.get("content-type") == "text/plain; charset=utf-8"
# TODO: make flask specific
def test_returning_flask_response_tuple(simple_app):
app_client = simple_app.test_client()
result = app_client.get("/v1.0/flask_response_tuple") # type: flask.Response
result = app_client.get("/v1.0/flask_response_tuple")
assert result.status_code == 201, result.text
assert result.content_type == "application/json"
result_data = json.loads(result.data.decode("utf-8", "replace"))
assert result.headers.get("content-type") == "application/json"
result_data = json.loads(result.text)
assert result_data == {"foo": "bar"}
def test_jsonifier(simple_app):
app_client = simple_app.test_client()
post_greeting = app_client.post(
"/v1.0/greeting/jsantos", data={}
) # type: flask.Response
post_greeting = app_client.post("/v1.0/greeting/jsantos", data={})
assert post_greeting.status_code == 200
assert post_greeting.content_type == "application/json"
greeting_reponse = json.loads(post_greeting.data.decode("utf-8", "replace"))
assert post_greeting.headers.get("content-type") == "application/json"
greeting_reponse = json.loads(post_greeting.text)
assert greeting_reponse["greeting"] == "Hello jsantos"
get_list_greeting = app_client.get(
"/v1.0/list/jsantos", data={}
) # type: flask.Response
get_list_greeting = app_client.get("/v1.0/list/jsantos", data={})
assert get_list_greeting.status_code == 200
assert get_list_greeting.content_type == "application/json"
greeting_reponse = json.loads(get_list_greeting.data.decode("utf-8", "replace"))
assert get_list_greeting.headers.get("content-type") == "application/json"
greeting_reponse = json.loads(get_list_greeting.text)
assert len(greeting_reponse) == 2
assert greeting_reponse[0] == "hello"
assert greeting_reponse[1] == "jsantos"
get_greetings = app_client.get(
"/v1.0/greetings/jsantos", data={}
) # type: flask.Response
get_greetings = app_client.get("/v1.0/greetings/jsantos", data={})
assert get_greetings.status_code == 200
assert get_greetings.content_type == "application/x.connexion+json"
greetings_reponse = json.loads(get_greetings.data.decode("utf-8", "replace"))
assert get_greetings.headers.get("content-type") == "application/x.connexion+json"
greetings_reponse = json.loads(get_greetings.text)
assert len(greetings_reponse) == 1
assert greetings_reponse["greetings"] == "Hello jsantos"
@@ -131,16 +122,17 @@ def test_not_content_response(simple_app):
get_no_content_response = app_client.get("/v1.0/test_no_content_response")
assert get_no_content_response.status_code == 204
assert get_no_content_response.content_length is None
assert get_no_content_response.headers.get("content-length") is None
def test_pass_through(simple_app):
app_client = simple_app.test_client()
response = app_client.get("/v1.0/multimime", data={}) # type: flask.Response
response = app_client.get("/v1.0/multimime", data={})
assert response.status_code == 500
detail = json.loads(response.text)["detail"]
assert (
response.json["detail"] == "Multiple response content types are defined in the "
detail == "Multiple response content types are defined in the "
"operation spec, but the handler response did not specify "
"which one to return."
)
@@ -149,19 +141,17 @@ def test_pass_through(simple_app):
def test_empty(simple_app):
app_client = simple_app.test_client()
response = app_client.get("/v1.0/empty") # type: flask.Response
response = app_client.get("/v1.0/empty")
assert response.status_code == 204
assert not response.data
assert not response.text
def test_exploded_deep_object_param_endpoint_openapi_simple(simple_openapi_app):
app_client = simple_openapi_app.test_client()
response = app_client.get(
"/v1.0/exploded-deep-object-param?id[foo]=bar"
) # type: flask.Response
response = app_client.get("/v1.0/exploded-deep-object-param?id[foo]=bar")
assert response.status_code == 200
response_data = json.loads(response.data.decode("utf-8", "replace"))
response_data = json.loads(response.text)
assert response_data == {"foo": "bar", "foo4": "blubb"}
@@ -172,9 +162,9 @@ def test_exploded_deep_object_param_endpoint_openapi_multiple_data_types(
response = app_client.get(
"/v1.0/exploded-deep-object-param?id[foo]=bar&id[fooint]=2&id[fooboo]=false"
) # type: flask.Response
)
assert response.status_code == 200, response.text
response_data = json.loads(response.data.decode("utf-8", "replace"))
response_data = json.loads(response.text)
assert response_data == {
"foo": "bar",
"fooint": 2,
@@ -190,9 +180,9 @@ def test_exploded_deep_object_param_endpoint_openapi_additional_properties(
response = app_client.get(
"/v1.0/exploded-deep-object-param-additional-properties?id[foo]=bar&id[fooint]=2"
) # type: flask.Response
)
assert response.status_code == 200
response_data = json.loads(response.data.decode("utf-8", "replace"))
response_data = json.loads(response.text)
assert response_data == {"foo": "bar", "fooint": "2"}
@@ -203,7 +193,7 @@ def test_exploded_deep_object_param_endpoint_openapi_additional_properties_false
response = app_client.get(
"/v1.0/exploded-deep-object-param?id[foo]=bar&id[foofoo]=barbar"
) # type: flask.Response
)
assert response.status_code == 400
@@ -212,9 +202,9 @@ def test_exploded_deep_object_param_endpoint_openapi_with_dots(simple_openapi_ap
response = app_client.get(
"/v1.0/exploded-deep-object-param-additional-properties?id[foo]=bar&id[foo.foo]=barbar"
) # type: flask.Response
)
assert response.status_code == 200
response_data = json.loads(response.data.decode("utf-8", "replace"))
response_data = json.loads(response.text)
assert response_data == {"foo": "bar", "foo.foo": "barbar"}
@@ -223,9 +213,9 @@ def test_nested_exploded_deep_object_param_endpoint_openapi(simple_openapi_app):
response = app_client.get(
"/v1.0/nested-exploded-deep-object-param?id[foo][foo2]=bar&id[foofoo]=barbar"
) # type: flask.Response
)
assert response.status_code == 200
response_data = json.loads(response.data.decode("utf-8", "replace"))
response_data = json.loads(response.text)
assert response_data == {
"foo": {"foo2": "bar", "foo3": "blubb"},
"foofoo": "barbar",
@@ -248,12 +238,12 @@ def test_default_object_body(simple_app):
app_client = simple_app.test_client()
resp = app_client.post("/v1.0/test-default-object-body")
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response["stack"] == {"image_version": "default_image"}
resp = app_client.post("/v1.0/test-default-integer-body")
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response == 1
@@ -265,7 +255,7 @@ def test_empty_object_body(simple_app):
headers={"Content-Type": "application/json"},
)
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response["stack"] == {}
@@ -277,10 +267,11 @@ def test_nested_additional_properties(simple_openapi_app):
headers={"Content-Type": "application/json"},
)
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response == {"nested": {"object": True}}
# TODO: make Flask specific
def test_custom_provider(simple_app):
class CustomProvider(FlaskJSONProvider):
def default(self, o):
@@ -294,7 +285,7 @@ def test_custom_provider(simple_app):
resp = app_client.get("/v1.0/custom-json-response")
assert resp.status_code == 200
response = json.loads(resp.data.decode("utf-8", "replace"))
response = json.loads(resp.text)
assert response["theResult"] == "cool result"
@@ -304,8 +295,15 @@ def test_content_type_not_json(simple_app):
resp = app_client.get("/v1.0/blob-response")
assert resp.status_code == 200
try:
# AsyncApp
content = resp.content
except AttributeError:
# FlaskApp
content = resp.data
# validate binary content
text, number = unpack("!4sh", resp.data)
text, number = unpack("!4sh", content)
assert text == b"cool"
assert number == 8
@@ -315,9 +313,17 @@ def test_maybe_blob_or_json(simple_app):
resp = app_client.get("/v1.0/binary-response")
assert resp.status_code == 200
assert resp.content_type == "application/octet-stream"
assert resp.headers.get("content-type") == "application/octet-stream"
try:
# AsyncApp
content = resp.content
except AttributeError:
# FlaskApp
content = resp.data
# validate binary content
text, number = unpack("!4sh", resp.data)
text, number = unpack("!4sh", content)
assert text == b"cool"
assert number == 8
@@ -353,21 +359,21 @@ def test_post_wrong_content_type(simple_app):
app_client = simple_app.test_client()
resp = app_client.post(
"/v1.0/post_wrong_content_type",
content_type="application/xml",
headers={"content-type": "application/xml"},
data=json.dumps({"some": "data"}),
)
assert resp.status_code == 415
resp = app_client.post(
"/v1.0/post_wrong_content_type",
content_type="application/x-www-form-urlencoded",
headers={"content-type": "application/x-www-form-urlencoded"},
data="a=1&b=2",
)
assert resp.status_code == 415
resp = app_client.post(
"/v1.0/post_wrong_content_type",
content_type="application/json",
headers={"content-type": "application/json"},
data="not a valid json",
)
assert (
@@ -379,7 +385,7 @@ def test_get_unicode_response(simple_app):
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/get_unicode_response")
actualJson = {"currency": "\xa3", "key": "leena"}
assert json.loads(resp.data.decode("utf-8", "replace")) == actualJson
assert json.loads(resp.text) == actualJson
def test_get_enum_response(simple_app):
@@ -403,6 +409,7 @@ def test_get_bad_default_response(simple_app):
assert resp.status_code == 500
# TODO: split per app
def test_streaming_response(simple_app):
app_client = simple_app.test_client()
resp = app_client.get("/v1.0/get_streaming_response")
@@ -412,29 +419,31 @@ def test_streaming_response(simple_app):
def test_oneof(simple_openapi_app):
app_client = simple_openapi_app.test_client()
post_greeting = app_client.post( # type: flask.Response
headers = {"Content-type": "application/json"}
post_greeting = app_client.post(
"/v1.0/oneof_greeting",
data=json.dumps({"name": 3}),
content_type="application/json",
headers=headers,
)
assert post_greeting.status_code == 200
assert post_greeting.content_type == "application/json"
greeting_reponse = json.loads(post_greeting.data.decode("utf-8", "replace"))
assert post_greeting.headers.get("content-type") == "application/json"
greeting_reponse = json.loads(post_greeting.text)
assert greeting_reponse["greeting"] == "Hello 3"
post_greeting = app_client.post( # type: flask.Response
post_greeting = app_client.post(
"/v1.0/oneof_greeting",
data=json.dumps({"name": True}),
content_type="application/json",
headers=headers,
)
assert post_greeting.status_code == 200
assert post_greeting.content_type == "application/json"
greeting_reponse = json.loads(post_greeting.data.decode("utf-8", "replace"))
assert post_greeting.headers.get("content-type") == "application/json"
greeting_reponse = json.loads(post_greeting.text)
assert greeting_reponse["greeting"] == "Hello True"
post_greeting = app_client.post( # type: flask.Response
post_greeting = app_client.post(
"/v1.0/oneof_greeting",
data=json.dumps({"name": "jsantos"}),
content_type="application/json",
headers=headers,
)
assert post_greeting.status_code == 400

154
tests/api/test_schema.py
View File

@@ -7,12 +7,10 @@ def test_schema(schema_app):
empty_request = app_client.post(
"/v1.0/test_schema", headers=headers, data=json.dumps({})
) # type: flask.Response
)
assert empty_request.status_code == 400
assert empty_request.content_type == "application/problem+json"
empty_request_response = json.loads(
empty_request.data.decode("utf-8", "replace")
) # type: dict
assert empty_request.headers.get("content-type") == "application/problem+json"
empty_request_response = json.loads(empty_request.text)
assert empty_request_response["title"] == "Bad Request"
assert empty_request_response["detail"].startswith(
"'image_version' is a required property"
@@ -20,23 +18,19 @@ def test_schema(schema_app):
bad_type = app_client.post(
"/v1.0/test_schema", headers=headers, data=json.dumps({"image_version": 22})
) # type: flask.Response
)
assert bad_type.status_code == 400
assert bad_type.content_type == "application/problem+json"
bad_type_response = json.loads(
bad_type.data.decode("utf-8", "replace")
) # type: dict
assert bad_type.headers.get("content-type") == "application/problem+json"
bad_type_response = json.loads(bad_type.text)
assert bad_type_response["title"] == "Bad Request"
assert bad_type_response["detail"].startswith("22 is not of type 'string'")
bad_type_path = app_client.post(
"/v1.0/test_schema", headers=headers, data=json.dumps({"image_version": 22})
) # type: flask.Response
)
assert bad_type_path.status_code == 400
assert bad_type_path.content_type == "application/problem+json"
bad_type_path_response = json.loads(
bad_type.data.decode("utf-8", "replace")
) # type: dict
assert bad_type_path.headers.get("content-type") == "application/problem+json"
bad_type_path_response = json.loads(bad_type.text)
assert bad_type_path_response["title"] == "Bad Request"
assert bad_type_path_response["detail"].endswith(" - 'image_version'")
@@ -44,32 +38,26 @@ def test_schema(schema_app):
"/v1.0/test_schema",
headers=headers,
data=json.dumps({"image_version": "version"}),
) # type: flask.Response
)
assert good_request.status_code == 200
good_request_response = json.loads(
good_request.data.decode("utf-8", "replace")
) # type: dict
good_request_response = json.loads(good_request.text)
assert good_request_response["image_version"] == "version"
good_request_extra = app_client.post(
"/v1.0/test_schema",
headers=headers,
data=json.dumps({"image_version": "version", "extra": "stuff"}),
) # type: flask.Response
)
assert good_request_extra.status_code == 200
good_request_extra_response = json.loads(
good_request.data.decode("utf-8", "replace")
) # type: dict
good_request_extra_response = json.loads(good_request.text)
assert good_request_extra_response["image_version"] == "version"
wrong_type = app_client.post(
"/v1.0/test_schema", headers=headers, data=json.dumps(42)
) # type: flask.Response
)
assert wrong_type.status_code == 400
assert wrong_type.content_type == "application/problem+json"
wrong_type_response = json.loads(
wrong_type.data.decode("utf-8", "replace")
) # type: dict
assert wrong_type.headers.get("content-type") == "application/problem+json"
wrong_type_response = json.loads(wrong_type.text)
assert wrong_type_response["title"] == "Bad Request"
assert wrong_type_response["detail"].startswith("42 is not of type 'object'")
@@ -79,59 +67,59 @@ def test_schema_response(schema_app):
request = app_client.get(
"/v1.0/test_schema/response/object/valid", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 200, request.text
request = app_client.get(
"/v1.0/test_schema/response/object/invalid_type", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 500, request.text
request = app_client.get(
"/v1.0/test_schema/response/object/invalid_requirements", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 500, request.text
request = app_client.get(
"/v1.0/test_schema/response/string/valid", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 200, request.text
request = app_client.get(
"/v1.0/test_schema/response/string/invalid", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 500, request.text
request = app_client.get(
"/v1.0/test_schema/response/integer/valid", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 200, request.text
request = app_client.get(
"/v1.0/test_schema/response/integer/invalid", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 500, request.text
request = app_client.get(
"/v1.0/test_schema/response/number/valid", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 200, request.text
request = app_client.get(
"/v1.0/test_schema/response/number/invalid", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 500, request.text
request = app_client.get(
"/v1.0/test_schema/response/boolean/valid", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 200, request.text
request = app_client.get(
"/v1.0/test_schema/response/boolean/invalid", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 500, request.text
request = app_client.get(
"/v1.0/test_schema/response/array/valid", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 200, request.text
request = app_client.get(
"/v1.0/test_schema/response/array/invalid_dict", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 500, request.text
request = app_client.get(
"/v1.0/test_schema/response/array/invalid_string", headers={}, data=None
) # type: flask.Response
)
assert request.status_code == 500, request.text
@@ -142,12 +130,10 @@ def test_schema_in_query(schema_app):
good_request = app_client.post(
"/v1.0/test_schema_in_query",
headers=headers,
query_string={"image_version": "version", "not_required": "test"},
) # type: flask.Response
params={"image_version": "version", "not_required": "test"},
)
assert good_request.status_code == 200
good_request_response = json.loads(
good_request.data.decode("utf-8", "replace")
) # type: dict
good_request_response = json.loads(good_request.text)
assert good_request_response["image_version"] == "version"
@@ -157,23 +143,19 @@ def test_schema_list(schema_app):
wrong_type = app_client.post(
"/v1.0/test_schema_list", headers=headers, data=json.dumps(42)
) # type: flask.Response
)
assert wrong_type.status_code == 400
assert wrong_type.content_type == "application/problem+json"
wrong_type_response = json.loads(
wrong_type.data.decode("utf-8", "replace")
) # type: dict
assert wrong_type.headers.get("content-type") == "application/problem+json"
wrong_type_response = json.loads(wrong_type.text)
assert wrong_type_response["title"] == "Bad Request"
assert wrong_type_response["detail"].startswith("42 is not of type 'array'")
wrong_items = app_client.post(
"/v1.0/test_schema_list", headers=headers, data=json.dumps([42])
) # type: flask.Response
)
assert wrong_items.status_code == 400
assert wrong_items.content_type == "application/problem+json"
wrong_items_response = json.loads(
wrong_items.data.decode("utf-8", "replace")
) # type: dict
assert wrong_items.headers.get("content-type") == "application/problem+json"
wrong_items_response = json.loads(wrong_items.text)
assert wrong_items_response["title"] == "Bad Request"
assert wrong_items_response["detail"].startswith("42 is not of type 'string'")
@@ -191,29 +173,25 @@ def test_schema_map(schema_app):
wrong_type = app_client.post(
"/v1.0/test_schema_map", headers=headers, data=json.dumps(42)
) # type: flask.Response
)
assert wrong_type.status_code == 400
assert wrong_type.content_type == "application/problem+json"
wrong_type_response = json.loads(
wrong_type.data.decode("utf-8", "replace")
) # type: dict
assert wrong_type.headers.get("content-type") == "application/problem+json"
wrong_type_response = json.loads(wrong_type.text)
assert wrong_type_response["title"] == "Bad Request"
assert wrong_type_response["detail"].startswith("42 is not of type 'object'")
wrong_items = app_client.post(
"/v1.0/test_schema_map", headers=headers, data=json.dumps(invalid_object)
) # type: flask.Response
)
assert wrong_items.status_code == 400
assert wrong_items.content_type == "application/problem+json"
wrong_items_response = json.loads(
wrong_items.data.decode("utf-8", "replace")
) # type: dict
assert wrong_items.headers.get("content-type") == "application/problem+json"
wrong_items_response = json.loads(wrong_items.text)
assert wrong_items_response["title"] == "Bad Request"
assert wrong_items_response["detail"].startswith("42 is not of type 'object'")
right_type = app_client.post(
"/v1.0/test_schema_map", headers=headers, data=json.dumps(valid_object)
) # type: flask.Response
)
assert right_type.status_code == 200
@@ -237,25 +215,25 @@ def test_schema_recursive(schema_app):
wrong_type = app_client.post(
"/v1.0/test_schema_recursive", headers=headers, data=json.dumps(42)
) # type: flask.Response
)
assert wrong_type.status_code == 400
assert wrong_type.content_type == "application/problem+json"
wrong_type_response = json.loads(wrong_type.data.decode("utf-8")) # type: dict
assert wrong_type.headers.get("content-type") == "application/problem+json"
wrong_type_response = json.loads(wrong_type.text)
assert wrong_type_response["title"] == "Bad Request"
assert wrong_type_response["detail"].startswith("42 is not of type 'object'")
wrong_items = app_client.post(
"/v1.0/test_schema_recursive", headers=headers, data=json.dumps(invalid_object)
) # type: flask.Response
)
assert wrong_items.status_code == 400
assert wrong_items.content_type == "application/problem+json"
wrong_items_response = json.loads(wrong_items.data.decode("utf-8")) # type: dict
assert wrong_items.headers.get("content-type") == "application/problem+json"
wrong_items_response = json.loads(wrong_items.text)
assert wrong_items_response["title"] == "Bad Request"
assert wrong_items_response["detail"].startswith("42 is not of type 'object'")
right_type = app_client.post(
"/v1.0/test_schema_recursive", headers=headers, data=json.dumps(valid_object)
) # type: flask.Response
)
assert right_type.status_code == 200
@@ -265,12 +243,10 @@ def test_schema_format(schema_app):
wrong_type = app_client.post(
"/v1.0/test_schema_format", headers=headers, data=json.dumps("xy")
) # type: flask.Response
)
assert wrong_type.status_code == 400
assert wrong_type.content_type == "application/problem+json"
wrong_type_response = json.loads(
wrong_type.data.decode("utf-8", "replace")
) # type: dict
assert wrong_type.headers.get("content-type") == "application/problem+json"
wrong_type_response = json.loads(wrong_type.text)
assert wrong_type_response["title"] == "Bad Request"
assert "'xy' is not a 'email'" in wrong_type_response["detail"]
@@ -281,12 +257,10 @@ def test_schema_array(schema_app):
array_request = app_client.post(
"/v1.0/schema_array", headers=headers, data=json.dumps(["list", "hello"])
) # type: flask.Response
)
assert array_request.status_code == 200
assert array_request.content_type == "application/json"
array_response = json.loads(
array_request.data.decode("utf-8", "replace")
) # type: list
assert array_request.headers.get("content-type") == "application/json"
array_response = json.loads(array_request.text)
assert array_response == ["list", "hello"]
@@ -296,19 +270,17 @@ def test_schema_int(schema_app):
array_request = app_client.post(
"/v1.0/schema_int", headers=headers, data=json.dumps(42)
) # type: flask.Response
)
assert array_request.status_code == 200
assert array_request.content_type == "application/json"
array_response = json.loads(
array_request.data.decode("utf-8", "replace")
) # type: list
assert array_request.headers.get("content-type") == "application/json"
array_response = json.loads(array_request.text) # type: list
assert array_response == 42
def test_global_response_definitions(schema_app):
app_client = schema_app.test_client()
resp = app_client.get("/v1.0/define_global_response")
assert json.loads(resp.data.decode("utf-8", "replace")) == ["general", "list"]
assert json.loads(resp.text) == ["general", "list"]
def test_media_range(schema_app):

120
tests/api/test_secure_api.py
View File

@@ -58,129 +58,109 @@ def test_security_over_nonexistent_endpoints(oauth_requests, secure_api_app):
headers = {"Authorization": "Bearer 300"}
get_inexistent_endpoint = app_client.get(
"/v1.0/does-not-exist-invalid-token", headers=headers
) # type: flask.Response
)
assert get_inexistent_endpoint.status_code == 401
assert get_inexistent_endpoint.content_type == "application/problem+json"
assert (
get_inexistent_endpoint.headers.get("content-type")
== "application/problem+json"
)
headers = {"Authorization": "Bearer 100"}
get_inexistent_endpoint = app_client.get(
"/v1.0/does-not-exist-valid-token", headers=headers
) # type: flask.Response
)
assert get_inexistent_endpoint.status_code == 404
assert get_inexistent_endpoint.content_type == "application/problem+json"
assert (
get_inexistent_endpoint.headers.get("content-type")
== "application/problem+json"
)
get_inexistent_endpoint = app_client.get(
"/v1.0/does-not-exist-no-token"
) # type: flask.Response
get_inexistent_endpoint = app_client.get("/v1.0/does-not-exist-no-token")
assert get_inexistent_endpoint.status_code == 401
headers = {"Authorization": "Bearer 100"}
post_greeting = app_client.post(
"/v1.0/greeting/rcaricio", data={}, headers=headers
) # type: flask.Response
post_greeting = app_client.post("/v1.0/greeting/rcaricio", data={}, headers=headers)
assert post_greeting.status_code == 200
post_greeting = app_client.post(
"/v1.0/greeting/rcaricio", data={}
) # type: flask.Response
post_greeting = app_client.post("/v1.0/greeting/rcaricio", data={})
assert post_greeting.status_code == 401
def test_security(oauth_requests, secure_endpoint_app):
app_client = secure_endpoint_app.test_client()
get_bye_no_auth = app_client.get("/v1.0/byesecure/jsantos") # type: flask.Response
get_bye_no_auth = app_client.get("/v1.0/byesecure/jsantos")
assert get_bye_no_auth.status_code == 401
assert get_bye_no_auth.content_type == "application/problem+json"
get_bye_no_auth_reponse = json.loads(
get_bye_no_auth.data.decode("utf-8", "replace")
) # type: dict
assert get_bye_no_auth.headers.get("content-type") == "application/problem+json"
get_bye_no_auth_reponse = json.loads(get_bye_no_auth.text)
assert get_bye_no_auth_reponse["detail"] == "No authorization token provided"
headers = {"Authorization": "Bearer 100"}
get_bye_good_auth = app_client.get(
"/v1.0/byesecure/jsantos", headers=headers
) # type: flask.Response
get_bye_good_auth = app_client.get("/v1.0/byesecure/jsantos", headers=headers)
assert get_bye_good_auth.status_code == 200
assert get_bye_good_auth.data == b"Goodbye jsantos (Secure: test-user)"
assert get_bye_good_auth.text == "Goodbye jsantos (Secure: test-user)"
headers = {"Authorization": "Bearer 200"}
get_bye_wrong_scope = app_client.get(
"/v1.0/byesecure/jsantos", headers=headers
) # type: flask.Response
get_bye_wrong_scope = app_client.get("/v1.0/byesecure/jsantos", headers=headers)
assert get_bye_wrong_scope.status_code == 403
assert get_bye_wrong_scope.content_type == "application/problem+json"
get_bye_wrong_scope_reponse = json.loads(
get_bye_wrong_scope.data.decode("utf-8", "replace")
) # type: dict
assert get_bye_wrong_scope.headers.get("content-type") == "application/problem+json"
get_bye_wrong_scope_reponse = json.loads(get_bye_wrong_scope.text)
assert (
get_bye_wrong_scope_reponse["detail"]
== "Provided token doesn't have the required scope"
)
headers = {"Authorization": "Bearer 300"}
get_bye_bad_token = app_client.get(
"/v1.0/byesecure/jsantos", headers=headers
) # type: flask.Response
get_bye_bad_token = app_client.get("/v1.0/byesecure/jsantos", headers=headers)
assert get_bye_bad_token.status_code == 401
assert get_bye_bad_token.content_type == "application/problem+json"
get_bye_bad_token_reponse = json.loads(
get_bye_bad_token.data.decode("utf-8", "replace")
) # type: dict
assert get_bye_bad_token.headers.get("content-type") == "application/problem+json"
get_bye_bad_token_reponse = json.loads(get_bye_bad_token.text)
assert get_bye_bad_token_reponse["detail"] == "Provided token is not valid"
response = app_client.get(
"/v1.0/more-than-one-security-definition"
) # type: flask.Response
response = app_client.get("/v1.0/more-than-one-security-definition")
assert response.status_code == 401
# also tests case-insensitivity
headers = {"X-AUTH": "mykey"}
response = app_client.get(
"/v1.0/more-than-one-security-definition", headers=headers
) # type: flask.Response
)
assert response.status_code == 200
headers = {"Authorization": "Bearer 100"}
get_bye_good_auth = app_client.get(
"/v1.0/byesecure-ignoring-context/hjacobs", headers=headers
) # type: flask.Response
)
assert get_bye_good_auth.status_code == 200
assert get_bye_good_auth.data == b"Goodbye hjacobs (Secure!)"
assert get_bye_good_auth.text == "Goodbye hjacobs (Secure!)"
headers = {"Authorization": "Bearer 100"}
get_bye_from_flask = app_client.get(
"/v1.0/byesecure-from-flask", headers=headers
) # type: flask.Response
assert get_bye_from_flask.data == b"Goodbye test-user (Secure!)"
get_bye_from_flask = app_client.get("/v1.0/byesecure-from-flask", headers=headers)
assert get_bye_from_flask.text == "Goodbye test-user (Secure!)"
headers = {"Authorization": "Bearer 100"}
get_bye_from_connexion = app_client.get(
"/v1.0/byesecure-from-connexion", headers=headers
) # type: flask.Response
assert get_bye_from_connexion.data == b"Goodbye test-user (Secure!)"
)
assert get_bye_from_connexion.text == "Goodbye test-user (Secure!)"
headers = {"Authorization": "Bearer 100"}
get_bye_from_connexion = app_client.get(
"/v1.0/byesecure-jwt/test-user", headers=headers
) # type: flask.Response
assert get_bye_from_connexion.data == b"Goodbye test-user (Secure: 100)"
)
assert get_bye_from_connexion.text == "Goodbye test-user (Secure: 100)"
# has optional auth
response = app_client.get("/v1.0/optional-auth") # type: flask.Response
response = app_client.get("/v1.0/optional-auth")
assert response.status_code == 200
assert response.data == b'"Unauthenticated"\n'
assert response.text == '"Unauthenticated"\n'
headers = {"X-AUTH": "mykey"}
response = app_client.get(
"/v1.0/optional-auth", headers=headers
) # type: flask.Response
response = app_client.get("/v1.0/optional-auth", headers=headers)
assert response.status_code == 200
assert response.data == b'"Authenticated"\n'
assert response.text == '"Authenticated"\n'
headers = {"X-AUTH": "wrong-key"}
response = app_client.get(
"/v1.0/optional-auth", headers=headers
) # type: flask.Response
assert response.data == b'"Unauthenticated"\n'
response = app_client.get("/v1.0/optional-auth", headers=headers)
assert response.text == '"Unauthenticated"\n'
assert response.status_code == 200
# security function throws exception
@@ -195,35 +175,25 @@ def test_checking_that_client_token_has_all_necessary_scopes(
# has only one of the required scopes
headers = {"Authorization": "Bearer has_myscope"}
response = app_client.get(
"/v1.0/more-than-one-scope", headers=headers
) # type: flask.Response
response = app_client.get("/v1.0/more-than-one-scope", headers=headers)
assert response.status_code == 403
# has none of the necessary scopes
headers = {"Authorization": "Bearer has_wrongscope"}
response = app_client.get(
"/v1.0/more-than-one-scope", headers=headers
) # type: flask.Response
response = app_client.get("/v1.0/more-than-one-scope", headers=headers)
assert response.status_code == 403
# is not auth
headers = {"Authorization": "Bearer is_not_invalid"}
response = app_client.get(
"/v1.0/more-than-one-scope", headers=headers
) # type: flask.Response
response = app_client.get("/v1.0/more-than-one-scope", headers=headers)
assert response.status_code == 401
# has all necessary scopes
headers = {"Authorization": "Bearer has_myscope_otherscope"}
response = app_client.get(
"/v1.0/more-than-one-scope", headers=headers
) # type: flask.Response
response = app_client.get("/v1.0/more-than-one-scope", headers=headers)
assert response.status_code == 200
# has all necessary scopes but under key 'scopes'
headers = {"Authorization": "Bearer has_scopes_in_scopes_with_s"}
response = app_client.get(
"/v1.0/more-than-one-scope", headers=headers
) # type: flask.Response
response = app_client.get("/v1.0/more-than-one-scope", headers=headers)
assert response.status_code == 200

6
tests/api/test_unordered_definition.py
View File

@@ -3,9 +3,7 @@ import json
def test_app(unordered_definition_app):
app_client = unordered_definition_app.test_client()
response = app_client.get(
"/v1.0/unordered-params/1?first=first&second=2"
) # type: flask.Response
response = app_client.get("/v1.0/unordered-params/1?first=first&second=2")
assert response.status_code == 400
response_data = json.loads(response.data.decode("utf-8", "replace"))
response_data = json.loads(response.text)
assert response_data["detail"].startswith("'first' is not of type 'integer'")

12
tests/conftest.py
View File

@@ -2,7 +2,7 @@ import logging
import pathlib
import pytest
from connexion import App
from connexion import AsyncApp, FlaskApp
from connexion.resolver import MethodResolver, MethodViewResolver
logging.basicConfig(level=logging.INFO)
@@ -14,6 +14,7 @@ OPENAPI2_SPEC = "swagger.yaml"
OPENAPI3_SPEC = "openapi.yaml"
SPECS = [OPENAPI2_SPEC, OPENAPI3_SPEC]
METHOD_VIEW_RESOLVERS = [MethodResolver, MethodViewResolver]
APP_CLASSES = [FlaskApp, AsyncApp]
@pytest.fixture
@@ -56,10 +57,15 @@ def method_view_resolver(request):
return request.param
@pytest.fixture(scope="session", params=APP_CLASSES)
def app_class(request):
return request.param
def build_app_from_fixture(
api_spec_folder, spec_file="openapi.yaml", middlewares=None, **kwargs
api_spec_folder, *, app_class, spec_file, middlewares=None, **kwargs
):
cnx_app = App(
cnx_app = app_class(
__name__,
specification_dir=FIXTURES_FOLDER / api_spec_folder,
middlewares=middlewares,

15
tests/test_flask_encoder.py
View File

@@ -2,7 +2,6 @@ import datetime
import json
import math
from decimal import Decimal
from unittest import mock
from connexion.frameworks.flask import FlaskJSONProvider
@@ -44,13 +43,15 @@ def test_json_encoder_datetime_with_timezone():
assert s.endswith('+00:00"')
def test_readonly(json_datetime_dir, spec):
app = build_app_from_fixture(json_datetime_dir, spec, validate_responses=True)
def test_readonly(json_datetime_dir, spec, app_class):
app = build_app_from_fixture(
json_datetime_dir, app_class=app_class, spec_file=spec, validate_responses=True
)
app_client = app.test_client()
res = app_client.get("/v1.0/" + spec.replace("yaml", "json"))
assert res.status_code == 200, f"Error is {res.data}"
spec_data = json.loads(res.data.decode())
spec_data = json.loads(res.text)
if spec == "openapi.yaml":
response_path = "responses.200.content.application/json.schema"
@@ -75,15 +76,15 @@ def test_readonly(json_datetime_dir, spec):
res = app_client.get("/v1.0/datetime")
assert res.status_code == 200, f"Error is {res.data}"
data = json.loads(res.data.decode())
data = json.loads(res.text)
assert data == {"value": "2000-01-02T03:04:05.000006Z"}
res = app_client.get("/v1.0/date")
assert res.status_code == 200, f"Error is {res.data}"
data = json.loads(res.data.decode())
data = json.loads(res.text)
assert data == {"value": "2000-01-02"}
res = app_client.get("/v1.0/uuid")
assert res.status_code == 200, f"Error is {res.data}"
data = json.loads(res.data.decode())
data = json.loads(res.text)
assert data == {"value": "e7ff66d0-3ec2-4c4e-bed0-6e4723c24c51"}

61
tests/test_json_validation.py
View File

@@ -41,63 +41,71 @@ def test_validator_map(json_validation_spec_dir, spec):
"/v1.0/minlength",
data=json.dumps({"foo": "bar"}),
content_type="application/json",
) # type: flask.Response
)
assert res.status_code == 200
res = app_client.post(
"/v1.0/minlength", data=json.dumps({"foo": ""}), content_type="application/json"
) # type: flask.Response
)
assert res.status_code == 400
def test_readonly(json_validation_spec_dir, spec):
def test_readonly(json_validation_spec_dir, spec, app_class):
app = build_app_from_fixture(
json_validation_spec_dir, spec, validate_responses=True
json_validation_spec_dir,
app_class=app_class,
spec_file=spec,
validate_responses=True,
)
app_client = app.test_client()
res = app_client.get("/v1.0/user") # type: flask.Response
headers = {"content-type": "application/json"}
res = app_client.get("/v1.0/user")
assert res.status_code == 200
assert json.loads(res.data.decode()).get("user_id") == 7
assert json.loads(res.text).get("user_id") == 7
res = app_client.post(
"/v1.0/user",
data=json.dumps({"name": "max", "password": "1234"}),
content_type="application/json",
) # type: flask.Response
headers=headers,
)
assert res.status_code == 200
assert json.loads(res.data.decode()).get("user_id") == 8
assert json.loads(res.text).get("user_id") == 8
res = app_client.post(
"/v1.0/user",
data=json.dumps({"user_id": 9, "name": "max"}),
content_type="application/json",
) # type: flask.Response
headers=headers,
)
assert res.status_code == 400
def test_writeonly(json_validation_spec_dir, spec):
def test_writeonly(json_validation_spec_dir, spec, app_class):
app = build_app_from_fixture(
json_validation_spec_dir, spec, validate_responses=True
json_validation_spec_dir,
app_class=app_class,
spec_file=spec,
validate_responses=True,
)
app_client = app.test_client()
res = app_client.post(
"/v1.0/user",
data=json.dumps({"name": "max", "password": "1234"}),
content_type="application/json",
) # type: flask.Response
headers={"content-type": "application/json"},
)
assert res.status_code == 200
assert "password" not in json.loads(res.data.decode())
assert "password" not in json.loads(res.text)
res = app_client.get("/v1.0/user") # type: flask.Response
res = app_client.get("/v1.0/user")
assert res.status_code == 200
assert "password" not in json.loads(res.data.decode())
assert "password" not in json.loads(res.text)
res = app_client.get("/v1.0/user_with_password") # type: flask.Response
res = app_client.get("/v1.0/user_with_password")
assert res.status_code == 500
assert (
json.loads(res.data.decode())["title"]
json.loads(res.text)["title"]
== "Response body does not conform to specification"
)
@@ -108,17 +116,20 @@ def test_nullable_default(json_validation_spec_dir, spec):
@pytest.mark.parametrize("spec", ["openapi.yaml"])
def test_multipart_form_json(json_validation_spec_dir, spec):
def test_multipart_form_json(json_validation_spec_dir, spec, app_class):
app = build_app_from_fixture(
json_validation_spec_dir, spec, validate_responses=True
json_validation_spec_dir,
app_class=app_class,
spec_file=spec,
validate_responses=True,
)
app_client = app.test_client()
res = app_client.post(
"/v1.0/multipart_form_json",
data={"x": json.dumps({"name": "joe", "age": 20})},
content_type="multipart/form-data",
headers={"content-type": "multipart/form-data"},
)
assert res.status_code == 200
assert json.loads(res.data.decode())["name"] == "joe-reply"
assert json.loads(res.data.decode())["age"] == 30
assert json.loads(res.text)["name"] == "joe-reply"
assert json.loads(res.text)["age"] == 30

6
tests/test_middleware.py
View File

@@ -31,9 +31,11 @@ class TestMiddleware:
@pytest.fixture(scope="session")
def middleware_app(spec):
def middleware_app(spec, app_class):
middlewares = ConnexionMiddleware.default_middlewares + [TestMiddleware]
return build_app_from_fixture("simple", spec, middlewares=middlewares)
return build_app_from_fixture(
"simple", app_class=app_class, spec_file=spec, middlewares=middlewares
)
def test_routing_middleware(middleware_app):

5
tests/test_resolver_methodview.py
View File

@@ -190,10 +190,11 @@ def test_methodview_resolve_with_default_module_name_will_resolve_resource_root_
assert operation.operation_id == "fakeapi.PetsView.post"
def test_method_view_resolver_integration(spec, method_view_resolver):
def test_method_view_resolver_integration(spec, app_class, method_view_resolver):
method_view_app = build_app_from_fixture(
"method_view",
spec,
app_class=app_class,
spec_file=spec,
resolver=MethodViewResolver("fakeapi.example_method_view"),
)