diff --git a/README.md b/README.md index eb1d4052..a7c20dc2 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,7 @@ from openapi_tester import SchemaTester schema_tester = SchemaTester(schema_file_path="./schemas/publishedSpecs.yaml") ``` -Once you've instantiated a tester, you can use it to test responses: +Once you've instantiated a tester, you can use it to test responses and request bodies: ```python from openapi_tester.schema_tester import SchemaTester @@ -53,6 +53,12 @@ def test_response_documentation(client): response = client.get('api/v1/test/1') assert response.status_code == 200 schema_tester.validate_response(response=response) + + +def test_request_documentation(client): + response = client.get('api/v1/test/1') + assert response.status_code == 200 + schema_tester.validate_request(response=response) ``` If you are using the Django testing framework, you can create a base `APITestCase` that incorporates schema validation: @@ -188,11 +194,11 @@ In case of issues with the schema itself, the validator will raise the appropria The library includes an `OpenAPIClient`, which extends Django REST framework's [`APIClient` class](https://www.django-rest-framework.org/api-guide/testing/#apiclient). -If you wish to validate each response against OpenAPI schema when writing +If you wish to validate each request and response against OpenAPI schema when writing unit tests - `OpenAPIClient` is what you need! To use `OpenAPIClient` simply pass `SchemaTester` instance that should be used -to validate responses and then use it like regular Django testing client: +to validate requests and responses and then use it like regular Django testing client: ```python schema_tester = SchemaTester() diff --git a/openapi_tester/clients.py b/openapi_tester/clients.py index b7da2d05..2727f13c 100644 --- a/openapi_tester/clients.py +++ b/openapi_tester/clients.py @@ -27,9 +27,15 @@ def __init__( def request(self, **kwargs) -> Response: # type: ignore[override] """Validate fetched response against given OpenAPI schema.""" response = super().request(**kwargs) + if self._is_successful_response(response): + self.schema_tester.validate_request(response) self.schema_tester.validate_response(response) return response + @staticmethod + def _is_successful_response(response: Response) -> bool: + return response.status_code < 400 + @staticmethod def _schema_tester_factory() -> SchemaTester: """Factory of default ``SchemaTester`` instances.""" diff --git a/openapi_tester/constants.py b/openapi_tester/constants.py index 512efb47..c6419e14 100644 --- a/openapi_tester/constants.py +++ b/openapi_tester/constants.py @@ -15,9 +15,9 @@ INVALID_PATTERN_ERROR = "String pattern is not valid regex: {pattern}" VALIDATE_ENUM_ERROR = "Expected: a member of the enum {enum}\n\nReceived: {received}" VALIDATE_TYPE_ERROR = 'Expected: {article} "{type}" type value\n\nReceived: {received}' -VALIDATE_MULTIPLE_OF_ERROR = "The response value {data} should be a multiple of {multiple}" -VALIDATE_MINIMUM_ERROR = "The response value {data} is lower than the specified minimum of {minimum}" -VALIDATE_MAXIMUM_ERROR = "The response value {data} exceeds the maximum allowed value of {maximum}" +VALIDATE_MULTIPLE_OF_ERROR = "The value {data} should be a multiple of {multiple}" +VALIDATE_MINIMUM_ERROR = "The value {data} is lower than the specified minimum of {minimum}" +VALIDATE_MAXIMUM_ERROR = "The value {data} exceeds the maximum allowed value of {maximum}" VALIDATE_MIN_LENGTH_ERROR = 'The length of "{data}" is shorter than the specified minimum length of {min_length}' VALIDATE_MAX_LENGTH_ERROR = 'The length of "{data}" exceeds the specified maximum length of {max_length}' VALIDATE_MIN_ARRAY_LENGTH_ERROR = ( @@ -32,9 +32,9 @@ ) VALIDATE_UNIQUE_ITEMS_ERROR = "The array {data} must contain unique items only" VALIDATE_NONE_ERROR = "Received a null value for a non-nullable schema object" -VALIDATE_MISSING_RESPONSE_KEY_ERROR = 'The following property is missing in the response data: "{missing_key}"' -VALIDATE_EXCESS_RESPONSE_KEY_ERROR = ( - 'The following property was found in the response, but is missing from the schema definition: "{excess_key}"' +VALIDATE_MISSING_KEY_ERROR = 'The following property is missing in the {http_message} data: "{missing_key}"' +VALIDATE_EXCESS_KEY_ERROR = ( + 'The following property was found in the {http_message}, but is missing from the schema definition: "{excess_key}"' ) VALIDATE_WRITE_ONLY_RESPONSE_KEY_ERROR = ( 'The following property was found in the response, but is documented as being "writeOnly": "{write_only_key}"' diff --git a/openapi_tester/loaders.py b/openapi_tester/loaders.py index 47683898..f520aaf6 100644 --- a/openapi_tester/loaders.py +++ b/openapi_tester/loaders.py @@ -129,6 +129,7 @@ def set_schema(self, schema: dict) -> None: """ de_referenced_schema = self.de_reference_schema(schema) self.validate_schema(de_referenced_schema) + self.schema = self.normalize_schema_paths(de_referenced_schema) @cached_property @@ -245,6 +246,7 @@ class StaticSchemaLoader(BaseSchemaLoader): def __init__(self, path: str, field_key_map: dict[str, str] | None = None): super().__init__(field_key_map=field_key_map) + self.path = path if not isinstance(path, pathlib.PosixPath) else str(path) def load_schema(self) -> dict[str, Any]: diff --git a/openapi_tester/schema_tester.py b/openapi_tester/schema_tester.py index 431d7b17..943b8319 100644 --- a/openapi_tester/schema_tester.py +++ b/openapi_tester/schema_tester.py @@ -13,8 +13,8 @@ INIT_ERROR, UNDOCUMENTED_SCHEMA_SECTION_ERROR, VALIDATE_ANY_OF_ERROR, - VALIDATE_EXCESS_RESPONSE_KEY_ERROR, - VALIDATE_MISSING_RESPONSE_KEY_ERROR, + VALIDATE_EXCESS_KEY_ERROR, + VALIDATE_MISSING_KEY_ERROR, VALIDATE_NONE_ERROR, VALIDATE_ONE_OF_ERROR, VALIDATE_WRITE_ONLY_RESPONSE_KEY_ERROR, @@ -135,6 +135,7 @@ def get_response_schema_section(self, response: Response) -> dict[str, Any]: :return dict """ schema = self.loader.get_schema() + response_method = response.request["REQUEST_METHOD"].lower() # type: ignore parameterized_path, _ = self.loader.resolve_path( response.request["PATH_INFO"], method=response_method # type: ignore @@ -198,6 +199,61 @@ def get_response_schema_section(self, response: Response) -> dict[str, Any]: ) return {} + def get_request_body_schema_section(self, request: dict[str, Any]) -> dict[str, Any]: + """ + Fetches the request section of a schema. + + :param response: DRF Request Instance + :return dict + """ + schema = self.loader.get_schema() + request_method = request["REQUEST_METHOD"].lower() + + parameterized_path, _ = self.loader.resolve_path(request["PATH_INFO"], method=request_method) + paths_object = self.get_key_value(schema, "paths") + + route_object = self.get_key_value( + paths_object, + parameterized_path, + f"\n\nUndocumented route {parameterized_path}.\n\nDocumented routes: " + "\n\t• ".join(paths_object.keys()), + ) + + method_object = self.get_key_value( + route_object, + request_method, + ( + f"\n\nUndocumented method: {request_method}.\n\nDocumented methods: " + f"{[method.lower() for method in route_object.keys() if method.lower() != 'parameters']}." + ), + ) + + if all(key in request for key in ["CONTENT_LENGTH", "CONTENT_TYPE", "wsgi.input"]): + if request["CONTENT_TYPE"] != "application/json": + return {} + + request_body_object = self.get_key_value( + method_object, + "requestBody", + f"\n\nNo request body documented for method: {request_method}, path: {parameterized_path}", + ) + content_object = self.get_key_value( + request_body_object, + "content", + f"\n\nNo content documented for method: {request_method}, path: {parameterized_path}", + ) + json_object = self.get_key_value( + content_object, + r"^application\/.*json$", + ( + "\n\nNo `application/json` requests documented for method: " + f"{request_method}, path: {parameterized_path}" + ), + use_regex=True, + ) + return self.get_key_value(json_object, "schema") + + return {} + def handle_one_of(self, schema_section: dict, data: Any, reference: str, **kwargs: Any) -> None: matches = 0 passed_schema_section_formats = set() @@ -226,6 +282,9 @@ def handle_any_of(self, schema_section: dict, data: Any, reference: str, **kwarg continue raise DocumentationError(f"{VALIDATE_ANY_OF_ERROR}\n\nReference: {reference}.anyOf") + def is_openapi_schema(self) -> bool: + return self.loader.get_schema().get("openapi") is not None + @staticmethod def test_is_nullable(schema_item: dict) -> bool: """ @@ -338,6 +397,7 @@ def test_openapi_object( reference: str, case_tester: Callable[[str], None] | None = None, ignore_case: list[str] | None = None, + http_message: str = "response", ) -> None: """ 1. Validate that casing is correct for both response and schema @@ -358,16 +418,17 @@ def test_openapi_object( self.test_key_casing(key, case_tester, ignore_case) if key in required_keys and key not in response_keys: raise DocumentationError( - f"{VALIDATE_MISSING_RESPONSE_KEY_ERROR.format(missing_key=key)}\n\nReference: {reference}." - f"object:key:{key}\n\nHint: Remove the key from your" - " OpenAPI docs, or include it in your API response" + f"{VALIDATE_MISSING_KEY_ERROR.format(missing_key=key, http_message=http_message)}\n\nReference:" + f" {reference}.object:key:{key}\n\nHint: Remove the key from your OpenAPI docs, or include it in" + " your API response" ) for key in response_keys: self.test_key_casing(key, case_tester, ignore_case) if key not in properties and not additional_properties_allowed: raise DocumentationError( - f"{VALIDATE_EXCESS_RESPONSE_KEY_ERROR.format(excess_key=key)}\n\nReference: {reference}.object:key:" - f"{key}\n\nHint: Remove the key from your API response, or include it in your OpenAPI docs" + f"{VALIDATE_EXCESS_KEY_ERROR.format(excess_key=key, http_message=http_message)}\n\nReference:" + f" {reference}.object:key:{key}\n\nHint: Remove the key from your API response, or include it in" + " your OpenAPI docs" ) if key in write_only_properties: raise DocumentationError( @@ -403,6 +464,37 @@ def test_openapi_array(self, schema_section: dict[str, Any], data: dict, referen **kwargs, ) + def validate_request( + self, + response: Response, + case_tester: Callable[[str], None] | None = None, + ignore_case: list[str] | None = None, + validators: list[Callable[[dict[str, Any], Any], str | None]] | None = None, + ) -> None: + """ + Verifies that an OpenAPI schema definition matches an API request body. + + :param request: The HTTP request + :param case_tester: Optional Callable that checks a string's casing + :param ignore_case: Optional list of keys to ignore in case testing + :param validators: Optional list of validator functions + :param **kwargs: Request keyword arguments + :raises: ``openapi_tester.exceptions.DocumentationError`` for inconsistencies in the API response and schema. + ``openapi_tester.exceptions.CaseError`` for case errors. + """ + if self.is_openapi_schema(): + # TODO: Implement for other schema types + request_body_schema = self.get_request_body_schema_section(response.request) # type: ignore + if request_body_schema: + self.test_schema_section( + schema_section=request_body_schema, + data=response.renderer_context["request"].data, # type: ignore + case_tester=case_tester or self.case_tester, + ignore_case=ignore_case, + validators=validators, + http_message="request", + ) + def validate_response( self, response: Response, @@ -427,4 +519,5 @@ def validate_response( case_tester=case_tester or self.case_tester, ignore_case=ignore_case, validators=validators, + http_message="response", ) diff --git a/test_project/api/serializers.py b/test_project/api/serializers.py index 3b7a8ba7..8dcac254 100644 --- a/test_project/api/serializers.py +++ b/test_project/api/serializers.py @@ -8,6 +8,11 @@ class Meta: vehicle_type = serializers.CharField(max_length=10) +class PetsSerializer(serializers.Serializer): + name = serializers.CharField(max_length=254) + tag = serializers.CharField(max_length=254, required=False) + + class ItemSerializer(serializers.Serializer): item_type = serializers.CharField(max_length=10) diff --git a/test_project/api/views/pets.py b/test_project/api/views/pets.py index e31f6e44..2b3bab27 100644 --- a/test_project/api/views/pets.py +++ b/test_project/api/views/pets.py @@ -6,6 +6,8 @@ from rest_framework.status import HTTP_200_OK from rest_framework.views import APIView +from test_project.api.serializers import PetsSerializer + if TYPE_CHECKING: from rest_framework.request import Request @@ -14,3 +16,8 @@ class Pet(APIView): def get(self, request: Request, petId: int) -> Response: pet = {"name": "doggie", "category": {"id": 1, "name": "Dogs"}, "photoUrls": [], "status": "available"} return Response(pet, HTTP_200_OK) + + def post(self, request) -> Response: + serializer = PetsSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + return Response({"id": 1, "name": request.data["name"]}, 201) diff --git a/test_project/urls.py b/test_project/urls.py index 1d5bdc3f..24b69bfc 100644 --- a/test_project/urls.py +++ b/test_project/urls.py @@ -35,6 +35,7 @@ path("api//snake-case/", SnakeCasedResponse.as_view()), # ^trailing slash is here on purpose path("api//router_generated/", include(router.urls)), + path("api/pets", Pet.as_view(), name="get-pets"), re_path(r"api/pet/(?P\d+)", Pet.as_view(), name="get-pet"), ] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..bb8c32ef --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +from copy import deepcopy +from typing import TYPE_CHECKING, Callable +from unittest.mock import MagicMock + +import pytest +from rest_framework.response import Response + +from tests.schema_converter import SchemaToPythonConverter +from tests.utils import TEST_ROOT + +if TYPE_CHECKING: + from pathlib import Path + + +@pytest.fixture() +def pets_api_schema() -> Path: + return TEST_ROOT / "schemas" / "openapi_v3_reference_schema.yaml" + + +@pytest.fixture() +def pets_post_request(): + request_body = MagicMock() + request_body.read.return_value = b'{"name": "doggie", "tag": "dog"}' + return { + "PATH_INFO": "/api/pets", + "REQUEST_METHOD": "POST", + "SERVER_PORT": "80", + "wsgi.url_scheme": "http", + "CONTENT_LENGTH": "70", + "CONTENT_TYPE": "application/json", + "wsgi.input": request_body, + "QUERY_STRING": "", + } + + +@pytest.fixture() +def invalid_pets_post_request(): + request_body = MagicMock() + request_body.read.return_value = b'{"surname": "doggie", "species": "dog"}' + return { + "PATH_INFO": "/api/pets", + "REQUEST_METHOD": "POST", + "SERVER_PORT": "80", + "wsgi.url_scheme": "http", + "CONTENT_LENGTH": "70", + "CONTENT_TYPE": "application/json", + "wsgi.input": request_body, + "QUERY_STRING": "", + } + + +@pytest.fixture() +def response_factory() -> Callable: + def response( + schema: dict | None, + url_fragment: str, + method: str, + status_code: int | str = 200, + response_body: dict | None = None, + ) -> Response: + converted_schema = None + if schema: + converted_schema = SchemaToPythonConverter(deepcopy(schema)).result + response = Response(status=int(status_code), data=converted_schema) + response.request = {"REQUEST_METHOD": method, "PATH_INFO": url_fragment} # type: ignore + if schema: + response.json = lambda: converted_schema # type: ignore + elif response_body: + response.request["CONTENT_LENGTH"] = len(response_body) # type: ignore + response.request["CONTENT_TYPE"] = "application/json" # type: ignore + response.request["wsgi.input"] = response_body # type: ignore + response.renderer_context = {"request": MagicMock(data=response_body)} # type: ignore + return response + + return response diff --git a/tests/schemas/openapi_v3_reference_schema.yaml b/tests/schemas/openapi_v3_reference_schema.yaml index d13d9b0d..460f4dfc 100644 --- a/tests/schemas/openapi_v3_reference_schema.yaml +++ b/tests/schemas/openapi_v3_reference_schema.yaml @@ -12,9 +12,9 @@ info: name: Apache 2.0 url: https://www.apache.org/licenses/LICENSE-2.0.html servers: - - url: http://petstore.swagger.io/api + - url: http://petstore.swagger.io paths: - /pets: + /api/pets: get: description: | Returns all pets from the system that the user has access to @@ -65,19 +65,31 @@ paths: schema: $ref: '#/components/schemas/NewPet' responses: - '200': + '201': description: pet response content: application/json: schema: $ref: '#/components/schemas/Pet' + '400': + description: Bad Request + content: + application/json: + schema: + type: object + properties: + error: + type: string + message: + type: string + additionalProperties: true default: description: unexpected error content: application/json: schema: $ref: '#/components/schemas/Error' - /pets/{id}: + /api/pets/{id}: get: description: Returns a user based on a single ID, if the user does not have access to the pet operationId: find pet by id diff --git a/tests/test_clients.py b/tests/test_clients.py index 88d92ee8..c1cf52cb 100644 --- a/tests/test_clients.py +++ b/tests/test_clients.py @@ -1,14 +1,18 @@ import functools import json +from typing import TYPE_CHECKING import pytest from django.test.testcases import SimpleTestCase from rest_framework import status from openapi_tester.clients import OpenAPIClient -from openapi_tester.exceptions import UndocumentedSchemaSectionError +from openapi_tester.exceptions import DocumentationError, UndocumentedSchemaSectionError from openapi_tester.schema_tester import SchemaTester +if TYPE_CHECKING: + from pathlib import Path + @pytest.fixture() def openapi_client(settings) -> OpenAPIClient: @@ -52,6 +56,55 @@ def test_request(openapi_client, generic_kwargs, expected_status_code): assert response.status_code == expected_status_code +@pytest.mark.parametrize( + ("generic_kwargs", "expected_status_code"), + [ + ( + { + "method": "POST", + "path": "/api/pets", + "data": json.dumps({"name": "doggie"}), + "content_type": "application/json", + }, + status.HTTP_201_CREATED, + ), + ( + { + "method": "POST", + "path": "/api/pets", + "data": json.dumps({"tag": "doggie"}), + "content_type": "application/json", + }, + status.HTTP_400_BAD_REQUEST, + ), + ], +) +def test_request_body(generic_kwargs, expected_status_code, pets_api_schema: "Path"): + """Ensure ``SchemaTester`` doesn't raise exception when request valid. + Additionally, request validation should be performed only in successful responses.""" + schema_tester = SchemaTester(schema_file_path=str(pets_api_schema)) + openapi_client = OpenAPIClient(schema_tester=schema_tester) + response = openapi_client.generic(**generic_kwargs) + + assert response.status_code == expected_status_code + + +def test_request_body_extra_non_documented_field(pets_api_schema: "Path"): + """Ensure ``SchemaTester`` raises exception when request is successfull but an + extra field non-documented was sent.""" + schema_tester = SchemaTester(schema_file_path=str(pets_api_schema)) + openapi_client = OpenAPIClient(schema_tester=schema_tester) + kwargs = { + "method": "POST", + "path": "/api/pets", + "data": json.dumps({"name": "doggie", "age": 1}), + "content_type": "application/json", + } + + with pytest.raises(DocumentationError): + openapi_client.generic(**kwargs) # type: ignore + + def test_request_on_empty_list(openapi_client): """Ensure ``SchemaTester`` doesn't raise exception when response is empty list.""" response = openapi_client.generic( diff --git a/tests/test_errors.py b/tests/test_errors.py index e27a70f4..3dbe5d60 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -62,29 +62,29 @@ def test_validate_unique_items_error(self): def test_validate_minimum_error(self): message = validate_minimum({"minimum": 2}, 0) - assert message == "The response value 0 is lower than the specified minimum of 2" + assert message == "The value 0 is lower than the specified minimum of 2" def test_validate_exclusive_minimum_error(self): message = validate_minimum({"minimum": 2, "exclusiveMinimum": True}, 2) - assert message == "The response value 2 is lower than the specified minimum of 3" + assert message == "The value 2 is lower than the specified minimum of 3" message = validate_minimum({"minimum": 2, "exclusiveMinimum": False}, 2) assert message is None def test_validate_maximum_error(self): message = validate_maximum({"maximum": 2}, 3) - assert message == "The response value 3 exceeds the maximum allowed value of 2" + assert message == "The value 3 exceeds the maximum allowed value of 2" def test_validate_exclusive_maximum_error(self): message = validate_maximum({"maximum": 2, "exclusiveMaximum": True}, 2) - assert message == "The response value 2 exceeds the maximum allowed value of 1" + assert message == "The value 2 exceeds the maximum allowed value of 1" message = validate_maximum({"maximum": 2, "exclusiveMaximum": False}, 2) assert message is None def test_validate_multiple_of_error(self): message = validate_multiple_of({"multipleOf": 2}, 3) - assert message == "The response value 3 should be a multiple of 2" + assert message == "The value 3 should be a multiple of 2" def test_validate_pattern_error(self): message = validate_pattern({"pattern": "^[a-z]$"}, "3") diff --git a/tests/test_schema_tester.py b/tests/test_schema_tester.py index df8ab74f..2c60d223 100644 --- a/tests/test_schema_tester.py +++ b/tests/test_schema_tester.py @@ -20,8 +20,8 @@ from openapi_tester.constants import ( INIT_ERROR, OPENAPI_PYTHON_MAPPING, - VALIDATE_EXCESS_RESPONSE_KEY_ERROR, - VALIDATE_MISSING_RESPONSE_KEY_ERROR, + VALIDATE_EXCESS_KEY_ERROR, + VALIDATE_MISSING_KEY_ERROR, VALIDATE_NONE_ERROR, VALIDATE_ONE_OF_ERROR, VALIDATE_WRITE_ONLY_RESPONSE_KEY_ERROR, @@ -30,9 +30,10 @@ from openapi_tester.loaders import UrlStaticSchemaLoader from test_project.models import Names from tests import example_object, example_schema_types -from tests.utils import TEST_ROOT, iterate_schema, mock_schema, response_factory +from tests.utils import TEST_ROOT, iterate_schema, mock_schema if TYPE_CHECKING: + from pathlib import Path from typing import Any tester = SchemaTester() @@ -154,7 +155,7 @@ def test_validate_response_failure_scenario_with_predefined_data(client): tester.validate_response(response) -def test_validate_response_failure_scenario_undocumented_path(monkeypatch): +def test_validate_response_failure_scenario_undocumented_path(monkeypatch, response_factory): schema = deepcopy(tester.loader.get_schema()) schema_section = schema["paths"][parameterized_path][method]["responses"][status]["content"]["application/json"][ "schema" @@ -169,7 +170,7 @@ def test_validate_response_failure_scenario_undocumented_path(monkeypatch): tester.validate_response(response) -def test_validate_response_failure_scenario_undocumented_method(monkeypatch): +def test_validate_response_failure_scenario_undocumented_method(monkeypatch, response_factory): schema = deepcopy(tester.loader.get_schema()) schema_section = schema["paths"][parameterized_path][method]["responses"][status]["content"]["application/json"][ "schema" @@ -184,7 +185,7 @@ def test_validate_response_failure_scenario_undocumented_method(monkeypatch): tester.validate_response(response) -def test_validate_response_failure_scenario_undocumented_status_code(monkeypatch): +def test_validate_response_failure_scenario_undocumented_status_code(monkeypatch, response_factory): schema = deepcopy(tester.loader.get_schema()) schema_section = schema["paths"][parameterized_path][method]["responses"][status]["content"]["application/json"][ "schema" @@ -214,6 +215,83 @@ def test_validate_response_failure_scenario_undocumented_content(client, monkeyp tester.validate_response(response) +def test_validate_request(response_factory, pets_api_schema: Path, pets_post_request: dict[str, Any]): + schema_tester = SchemaTester(schema_file_path=str(pets_api_schema)) + response = response_factory( + schema=None, + url_fragment="/api/pets", + method="POST", + status_code=201, + response_body={"name": "doggie", "tag": "dog"}, + ) + schema_tester.validate_request(response) + + +def test_validate_request_invalid(response_factory, pets_api_schema: Path, pets_post_request: dict[str, Any]): + schema_tester = SchemaTester(schema_file_path=str(pets_api_schema)) + response = response_factory( + schema=None, + url_fragment="/api/pets", + method="POST", + status_code=201, + response_body={"tag": "dog"}, + ) + + with pytest.raises(DocumentationError): + schema_tester.validate_request(response) + + +def test_validate_request_no_application_json( + response_factory, pets_api_schema: Path, pets_post_request: dict[str, Any] +): + schema_tester = SchemaTester(schema_file_path=str(pets_api_schema)) + response = response_factory( + schema=None, + url_fragment="/api/pets", + method="POST", + status_code=201, + response_body={"tag": "dog"}, + ) + response.request["CONTENT_TYPE"] = "application/xml" + schema_tester.validate_request(response) + + +def test_is_openapi_schema(pets_api_schema: Path): + schema_tester = SchemaTester(schema_file_path=str(pets_api_schema)) + assert schema_tester.is_openapi_schema() is True + + +def test_is_openapi_schema_false(): + schema_tester = SchemaTester() + assert schema_tester.is_openapi_schema() is False + + +def test_get_request_body_schema_section(pets_post_request: dict[str, Any], pets_api_schema: Path): + schema_tester = SchemaTester(schema_file_path=str(pets_api_schema)) + schema_section = schema_tester.get_request_body_schema_section(pets_post_request) + assert schema_section == { + "type": "object", + "required": ["name"], + "properties": {"name": {"type": "string"}, "tag": {"type": "string"}}, + } + + +def test_get_request_body_schema_section_content_type_no_application_json( + pets_post_request: dict[str, Any], pets_api_schema: Path +): + schema_tester = SchemaTester(schema_file_path=str(pets_api_schema)) + pets_post_request["CONTENT_TYPE"] = "application/xml" + schema_section = schema_tester.get_request_body_schema_section(pets_post_request) + assert schema_section == {} + + +def test_get_request_body_schema_section_no_content_request(pets_post_request: dict[str, Any], pets_api_schema: Path): + schema_tester = SchemaTester(schema_file_path=str(pets_api_schema)) + del pets_post_request["wsgi.input"] + schema_section = schema_tester.get_request_body_schema_section(pets_post_request) + assert schema_section == {} + + def test_validate_response_global_case_tester(client): response = client.get(de_parameterized_path) with pytest.raises(CaseError, match="is not properly PascalCased"): @@ -221,7 +299,7 @@ def test_validate_response_global_case_tester(client): @pytest.mark.parametrize("empty_schema", [None, {}]) -def test_validate_response_empty_content(empty_schema, client, monkeypatch): +def test_validate_response_empty_content(empty_schema, client, monkeypatch, response_factory): schema = deepcopy(tester.loader.get_schema()) del schema["paths"][parameterized_path][method]["responses"][status]["content"] monkeypatch.setattr(tester.loader, "get_schema", mock_schema(schema)) @@ -394,7 +472,9 @@ def test_one_of_validation(): def test_missing_keys_validation(): # If a required key is missing, we should raise an error required_key = {"type": "object", "properties": {"value": {"type": "integer"}}, "required": ["value"]} - with pytest.raises(DocumentationError, match=VALIDATE_MISSING_RESPONSE_KEY_ERROR.format(missing_key="value")): + with pytest.raises( + DocumentationError, match=VALIDATE_MISSING_KEY_ERROR.format(http_message="response", missing_key="value") + ): tester.test_schema_section(required_key, {}) # If not required, it should pass @@ -406,7 +486,7 @@ def test_excess_keys_validation(): schema = {"type": "object", "properties": {}} with pytest.raises( DocumentationError, - match=VALIDATE_EXCESS_RESPONSE_KEY_ERROR.format(excess_key="value"), + match=VALIDATE_EXCESS_KEY_ERROR.format(http_message="response", excess_key="value"), ): tester.test_schema_section(schema, example_object) diff --git a/tests/test_validators.py b/tests/test_validators.py index 7734a477..bc668dc6 100644 --- a/tests/test_validators.py +++ b/tests/test_validators.py @@ -9,7 +9,7 @@ from openapi_tester import SchemaTester from openapi_tester.constants import ( OPENAPI_PYTHON_MAPPING, - VALIDATE_EXCESS_RESPONSE_KEY_ERROR, + VALIDATE_EXCESS_KEY_ERROR, VALIDATE_MAX_ARRAY_LENGTH_ERROR, VALIDATE_MAX_LENGTH_ERROR, VALIDATE_MAXIMUM_ERROR, @@ -155,7 +155,9 @@ def test_additional_properties_specified_as_empty_object_allowed(): def test_additional_properties_not_allowed_by_default(): schema = {"type": "object", "properties": {"oneKey": {"type": "string"}}} - with pytest.raises(DocumentationError, match=VALIDATE_EXCESS_RESPONSE_KEY_ERROR[:90]): + with pytest.raises( + DocumentationError, match=VALIDATE_EXCESS_KEY_ERROR.format(http_message="response", excess_key="twoKey") + ): tester.test_schema_section(schema, {"oneKey": "test", "twoKey": "test2"})