Add example of the Starlette test client on a simple Connexion REST app

The example Connexion app processes JSON requests and responses as specified
with OpenAPI v2 (aka Swagger) or OpenAPI v3 file format.
JSON responses are validated against the spec.
An error handler catches exceptions raised while processing a request.
Tests are run by `tox` which also reports code coverage.

Author: chrisinmtown

examples/testclient/README.rst

@@ -0,0 +1,45 @@
+===================
+Test Client Example
+===================
+
+This directory offers an example of using the Starlette test client
+to test a Connexion app. The app processes JSON requests and responses
+as specified with OpenAPI v2 (aka Swagger) or OpenAPI v3 file format.
+In addition, the responses are validated against the spec, and an error
+handler catches exceptions raised while processing a request. The tests
+are run by `tox` which also reports code coverage.
+
+Preparing
+---------
+
+Create a new virtual environment and install the required libraries
+with these commands:
+
+.. code-block:: bash
+
+    $ python -m venv my-venv
+    $ source my-venv/bin/activate
+    $ pip install 'connexion[flask,swagger-ui,uvicorn]>=3.1.0' tox
+
+Testing
+-------
+
+Run the test suite and generate the coverage report with this command:
+
+.. code-block:: bash
+
+    $ tox
+
+Running
+-------
+
+Launch the connexion server with this command:
+
+.. code-block:: bash
+
+    $ python -m hello.app
+
+Now open your browser and view the Swagger UI for these specification files:
+
+* http://localhost:8080/openapi/ui/ for the OpenAPI 3 spec
+* http://localhost:8080/swagger/ui/ for the Swagger 2 spec

examples/testclient/hello/__init__.py

@@ -0,0 +1,47 @@
+import logging
+
+from connexion import FlaskApp
+from connexion.lifecycle import ConnexionRequest, ConnexionResponse
+from connexion.problem import problem
+
+logger = logging.getLogger(__name__)
+
+
+async def handle_error(request: ConnexionRequest, ex: Exception) -> ConnexionResponse:
+    """
+    Report an error that happened while processing a request.
+    See: https://connexion.readthedocs.io/en/latest/exceptions.html
+
+    This function is defined as `async` so it can be called by the
+    Connexion asynchronous middleware framework without a wrapper.
+    If a plain function is provided, the framework calls the function
+    from a threadpool and the exception stack trace is not available.
+
+    :param request: Request that failed
+    :parm ex: Exception that was raised
+    :return: ConnexionResponse with RFC7087 problem details
+    """
+    # log the request URL, exception and stack trace
+    logger.exception("Request to %s caused exception", request.url)
+    return problem(title="Error", status=500, detail=repr(ex))
+
+
+def create_app() -> FlaskApp:
+    """
+    Create the Connexion FlaskApp, which wraps a Flask app.
+
+    :return Newly created FlaskApp
+    """
+    app = FlaskApp(__name__, specification_dir="spec/")
+    # hook the functions to the OpenAPI spec
+    title = {"title": "Hello World Plus Example"}
+    app.add_api("openapi.yaml", arguments=title, validate_responses=True)
+    app.add_api("swagger.yaml", arguments=title, validate_responses=True)
+    # hook an async function that is invoked on any exception
+    app.add_error_handler(Exception, handle_error)
+    # return the fully initialized FlaskApp
+    return app
+
+
+# create and publish for import by other modules
+conn_app = create_app()

examples/testclient/hello/app.py

@@ -0,0 +1,30 @@
+import logging
+
+from . import conn_app
+
+# reuse the configured logger
+logger = logging.getLogger("uvicorn.error")
+
+
+def post_greeting(name: str, body: dict) -> tuple:
+    logger.info(
+        "%s: name len %d, body items %d", post_greeting.__name__, len(name), len(body)
+    )
+    # the body is optional
+    message = body.get("message", None)
+    if "crash" == message:
+        raise ValueError(f"Raise exception for {name}")
+    if "invalid" == message:
+        return {"bogus": "response"}
+    return {"greeting": f"Hello {name}"}, 200
+
+
+def main() -> None:
+    # ensure logging is configured
+    logging.basicConfig(level=logging.DEBUG)
+    # launch the app using the dev server
+    conn_app.run("hello:conn_app", port=8080)
+
+
+if __name__ == "__main__":
+    main()

examples/testclient/hello/spec/openapi.yaml

