Merge pull request #33 from maticardenas/31-add-django-ninja-test-client

feat: adding support for Django Ninja test client

README.md

@@ -256,6 +256,30 @@ class MySimpleTestCase(SimpleTestCase):
 This will ensure you all newly implemented views will be validated against
 the OpenAPI schema.
 
+
+### Django Ninja Test Client
+
+In case you are using `Django Ninja` and its corresponding [test client](https://github.com/vitalik/django-ninja/blob/master/ninja/testing/client.py#L159), you can use the `OpenAPINinjaClient`, which extends from it, in the same way as the `OpenAPIClient`:
+
+```python
+schema_tester = SchemaTester()
+client = OpenAPINinjaClient(
+        router_or_app=router,
+        schema_tester=schema_tester,
+    )
+response = client.get('/api/v1/tests/123/')
+```
+
+Given that the Django Ninja test client works separately from the django url resolver, you can pass the `path_prefix` argument to the `OpenAPINinjaClient` to specify the prefix of the path that should be used to look into the OpenAPI schema.
+
+```python
+client = OpenAPINinjaClient(
+        router_or_app=router,
+        path_prefix='/api/v1',
+        schema_tester=schema_tester,
+    )
+```
+
 ## Known Issues
 
 * We are using [prance](https://github.com/jfinkhaeuser/prance) as a schema resolver, and it has some issues with the

openapi_tester/clients.py

@@ -2,12 +2,17 @@
 
 from __future__ import annotations
 
-import json
+import http
 from typing import TYPE_CHECKING
 
+# pylint: disable=import-error
+from ninja import NinjaAPI, Router
+from ninja.testing import TestClient
 from rest_framework.test import APIClient
 
+from .response_handler_factory import ResponseHandlerFactory
 from .schema_tester import SchemaTester
+from .utils import serialize_json
 
 if TYPE_CHECKING:
     from rest_framework.response import Response
@@ -26,132 +31,125 @@ def __init__(
         super().__init__(*args, **kwargs)
         self.schema_tester = schema_tester or self._schema_tester_factory()
 
-    def request(self, **kwargs) -> Response:  # type: ignore[override]
+    def request(self, *args, **kwargs) -> Response:  # type: ignore[override]
         """Validate fetched response against given OpenAPI schema."""
         response = super().request(**kwargs)
+        response_handler = ResponseHandlerFactory.create(
+            *args, response=response, **kwargs
+        )
         if self._is_successful_response(response):
-            self.schema_tester.validate_request(response)
-        self.schema_tester.validate_response(response)
+            self.schema_tester.validate_request(response_handler=response_handler)
+        self.schema_tester.validate_response(response_handler=response_handler)
         return response
 
     # pylint: disable=W0622
+    @serialize_json
     def post(
         self,
-        path,
-        data=None,
-        format=None,
+        *args,
         content_type="application/json",
-        follow=False,
-        **extra,
+        **kwargs,
     ):
-        if data and content_type == "application/json":
-            data = self._serialize(data)
         return super().post(
-            path,
-            data=data,
-            format=format,
+            *args,
             content_type=content_type,
-            follow=follow,
-            **extra,
+            **kwargs,
         )
 
     # pylint: disable=W0622
+    @serialize_json
     def put(
         self,
-        path,
-        data=None,
-        format=None,
+        *args,
         content_type="application/json",
-        follow=False,
-        **extra,
+        **kwargs,
     ):
-        if data and content_type == "application/json":
-            data = self._serialize(data)
         return super().put(
-            path,
-            data=data,
-            format=format,
+            *args,
             content_type=content_type,
-            follow=follow,
-            **extra,
+            **kwargs,
         )
 
     # pylint: disable=W0622
-    def patch(
-        self,
-        path,
-        data=None,
-        format=None,
-        content_type="application/json",
-        follow=False,
-        **extra,
-    ):
-        if data and content_type == "application/json":
-            data = self._serialize(data)
+    @serialize_json
+    def patch(self, *args, content_type="application/json", **kwargs):
         return super().patch(
-            path,
-            data=data,
-            format=format,
+            *args,
             content_type=content_type,
-            follow=follow,
-            **extra,
+            **kwargs,
         )
 
     # pylint: disable=W0622
+    @serialize_json
     def delete(
         self,
-        path,
-        data=None,
-        format=None,
+        *args,
         content_type="application/json",
-        follow=False,
-        **extra,
+        **kwargs,
     ):
-        if data and content_type == "application/json":
-            data = self._serialize(data)
         return super().delete(
-            path,
-            data=data,
-            format=format,
+            *args,
             content_type=content_type,
-            follow=follow,
-            **extra,
+            **kwargs,
         )
 
     # pylint: disable=W0622
+    @serialize_json
     def options(
         self,
-        path,
-        data=None,
-        format=None,
+        *args,
         content_type="application/json",
-        follow=False,
-        **extra,
+        **kwargs,
     ):
-        if data and content_type == "application/json":
-            data = self._serialize(data)
         return super().options(
-            path,
-            data=data,
-            format=format,
+            *args,
             content_type=content_type,
-            follow=follow,
-            **extra,
+            **kwargs,
         )
 
     @staticmethod
     def _is_successful_response(response: Response) -> bool:
-        return response.status_code < 400
+        return response.status_code < http.HTTPStatus.BAD_REQUEST
 
     @staticmethod
     def _schema_tester_factory() -> SchemaTester:
         """Factory of default ``SchemaTester`` instances."""
         return SchemaTester()
 
+
+# pylint: disable=R0903
+class OpenAPINinjaClient(TestClient):
+    """``APINinjaClient`` validating responses against OpenAPI schema."""
+
+    def __init__(
+        self,
+        *args,
+        router_or_app: NinjaAPI | Router,
+        path_prefix: str = "",
+        schema_tester: SchemaTester | None = None,
+        **kwargs,
+    ) -> None:
+        """Initialize ``OpenAPIClient`` instance."""
+        super().__init__(*args, router_or_app=router_or_app, **kwargs)
+        self.schema_tester = schema_tester or self._schema_tester_factory()
+        self._ninja_path_prefix = path_prefix
+
+    def request(self, *args, **kwargs) -> Response:
+        """Validate fetched response against given OpenAPI schema."""
+        response = super().request(*args, **kwargs)
+        response_handler = ResponseHandlerFactory.create(
+            *args, response=response, path_prefix=self._ninja_path_prefix, **kwargs
+        )
+        if self._is_successful_response(response):
+            self.schema_tester.validate_request(response_handler=response_handler)
+        self.schema_tester.validate_response(response_handler)
+        return response
+
+    @staticmethod
+    def _is_successful_response(response: Response) -> bool:
+        return response.status_code < http.HTTPStatus.BAD_REQUEST
+
     @staticmethod
-    def _serialize(data):
-        try:
-            return json.dumps(data)
-        except (TypeError, OverflowError):
-            # Data is already serialized
-            return data
+    def _schema_tester_factory() -> SchemaTester:
+        """Factory of default ``SchemaTester`` instances."""
+        return SchemaTester()

openapi_tester/response_handler.py

@@ -3,14 +3,26 @@
 """
 
 import json
-from typing import TYPE_CHECKING, Optional, Union
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any, Optional, Union
 
 if TYPE_CHECKING:
     from django.http.response import HttpResponse
     from rest_framework.response import Response
 
 
-class ResponseHandler:
+@dataclass
+class GenericRequest:
+    """Generic request class for both DRF and Django Ninja."""
+
+    path: str
+    method: str
+    data: dict = field(default_factory=dict)
+    headers: dict = field(default_factory=dict)
+
+
+class ResponseHandler(ABC):
     """
     This class is used to handle the response and request data
     from both DRF and Django HTTP (Django Ninja) responses.
@@ -24,10 +36,12 @@ def response(self) -> Union["Response", "HttpResponse"]:
         return self._response
 
     @property
-    def data(self) -> Optional[dict]: ...
+    @abstractmethod
+    def request(self) -> GenericRequest: ...
 
     @property
-    def request_data(self) -> Optional[dict]: ...
+    @abstractmethod
+    def data(self) -> Optional[dict]: ...
 
 
 class DRFResponseHandler(ResponseHandler):
@@ -43,23 +57,44 @@ def data(self) -> Optional[dict]:
         return self.response.json() if self.response.data is not None else None  # type: ignore[attr-defined]
 
     @property
-    def request_data(self) -> Optional[dict]:
-        return self.response.renderer_context["request"].data  # type: ignore[attr-defined]
+    def request(self) -> GenericRequest:
+        return GenericRequest(
+            path=self.response.renderer_context["request"].path,  # type: ignore[attr-defined]
+            method=self.response.renderer_context["request"].method,  # type: ignore[attr-defined]
+            data=self.response.renderer_context["request"].data,  # type: ignore[attr-defined]
+            headers=self.response.renderer_context["request"].headers,  # type: ignore[attr-defined]
+        )
 
 
 class DjangoNinjaResponseHandler(ResponseHandler):
     """
     Handles the response and request data from Django Ninja responses.
     """
 
-    def __init__(self, response: "HttpResponse") -> None:
+    def __init__(
+        self, *request_args, response: "HttpResponse", path_prefix: str = "", **kwargs
+    ) -> None:
         super().__init__(response)
+        self._request_method = request_args[0]
+        self._request_path = f"{path_prefix}{request_args[1]}"
+        self._request_data = self._build_request_data(request_args[2])
+        self._request_headers = kwargs
 
     @property
     def data(self) -> Optional[dict]:
         return self.response.json() if self.response.content else None  # type: ignore[attr-defined]
 
     @property
-    def request_data(self) -> Optional[dict]:
-        response_body = self.response.wsgi_request.body  # type: ignore[attr-defined]
-        return json.loads(response_body) if response_body else None
+    def request(self) -> GenericRequest:
+        return GenericRequest(
+            path=self._request_path,
+            method=self._request_method,
+            data=self._request_data,
+            headers=self._request_headers,
+        )
+
+    def _build_request_data(self, request_data: Any) -> dict:
+        try:
+            return json.loads(request_data)
+        except (json.JSONDecodeError, TypeError, ValueError):
+            return {}

openapi_tester/response_handler_factory.py

@@ -25,7 +25,9 @@ class ResponseHandlerFactory:
     """
 
     @staticmethod
-    def create(response: Union[Response, "HttpResponse"]) -> "ResponseHandler":
+    def create(
+        *request_args, response: Union[Response, "HttpResponse"], **kwargs
+    ) -> "ResponseHandler":
         if isinstance(response, Response):
-            return DRFResponseHandler(response)
-        return DjangoNinjaResponseHandler(response)
+            return DRFResponseHandler(response=response)
+        return DjangoNinjaResponseHandler(*request_args, response=response, **kwargs)