Skip to content
This repository has been archived by the owner on Nov 19, 2023. It is now read-only.

Commit

Permalink
Add tests (#199)
Browse files Browse the repository at this point in the history
  • Loading branch information
sondrelg authored Feb 11, 2021
1 parent 5f30fbe commit 809fbb9
Show file tree
Hide file tree
Showing 6 changed files with 734 additions and 587 deletions.
1 change: 1 addition & 0 deletions openapi_tester/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,3 +47,4 @@
UNDOCUMENTED_SCHEMA_SECTION_ERROR = "Error: Unsuccessfully tried to index the OpenAPI schema by `{key}`. {error_addon}"
ONE_OF_ERROR = "Expected data to match one and only one of the oneOf schema types; found {matches} matches."
ANY_OF_ERROR = "Expected data to match one or more of the documented anyOf schema types, but found no matches."
INIT_ERROR = "Unable to configure loader."
5 changes: 4 additions & 1 deletion openapi_tester/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,10 @@ def _sort_data(data_object: Any) -> Any:
if isinstance(data_object, dict):
return dict(sorted(data_object.items()))
if isinstance(data_object, list):
return sorted(data_object)
try:
return sorted(data_object)
except TypeError:
pass
return data_object

@staticmethod
Expand Down
142 changes: 71 additions & 71 deletions openapi_tester/schema_tester.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
from openapi_tester.constants import (
ANY_OF_ERROR,
EXCESS_RESPONSE_KEY_ERROR,
INIT_ERROR,
INVALID_PATTERN_ERROR,
MISSING_RESPONSE_KEY_ERROR,
NONE_ERROR,
Expand Down Expand Up @@ -69,7 +70,70 @@ def __init__(
elif "drf_yasg" in settings.INSTALLED_APPS:
self.loader = DrfYasgSchemaLoader()
else:
raise ImproperlyConfigured("No loader is configured.")
raise ImproperlyConfigured(INIT_ERROR)

@staticmethod
def _get_key_value(schema: dict, key: str, error_addon: str = "") -> dict:
"""
Returns the value of a given key
"""
try:
return schema[key]
except KeyError as e:
raise UndocumentedSchemaSectionError(
UNDOCUMENTED_SCHEMA_SECTION_ERROR.format(key=key, error_addon=error_addon)
) from e

@staticmethod
def _get_status_code(schema: dict, status_code: Union[str, int], error_addon: str = "") -> dict:
"""
Returns the status code section of a schema, handles both str and int status codes
"""
if str(status_code) in schema:
return schema[str(status_code)]
if int(status_code) in schema:
return schema[int(status_code)]
raise UndocumentedSchemaSectionError(
UNDOCUMENTED_SCHEMA_SECTION_ERROR.format(key=status_code, error_addon=error_addon)
)

def get_response_schema_section(self, response: td.Response) -> dict:
"""
Fetches the response section of a schema, wrt. the route, method, status code, and schema version.
:param response: DRF Response Instance
:return dict
"""
schema = self.loader.get_schema()
parameterized_path = self.loader.parameterize_path(response.request["PATH_INFO"])
paths_object = self._get_key_value(schema, "paths")

pretty_routes = "\n\t• ".join(paths_object.keys())
route_object = self._get_key_value(
paths_object,
parameterized_path,
f"\n\nFor debugging purposes, other valid routes include: \n\n\t{pretty_routes}",
)

str_methods = ", ".join(method.upper() for method in route_object.keys() if method.upper() != "PARAMETERS")
method_object = self._get_key_value(
route_object, response.request["REQUEST_METHOD"].lower(), f"\n\nAvailable methods include: {str_methods}."
)

responses_object = self._get_key_value(method_object, "responses")
keys = ", ".join(str(key) for key in responses_object.keys())
status_code_object = self._get_status_code(
responses_object,
response.status_code,
f"\n\nUndocumented status code: {response.status_code}.\n\nDocumented responses include: {keys}. ",
)

if "openapi" not in schema:
# openapi 2.0, i.e. "swagger" has a different structure than openapi 3.0 status sub-schemas
return self._get_key_value(status_code_object, "schema")
content_object = self._get_key_value(status_code_object, "content")
json_object = self._get_key_value(content_object, "application/json")
return self._get_key_value(json_object, "schema")

def handle_all_of(
self,
Expand Down Expand Up @@ -150,69 +214,6 @@ def handle_any_of(
hint="",
)

@staticmethod
def _get_key_value(schema: dict, key: str, error_addon: str = "") -> dict:
"""
Returns the value of a given key
"""
try:
return schema[key]
except KeyError as e:
raise UndocumentedSchemaSectionError(
UNDOCUMENTED_SCHEMA_SECTION_ERROR.format(key=key, error_addon=error_addon)
) from e

@staticmethod
def _get_status_code(schema: dict, status_code: Union[str, int], error_addon: str = "") -> dict:
"""
Returns the status code section of a schema, handles both str and int status codes
"""
if str(status_code) in schema:
return schema[str(status_code)]
if int(status_code) in schema:
return schema[int(status_code)]
raise UndocumentedSchemaSectionError(
UNDOCUMENTED_SCHEMA_SECTION_ERROR.format(key=status_code, error_addon=error_addon)
)

def get_response_schema_section(self, response: td.Response) -> dict:
"""
Fetches the response section of a schema, wrt. the route, method, status code, and schema version.
:param response: DRF Response Instance
:return dict
"""
schema = self.loader.get_schema()
parameterized_path = self.loader.parameterize_path(response.request["PATH_INFO"])
paths_object = self._get_key_value(schema, "paths")

pretty_routes = "\n\t• ".join(paths_object.keys())
route_object = self._get_key_value(
paths_object,
parameterized_path,
f"\n\nFor debugging purposes, other valid routes include: \n\n\t{pretty_routes}",
)

str_methods = ", ".join(method.upper() for method in route_object.keys() if method.upper() != "PARAMETERS")
method_object = self._get_key_value(
route_object, response.request["REQUEST_METHOD"].lower(), f"\n\nAvailable methods include: {str_methods}."
)

responses_object = self._get_key_value(method_object, "responses")
keys = ", ".join(str(key) for key in responses_object.keys())
status_code_object = self._get_status_code(
responses_object,
response.status_code,
f"\n\nUndocumented status code: {response.status_code}.\n\nDocumented responses include: {keys}. ",
)

if "openapi" not in schema:
# openapi 2.0, i.e. "swagger" has a different structure than openapi 3.0 status sub-schemas
return self._get_key_value(status_code_object, "schema")
content_object = self._get_key_value(status_code_object, "content")
json_object = self._get_key_value(content_object, "application/json")
return self._get_key_value(json_object, "schema")

@staticmethod
def is_nullable(schema_item: dict) -> bool:
"""
Expand Down Expand Up @@ -267,11 +268,7 @@ def _validate_format(schema_section: dict, data: Any) -> Optional[str]:
valid = isinstance(data, bytes)
elif schema_format in ["date", "date-time"]:
parser = parse_date if schema_format == "date" else parse_datetime
try:
result = parser(data)
valid = result is not None
except ValueError:
valid = False
valid = parser(data) is not None
return None if valid else VALIDATE_FORMAT_ERROR.format(expected=schema_section["format"], received=str(data))

def _validate_openapi_type(self, schema_section: dict, data: Any) -> Optional[str]:
Expand All @@ -281,14 +278,17 @@ def _validate_openapi_type(self, schema_section: dict, data: Any) -> Optional[st
return None
if schema_type in ["file", "string"]:
valid = isinstance(data, (str, bytes))
elif schema_type == "boolean":
valid = isinstance(data, bool)
elif schema_type == "integer":
valid = isinstance(data, int)
valid = isinstance(data, int) and not isinstance(data, bool)
elif schema_type == "number":
valid = isinstance(data, (int, float))
valid = isinstance(data, (int, float)) and not isinstance(data, bool)
elif schema_type == "object":
valid = isinstance(data, dict)
elif schema_type == "array":
valid = isinstance(data, list)

return (
None
if valid
Expand Down
9 changes: 9 additions & 0 deletions tests/test_exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,3 +45,12 @@ def test_documentation_error_message():
def test_case_error_message():
error = CaseError(key="test-key", case="camelCase", expected="testKey")
assert error.args[0].strip() == "The response key `test-key` is not properly camelCase. Expected value: testKey"


def test_documentation_error_sort_data_type():
assert DocumentationError._sort_data([1, 3, 2]) == [1, 2, 3] # list
assert DocumentationError._sort_data({"1", "3", "2"}) == {"1", "2", "3"} # set
assert DocumentationError._sort_data({"1": "a", "3": "a", "2": "a"}) == {"1": "a", "2": "a", "3": "a"} # dict

# Test sort failure scenario - expect the method to succeed and default to no reordering
assert DocumentationError._sort_data(["1", {}, []]) == ["1", {}, []]
Loading

0 comments on commit 809fbb9

Please sign in to comment.