Added feature tests

Author: eadwinCode

tests/test_async_auth.py

@@ -0,0 +1,148 @@
+import django
+import pytest
+from ninja.testing import TestAsyncClient
+
+from ninja_extra import NinjaExtraAPI
+from ninja_extra.security import (
+    AsyncAPIKeyCookie,
+    AsyncAPIKeyHeader,
+    AsyncAPIKeyQuery,
+    AsyncHttpBasicAuth,
+    AsyncHttpBearer,
+    async_django_auth,
+)
+
+
+async def callable_auth(request):
+    return request.GET.get("auth")
+
+
+class KeyQuery(AsyncAPIKeyQuery):
+    async def authenticate(self, request, key):
+        if key == "keyquerysecret":
+            return key
+
+
+class KeyHeader(AsyncAPIKeyHeader):
+    async def authenticate(self, request, key):
+        if key == "keyheadersecret":
+            return key
+
+
+class KeyCookie(AsyncAPIKeyCookie):
+    async def authenticate(self, request, key):
+        if key == "keycookiersecret":
+            return key
+
+
+class BasicAuth(AsyncHttpBasicAuth):
+    async def authenticate(self, request, username, password):
+        if username == "admin" and password == "secret":
+            return username
+
+
+class BearerAuth(AsyncHttpBearer):
+    async def authenticate(self, request, token):
+        if token == "bearertoken":
+            return token
+
+
+async def demo_operation(request):
+    return {"auth": request.auth}
+
+
+api = NinjaExtraAPI(csrf=True, urls_namespace="async_auth")
+
+for path, auth in [
+    ("django_auth", async_django_auth),
+    ("callable", callable_auth),
+    ("apikeyquery", KeyQuery()),
+    ("apikeyheader", KeyHeader()),
+    ("apikeycookie", KeyCookie()),
+    ("basic", BasicAuth()),
+    ("bearer", BearerAuth()),
+]:
+    api.get(f"/{path}", auth=auth, operation_id=path)(demo_operation)
+
+
+client = TestAsyncClient(api)
+
+
+class MockUser(str):
+    is_authenticated = True
+
+
+BODY_UNAUTHORIZED_DEFAULT = dict(detail="Unauthorized")
+
+
+@pytest.mark.skipif(django.VERSION < (3, 1), reason="requires django 3.1 or higher")
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "path,kwargs,expected_code,expected_body",
+    [
+        ("/django_auth", {}, 401, BODY_UNAUTHORIZED_DEFAULT),
+        ("/django_auth", dict(user=MockUser("admin")), 200, dict(auth="admin")),
+        ("/callable", {}, 401, BODY_UNAUTHORIZED_DEFAULT),
+        ("/callable?auth=demo", {}, 200, dict(auth="demo")),
+        ("/apikeyquery", {}, 401, BODY_UNAUTHORIZED_DEFAULT),
+        ("/apikeyquery?key=keyquerysecret", {}, 200, dict(auth="keyquerysecret")),
+        ("/apikeyheader", {}, 401, BODY_UNAUTHORIZED_DEFAULT),
+        (
+            "/apikeyheader",
+            dict(headers={"key": "keyheadersecret"}),
+            200,
+            dict(auth="keyheadersecret"),
+        ),
+        ("/apikeycookie", {}, 401, BODY_UNAUTHORIZED_DEFAULT),
+        (
+            "/apikeycookie",
+            dict(COOKIES={"key": "keycookiersecret"}),
+            200,
+            dict(auth="keycookiersecret"),
+        ),
+        ("/basic", {}, 401, BODY_UNAUTHORIZED_DEFAULT),
+        (
+            "/basic",
+            dict(headers={"Authorization": "Basic YWRtaW46c2VjcmV0"}),
+            200,
+            dict(auth="admin"),
+        ),
+        (
+            "/basic",
+            dict(headers={"Authorization": "YWRtaW46c2VjcmV0"}),
+            200,
+            dict(auth="admin"),
+        ),
+        (
+            "/basic",
+            dict(headers={"Authorization": "Basic invalid"}),
+            401,
+            BODY_UNAUTHORIZED_DEFAULT,
+        ),
+        (
+            "/basic",
+            dict(headers={"Authorization": "some invalid value"}),
+            401,
+            BODY_UNAUTHORIZED_DEFAULT,
+        ),
+        ("/bearer", {}, 401, BODY_UNAUTHORIZED_DEFAULT),
+        (
+            "/bearer",
+            dict(headers={"Authorization": "Bearer bearertoken"}),
+            200,
+            dict(auth="bearertoken"),
+        ),
+        (
+            "/bearer",
+            dict(headers={"Authorization": "Invalid bearertoken"}),
+            401,
+            BODY_UNAUTHORIZED_DEFAULT,
+        ),
+    ],
+)
+async def test_auth(path, kwargs, expected_code, expected_body, settings):
+    for debug in (False, True):
+        settings.DEBUG = debug
+        response = await client.get(path, **kwargs)
+        assert response.status_code == expected_code
+        assert response.json() == expected_body

