Base API and tables for RBAC support.

tests/api/routes/controller/test_groups.py

@@ -22,7 +22,10 @@ from httpx import AsyncClient
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from gns3server.db.repositories.users import UsersRepository
+from gns3server.db.repositories.rbac import RbacRepository
 from gns3server.schemas.controller.users import User
+from gns3server.schemas.controller.rbac import Role
+from gns3server import schemas
 
 pytestmark = pytest.mark.asyncio
 
@@ -103,6 +106,9 @@ class TestGroupRoutes:
         response = await client.delete(app.url_path_for("delete_user_group", user_group_id=group_in_db.user_group_id))
         assert response.status_code == status.HTTP_403_FORBIDDEN
 
+
+class TestGroupMembersRoutes:
+
     async def test_add_member_to_group(
             self,
             app: FastAPI,
@@ -163,3 +169,81 @@ class TestGroupRoutes:
         assert response.status_code == status.HTTP_204_NO_CONTENT
         members = await user_repo.get_user_group_members(group_in_db.user_group_id)
         assert len(members) == 0
+
+
+@pytest.fixture
+async def test_role(db_session: AsyncSession) -> Role:
+
+    new_role = schemas.RoleCreate(
+        name="TestRole",
+        description="This is my test role"
+    )
+    rbac_repo = RbacRepository(db_session)
+    existing_role = await rbac_repo.get_role_by_name(new_role.name)
+    if existing_role:
+        return existing_role
+    return await rbac_repo.create_role(new_role)
+
+
+class TestGroupRolesRoutes:
+
+    async def test_add_role_to_group(
+            self,
+            app: FastAPI,
+            client: AsyncClient,
+            test_role: Role,
+            db_session: AsyncSession
+    ) -> None:
+
+        user_repo = UsersRepository(db_session)
+        group_in_db = await user_repo.get_user_group_by_name("Users")
+        response = await client.put(
+            app.url_path_for(
+                "add_role_to_group",
+                user_group_id=group_in_db.user_group_id,
+                role_id=str(test_role.role_id)
+            )
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+        roles = await user_repo.get_user_group_roles(group_in_db.user_group_id)
+        assert len(roles) == 1
+        assert roles[0].name == test_role.name
+
+    async def test_get_user_group_roles(
+            self,
+            app: FastAPI,
+            client: AsyncClient,
+            db_session: AsyncSession
+    ) -> None:
+
+        user_repo = UsersRepository(db_session)
+        group_in_db = await user_repo.get_user_group_by_name("Users")
+        response = await client.get(
+            app.url_path_for(
+                "get_user_group_roles",
+                user_group_id=group_in_db.user_group_id)
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.json()) == 1
+
+    async def test_remove_role_from_group(
+            self,
+            app: FastAPI,
+            client: AsyncClient,
+            test_role: Role,
+            db_session: AsyncSession
+    ) -> None:
+
+        user_repo = UsersRepository(db_session)
+        group_in_db = await user_repo.get_user_group_by_name("Users")
+
+        response = await client.delete(
+            app.url_path_for(
+                "remove_role_from_group",
+                user_group_id=group_in_db.user_group_id,
+                role_id=test_role.role_id
+            ),
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+        roles = await user_repo.get_user_group_roles(group_in_db.user_group_id)
+        assert len(roles) == 0