-
Notifications
You must be signed in to change notification settings - Fork 46
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add ability to check openapi responses.
This can be used to ensure all API endpoints support common responses. Path to responses.yaml needs to be provied via config `pyramid_openapi3_responses_config`
- Loading branch information
1 parent
5d77dcb
commit a4c33b2
Showing
4 changed files
with
244 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,89 @@ | ||
"""Verify that endpoints have defined required responses.""" | ||
|
||
from .exceptions import ResponseValidationError | ||
from openapi_core.schema.specs.models import Spec | ||
from pathlib import Path | ||
from pyramid.config import Configurator | ||
|
||
import openapi_core | ||
import typing as t | ||
import yaml | ||
|
||
|
||
def get_config(config_path: str) -> t.Dict[str, str]: | ||
"""Read config from file.""" | ||
config = Path(config_path) | ||
|
||
if not config.is_file(): | ||
raise Exception(f"ERROR Config file not found on: {config}\n") | ||
|
||
with open(config, "r") as f: | ||
return yaml.safe_load(f) | ||
|
||
|
||
def get_spec(schema_path: str) -> Spec: | ||
"""Create openapi spec from schema.""" | ||
schema = Path(schema_path) | ||
|
||
if not schema.is_file(): | ||
raise Exception(f"ERROR schema file not found on: {schema}\n") | ||
|
||
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_resp: 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_resp = required_resp.union(required_params) | ||
allowed_missing: t.Set = set( | ||
config.get("allowed_missing_responses", {}).get(endpoint, {}).get(method, []) | ||
) | ||
required_resp = required_resp - allowed_missing | ||
return required_resp | ||
|
||
|
||
def validate_required_responses(spec_file: str, config: Configurator) -> None: | ||
"""Verify that all endpoints have defined required responses.""" | ||
check_failed: bool = False | ||
missing_responses_count: int = 0 | ||
errors: t.List = [] | ||
|
||
filepath: str = config.registry.settings.get("pyramid_openapi3_responses_config") | ||
if not filepath: | ||
return | ||
|
||
responses_config: t.Dict[str, str] = get_config(filepath) | ||
|
||
spec: Spec = get_spec(spec_file) | ||
for path in spec.paths.values(): | ||
for operation in path.operations.values(): | ||
operation_responses = operation.responses.keys() | ||
method: str = operation.http_method | ||
endpoint: str = operation.path_name | ||
has_params: bool = len(operation.parameters) > 0 | ||
required: t.Set = required_responses( | ||
responses_config, endpoint, method, has_params | ||
) | ||
|
||
missing_responses: t.Set = required - operation_responses | ||
for missing_response in missing_responses: | ||
check_failed = True | ||
missing_responses_count += 1 | ||
errors.append( | ||
"ERROR missing response " | ||
f"'{missing_response}' for '{method}' request on path " | ||
f"'{endpoint}'\n" | ||
) | ||
if check_failed: | ||
errors.append( | ||
"\nFAILED: Openapi responses check: " | ||
f"{missing_responses_count} missing response definitions. \n" | ||
) | ||
raise ResponseValidationError(errors=errors) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,146 @@ | ||
"""Tests validation of openapi responses.""" | ||
|
||
from pyramid.testing import testConfig | ||
from pyramid_openapi3.check_openapi_responses import validate_required_responses | ||
from pyramid_openapi3.exceptions import ResponseValidationError | ||
|
||
import pytest | ||
import tempfile | ||
|
||
VALIDATION_DOCUMENT = b""" | ||
required_responses: | ||
get: | ||
- '200' | ||
- '400' | ||
# Additional required response definitions for endpoints with parameters | ||
required_responses_params: | ||
get: | ||
- '404' | ||
""" | ||
|
||
|
||
def test_get_config_no_file() -> None: | ||
"""Test get_config when config is not a file object.""" | ||
from pyramid_openapi3.check_openapi_responses import get_config | ||
|
||
with pytest.raises(Exception): | ||
get_config("file_does_not_exist") | ||
|
||
|
||
def test_get_spec_no_file() -> None: | ||
"""Test get_spec when config is not a file object.""" | ||
from pyramid_openapi3.check_openapi_responses import get_spec | ||
|
||
with pytest.raises(Exception): | ||
get_spec("file_does_not_exist") | ||
|
||
|
||
def test_response_validation_error() -> None: | ||
"""Test that ResponseValidationError is raised when 404 is missing.""" | ||
|
||
SPEC_DOCUMENT = b""" | ||
openapi: "3.0.0" | ||
info: | ||
version: "1.0.0" | ||
title: Foo API | ||
paths: | ||
/foo: | ||
get: | ||
parameters: | ||
- name: bar | ||
in: query | ||
schema: | ||
type: integer | ||
responses: | ||
"200": | ||
description: A foo | ||
"400": | ||
description: Bad Request | ||
""" | ||
|
||
with testConfig() as config: | ||
config.include("pyramid_openapi3") | ||
|
||
with tempfile.NamedTemporaryFile() as document: | ||
with tempfile.NamedTemporaryFile() as spec: | ||
spec.write(SPEC_DOCUMENT) | ||
spec.seek(0) | ||
document.write(VALIDATION_DOCUMENT) | ||
document.seek(0) | ||
|
||
config.add_settings(pyramid_openapi3_responses_config=document.name) | ||
with pytest.raises(ResponseValidationError): | ||
validate_required_responses(spec.name, config) | ||
|
||
|
||
def test_happy_path() -> None: | ||
"""Test validation of openapi responses.""" | ||
|
||
SPEC_DOCUMENT = b""" | ||
openapi: "3.0.0" | ||
info: | ||
version: "1.0.0" | ||
title: Foo API | ||
paths: | ||
/foo: | ||
get: | ||
parameters: | ||
- name: bar | ||
in: query | ||
schema: | ||
type: integer | ||
responses: | ||
"200": | ||
description: A foo | ||
"400": | ||
description: Bad Request | ||
"404": | ||
description: Not Found | ||
""" | ||
|
||
with testConfig() as config: | ||
config.include("pyramid_openapi3") | ||
|
||
with tempfile.NamedTemporaryFile() as document: | ||
with tempfile.NamedTemporaryFile() as spec: | ||
spec.write(SPEC_DOCUMENT) | ||
spec.seek(0) | ||
document.write(VALIDATION_DOCUMENT) | ||
document.seek(0) | ||
|
||
config.add_settings(pyramid_openapi3_responses_config=document.name) | ||
validate_required_responses(spec.name, config) | ||
|
||
|
||
def test_no_params() -> None: | ||
"""Test spec without parameters.""" | ||
|
||
SPEC_DOCUMENT = b""" | ||
openapi: "3.0.0" | ||
info: | ||
version: "1.0.0" | ||
title: Foo API | ||
paths: | ||
/foo: | ||
get: | ||
responses: | ||
"200": | ||
description: A foo | ||
"400": | ||
description: Bad Request | ||
"404": | ||
description: Not Found | ||
""" | ||
|
||
with testConfig() as config: | ||
config.include("pyramid_openapi3") | ||
|
||
with tempfile.NamedTemporaryFile() as document: | ||
with tempfile.NamedTemporaryFile() as spec: | ||
spec.write(SPEC_DOCUMENT) | ||
spec.seek(0) | ||
document.write(VALIDATION_DOCUMENT) | ||
document.seek(0) | ||
|
||
config.add_settings(pyramid_openapi3_responses_config=document.name) | ||
validate_required_responses(spec.name, config) |