tests/test_controller.py

@@ -1,19 +1,21 @@
 from unittest.mock import Mock, patch
 
+import django
 import pytest
 from django.contrib.auth.models import Group
 
 from ninja_extra import NinjaExtraAPI, api_controller, exceptions, http_get, testing
 from ninja_extra.controllers import ControllerBase, RouteContext, RouteFunction
 from ninja_extra.controllers.base import (
     APIController,
-    MissingAPIControllerDecoratorException,
     compute_api_route_function,
     get_route_functions,
 )
 from ninja_extra.controllers.response import Detail, Id, Ok
 from ninja_extra.permissions.common import AllowAny
 
+from .utils import AsyncFakeAuth, FakeAuth
+
 
 @api_controller
 class SomeController:
@@ -57,11 +59,16 @@ class DisableAutoImportController:
 
 class TestAPIController:
     def test_api_controller_as_decorator(self):
-        api_controller_instance = api_controller("prefix", tags="new_tag")
+        api_controller_instance = api_controller(
+            "prefix", tags="new_tag", auth=FakeAuth()
+        )
 
+        assert not api_controller_instance.has_auth_async
+        assert not api_controller_instance._prefix_has_route_param
         assert api_controller_instance.prefix == "prefix"
         assert api_controller_instance.tags == ["new_tag"]
         assert api_controller_instance.permission_classes == [AllowAny]
+
         api_controller_instance = api_controller()
         assert api_controller_instance.prefix == ""
         assert api_controller_instance.tags is None
@@ -183,6 +190,35 @@ def test_controller_base_get_object_or_none_works(self):
                 assert isinstance(ex, exceptions.PermissionDenied)
 
 
+@pytest.mark.skipif(django.VERSION < (3, 1), reason="requires django 3.1 or higher")
+def test_async_controller():
+    api_controller_instance = api_controller(
+        "prefix", tags="any_Tag", auth=AsyncFakeAuth()
+    )
+    assert api_controller_instance.has_auth_async
+
+    with pytest.raises(Exception) as ex:
+
+        @api_controller_instance
+        class NonAsyncRouteInControllerWithAsyncAuth:
+            @http_get("/example")
+            def example(self):
+                pass
+
+    assert "NonAsyncRouteInControllerWithAsyncAuth" in str(ex) and "example" in str(ex)
+
+    @api_controller_instance
+    class AsyncRouteInControllerWithAsyncAuth:
+        @http_get("/example")
+        async def example(self):
+            pass
+
+    assert isinstance(
+        AsyncRouteInControllerWithAsyncAuth.example.operation.auth_callbacks[0],
+        AsyncFakeAuth,
+    )
+
+
 class TestAPIControllerResponse:
     ok_response = Ok("OK")
     id_response = Id("ID")

tests/test_django_ninja_router.py

