Refactor tests (#1631)

This PR refactors our tests to prepare the activation of our tests for
our AsyncApp.

The separate commits are atomic, which can be useful for reviewing.

---------

Co-authored-by: Ruwann <ruwanlambrichts@gmail.com>

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)

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"))

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()