feat(role_management): Add apitest for test_assign_roles

apps/authentication/tests/api_tests/user/conftest.py

@@ -0,0 +1,21 @@
+import pytest
+
+from tests.base.authentication_web import AuthenticationWeb
+
+
+@pytest.fixture(scope="session")
+def authentication_web() -> AuthenticationWeb:
+    authentication_web = AuthenticationWeb()
+    authentication_web.login()
+    return authentication_web
+
+
+@pytest.fixture(scope="session")
+def authentication_web_of_temp_user1() -> AuthenticationWeb:
+    authentication_web = AuthenticationWeb()
+    user = authentication_web.create_temporary_user()
+    authentication_web.user_email = user["email"]
+    authentication_web.password = user["password"]
+    authentication_web.user_id = user["user_id"]
+    authentication_web.login()
+    return authentication_web

apps/authentication/tests/api_tests/user/test_assign_roles.py

@@ -0,0 +1,100 @@
+import pytest
+import random
+from backend.models.permission.constants import DefaultRoleEnum
+from tests.base.authentication_web import AuthenticationWeb
+
+
+class TestAssignRolesToUser:
+    @pytest.mark.asyncio
+    async def test_assign_roles_success_by_admin(self, authentication_web: AuthenticationWeb):
+        """Test assigning roles to a user successfully by admin user."""
+        # Create a temporary user
+        temp_user = authentication_web.create_temporary_user()
+        # Create a new role
+        suffix = str(random.randint(10000, 99999))
+        role_resp = await authentication_web.create_role({
+            "role_key": f"assignrole_role_{suffix}",
+            "role_name": f"AssignRole Role {suffix}",
+            "role_description": "desc",
+            "role_level": 1
+        })
+        role_id = role_resp.json()["id"]
+        # Assign role to user
+        resp = await authentication_web.assign_roles_to_user({
+            "user_id": temp_user["user_id"], "role_ids": [role_id]
+        })
+        assert resp.status_code == 200
+        json = resp.json()
+        assert json["user_id"] == temp_user["user_id"]
+        assert json["role_ids"] == [role_id]
+
+    @pytest.mark.asyncio
+    async def test_assign_roles_fail_by_non_admin(self, authentication_web_of_temp_user1: AuthenticationWeb):
+        """Test assigning roles to a user fails by non-admin user (no permission)."""
+        # Create another temporary user
+        temp_user = authentication_web_of_temp_user1.create_temporary_user()
+        # Query default admin role
+        resp = await authentication_web_of_temp_user1.query_roles({"role_key": DefaultRoleEnum.ADMIN.value.role_key})
+        admin_role_id = resp.json()["items"][0]["id"]
+        # Try to assign admin role to another user
+        resp = await authentication_web_of_temp_user1.assign_roles_to_user({
+            "user_id": temp_user["user_id"], "role_ids": [admin_role_id]
+        })
+        assert resp.status_code == 403 or resp.status_code == 401
+
+    @pytest.mark.asyncio
+    async def test_assign_roles_fail_role_not_found(self, authentication_web: AuthenticationWeb):
+        """Test assigning roles fails when role_id does not exist."""
+        # Create a temporary user
+        temp_user = authentication_web.create_temporary_user()
+        # Try to assign non-existent role
+        resp = await authentication_web.assign_roles_to_user({
+            "user_id": temp_user["user_id"], "role_ids": ["000000000000000000000000"]
+        })
+        assert resp.status_code == 422 or resp.status_code == 400
+
+    @pytest.mark.asyncio
+    async def test_assign_roles_fail_empty_role_ids(self, authentication_web: AuthenticationWeb):
+        """Test assigning roles fails when role_ids is empty."""
+        # Create a temporary user
+        temp_user = authentication_web.create_temporary_user()
+        resp = await authentication_web.assign_roles_to_user({
+            "user_id": temp_user["user_id"], "role_ids": []
+        })
+        assert resp.status_code == 422 or resp.status_code == 400
+
+    @pytest.mark.asyncio
+    async def test_assign_roles_fail_empty_user_id(self, authentication_web: AuthenticationWeb):
+        """Test assigning roles fails when user_id is empty."""
+        # Query default admin role
+        resp = await authentication_web.query_roles({"role_key": DefaultRoleEnum.ADMIN.value.role_key})
+        admin_role_id = resp.json()["items"][0]["id"]
+        resp = await authentication_web.assign_roles_to_user({
+            "user_id": "", "role_ids": [admin_role_id]
+        })
+        assert resp.status_code == 422 or resp.status_code == 400
+
+    @pytest.mark.asyncio
+    async def test_assign_roles_remove_duplicates(self, authentication_web: AuthenticationWeb):
+        """Test assigning roles with duplicate role_ids removes duplicates."""
+        # Create a temporary user
+        temp_user = authentication_web.create_temporary_user()
+        # Create a new role
+        suffix = str(random.randint(10000, 99999))
+        role_resp = await authentication_web.create_role({
+            "role_key": f"assignrole_role_dup_{suffix}",
+            "role_name": f"AssignRole RoleDup {suffix}",
+            "role_description": "desc",
+            "role_level": 1
+        })
+        role_id = role_resp.json()["id"]
+        # Assign duplicate role_ids
+        resp = await authentication_web.assign_roles_to_user({
+            "user_id": temp_user["user_id"], "role_ids": [role_id, role_id, role_id]
+        })
+        assert resp.status_code == 200
+        json = resp.json()
+        assert json["role_ids"] == [role_id]
+
+if __name__ == '__main__':
+    pytest.main([__file__])

apps/authentication/tests/base/authentication_web.py

@@ -10,6 +10,7 @@ class AuthenticationWeb:
     def __init__(self, user_email: str = USER_EMAIL, password: str = USER_PASSWORD, base_url: str = BASE_URL):
         self.user_email = user_email
         self.password = password
+        self.user_id = None
         self.base_url = base_url
         self.token: Optional[str] = None
 
@@ -42,6 +43,7 @@ class AuthenticationWeb:
         return {
             "email": email,
             "password": password,
+            "user_id": response2.json()["identity"]
         }
 
     def update_new_user_flid(self, params: dict, token: str = None):
@@ -144,6 +146,10 @@ class AuthenticationWeb:
         """Assign permissions to a role via API"""
         return await self.request("POST", "/api/auth/role/assign-permissions", json=data)
 
+    async def assign_roles_to_user(self, data: dict):
+        """Assign roles to a user via API"""
+        return await self.request("POST", "/api/auth/user/assign-roles", json=data)
+
 
 if __name__ == '__main__':
     authentication = AuthenticationWeb()