diff --git a/tests/api/conftest.py b/tests/api/conftest.py new file mode 100644 index 000000000..6022d43ba --- /dev/null +++ b/tests/api/conftest.py @@ -0,0 +1,108 @@ +import logging + +import pytest +from starlette.types import Receive, Scope, Send + +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) + + +@pytest.fixture(scope="session") +def simple_openapi_app(): + return build_app_from_fixture("simple", OPENAPI3_SPEC, validate_responses=True) + + +@pytest.fixture(scope="session") +def reverse_proxied_app(spec): + class ReverseProxied: + def __init__(self, app, root_path=None, scheme=None, server=None): + self.app = app + self.root_path = root_path + self.scheme = scheme + self.server = server + + async def __call__(self, scope: Scope, receive: Receive, send: Send): + logging.warning( + "this demo is not secure by default!! " + "You'll want to make sure these headers are coming from your proxy, " + "and not directly from users on the web!" + ) + root_path = scope.get("root_path") or self.root_path + for header, value in scope.get("headers", []): + if header == b"x-forwarded-path": + root_path = value.decode() + break + if root_path: + scope["root_path"] = "/" + root_path.strip("/") + path_info = scope.get("PATH_INFO", scope.get("path")) + if path_info.startswith(root_path): + scope["PATH_INFO"] = path_info[len(root_path) :] + + scope["scheme"] = scope.get("scheme") or self.scheme + scope["server"] = scope.get("server") or (self.server, None) + + return await self.app(scope, receive, send) + + app = build_app_from_fixture("simple", spec, validate_responses=True) + app.middleware = ReverseProxied(app.middleware, root_path="/reverse_proxied/") + return app + + +@pytest.fixture(scope="session") +def snake_case_app(spec): + return build_app_from_fixture( + "snake_case", 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): + return build_app_from_fixture( + "simple", spec, validate_responses=True, strict_validation=True + ) + + +@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): + return build_app_from_fixture( + "secure_endpoint", + spec, + validate_responses=True, + ) + + +@pytest.fixture(scope="session") +def secure_api_app(spec): + options = {"swagger_ui": False} + return build_app_from_fixture( + "secure_api", spec, options=options, auth_all_paths=True + ) + + +@pytest.fixture(scope="session") +def unordered_definition_app(spec): + return build_app_from_fixture("unordered_definition", spec) + + +@pytest.fixture(scope="session") +def bad_operations_app(spec): + return build_app_from_fixture("bad_operations", spec, resolver_error=501) diff --git a/tests/api/test_bootstrap.py b/tests/api/test_bootstrap.py index 4aace725e..9e2e1255c 100644 --- a/tests/api/test_bootstrap.py +++ b/tests/api/test_bootstrap.py @@ -12,10 +12,7 @@ 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( @@ -30,7 +27,6 @@ def test_app_with_relative_path(simple_api_spec_dir, spec): 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 @@ -63,7 +59,6 @@ def test_app_with_different_uri_parser(simple_api_spec_dir): 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) @@ -76,7 +71,6 @@ def test_swagger_ui(simple_api_spec_dir, 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} @@ -93,7 +87,6 @@ def test_swagger_ui_with_config(simple_api_spec_dir, 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( @@ -114,7 +107,6 @@ def test_no_swagger_ui(simple_api_spec_dir, spec): 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} @@ -134,7 +126,6 @@ def test_swagger_ui_config_json(simple_api_spec_dir, spec): ) -@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) @@ -145,7 +136,6 @@ def test_no_swagger_ui_config_json(simple_api_spec_dir, spec): 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) @@ -157,7 +147,6 @@ def test_swagger_json_app(simple_api_spec_dir, spec): 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) @@ -169,7 +158,6 @@ def test_swagger_yaml_app(simple_api_spec_dir, spec): 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} @@ -187,7 +175,6 @@ def test_no_swagger_json_app(simple_api_spec_dir, spec): 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 @@ -210,7 +197,6 @@ def test_dict_as_yaml_path(simple_api_spec_dir, spec): 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) @@ -222,7 +208,6 @@ def test_swagger_json_api(simple_api_spec_dir, spec): 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) @@ -282,7 +267,6 @@ def test_resolve_classmethod(simple_app): 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")) diff --git a/tests/api/test_secure_api.py b/tests/api/test_secure_api.py index 9ac05bd0a..c3f12163b 100644 --- a/tests/api/test_secure_api.py +++ b/tests/api/test_secure_api.py @@ -1,5 +1,57 @@ import json +import pytest +from connexion.security import SecurityHandlerFactory + + +class FakeResponse: + def __init__(self, status_code, text): + """ + :type status_code: int + :type text: str + """ + self.status_code = status_code + self.text = text + self.ok = status_code == 200 + + def json(self): + return json.loads(self.text) + + +@pytest.fixture +def oauth_requests(monkeypatch): + class FakeClient: + @staticmethod + async def get(url, params=None, headers=None, timeout=None): + """ + :type url: str + :type params: dict| None + """ + headers = headers or {} + if url == "https://oauth.example/token_info": + token = headers.get("Authorization", "invalid").split()[-1] + if token in ["100", "has_myscope"]: + return FakeResponse( + 200, '{"uid": "test-user", "scope": ["myscope"]}' + ) + if token in ["200", "has_wrongscope"]: + return FakeResponse( + 200, '{"uid": "test-user", "scope": ["wrongscope"]}' + ) + if token == "has_myscope_otherscope": + return FakeResponse( + 200, '{"uid": "test-user", "scope": ["myscope", "otherscope"]}' + ) + if token in ["300", "is_not_invalid"]: + return FakeResponse(404, "") + if token == "has_scopes_in_scopes_with_s": + return FakeResponse( + 200, '{"uid": "test-user", "scopes": ["myscope", "otherscope"]}' + ) + return url + + monkeypatch.setattr(SecurityHandlerFactory, "client", FakeClient()) + def test_security_over_nonexistent_endpoints(oauth_requests, secure_api_app): app_client = secure_api_app.test_client() diff --git a/tests/conftest.py b/tests/conftest.py index c6e9e9c10..afb8f4629 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,123 +1,21 @@ -import json import logging import pathlib import pytest from connexion import App from connexion.resolver import MethodResolver, MethodViewResolver -from connexion.security import SecurityHandlerFactory -from starlette.types import Receive, Scope, Send -from werkzeug.test import Client, EnvironBuilder logging.basicConfig(level=logging.INFO) TEST_FOLDER = pathlib.Path(__file__).parent FIXTURES_FOLDER = TEST_FOLDER / "fixtures" SPEC_FOLDER = TEST_FOLDER / "fakeapi" -OPENAPI2_SPEC = ["swagger.yaml"] -OPENAPI3_SPEC = ["openapi.yaml"] -SPECS = OPENAPI2_SPEC + OPENAPI3_SPEC +OPENAPI2_SPEC = "swagger.yaml" +OPENAPI3_SPEC = "openapi.yaml" +SPECS = [OPENAPI2_SPEC, OPENAPI3_SPEC] METHOD_VIEW_RESOLVERS = [MethodResolver, MethodViewResolver] -class FakeResponse: - def __init__(self, status_code, text): - """ - :type status_code: int - :type text: ste - """ - self.status_code = status_code - self.text = text - self.ok = status_code == 200 - - def json(self): - return json.loads(self.text) - - -def fixed_get_environ(): - """See https://github.com/pallets/werkzeug/issues/2347""" - - original_get_environ = EnvironBuilder.get_environ - - def f(self): - result = original_get_environ(self) - result.pop("HTTP_CONTENT_TYPE", None) - result.pop("HTTP_CONTENT_LENGTH", None) - return result - - return f - - -EnvironBuilder.get_environ = fixed_get_environ() - - -def buffered_open(): - """For use with ASGI middleware""" - - original_open = Client.open - - def f(*args, **kwargs): - kwargs["buffered"] = True - return original_open(*args, **kwargs) - - return f - - -Client.open = buffered_open() - - -# Helper fixtures functions -# ========================= - - -@pytest.fixture -def oauth_requests(monkeypatch): - class FakeClient: - @staticmethod - async def get(url, params=None, headers=None, timeout=None): - """ - :type url: str - :type params: dict| None - """ - headers = headers or {} - if url == "https://oauth.example/token_info": - token = headers.get("Authorization", "invalid").split()[-1] - if token in ["100", "has_myscope"]: - return FakeResponse( - 200, '{"uid": "test-user", "scope": ["myscope"]}' - ) - if token in ["200", "has_wrongscope"]: - return FakeResponse( - 200, '{"uid": "test-user", "scope": ["wrongscope"]}' - ) - if token == "has_myscope_otherscope": - return FakeResponse( - 200, '{"uid": "test-user", "scope": ["myscope", "otherscope"]}' - ) - if token in ["300", "is_not_invalid"]: - return FakeResponse(404, "") - if token == "has_scopes_in_scopes_with_s": - return FakeResponse( - 200, '{"uid": "test-user", "scopes": ["myscope", "otherscope"]}' - ) - return url - - monkeypatch.setattr(SecurityHandlerFactory, "client", FakeClient()) - - -@pytest.fixture -def security_handler_factory(): - security_handler_factory = SecurityHandlerFactory() - yield security_handler_factory - - -@pytest.fixture -def app(): - cnx_app = App(__name__, specification_dir=SPEC_FOLDER) - cnx_app.add_api("api.yaml", validate_responses=True) - return cnx_app - - @pytest.fixture def simple_api_spec_dir(): return FIXTURES_FOLDER / "simple" @@ -148,14 +46,19 @@ def json_datetime_dir(): return FIXTURES_FOLDER / "datetime_support" +@pytest.fixture(scope="session", params=SPECS) +def spec(request): + return request.param + + +@pytest.fixture(scope="session", params=METHOD_VIEW_RESOLVERS) +def method_view_resolver(request): + return request.param + + def build_app_from_fixture( api_spec_folder, spec_file="openapi.yaml", middlewares=None, **kwargs ): - debug = True - if "debug" in kwargs: - debug = kwargs["debug"] - del kwargs["debug"] - cnx_app = App( __name__, specification_dir=FIXTURES_FOLDER / api_spec_folder, @@ -165,121 +68,3 @@ def build_app_from_fixture( cnx_app.add_api(spec_file, **kwargs) cnx_app._spec_file = spec_file return cnx_app - - -@pytest.fixture(scope="session", params=SPECS) -def simple_app(request): - return build_app_from_fixture("simple", request.param, validate_responses=True) - - -@pytest.fixture(scope="session", params=OPENAPI3_SPEC) -def simple_openapi_app(request): - return build_app_from_fixture("simple", request.param, validate_responses=True) - - -@pytest.fixture(scope="session", params=SPECS) -def reverse_proxied_app(request): - class ReverseProxied: - def __init__(self, app, root_path=None, scheme=None, server=None): - self.app = app - self.root_path = root_path - self.scheme = scheme - self.server = server - - async def __call__(self, scope: Scope, receive: Receive, send: Send): - logging.warning( - "this demo is not secure by default!! " - "You'll want to make sure these headers are coming from your proxy, " - "and not directly from users on the web!" - ) - root_path = scope.get("root_path") or self.root_path - for header, value in scope.get("headers", []): - if header == b"x-forwarded-path": - root_path = value.decode() - break - if root_path: - scope["root_path"] = "/" + root_path.strip("/") - path_info = scope.get("PATH_INFO", scope.get("path")) - if path_info.startswith(root_path): - scope["PATH_INFO"] = path_info[len(root_path) :] - - scope["scheme"] = scope.get("scheme") or self.scheme - scope["server"] = scope.get("server") or (self.server, None) - - return await self.app(scope, receive, send) - - app = build_app_from_fixture("simple", request.param, validate_responses=True) - app.middleware = ReverseProxied(app.middleware, root_path="/reverse_proxied/") - return app - - -@pytest.fixture(scope="session", params=SPECS) -def snake_case_app(request): - return build_app_from_fixture( - "snake_case", request.param, validate_responses=True, pythonic_params=True - ) - - -@pytest.fixture(scope="session", params=SPECS) -def invalid_resp_allowed_app(request): - return build_app_from_fixture("simple", request.param, validate_responses=False) - - -@pytest.fixture(scope="session", params=SPECS) -def strict_app(request): - return build_app_from_fixture( - "simple", request.param, validate_responses=True, strict_validation=True - ) - - -@pytest.fixture(scope="session", params=SPECS) -def problem_app(request): - return build_app_from_fixture("problem", request.param, validate_responses=True) - - -@pytest.fixture(scope="session", params=SPECS) -def schema_app(request): - return build_app_from_fixture( - "different_schemas", request.param, validate_responses=True - ) - - -@pytest.fixture(scope="session", params=SPECS) -def secure_endpoint_app(request): - return build_app_from_fixture( - "secure_endpoint", - request.param, - validate_responses=True, - ) - - -@pytest.fixture(scope="session", params=SPECS) -def secure_api_app(request): - options = {"swagger_ui": False} - return build_app_from_fixture( - "secure_api", request.param, options=options, auth_all_paths=True - ) - - -@pytest.fixture(scope="session", params=SPECS) -def unordered_definition_app(request): - return build_app_from_fixture("unordered_definition", request.param) - - -@pytest.fixture(scope="session", params=SPECS) -def bad_operations_app(request): - return build_app_from_fixture("bad_operations", request.param, resolver_error=501) - - -@pytest.fixture(scope="session", params=SPECS) -def method_view_app(request): - return build_app_from_fixture( - "method_view", - request.param, - resolver=MethodViewResolver("fakeapi.example_method_view"), - ) - - -@pytest.fixture(scope="session", params=METHOD_VIEW_RESOLVERS) -def method_view_resolver(request): - return request.param diff --git a/tests/decorators/test_security.py b/tests/decorators/test_security.py index a9d430465..26c5b3159 100644 --- a/tests/decorators/test_security.py +++ b/tests/decorators/test_security.py @@ -1,4 +1,3 @@ -import inspect import json from unittest.mock import MagicMock @@ -14,7 +13,8 @@ from connexion.security import SecurityHandlerFactory -def test_get_tokeninfo_url(monkeypatch, security_handler_factory): +def test_get_tokeninfo_url(monkeypatch): + security_handler_factory = SecurityHandlerFactory() security_handler_factory.get_token_info_remote = MagicMock( return_value="get_token_info_remote_result" ) @@ -45,10 +45,11 @@ def test_get_tokeninfo_url(monkeypatch, security_handler_factory): logger.warn.assert_not_called() -def test_verify_oauth_missing_auth_header(security_handler_factory): +def test_verify_oauth_missing_auth_header(): def somefunc(token): return None + security_handler_factory = SecurityHandlerFactory() wrapped_func = security_handler_factory.verify_oauth( somefunc, security_handler_factory.validate_scope, ["admin"] ) @@ -59,7 +60,7 @@ def somefunc(token): assert wrapped_func(request) is security_handler_factory.no_value -async def test_verify_oauth_scopes_remote(monkeypatch, security_handler_factory): +async def test_verify_oauth_scopes_remote(monkeypatch): tokeninfo = dict(uid="foo", scope="scope1 scope2") async def get_tokeninfo_response(*args, **kwargs): @@ -68,6 +69,7 @@ async def get_tokeninfo_response(*args, **kwargs): tokeninfo_response._content = json.dumps(tokeninfo).encode() return tokeninfo_response + security_handler_factory = SecurityHandlerFactory() token_info_func = security_handler_factory.get_tokeninfo_func( {"x-tokenInfoUrl": "https://example.org/tokeninfo"} ) @@ -102,10 +104,11 @@ async def get_tokeninfo_response(*args, **kwargs): assert await wrapped_func(request) is not None -async def test_verify_oauth_invalid_local_token_response_none(security_handler_factory): +async def test_verify_oauth_invalid_local_token_response_none(): def somefunc(token): return None + security_handler_factory = SecurityHandlerFactory() wrapped_func = security_handler_factory.verify_oauth( somefunc, security_handler_factory.validate_scope, ["admin"] ) @@ -117,12 +120,13 @@ def somefunc(token): await wrapped_func(request) -async def test_verify_oauth_scopes_local(security_handler_factory): +async def test_verify_oauth_scopes_local(): tokeninfo = dict(uid="foo", scope="scope1 scope2") def token_info(token): return tokeninfo + security_handler_factory = SecurityHandlerFactory() wrapped_func = security_handler_factory.verify_oauth( token_info, security_handler_factory.validate_scope, ["admin"] ) @@ -150,10 +154,11 @@ def token_info(token): assert await wrapped_func(request) is not None -def test_verify_basic_missing_auth_header(security_handler_factory): +def test_verify_basic_missing_auth_header(): def somefunc(username, password, required_scopes=None): return None + security_handler_factory = SecurityHandlerFactory() wrapped_func = security_handler_factory.verify_basic(somefunc) request = MagicMock() @@ -162,12 +167,13 @@ def somefunc(username, password, required_scopes=None): assert wrapped_func(request) is security_handler_factory.no_value -async def test_verify_basic(security_handler_factory): +async def test_verify_basic(): def basic_info(username, password, required_scopes=None): if username == "foo" and password == "bar": return {"sub": "foo"} return None + security_handler_factory = SecurityHandlerFactory() wrapped_func = security_handler_factory.verify_basic(basic_info) request = MagicMock() @@ -176,12 +182,13 @@ def basic_info(username, password, required_scopes=None): assert await wrapped_func(request) is not None -async def test_verify_apikey_query(security_handler_factory): +async def test_verify_apikey_query(): def apikey_info(apikey, required_scopes=None): if apikey == "foobar": return {"sub": "foo"} return None + security_handler_factory = SecurityHandlerFactory() wrapped_func = security_handler_factory.verify_api_key(apikey_info, "query", "auth") request = MagicMock() @@ -190,12 +197,13 @@ def apikey_info(apikey, required_scopes=None): assert await wrapped_func(request) is not None -async def test_verify_apikey_header(security_handler_factory): +async def test_verify_apikey_header(): def apikey_info(apikey, required_scopes=None): if apikey == "foobar": return {"sub": "foo"} return None + security_handler_factory = SecurityHandlerFactory() wrapped_func = security_handler_factory.verify_api_key( apikey_info, "header", "X-Auth" ) @@ -206,7 +214,7 @@ def apikey_info(apikey, required_scopes=None): assert await wrapped_func(request) is not None -async def test_multiple_schemes(security_handler_factory): +async def test_multiple_schemes(): def apikey1_info(apikey, required_scopes=None): if apikey == "foobar": return {"sub": "foo"} @@ -217,6 +225,7 @@ def apikey2_info(apikey, required_scopes=None): return {"sub": "bar"} return None + security_handler_factory = SecurityHandlerFactory() wrapped_func_key1 = security_handler_factory.verify_api_key( apikey1_info, "header", "X-Auth-1" ) @@ -251,8 +260,9 @@ def apikey2_info(apikey, required_scopes=None): assert await wrapped_func(request) == expected_token_info -async def test_verify_security_oauthproblem(security_handler_factory): +async def test_verify_security_oauthproblem(): """Tests whether verify_security raises an OAuthProblem if there are no auth_funcs.""" + security_handler_factory = SecurityHandlerFactory() security_func = security_handler_factory.verify_security([]) request = MagicMock() @@ -285,8 +295,8 @@ async def test_verify_security_oauthproblem(security_handler_factory): ([ConnexionException()], ConnexionException), ], ) -def test_raise_most_specific(errors, most_specific, security_handler_factory): +def test_raise_most_specific(errors, most_specific): """Tests whether most specific exception is raised from a list.""" - + security_handler_factory = SecurityHandlerFactory() with pytest.raises(most_specific): security_handler_factory._raise_most_specific(errors) diff --git a/tests/test_api.py b/tests/test_api.py index f102a010e..acace4253 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -5,7 +5,7 @@ import pytest from connexion import FlaskApi -from connexion.exceptions import InvalidSpecification, ResolverError +from connexion.exceptions import InvalidSpecification from connexion.spec import canonical_base_path from yaml import YAMLError diff --git a/tests/test_flask_encoder.py b/tests/test_flask_encoder.py index fd9634d7c..0ffc7fb6f 100644 --- a/tests/test_flask_encoder.py +++ b/tests/test_flask_encoder.py @@ -2,35 +2,37 @@ import json import math from decimal import Decimal +from unittest import mock -import pytest from connexion.frameworks.flask import FlaskJSONProvider from conftest import build_app_from_fixture -SPECS = ["swagger.yaml", "openapi.yaml"] +def test_json_encoder(): + json_encoder = json.JSONEncoder + json_encoder.default = FlaskJSONProvider.default -def test_json_encoder(simple_app): - flask_app = simple_app.app - - s = FlaskJSONProvider(flask_app).dumps({1: 2}) + s = json.dumps({1: 2}, cls=json_encoder) assert '{"1": 2}' == s - s = FlaskJSONProvider(flask_app).dumps(datetime.date.today()) + s = json.dumps(datetime.date.today(), cls=json_encoder) assert len(s) == 12 - s = FlaskJSONProvider(flask_app).dumps(datetime.datetime.utcnow()) + s = json.dumps(datetime.datetime.utcnow(), cls=json_encoder) assert s.endswith('Z"') - s = FlaskJSONProvider(flask_app).dumps(Decimal(1.01)) + s = json.dumps(Decimal(1.01), cls=json_encoder) assert s == "1.01" - s = FlaskJSONProvider(flask_app).dumps(math.expm1(1e-10)) + s = json.dumps(math.expm1(1e-10), cls=json_encoder) assert s == "1.00000000005e-10" -def test_json_encoder_datetime_with_timezone(simple_app): +def test_json_encoder_datetime_with_timezone(): + json_encoder = json.JSONEncoder + json_encoder.default = FlaskJSONProvider.default + class DummyTimezone(datetime.tzinfo): def utcoffset(self, dt): return datetime.timedelta(0) @@ -38,12 +40,10 @@ def utcoffset(self, dt): def dst(self, dt): return datetime.timedelta(0) - flask_app = simple_app.app - s = FlaskJSONProvider(flask_app).dumps(datetime.datetime.now(DummyTimezone())) + s = json.dumps(datetime.datetime.now(DummyTimezone()), cls=json_encoder) assert s.endswith('+00:00"') -@pytest.mark.parametrize("spec", SPECS) def test_readonly(json_datetime_dir, spec): app = build_app_from_fixture(json_datetime_dir, spec, validate_responses=True) app_client = app.test_client() diff --git a/tests/test_json_validation.py b/tests/test_json_validation.py index 4457a7e08..da89650de 100644 --- a/tests/test_json_validation.py +++ b/tests/test_json_validation.py @@ -10,10 +10,7 @@ from conftest import build_app_from_fixture -SPECS = ["swagger.yaml", "openapi.yaml"] - -@pytest.mark.parametrize("spec", SPECS) def test_validator_map(json_validation_spec_dir, spec): def validate_type(validator, types, instance, schema): types = _utils.ensure_list(types) @@ -53,7 +50,6 @@ def __init__(self, *args, **kwargs): assert res.status_code == 400 -@pytest.mark.parametrize("spec", SPECS) def test_readonly(json_validation_spec_dir, spec): app = build_app_from_fixture( json_validation_spec_dir, spec, validate_responses=True @@ -80,7 +76,6 @@ def test_readonly(json_validation_spec_dir, spec): assert res.status_code == 400 -@pytest.mark.parametrize("spec", SPECS) def test_writeonly(json_validation_spec_dir, spec): app = build_app_from_fixture( json_validation_spec_dir, spec, validate_responses=True @@ -107,7 +102,6 @@ def test_writeonly(json_validation_spec_dir, spec): ) -@pytest.mark.parametrize("spec", SPECS) def test_nullable_default(json_validation_spec_dir, spec): spec_path = pathlib.Path(json_validation_spec_dir) / spec Specification.load(spec_path) diff --git a/tests/test_middleware.py b/tests/test_middleware.py index 0d23193b1..aa966be48 100644 --- a/tests/test_middleware.py +++ b/tests/test_middleware.py @@ -2,7 +2,7 @@ from connexion.middleware import ConnexionMiddleware from starlette.datastructures import MutableHeaders -from conftest import SPECS, build_app_from_fixture +from conftest import build_app_from_fixture class TestMiddleware: @@ -30,10 +30,10 @@ async def patched_send(message): await self.app(scope, receive, patched_send) -@pytest.fixture(scope="session", params=SPECS) -def middleware_app(request): +@pytest.fixture(scope="session") +def middleware_app(spec): middlewares = ConnexionMiddleware.default_middlewares + [TestMiddleware] - return build_app_from_fixture("simple", request.param, middlewares=middlewares) + return build_app_from_fixture("simple", spec, middlewares=middlewares) def test_routing_middleware(middleware_app): diff --git a/tests/test_operation2.py b/tests/test_operation2.py index ddcb5928f..f60c80075 100644 --- a/tests/test_operation2.py +++ b/tests/test_operation2.py @@ -12,6 +12,7 @@ from connexion.middleware.security import SecurityOperation from connexion.operations import Swagger2Operation from connexion.resolver import Resolver +from connexion.security import SecurityHandlerFactory TEST_FOLDER = pathlib.Path(__file__).parent @@ -376,9 +377,9 @@ @pytest.fixture -def api(security_handler_factory): +def api(): api = mock.MagicMock(jsonifier=Jsonifier) - api.security_handler_factory = security_handler_factory + api.security_handler_factory = SecurityHandlerFactory() yield api @@ -394,7 +395,7 @@ def make_operation(op, definitions=True, parameters=True): return resolve_refs(new_op)["wrapper"] -def test_operation(api, security_handler_factory): +def test_operation(api): op_spec = make_operation(OPERATION1) operation = Swagger2Operation( api=api, @@ -417,7 +418,9 @@ def test_operation(api, security_handler_factory): assert operation.body_schema() == expected_body_schema -def test_operation_remote_token_info(security_handler_factory): +def test_operation_remote_token_info(): + security_handler_factory = SecurityHandlerFactory() + verify_oauth = mock.MagicMock(return_value="verify_oauth_result") security_handler_factory.verify_oauth = verify_oauth security_handler_factory.get_token_info_remote = mock.MagicMock( @@ -490,7 +493,9 @@ def test_operation_composed_definition(api): assert operation.body_schema() == expected_body_schema -def test_operation_local_security_oauth2(security_handler_factory): +def test_operation_local_security_oauth2(): + security_handler_factory = SecurityHandlerFactory() + verify_oauth = mock.MagicMock(return_value="verify_oauth_result") security_handler_factory.verify_oauth = verify_oauth @@ -506,7 +511,9 @@ def test_operation_local_security_oauth2(security_handler_factory): ) -def test_operation_local_security_duplicate_token_info(security_handler_factory): +def test_operation_local_security_duplicate_token_info(): + security_handler_factory = SecurityHandlerFactory() + verify_oauth = mock.MagicMock(return_value="verify_oauth_result") security_handler_factory.verify_oauth = verify_oauth @@ -546,7 +553,8 @@ def test_multi_body(api): ) -def test_no_token_info(security_handler_factory): +def test_no_token_info(): + security_handler_factory = SecurityHandlerFactory() SecurityOperation( next_app=mock.Mock, security_handler_factory=security_handler_factory, @@ -555,12 +563,13 @@ def test_no_token_info(security_handler_factory): ) -def test_multiple_security_schemes_and(security_handler_factory): +def test_multiple_security_schemes_and(): """Tests an operation with multiple security schemes in AND fashion.""" def return_api_key_name(func, in_, name): return name + security_handler_factory = SecurityHandlerFactory() verify_api_key = mock.MagicMock(side_effect=return_api_key_name) security_handler_factory.verify_api_key = verify_api_key verify_multiple = mock.MagicMock(return_value="verify_multiple_result") @@ -583,11 +592,13 @@ def return_api_key_name(func, in_, name): verify_multiple.assert_called_with({"key1": "X-Auth-1", "key2": "X-Auth-2"}) -def test_multiple_oauth_in_and(security_handler_factory, caplog): +def test_multiple_oauth_in_and(caplog): """Tests an operation with multiple oauth security schemes in AND fashion. These should be ignored and raise a warning. """ caplog.set_level(logging.WARNING, logger="connexion.operations.secure") + security_handler_factory = SecurityHandlerFactory() + verify_oauth = mock.MagicMock(return_value="verify_oauth_result") security_handler_factory.verify_oauth = verify_oauth @@ -683,8 +694,10 @@ def test_get_path_parameter_types(api): } == operation.get_path_parameter_types() -def test_oauth_scopes_in_or(security_handler_factory): +def test_oauth_scopes_in_or(): """Tests whether an OAuth security scheme with 2 different possible scopes is correctly handled.""" + security_handler_factory = SecurityHandlerFactory() + verify_oauth = mock.MagicMock(return_value="verify_oauth_result") security_handler_factory.verify_oauth = verify_oauth diff --git a/tests/test_resolver_methodview.py b/tests/test_resolver_methodview.py index e52944f58..502fed263 100644 --- a/tests/test_resolver_methodview.py +++ b/tests/test_resolver_methodview.py @@ -1,5 +1,7 @@ from connexion.operations import OpenAPIOperation -from connexion.resolver import Resolver +from connexion.resolver import MethodViewResolver, Resolver + +from conftest import build_app_from_fixture COMPONENTS = {"parameters": {"myparam": {"in": "path", "schema": {"type": "integer"}}}} @@ -188,7 +190,13 @@ 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(method_view_app): +def test_method_view_resolver_integration(spec, method_view_resolver): + method_view_app = build_app_from_fixture( + "method_view", + spec, + resolver=MethodViewResolver("fakeapi.example_method_view"), + ) + client = method_view_app.test_client() r = client.get("/v1.0/pets")