@@ -0,0 +1,100 @@
+import pytest
+from ninja.testing import TestClient
+
+from ninja_extra import NinjaExtraAPI, Router
+from ninja_extra.operation import PathView
+
+api = NinjaExtraAPI(urls_namespace="ninja_router")
+
+
+@api.get("/endpoint")
+# view->api
+def global_op(request):
+    return "global"
+
+
+first_router = Router()
+
+
+@first_router.get("/endpoint_1")
+# view->router, router->api
+def router_op1(request):
+    return "first 1"
+
+
+second_router_one = Router()
+
+
+@second_router_one.get("endpoint_1")
+# view->router2, router2->router1, router1->api
+def router_op2(request):
+    return "second 1"
+
+
+second_router_two = Router()
+
+
+@second_router_two.get("endpoint_2")
+# view->router2, router2->router1, router1->api
+def router2_op3(request):
+    return "second 2"
+
+
+first_router.add_router("/second", second_router_one, tags=["one"])
+first_router.add_router("/second", second_router_two, tags=["two"])
+api.add_router("/first", first_router, tags=["global"])
+
+
+@first_router.get("endpoint_2")
+# router->api, view->router
+def router1_op1(request):
+    return "first 2"
+
+
+@second_router_one.get("endpoint_3")
+# router2->router1, router1->api, view->router2
+def router21_op3(request, path_param: int = None):
+    return "second 3" if path_param is None else f"second 3: {path_param}"
+
+
+second_router_three = Router()
+
+
+@second_router_three.get("endpoint_4")
+# router1->api, view->router2, router2->router1
+def router_op3(request, path_param: int = None):
+    return "second 4" if path_param is None else f"second 4: {path_param}"
+
+
+first_router.add_router("/second", second_router_three, tags=["three"])
+
+
+client = TestClient(api)
+
+
+@pytest.mark.parametrize(
+    "path,expected_status,expected_response",
+    [
+        ("/endpoint", 200, "global"),
+        ("/first/endpoint_1", 200, "first 1"),
+        ("/first/endpoint_2", 200, "first 2"),
+        ("/first/second/endpoint_1", 200, "second 1"),
+        ("/first/second/endpoint_2", 200, "second 2"),
+        ("/first/second/endpoint_3", 200, "second 3"),
+        ("/first/second/endpoint_4", 200, "second 4"),
+    ],
+)
+def test_inheritance_responses(path, expected_status, expected_response):
+    response = client.get(path)
+    assert response.status_code == expected_status, response.content
+    assert response.json() == expected_response
+
+
+def test_router_path_view():
+    router_op1_path_view = first_router.path_operations.get("/endpoint_1")
+    assert router_op1_path_view
+    assert isinstance(router_op1_path_view, PathView)
+
+    global_op_path_view = api.default_router.path_operations.get("/endpoint")
+    assert router_op1_path_view
+    assert isinstance(global_op_path_view, PathView)

tests/test_operation.py

@@ -2,9 +2,11 @@
 import pytest
 
 from ninja_extra import api_controller, route
+from ninja_extra.controllers import AsyncRouteFunction, RouteFunction
+from ninja_extra.operation import AsyncControllerOperation, ControllerOperation
 from ninja_extra.testing import TestAsyncClient, TestClient
 
-from .utils import mock_log_call, mock_signal_call
+from .utils import AsyncFakeAuth, FakeAuth, mock_log_call, mock_signal_call
 
 
 class TestOperation:
@@ -35,6 +37,40 @@ def test_route_operation_execution_should_log_execution(self):
             client.get(str(self.SomeTestController.example_exception))
 
 
+@pytest.mark.skipif(django.VERSION < (3, 1), reason="requires django 3.1 or higher")
+def test_operation_auth_configs():
+    api_controller_instance = api_controller("prefix", tags="any_Tag")
+
+    async def async_endpoint(self, request):
+        pass
+
+    def sync_endpoint(self, request):
+        pass
+
+    sync_auth_http_get = route.get("/example", auth=[FakeAuth()])
+    async_auth_http_get = route.get("/example/async", auth=[AsyncFakeAuth()])
+
+    route_function = sync_auth_http_get(async_endpoint)
+    assert isinstance(route_function, AsyncRouteFunction)
+    async_route_function = async_auth_http_get(async_endpoint)
+
+    api_controller_instance.add_operation_from_route_function(route_function)
+    assert isinstance(route_function.operation, AsyncControllerOperation)
+    api_controller_instance.add_operation_from_route_function(async_route_function)
+    assert isinstance(async_route_function.operation, AsyncControllerOperation)
+
+    sync_route_function = sync_auth_http_get(sync_endpoint)
+    api_controller_instance.add_operation_from_route_function(sync_route_function)
+    assert isinstance(sync_route_function.operation, ControllerOperation)
+    assert isinstance(sync_route_function, RouteFunction)
+
+    with pytest.raises(Exception) as ex:
+        api_controller_instance.add_operation_from_route_function(
+            async_auth_http_get(sync_endpoint)
+        )
+    assert "sync_endpoint" in str(ex) and "AsyncFakeAuth" in str(ex)
+
+
 @pytest.mark.skipif(django.VERSION < (3, 1), reason="requires django 3.1 or higher")
 @pytest.mark.asyncio
 class TestAsyncOperations: