fix: modify code such that a user has admin role type in case

Signed-off-by: Youngjin Jo <[email protected]>

src/spaceone/identity/service/user_profile_service.py

@@ -423,7 +423,6 @@ def get_workspace_groups(
         Returns:
             MyWorkspaceResponse:
         """
-
         rb_mgr = RoleBindingManager()
         allow_all = False
 
@@ -436,16 +435,24 @@ def get_workspace_groups(
             workspace_group_vos = self.workspace_group_mgr.filter_workspace_groups(
                 domain_id=params.domain_id
             )
+            workspace_group_infos = [
+                workspace_group_vo.to_dict()
+                for workspace_group_vo in workspace_group_vos
+            ]
         else:
-            # TODO: Need to check users__user_id is correct
-            workspace_group_vos = self.workspace_group_mgr.filter_workspace_groups(
-                users__user_id=params.user_id,
-                domain_id=params.domain_id,
+            query_filter = {
+                "filter": [
+                    {"key": "users.user_id", "value": params.user_id, "operator": "eq"},
+                    {"key": "domain_id", "value": params.domain_id, "operator": "eq"},
+                ]
+            }
+            workspace_group_infos, _ = self.workspace_group_mgr.list_workspace_groups(
+                query_filter
             )
 
         workspace_group_ids = [
-            workspace_group_vo.workspace_group_id
-            for workspace_group_vo in workspace_group_vos
+            workspace_group_info["workspace_group_id"]
+            for workspace_group_info in workspace_group_infos
         ]
 
         rb_vos = rb_mgr.filter_role_bindings(
@@ -457,16 +464,23 @@ def get_workspace_groups(
         role_bindings_info_map = {rb.workspace_group_id: rb.to_dict() for rb in rb_vos}
 
         workspace_group_user_ids = []
-        for workspace_group in workspace_group_vos:
-            if workspace_group.users:
-                for user in workspace_group.users:
-                    workspace_group_user_ids.append(user.user_id)
+        for workspace_group_info in workspace_group_infos:
+            if not isinstance(workspace_group_info, dict):
+                workspace_group_info = workspace_group_info.to_dict()
+            if users := workspace_group_info.get("users", []) or []:
+                for user in users:
+                    if isinstance(user, dict):
+                        workspace_group_user_ids.append(user.get("user_id"))
+                    elif hasattr(user, "user_id"):
+                        workspace_group_user_ids.append(user.user_id)
 
         workspace_groups_info = []
-        for workspace_group_vo in workspace_group_vos:
+        for workspace_group_info in workspace_group_infos:
             workspace_group_dict = (
                 self.workspace_group_svc.add_user_name_and_state_to_users(
-                    workspace_group_user_ids, workspace_group_vo, params.domain_id
+                    workspace_group_user_ids,
+                    workspace_group_info,
+                    params.domain_id,
                 )
             )
             workspace_groups_info.append(workspace_group_dict)

src/spaceone/identity/service/workspace_group_service.py

@@ -474,7 +474,8 @@ def delete_workspace_users_role_binding(
             workspace_id=workspace_group_workspace_ids,
             domain_id=domain_id,
         )
-        rb_vos.delete()
+        for rb_vo in rb_vos:
+            self.rb_mgr.delete_role_binding_by_vo(rb_vo)
 
     def add_users_to_workspace_group(
         self,
@@ -534,19 +535,24 @@ def add_users_to_workspace_group(
     def add_user_name_and_state_to_users(
         self,
         workspace_group_user_ids: List[str],
-        workspace_group_vo: WorkspaceGroup,
+        workspace_group_info: Union[WorkspaceGroup, Dict[str, str]],
         domain_id: str,
     ) -> Dict[str, str]:
         """Add user's name and state to users in workspace group.
         Since the user's name and state are not in user of workspace group in database,
         we need to add user's name and state to users in the Application layer.
         Args:
             workspace_group_user_ids: 'List[str]'
-            workspace_group_vo: 'WorkspaceGroup'
+            workspace_group_info: 'Union[WorkspaceGroup, Dict[str, str]]'
             domain_id: 'str'
         Returns:
             workspace_group_dict:
         """
+        if isinstance(workspace_group_info, dict):
+            wg_users = workspace_group_info.get("users", [])
+        else:
+            wg_users = workspace_group_info.users or []
+
         user_vos = self.user_mgr.filter_users(
             user_id=workspace_group_user_ids, domain_id=domain_id
         )
@@ -557,20 +563,32 @@ def add_user_name_and_state_to_users(
                 "name": user_vo.name,
                 "state": user_vo.state,
             }
-
-        workspace_group_info = workspace_group_vo.to_dict()
-
-        if workspace_group_info.get("users", []):
-            users = []
-            for user in workspace_group_info["users"]:
-                user_id = user["user_id"]
-                user["user_name"] = user_info_map[user_id]["name"]
-                user["state"] = user_info_map[user_id]["state"]
+        if wg_users is None:
+            wg_users = []
+        users = []
+        for user in wg_users:
+            if isinstance(user, dict):
+                user_id = user.get("user_id", "")
+                user_name = user_info_map.get(user_id, {}).get("name", "")
+                user_state = user_info_map.get(user_id, {}).get("state", "")
+                user["user_name"] = user_name
+                user["state"] = user_state
                 users.append(user)
-
+            else:
+                user_id = getattr(user, "user_id", "") or ""
+                user_name = user_info_map.get(user_id, {}).get("name", "")
+                user_state = user_info_map.get(user_id, {}).get("state", "")
+                user_dict = user.to_mongo().to_dict()
+                user_dict["user_name"] = user_name
+                user_dict["state"] = user_state
+                users.append(user_dict)
+
+        if isinstance(workspace_group_info, dict):
             workspace_group_info["users"] = users
-
-        return workspace_group_info
+            return workspace_group_info
+        else:
+            workspace_group_info.users = users
+            return workspace_group_info.to_dict()
 
     def remove_users_from_workspace_group(
         self,

src/spaceone/identity/service/workspace_group_user_service.py

@@ -250,7 +250,7 @@ def update_role(
         self.workspace_group_svc.check_role_type(role_type)
 
         self.workspace_group_svc.update_user_role_of_workspace_group(
-            role_id, role_type, user_id, workspace_group_id, domain_id
+            role_id, role_type, target_user_id, workspace_group_id, domain_id
         )
 
         update_workspace_group_params = {"users": workspace_group_vo.users or []}
@@ -404,7 +404,9 @@ def list(
 
             workspace_group_dict = (
                 self.workspace_group_svc.add_user_name_and_state_to_users(
-                    workspace_group_user_ids, workspace_group_vo, params.domain_id
+                    workspace_group_user_ids,
+                    workspace_group_vo,
+                    params.domain_id,
                 )
             )
             workspace_groups_info.append(workspace_group_dict)

src/spaceone/identity/service/workspace_service.py

@@ -124,8 +124,8 @@ def change_workspace_group(
             workspace_group_vo = self.workspace_group_mgr.get_workspace_group(
                 old_workspace_group_id, domain_id
             )
-            self._remove_workspace_from_group(
-                workspace_id, old_workspace_group_id, domain_id
+            self._remove_workspace_from_group_with_workspace_vo(
+                workspace_vo, old_workspace_group_id, domain_id
             )
 
         if is_updatable:
@@ -416,7 +416,9 @@ def _add_workspace_to_group(
 
         if old_workspace_group_id:
             if old_workspace_group_id != workspace_group_id:
-                self._delete_role_bindings(old_workspace_group_id, domain_id)
+                self._delete_role_bindings(
+                    workspace_id, domain_id, old_workspace_group_id
+                )
 
                 self._create_role_bindings(
                     workspace_group_vo.users,
@@ -457,6 +459,10 @@ def _add_workspace_to_group(
             else:
                 is_updatable = False
         else:
+            workspace_group_dict = workspace_group_vo.to_dict()
+            users = workspace_group_dict.get("users", [])
+            user_ids = [user.get("user_id") for user in users]
+            self._delete_role_bindings(workspace_id, domain_id, user_ids=user_ids)
             self._create_role_bindings(
                 workspace_group_vo.users,
                 workspace_id,
@@ -472,46 +478,47 @@ def _add_workspace_to_group(
 
         return is_updatable
 
-    def _remove_workspace_from_group(
-        self, workspace_id: str, old_workspace_group_id: str, domain_id: str
+    def _remove_workspace_from_group_with_workspace_vo(
+        self, workspace_vo: Workspace, old_workspace_group_id: str, domain_id: str
     ) -> None:
-        self._delete_role_bindings(workspace_id, old_workspace_group_id, domain_id)
-
-        if old_workspace_group_id:
-            workspace_vo = self.workspace_mgr.get_workspace(
-                workspace_id=workspace_id, domain_id=domain_id
-            )
-            if workspace_vo:
-                workspace_vo.changed_at = datetime.utcnow()
-                workspace_vo.workspace_group_id = None
-
-                user_rb_ids = self.rb_mgr.stat_role_bindings(
-                    query={
-                        "distinct": "user_id",
-                        "filter": [
-                            {"k": "workspace_id", "v": workspace_id, "o": "eq"},
-                            {"k": "domain_id", "v": domain_id, "o": "eq"},
-                        ],
-                    }
-                ).get("results", [])
-                user_rb_total_count = len(user_rb_ids)
-
-                self.workspace_mgr.update_workspace_by_vo(
-                    {
-                        "user_count": user_rb_total_count,
-                        "changed_at": workspace_vo.changed_at,
-                        "workspace_group_id": None,
-                    },
-                    workspace_vo,
-                )
+        workspace_id = workspace_vo.workspace_id
+        self._delete_role_bindings(workspace_id, domain_id, old_workspace_group_id)
+
+        workspace_vo.changed_at = datetime.utcnow()
+        workspace_vo.workspace_group_id = None
+
+        user_rb_ids = self.rb_mgr.stat_role_bindings(
+            query={
+                "distinct": "user_id",
+                "filter": [
+                    {"k": "workspace_id", "v": workspace_id, "o": "eq"},
+                    {"k": "domain_id", "v": domain_id, "o": "eq"},
+                ],
+            }
+        ).get("results", [])
+        user_rb_total_count = len(user_rb_ids)
+
+        self.workspace_mgr.update_workspace_by_vo(
+            {
+                "user_count": user_rb_total_count,
+                "changed_at": workspace_vo.changed_at,
+                "workspace_group_id": None,
+            },
+            workspace_vo,
+        )
 
     def _delete_role_bindings(
-        self, workspace_id: str, existing_workspace_group_id: str, domain_id: str
+        self,
+        workspace_id: str,
+        domain_id: str,
+        existing_workspace_group_id: str = None,
+        user_ids: List[str] = None,
     ):
         rb_vos = self.rb_mgr.filter_role_bindings(
             workspace_id=workspace_id,
-            workspace_group_id=existing_workspace_group_id,
             domain_id=domain_id,
+            workspace_group_id=existing_workspace_group_id,
+            user_id=user_ids,
         )
         for rb_vo in rb_vos:
             self.rb_mgr.delete_role_binding_by_vo(rb_vo)