Enforce common responses

[zupo]

Use case: I want to make sure that all of my endpoints define at least some responses. For example, I want all endpoints to define 200, 404, 403 and 500. If not, fail app startup.

We have a similar thing already built for https://app.woocart.com/api/v1/, but it's a script that we run in CI. And I guess other people probably have a similar use case. I'm pasting the files below, as a starting point.

openapi_responses.yaml:

# Required response definitions
required_responses:
  get:
    - '200'
    - '400'
    - '403'
    - '5XX'
  post:
    - '200'
    - '400'
    - '403'
    - '5XX'
  put:
    - '200'
    - '400'
    - '403'
    - '5XX'
  patch:
    - '200'
    - '400'
    - '403'
    - '5XX'
  delete:
    - '200'
    - '400'
    - '403'
    - '5XX'

# Additional required response definitions for endpoints with parameters
required_responses_params:
  get:
    - '404'
  post:
    - '404'
  put:
    - '404'
  patch:
    - '404'
  delete:
    - '404'

# Allowed missing response definitions
allowed_missing_responses:
  /user/login/:
    post:
      - '403'
  /user/logout/:
    post:
      - '200'
      - '403'
  /openapi_validation/{formatter_name}/:
    post:
      - '403'
  /stores/{storeId}/backups/{backupId}/download:
    get:
      - '200'

check_openapi_responses.py

"""Verify that endpoints have defined required responses."""

from pathlib import Path

import argparse
import openapi_core
import sys
import typing as t
import yaml


def get_config(config_path: str) -> t.Dict:
    """Read config from file."""
    config = Path(config_path)

    if not config.is_file():
        sys.stdout.write(f"ERROR Config file not found on: {config}\n")
        sys.exit(1)

    with open(config, "r") as f:
        return yaml.safe_load(f)


def get_spec(schema_path: str) -> t.Any:
    """Create openapi spec from schema."""
    schema = Path(schema_path)

    if not schema.is_file():
        sys.stdout.write(f"ERROR schema file not found on: {schema}\n")
        sys.exit(1)

    with open(schema, "r") as f:
        return openapi_core.create_spec(yaml.safe_load(f))


def required_responses(
    config: t.Dict, endpoint: str, method: str, has_params: bool
) -> t.Set:
    """Get required responses for given method on endpoint."""
    required_responses: t.Set = set(
        config.get("required_responses", {}).get(method, [])
    )
    if has_params:
        required_params: t.Set = set(
            config.get("required_responses_params", {}).get(method, [])
        )
        required_responses = required_responses.union(required_params)
    allowed_missing: t.Set = set(
        config.get("allowed_missing_responses", {}).get(endpoint, {}).get(method, [])
    )
    required_responses = required_responses - allowed_missing
    return required_responses


def check_responses(spec, config: t.Dict) -> None:
    """Verify that all endpoints have defined required responses."""
    check_failed = False
    missing_responses_count = 0

    for path in spec.paths.values():
        for operation in path.operations.values():
            operation_responses = operation.responses.keys()
            method = operation.http_method
            endpoint = operation.path_name
            has_params = len(operation.parameters) > 0
            required = required_responses(config, endpoint, method, has_params)

            missing_responses = required - operation_responses
            for missing_response in missing_responses:
                check_failed = True
                missing_responses_count += 1
                sys.stdout.write(
                    "ERROR missing response "
                    f"'{missing_response}' for '{method}' request on path "
                    f"'{endpoint}'\n"
                )
    if check_failed:
        sys.stdout.write(
            "\nFAILED: Openapi responses check: "
            f"{missing_responses_count} missing response definitions. \n"
        )
        sys.exit(1)
    else:
        sys.exit(0)


def main(argv=sys.argv) -> None:  # pragma: no cover
    """Run endpoints responses check."""
    parser = argparse.ArgumentParser(
        usage=(
            "python openapi_responses.py --schema openapi.yaml --config responses.yaml \n"
        )
    )
    parser.add_argument(
        "--schema",
        "-s",
        required=True,
        type=str,
        metavar="<schema>",
        help="File with openapi schema.",
    )
    parser.add_argument(
        "--config",
        "-c",
        nargs="?",
        type=str,
        metavar="<config>",
        help="File with exceptions for responses.",
    )
    args = parser.parse_args()

    config = {}
    if args.config:
        config = get_config(args.config)
    spec = get_spec(args.schema)

    check_responses(spec, config)


if __name__ == "__main__":
    main()

[zupo]

I've drafted how I want the configuration for this functionality to look & feel: #103

Instead of requiring the openapi_responses.yaml file, I opted for simple defaults that can be overridden. In the case of the openapi_responses.yaml provided above, the configuration would look like this (assuming defaults defined in https://github.com/Pylons/pyramid_openapi3/pull/103/files#diff-a8182b0c3377a86ad3bb94d2301b04ff58798bbf01c9257be61d5aa088887a7aR358-R366):

    config.endpoint_validation_overrides = 
    {
        "/user/logout": {"post": [202, 400, 500]},
        "/stores/{storeId}/backups/{backupId}/download:": {"post": [202, 400, 500]},
    }

It's not ideal to not be able to support endpoint_validation_overrides via .ini files. Anyone has a better idea?

Would love to hear feedback and ideas on how to improve the UX of this new check/feature!

[damonhook]

There are 2 cases covered by the OpenApi specification that I just wanted to bring to light here (and maybe discuss how / if they should be handled, or just documented as edge cases in the readme / upcoming docs):

1. The response HTTP code is actually a patterned field. For example, you could have a 4XX response key, which will handle [400-499] response codes. Ideally, if 400 and 404 are required, and endpoint that has defined 4XX should satisfy these

2. This one I have not seen as often, but you are allowed to define a default key (which will satisfy all response codes not covered individually).

---

The first point (4XX / 5XX) would fail when a set intersection is done.

The second point (default) could be easily handled as if an endpoint has one, it automatically passes the required status code check.

[damonhook]

One option would be to test it ourself.

Another possibility may be to leverage openapi-core for it: If you call the .get_response(self, http_status) function for each required response on each operation. It will throw an exception if it cannot find a matching path in the spec (source)

This way it would solve both points from the previous comment without relying on specialized code inside this library

[zupo]

Fantastic insights, thanks!