fix: fix bugs in WorkspaceUser.stat API

cloudforet-io · Dec 31, 2023 · 47949ed · 47949ed
1 parent 84a507e
commit 47949ed
Show file tree

Hide file tree

Showing 3 changed files with 35 additions and 36 deletions.
diff --git a/src/spaceone/identity/interface/grpc/workspace_user.py b/src/spaceone/identity/interface/grpc/workspace_user.py
@@ -34,5 +34,5 @@ def list(self, request, context):
     def stat(self, request, context):
         params, metadata = self.parse_request(request, context)
         workspace_user_svc = WorkspaceUserService(metadata)
-        response: dict = workspace_user_svc.list(params)
+        response: dict = workspace_user_svc.stat(params)
         return self.dict_to_message(response)
diff --git a/src/spaceone/identity/manager/workspace_user_manager.py b/src/spaceone/identity/manager/workspace_user_manager.py
@@ -40,24 +40,9 @@ def create_workspace_user(self, params: dict) -> User:
 
         return user_vo
 
-    def get_workspace_user(
-        self, user_id: str, workspace_id: str, domain_id: str
-    ) -> dict:
-        user_ids, user_rb_map = self._get_role_binding_map_in_workspace(
-            workspace_id, domain_id
-        )
-        if user_id not in user_ids:
-            raise ERROR_NOT_FOUND(key="user_id", value=user_id)
-
-        user_vo = self.user_mgr.get_user(user_id, domain_id)
-        user_info = user_vo.to_dict()
-        user_info["role_binding_info"] = user_rb_map[user_id]
-
-        return user_info
-
     def check_workspace_users(
         self, users: List[str], workspace_id: str, domain_id: str
-    ) -> bool:
+    ) -> None:
         rb_vos = self.rb_mgr.filter_role_bindings(
             user_id=users,
             workspace_id=[workspace_id, "*"],
@@ -73,17 +58,29 @@ def check_workspace_users(
                 user_id=user_id, workspace_id=workspace_id
             )
 
+    def get_workspace_user(
+        self, user_id: str, workspace_id: str, domain_id: str
+    ) -> dict:
+        user_ids, user_rb_map = self._get_role_binding_map_in_workspace(
+            workspace_id, domain_id
+        )
+        if user_id not in user_ids:
+            raise ERROR_NOT_FOUND(key="user_id", value=user_id)
+
+        user_vo = self.user_mgr.get_user(user_id, domain_id)
+        user_info = user_vo.to_dict()
+        user_info["role_binding_info"] = user_rb_map[user_id]
+
+        return user_info
+
     def list_workspace_users(
-        self, query: dict, domain_id: str, workspace_id: str = None
+        self, query: dict, domain_id: str, workspace_id: str
     ) -> Tuple[list, int]:
-        user_rb_map = None
-
-        if workspace_id:
-            user_ids, user_rb_map = self._get_role_binding_map_in_workspace(
-                workspace_id, domain_id
-            )
-            query["filter"] = query.get("filter", [])
-            query["filter"].append({"k": "user_id", "v": user_ids, "o": "in"})
+        user_ids, user_rb_map = self._get_role_binding_map_in_workspace(
+            workspace_id, domain_id
+        )
+        query["filter"] = query.get("filter", [])
+        query["filter"].append({"k": "user_id", "v": user_ids, "o": "in"})
 
         user_vos, total_count = self.user_mgr.list_users(query)
 
@@ -99,16 +96,15 @@ def list_workspace_users(
         return users_info, total_count
 
     def stat_workspace_users(
-        self, query: dict, domain_id: str, workspace_id: str = None
+        self, query: dict, domain_id: str, workspace_id: str
     ) -> dict:
-        if workspace_id:
-            user_ids, user_rb_map = self._get_role_binding_map_in_workspace(
-                workspace_id, domain_id
-            )
-            query["filter"] = query.get("filter", [])
-            query["filter"].append({"k": "user_id", "v": user_ids, "o": "in"})
+        user_ids, user_rb_map = self._get_role_binding_map_in_workspace(
+            workspace_id, domain_id
+        )
+        query["filter"] = query.get("filter", [])
+        query["filter"].append({"k": "user_id", "v": user_ids, "o": "in"})
 
-        return self.user_mgr.stat(**query)
+        return self.user_mgr.stat_users(query)
 
     def _get_role_binding_map_in_workspace(
         self, workspace_id: str, domain_id: str

diff --git a/src/spaceone/identity/service/workspace_user_service.py b/src/spaceone/identity/service/workspace_user_service.py
@@ -82,6 +82,7 @@ def find(
         )
         workspace_user_ids = list(set([rb.user_id for rb in rb_vos]))
 
+        # default query
         query = {
             "filter": [
                 {"k": "domain_id", "v": params.domain_id, "o": "eq"},
@@ -92,12 +93,14 @@ def find(
             "only": ["user_id", "name", "state"],
         }
 
+        # append keyword filter
         if params.keyword:
             query["filter_or"] = [
                 {"k": "user_id", "v": params.keyword, "o": "contain"},
                 {"k": "name", "v": params.keyword, "o": "contain"},
             ]
 
+        # append state filter
         if params.state:
             query["filter"].append({"k": "state", "v": params.state, "o": "eq"})
 
@@ -169,9 +172,9 @@ def list(
 
     @transaction(
         permission="identity:WorkspaceUser.read",
-        role_types=["WORKSPACE_OWNER", "WORKSPACE_MEMBER"],
+        role_types=["DOMAIN", "WORKSPACE_OWNER", "WORKSPACE_MEMBER"],
     )
-    @append_query_filter(["domain_id", "workspace_id"])
+    @append_query_filter(["domain_id"])
     @convert_model
     def stat(self, params: WorkspaceUserStatQueryRequest) -> dict:
         """stat users in workspace