Add support of DRF OR, AND, NOT operands (#202)

djangochannelsrestframework/consumers.py

@@ -17,7 +17,7 @@
 from django.http.response import Http404
 from django.template.response import SimpleTemplateResponse
 from rest_framework.exceptions import PermissionDenied, MethodNotAllowed, APIException
-from rest_framework.permissions import BasePermission as DRFBasePermission
+from rest_framework.permissions import BasePermission as DRFBasePermission, OR, AND, NOT
 from rest_framework.response import Response
 
 from djangochannelsrestframework.settings import api_settings
@@ -124,7 +124,7 @@ async def get_permissions(self, action: str, **kwargs):
             instance = permission_class()
 
             # If the permission is an DRF permission instance
-            if isinstance(instance, DRFBasePermission):
+            if isinstance(instance, (DRFBasePermission, OR, AND, NOT)):
                 instance = WrappedDRFPermission(instance)
             permission_instances.append(instance)
 

tests/test_permission.py

@@ -93,3 +93,135 @@ class AConsumer(AsyncAPIConsumer):
     connected, _ = await communicator.connect()
     assert connected
     assert called == {"has_permission": True}
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.asyncio
+async def test_users_drf_or_permission(settings):
+
+    called = {}
+
+    class TestPermissionA(DRFBasePermission):
+        def has_permission(self, request, view):
+            called["has_permission_a"] = False
+            return False
+
+    class TestPermissionB(DRFBasePermission):
+        def has_permission(self, request, view):
+            called["has_permission_b"] = True
+            return True
+
+    class AConsumer(AsyncAPIConsumer):
+        permission_classes = [TestPermissionA | TestPermissionB]
+        pass
+
+    # Test a normal connection
+    communicator = WebsocketCommunicator(AConsumer(), "/testws/")
+
+    connected, _ = await communicator.connect()
+    assert connected
+    assert called == {"has_permission_a": False, "has_permission_b": True}
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.asyncio
+async def test_users_drf_and_permission(settings):
+
+    called = {}
+
+    class TestPermissionA(DRFBasePermission):
+        def has_permission(self, request, view):
+            called["has_permission_a"] = True
+            return True
+
+    class TestPermissionB(DRFBasePermission):
+        def has_permission(self, request, view):
+            called["has_permission_b"] = True
+            return True
+
+    class AConsumer(AsyncAPIConsumer):
+        permission_classes = [TestPermissionA & TestPermissionB]
+        pass
+
+    # Test a normal connection
+    communicator = WebsocketCommunicator(AConsumer(), "/testws/")
+
+    connected, _ = await communicator.connect()
+    assert connected
+    assert called == {"has_permission_a": True, "has_permission_b": True}
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.asyncio
+async def test_users_drf_not_permission(settings):
+
+    called = {}
+
+    class TestPermission(DRFBasePermission):
+        def has_permission(self, request, view):
+            called["has_permission"] = False
+            return False
+
+    class AConsumer(AsyncAPIConsumer):
+        permission_classes = [~TestPermission]
+        pass
+
+    # Test a normal connection
+    communicator = WebsocketCommunicator(AConsumer(), "/testws/")
+
+    connected, _ = await communicator.connect()
+    assert connected
+    assert called == {"has_permission": False}
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.asyncio
+async def test_users_drf_complex_permission(settings):
+
+    called = {}
+
+    class TestPermissionA(DRFBasePermission):
+        def has_permission(self, request, view):
+            called["has_permission_a"] = False
+            return False
+
+    class TestPermissionB(DRFBasePermission):
+        def has_permission(self, request, view):
+            called["has_permission_b"] = True
+            return True
+
+    class TestPermissionC(DRFBasePermission):
+        def has_permission(self, request, view):
+            called["has_permission_c"] = True
+            return True
+
+    class TestPermissionD(DRFBasePermission):
+        def has_permission(self, request, view):
+            called["has_permission_d"] = True
+            return True
+
+    class TestEPermission(DRFBasePermission):
+        def has_permission(self, request, view):
+            called["has_permission_e"] = False
+            return False
+
+    class AConsumer(AsyncAPIConsumer):
+        permission_classes = (
+            TestPermissionA | TestPermissionB,
+            TestPermissionC & TestPermissionD,
+            ~TestEPermission,
+        )
+        pass
+
+    # Test a normal connection
+    communicator = WebsocketCommunicator(AConsumer(), "/testws/")
+
+    connected, _ = await communicator.connect()
+    assert connected
+    assert called == {
+        "has_permission_a": False,
+        "has_permission_b": True,
+        "has_permission_c": True,
+        "has_permission_d": True,
+        "has_permission_e": False
+    }