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..e287d737 100644 --- a/openapi_tester/schema_tester.py +++ b/openapi_tester/schema_tester.py @@ -2,6 +2,7 @@ from __future__ import annotations import re +from dataclasses import dataclass from itertools import chain from typing import TYPE_CHECKING, Any, Callable, cast @@ -13,8 +14,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, @@ -50,6 +51,17 @@ from rest_framework.response import Response +@dataclass +class OpenAPITestConfig: + """Configuration dataclass for schema section test.""" + + case_tester: Callable[[str], None] | None = None + ignore_case: list[str] | None = None + validators: list[Callable[[dict[str, Any], Any], str | None]] | None = None + reference: str = "init" + http_message: str = "response" + + class SchemaTester: """Schema Tester: this is the base class of the library.""" @@ -135,6 +147,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,12 +211,72 @@ def get_response_schema_section(self, response: Response) -> dict[str, Any]: ) return {} - def handle_one_of(self, schema_section: dict, data: Any, reference: str, **kwargs: Any) -> None: + 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, test_config: OpenAPITestConfig) -> None: matches = 0 passed_schema_section_formats = set() for option in schema_section["oneOf"]: try: - self.test_schema_section(schema_section=option, data=data, reference=f"{reference}.oneOf", **kwargs) + test_config.reference = f"{test_config.reference}.oneOf" + self.test_schema_section( + schema_section=option, + data=data, + test_config=test_config, + ) matches += 1 passed_schema_section_formats.add(option.get("format")) except DocumentationError: @@ -216,16 +289,24 @@ def handle_one_of(self, schema_section: dict, data: Any, reference: str, **kwarg if matches != 1: raise DocumentationError(f"{VALIDATE_ONE_OF_ERROR.format(matches=matches)}\n\nReference: {reference}.oneOf") - def handle_any_of(self, schema_section: dict, data: Any, reference: str, **kwargs: Any) -> None: + def handle_any_of(self, schema_section: dict, data: Any, reference: str, test_config: OpenAPITestConfig) -> None: any_of: list[dict[str, Any]] = schema_section.get("anyOf", []) for schema in chain(any_of, lazy_combinations(any_of)): + test_config.reference = f"{test_config.reference}.anyOf" try: - self.test_schema_section(schema_section=schema, data=data, reference=f"{reference}.anyOf", **kwargs) + self.test_schema_section( + schema_section=schema, + data=data, + test_config=test_config, + ) return except DocumentationError: 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: """ @@ -273,28 +354,31 @@ def test_schema_section( self, schema_section: dict, data: Any, - reference: str = "init", - validators: list[Callable[[dict, dict], str | None]] | None = None, - **kwargs: Any, + test_config: OpenAPITestConfig | None = None, ) -> None: """ This method orchestrates the testing of a schema section """ + test_config = test_config or OpenAPITestConfig() if data is None: if self.test_is_nullable(schema_section): # If data is None and nullable, we return early return raise DocumentationError( f"{VALIDATE_NONE_ERROR}\n\n" - f"Reference: {reference}\n\n" + f"Reference: {test_config.reference}\n\n" "Hint: Return a valid type, or document the value as nullable" ) schema_section = normalize_schema_section(schema_section) if "oneOf" in schema_section: - self.handle_one_of(schema_section=schema_section, data=data, reference=reference, **kwargs) + self.handle_one_of( + schema_section=schema_section, data=data, reference=test_config.reference, test_config=test_config + ) return if "anyOf" in schema_section: - self.handle_any_of(schema_section=schema_section, data=data, reference=reference, **kwargs) + self.handle_any_of( + schema_section=schema_section, data=data, reference=test_config.reference, test_config=test_config + ) return schema_section_type = self.get_schema_type(schema_section) @@ -318,97 +402,124 @@ def test_schema_section( validate_min_properties, validate_enum, *self.validators, - *(validators or []), + *(test_config.validators or []), ], ) for validator in combined_validators: error = validator(schema_section, data) if error: - raise DocumentationError(f"\n\n{error}\n\nReference: {reference}") + raise DocumentationError(f"\n\n{error}\n\nReference: {test_config.reference}") if schema_section_type == "object": - self.test_openapi_object(schema_section=schema_section, data=data, reference=reference, **kwargs) + self.test_openapi_object(schema_section=schema_section, data=data, test_config=test_config) elif schema_section_type == "array": - self.test_openapi_array(schema_section=schema_section, data=data, reference=reference, **kwargs) + self.test_openapi_array(schema_section=schema_section, data=data, test_config=test_config) def test_openapi_object( self, schema_section: dict, data: dict, - reference: str, - case_tester: Callable[[str], None] | None = None, - ignore_case: list[str] | None = None, + test_config: OpenAPITestConfig, ) -> None: """ - 1. Validate that casing is correct for both response and schema - 2. Check if any required key is missing from the response - 3. Check if any response key is not in the schema + 1. Validate that casing is correct for both request/response and schema + 2. Check if any required key is missing from the request/response + 3. Check if any request/response key is not in the schema 4. Validate sub-schema/nested data """ - properties = schema_section.get("properties", {}) write_only_properties = [key for key in properties.keys() if properties[key].get("writeOnly")] required_keys = [key for key in schema_section.get("required", []) if key not in write_only_properties] - response_keys = data.keys() + request_response_keys = data.keys() additional_properties: bool | dict | None = schema_section.get("additionalProperties") additional_properties_allowed = additional_properties is not None if additional_properties_allowed and not isinstance(additional_properties, (bool, dict)): raise OpenAPISchemaError("Invalid additionalProperties type") for key in properties.keys(): - self.test_key_casing(key, case_tester, ignore_case) - if key in required_keys and key not in response_keys: + self.test_key_casing(key, test_config.case_tester, test_config.ignore_case) + if key in required_keys and key not in request_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=test_config.http_message)}" + "\n\nReference:" + f" {test_config.reference}.object:key:{key}\n\nHint: Remove the key from your OpenAPI docs, or" + f" include it in your API {test_config.http_message}" ) - for key in response_keys: - self.test_key_casing(key, case_tester, ignore_case) + for key in request_response_keys: + self.test_key_casing(key, test_config.case_tester, test_config.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=test_config.http_message)}" + "\n\nReference:" + f" {test_config.reference}.object:key:{key}\n\nHint: Remove the key from your API" + f" {test_config.http_message}, or include it in your OpenAPI docs" ) if key in write_only_properties: raise DocumentationError( - f"{VALIDATE_WRITE_ONLY_RESPONSE_KEY_ERROR.format(write_only_key=key)}\n\nReference: {reference}" - f".object:key:{key}\n\nHint: Remove the key from your API response, or remove the " - '"WriteOnly" restriction' + f"{VALIDATE_WRITE_ONLY_RESPONSE_KEY_ERROR.format(write_only_key=key)}\n\nReference:" + f" {test_config.reference}.object:key:{key}\n\nHint:" + f" Remove the key from your API {test_config.http_message}, or" + ' remove the "WriteOnly" restriction' ) for key, value in data.items(): if key in properties: + test_config.reference = f"{test_config.reference}.object:key:{key}" self.test_schema_section( schema_section=properties[key], data=value, - reference=f"{reference}.object:key:{key}", - case_tester=case_tester, - ignore_case=ignore_case, + test_config=test_config, ) elif isinstance(additional_properties, dict): + test_config.reference = f"{test_config.reference}.object:key:{key}" self.test_schema_section( schema_section=additional_properties, data=value, - reference=f"{reference}.object:key:{key}", - case_tester=case_tester, - ignore_case=ignore_case, + test_config=test_config, ) - def test_openapi_array(self, schema_section: dict[str, Any], data: dict, reference: str, **kwargs: Any) -> None: + def test_openapi_array(self, schema_section: dict[str, Any], data: dict, test_config: OpenAPITestConfig) -> None: for datum in data: + test_config.reference = f"{test_config.reference}.array.item" self.test_schema_section( # the items keyword is required in arrays schema_section=schema_section["items"], data=datum, - reference=f"{reference}.array.item", - **kwargs, + test_config=test_config, ) + def validate_request( + self, + response: Response, + test_config: OpenAPITestConfig | 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 + if test_config: + test_config.http_message = "request" + else: + test_config = OpenAPITestConfig(http_message="request") + 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 + test_config=test_config, + ) + def validate_response( 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, + test_config: OpenAPITestConfig | None = None, ) -> None: """ Verifies that an OpenAPI schema definition matches an API response. @@ -420,11 +531,13 @@ def validate_response( :raises: ``openapi_tester.exceptions.DocumentationError`` for inconsistencies in the API response and schema. ``openapi_tester.exceptions.CaseError`` for case errors. """ + if test_config: + test_config.http_message = "response" + else: + test_config = OpenAPITestConfig(http_message="response") response_schema = self.get_response_schema_section(response) self.test_schema_section( schema_section=response_schema, data=response.json() if response.data is not None else {}, # type: ignore - case_tester=case_tester or self.case_tester, - ignore_case=ignore_case, - validators=validators, + test_config=test_config, ) 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..b19b88f1 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -2,6 +2,7 @@ from openapi_tester import SchemaTester from openapi_tester.exceptions import CaseError, DocumentationError +from openapi_tester.schema_tester import OpenAPITestConfig from openapi_tester.validators import ( validate_enum, validate_format, @@ -62,29 +63,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") @@ -159,7 +160,9 @@ def test_missing_response_key_error(self): tester = SchemaTester() with pytest.raises(DocumentationError, match=expected_error_message): tester.test_openapi_object( - {"required": ["one"], "properties": {"one": {"type": "int"}}}, {"two": 2}, reference="init" + {"required": ["one"], "properties": {"one": {"type": "int"}}}, + {"two": 2}, + OpenAPITestConfig(reference="init"), ) def test_missing_schema_key_error(self): @@ -171,7 +174,9 @@ def test_missing_schema_key_error(self): tester = SchemaTester() with pytest.raises(DocumentationError, match=expected_error_message): tester.test_openapi_object( - {"required": ["one"], "properties": {"one": {"type": "int"}}}, {"one": 1, "two": 2}, reference="init" + {"required": ["one"], "properties": {"one": {"type": "int"}}}, + {"one": 1, "two": 2}, + OpenAPITestConfig(reference="init"), ) def test_key_in_write_only_properties_error(self): @@ -185,7 +190,7 @@ def test_key_in_write_only_properties_error(self): tester.test_openapi_object( {"properties": {"one": {"type": "int", "writeOnly": True}}}, {"one": 1}, - reference="init", + OpenAPITestConfig(reference="init"), ) @@ -197,7 +202,7 @@ def test_null_error(): ) tester = SchemaTester() with pytest.raises(DocumentationError, match=expected_error_message): - tester.test_schema_section({"type": "object"}, None, reference="init") + tester.test_schema_section({"type": "object"}, None, OpenAPITestConfig(reference="init")) def test_any_of_error(): @@ -207,7 +212,7 @@ def test_any_of_error(): ) tester = SchemaTester() with pytest.raises(DocumentationError, match=expected_error_message): - tester.test_schema_section({"anyOf": []}, {}, reference="init") + tester.test_schema_section({"anyOf": []}, {}, OpenAPITestConfig(reference="init")) def test_one_of_error(): @@ -216,4 +221,4 @@ def test_one_of_error(): ) tester = SchemaTester() with pytest.raises(DocumentationError, match=expected_error_message): - tester.test_schema_section({"oneOf": []}, {}, reference="init") + tester.test_schema_section({"oneOf": []}, {}, OpenAPITestConfig(reference="init")) diff --git a/tests/test_schema_tester.py b/tests/test_schema_tester.py index df8ab74f..9a75a4b5 100644 --- a/tests/test_schema_tester.py +++ b/tests/test_schema_tester.py @@ -20,19 +20,21 @@ 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, ) from openapi_tester.exceptions import CaseError, DocumentationError, UndocumentedSchemaSectionError from openapi_tester.loaders import UrlStaticSchemaLoader +from openapi_tester.schema_tester import OpenAPITestConfig 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 +156,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 +171,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 +186,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 +216,95 @@ 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_with_config(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, OpenAPITestConfig(case_tester=is_pascal_case, ignore_case=["name", "tag"])) + + +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 +312,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)) @@ -239,13 +330,16 @@ def test_validate_response_global_ignored_case(client): def test_validate_response_passed_in_case_tester(client): response = client.get(de_parameterized_path) with pytest.raises(CaseError, match="The response key `name` is not properly PascalCased. Expected value: Name"): - tester.validate_response(response=response, case_tester=is_pascal_case) + tester.validate_response(response=response, test_config=OpenAPITestConfig(case_tester=is_pascal_case)) def test_validate_response_passed_in_ignored_case(client): response = client.get(de_parameterized_path) tester.validate_response( - response=response, case_tester=is_pascal_case, ignore_case=["name", "color", "height", "width", "length"] + response=response, + test_config=OpenAPITestConfig( + case_tester=is_pascal_case, ignore_case=["name", "color", "height", "width", "length"] + ), ) @@ -394,7 +488,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 +502,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) @@ -446,10 +542,17 @@ def uuid_1_validator(schema_section: dict, data: Any) -> str | None: # pragma: ): tester_with_custom_validator.test_schema_section(uid4_schema, uid1) - assert tester_with_custom_validator.test_schema_section(uid1_schema, uid1, validators=[uuid_1_validator]) is None + assert ( + tester_with_custom_validator.test_schema_section( + uid1_schema, uid1, test_config=OpenAPITestConfig(validators=[uuid_1_validator]) + ) + is None + ) with pytest.raises( DocumentationError, match=f"Expected uuid1, but received {uid4}", ): - tester_with_custom_validator.test_schema_section(uid1_schema, uid4, validators=[uuid_1_validator]) + tester_with_custom_validator.test_schema_section( + uid1_schema, uid4, test_config=OpenAPITestConfig(validators=[uuid_1_validator]) + ) 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"})