Add custom permission/rotes (#56)

Author: ferhatelmas

stream_chat/async_chat/client.py

@@ -360,6 +360,67 @@ async def check_sqs(self, sqs_key=None, sqs_secret=None, sqs_url=None):
         data = {"sqs_key": sqs_key, "sqs_secret": sqs_secret, "sqs_url": sqs_url}
         return await self.post("check_sqs", data=data)
 
+    async def get_permission(self, name):
+        """
+        Get the definition for a permission
+
+        :param name: Name of the permission
+        """
+        return await self.get(f"custom_permission/{name}")
+
+    async def create_permission(self, permission):
+        """
+        Create a custom permission
+
+        :param permission: Definition of the permission
+        """
+        return await self.post("custom_permission", data=permission)
+
+    async def update_permission(self, name, permission):
+        """
+        Update a custom permission
+
+        :param name: Name of the permission
+        :param permission: New definition of the permission
+        """
+        return await self.post(f"custom_permission/{name}", data=permission)
+
+    async def delete_permission(self, name):
+        """
+        Delete a custom permission
+
+        :param name: Name of the permission
+        """
+        return await self.delete(f"custom_permission/{name}")
+
+    async def list_permissions(self):
+        """
+        List custom permissions of the app
+        """
+        return await self.get("custom_permission")
+
+    async def create_role(self, name):
+        """
+        Create a custom role
+
+        :param name: Name of the role
+        """
+        return await self.post("custom_role", data={"name": name})
+
+    async def delete_role(self, name):
+        """
+        Delete a custom role
+
+        :param name: Name of the role
+        """
+        return await self.delete(f"custom_role/{name}")
+
+    async def list_roles(self):
+        """
+        List custom roles of the app
+        """
+        return await self.get("custom_role")
+
     async def close(self):
         await self.session.close()
 

stream_chat/base/client.py

@@ -318,7 +318,7 @@ def delete_blocklist(self, name):
         pass
 
     @abc.abstractmethod
-    async def check_sqs(self, sqs_key=None, sqs_secret=None, sqs_url=None):
+    def check_sqs(self, sqs_key=None, sqs_secret=None, sqs_url=None):
         """
         Check SQS Push settings
 
@@ -330,3 +330,72 @@ async def check_sqs(self, sqs_key=None, sqs_secret=None, sqs_url=None):
         :return:
         """
         pass
+
+    @abc.abstractmethod
+    def get_permission(self, name):
+        """
+        Get the definition for a permission
+
+        :param name: Name of the permission
+        """
+        pass
+
+    @abc.abstractmethod
+    def create_permission(self, permission):
+        """
+        Create a custom permission
+
+        :param permission: Definition of the permission
+        """
+        pass
+
+    @abc.abstractmethod
+    def update_permission(self, name, permission):
+        """
+        Update a custom permission
+
+        :param name: Name of the permission
+        :param permission: New definition of the permission
+        """
+        pass
+
+    @abc.abstractmethod
+    def delete_permission(self, name):
+        """
+        Delete a custom permission
+
+        :param name: Name of the permission
+        """
+        pass
+
+    @abc.abstractmethod
+    def list_permissions(self):
+        """
+        List custom permissions of the app
+        """
+        pass
+
+    @abc.abstractmethod
+    def create_role(self, name):
+        """
+        Create a custom role
+
+        :param name: Name of the role
+        """
+        pass
+
+    @abc.abstractmethod
+    def delete_role(self, name):
+        """
+        Delete a custom role
+
+        :param name: Name of the role
+        """
+        pass
+
+    @abc.abstractmethod
+    def list_roles(self):
+        """
+        List custom roles of the app
+        """
+        pass

stream_chat/client.py

@@ -347,3 +347,64 @@ def check_sqs(self, sqs_key=None, sqs_secret=None, sqs_url=None):
         """
         data = {"sqs_key": sqs_key, "sqs_secret": sqs_secret, "sqs_url": sqs_url}
         return self.post("check_sqs", data=data)
+
+    def get_permission(self, name):
+        """
+        Get the definition for a permission
+
+        :param name: Name of the permission
+        """
+        return self.get(f"custom_permission/{name}")
+
+    def create_permission(self, permission):
+        """
+        Create a custom permission
+
+        :param permission: Definition of the permission
+        """
+        return self.post("custom_permission", data=permission)
+
+    def update_permission(self, name, permission):
+        """
+        Update a custom permission
+
+        :param name: Name of the permission
+        :param permission: New definition of the permission
+        """
+        return self.post(f"custom_permission/{name}", data=permission)
+
+    def delete_permission(self, name):
+        """
+        Delete a custom permission
+
+        :param name: Name of the permission
+        """
+        return self.delete(f"custom_permission/{name}")
+
+    def list_permissions(self):
+        """
+        List custom permissions of the app
+        """
+        return self.get("custom_permission")
+
+    def create_role(self, name):
+        """
+        Create a custom role
+
+        :param name: Name of the role
+        """
+        return self.post("custom_role", data={"name": name})
+
+    def delete_role(self, name):
+        """
+        Delete a custom role
+
+        :param name: Name of the role
+        """
+        return self.delete(f"custom_role/{name}")
+
+    def list_roles(self):
+        """
+        List custom roles of the app
+        """
+        return self.get("custom_role")

stream_chat/tests/async_chat/test_client.py

@@ -330,3 +330,52 @@ async def test_check_sqs(self, client):
         response = await client.check_sqs("key", "secret", "https://foo.com/bar")
         assert response["status"] == "error"
         assert "invalid SQS url" in response["error"]
+
+    @pytest.mark.asyncio
+    async def test_custom_permission_and_roles(self, client):
+        name, role = "Something restricted", "god"
+
+        try:
+            client.delete_permission(name)
+            client.delete_role(role)
+        except:  # noqa
+            pass
+
+        custom = {
+            "name": name,
+            "resource": "DeleteChannel",
+            "owner": False,
+            "same_team": True,
+        }
+
+        await client.create_permission(custom)
+        response = await client.get_permission(name)
+        print(response)
+        assert response["permission"]["name"] == name
+        assert response["permission"]["custom"]
+        assert not response["permission"]["owner"]
+        assert response["permission"]["resource"] == custom["resource"]
+
+        custom["owner"] = True
+        await client.update_permission(name, custom)
+
+        response = await client.get_permission(name)
+        print(response)
+        assert response["permission"]["name"] == name
+        assert response["permission"]["custom"]
+        assert response["permission"]["owner"]
+        assert response["permission"]["resource"] == custom["resource"]
+
+        response = await client.list_permissions()
+        assert len(response["permissions"]) == 1
+        assert response["permissions"][0]["name"] == name
+        await client.delete_permission(name)
+        response = await client.list_permissions()
+        assert len(response["permissions"]) == 0
+
+        await client.create_role(role)
+        response = await client.list_roles()
+        assert role in response["roles"]
+        await client.delete_role(role)
+        response = await client.list_roles()
+        assert role not in response["roles"]

stream_chat/tests/test_client.py

@@ -270,3 +270,51 @@ def test_check_sqs(self, client):
         response = client.check_sqs("key", "secret", "https://foo.com/bar")
         assert response["status"] == "error"
         assert "invalid SQS url" in response["error"]
+
+    def test_custom_permission_and_roles(self, client):
+        name, role = "Something restricted", "god"
+
+        try:
+            client.delete_permission(name)
+            client.delete_role(role)
+        except:  # noqa
+            pass
+
+        custom = {
+            "name": name,
+            "resource": "DeleteChannel",
+            "owner": False,
+            "same_team": True,
+        }
+
+        client.create_permission(custom)
+        response = client.get_permission(name)
+        print(response)
+        assert response["permission"]["name"] == name
+        assert response["permission"]["custom"]
+        assert not response["permission"]["owner"]
+        assert response["permission"]["resource"] == custom["resource"]
+
+        custom["owner"] = True
+        client.update_permission(name, custom)
+
+        response = client.get_permission(name)
+        print(response)
+        assert response["permission"]["name"] == name
+        assert response["permission"]["custom"]
+        assert response["permission"]["owner"]
+        assert response["permission"]["resource"] == custom["resource"]
+
+        response = client.list_permissions()
+        assert len(response["permissions"]) == 1
+        assert response["permissions"][0]["name"] == name
+        client.delete_permission(name)
+        response = client.list_permissions()
+        assert len(response["permissions"]) == 0
+
+        client.create_role(role)
+        response = client.list_roles()
+        assert role in response["roles"]
+        client.delete_role(role)
+        response = client.list_roles()
+        assert role not in response["roles"]