@@ -0,0 +1,48 @@
+openapi: "3.0.0"
+
+info:
+  title: Hello World
+  version: "1.0"
+
+servers:
+  - url: /openapi
+
+paths:
+  /greeting/{name}:
+    post:
+      summary: Generate greeting
+      description: Generates a greeting message.
+      operationId: hello.app.post_greeting
+      parameters:
+        - name: name
+          in: path
+          description: Name of the person to greet.
+          required: true
+          schema:
+            type: string
+            example: "dave"
+      requestBody:
+        description: >
+          Optional body with a message.
+          Send message "crash" or "invalid" to simulate an error.
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                message:
+                  type: string
+                  example: "hi"
+      responses:
+        '200':
+          description: greeting response
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  greeting:
+                    type: string
+                    example: "Hello John"
+                required:
+                  - greeting

examples/testclient/hello/spec/swagger.yaml

@@ -0,0 +1,44 @@
+swagger: "2.0"
+
+info:
+  title: "{{title}}"
+  version: "1.0"
+
+basePath: /swagger
+
+paths:
+  /greeting/{name}:
+    post:
+      summary: Generate greeting
+      operationId: hello.app.post_greeting
+      parameters:
+        - name: name
+          in: path
+          description: Name of the person to greet.
+          required: true
+          type: string
+        - name: body
+          in: body
+          description: >
+            Optional body with a message.
+            Send message "crash" or "invalid" to simulate an error.
+          schema:
+            type: object
+            properties:
+              message:
+                type: string
+                example: "hi"
+      produces:
+        - application/json
+      responses:
+        '200':
+          description: greeting response
+          schema:
+            type: object
+            properties:
+              greeting:
+                type: string
+            required:
+              - greeting
+            example:
+              greeting: "Hello John"

examples/testclient/requirements.txt

@@ -0,0 +1 @@
+connexion[flask,swagger-ui,uvicorn]

examples/testclient/tests/__init__.py

@@ -0,0 +1,2 @@
+# empty __init__.py so that pytest can add correct path to coverage report
+# https://github.com/pytest-dev/pytest-cov/issues/98#issuecomment-451344057

examples/testclient/tests/conftest.py

@@ -0,0 +1,17 @@
+"""
+fixtures available for injection to tests by pytest
+"""
+import pytest
+from starlette.testclient import TestClient
+from hello.app import conn_app
+
+
+@pytest.fixture
+def client():
+    """
+    Create a Connexion test_client from the Connexion app.
+
+    https://connexion.readthedocs.io/en/stable/testing.html
+    """
+    client: TestClient = conn_app.test_client()
+    yield client

examples/testclient/tests/test_app.py

@@ -0,0 +1,47 @@
+from httpx import Response
+from pytest_mock import MockerFixture
+from starlette.testclient import TestClient
+from hello import app
+
+greeting = "greeting"
+prefixes = ["openapi", "swagger"]
+
+
+def test_greeting_success(client: TestClient):
+    name = "dave"
+    for prefix in prefixes:
+        # a body is required in the POST
+        res: Response = client.post(
+            f"/{prefix}/{greeting}/{name}", json={"message": "hi"}
+        )
+        assert res.status_code == 200
+        assert name in res.json()[greeting]
+
+
+def test_greeting_exception(client: TestClient):
+    name = "dave"
+    for prefix in prefixes:
+        # a body is required in the POST
+        res: Response = client.post(
+            f"/{prefix}/{greeting}/{name}", json={"message": "crash"}
+        )
+        assert res.status_code == 500
+        assert name in res.json()["detail"]
+
+
+def test_greeting_invalid(client: TestClient):
+    name = "dave"
+    for prefix in prefixes:
+        # a body is required in the POST
+        res: Response = client.post(
+            f"/{prefix}/{greeting}/{name}", json={"message": "invalid"}
+        )
+        assert res.status_code == 500
+        assert "Response body does not conform" in res.json()["detail"]
+
+
+def test_main(mocker: MockerFixture):
+    # patch the run-app function to do nothing
+    mock_run = mocker.patch("hello.app.conn_app.run")
+    app.main()
+    mock_run.assert_called()

examples/testclient/tests/test_init.py

@@ -0,0 +1,18 @@
+from connexion.lifecycle import ConnexionResponse
+import json
+import pytest
+from unittest.mock import Mock
+from hello import handle_error
+
+
+@pytest.mark.asyncio
+async def test_handle_error():
+    # Mock the ConnexionRequest object
+    mock_req = Mock()
+    mock_req.url = "http://some/url"
+    # call the function
+    conn_resp: ConnexionResponse = await handle_error(mock_req, ValueError("Value"))
+    assert 500 == conn_resp.status_code
+    # check structure of response
+    resp_dict = json.loads(conn_resp.body)
+    assert "Error" == resp_dict["title"]