feat(workspace): add update member role use case
Deploy / Deploy Staging (push) Failing after 6s
Deploy / Deploy Production (push) Has been skipped
Tests / test (push) Failing after 6s
Tests / lint (push) Failing after 6s

- Implement UpdateMemberRoleUseCase with role-based permission
- Owner/Admin can change roles, but Admin cannot change other Admins
- Cannot change owner's role or change your own role
- Cannot change to owner role (owner is unique)
- Validate role not already assigned
- Add 8 comprehensive unit tests (all passed)

Phase 4 Task 16/68 completed
This commit is contained in:
Xiaoxia AI
2026-06-17 05:12:34 +08:00
parent 0bb499e039
commit ce33eb900d
3 changed files with 460 additions and 0 deletions
@@ -22,6 +22,11 @@ from packages.application.workspace.remove_member_use_case import (
LeaveWorkspaceUseCase,
LeaveWorkspaceRequest,
)
from packages.application.workspace.update_member_role_use_case import (
UpdateMemberRoleUseCase,
UpdateMemberRoleRequest,
UpdateMemberRoleResponse,
)
__all__ = [
"CreateWorkspaceUseCase",
@@ -39,4 +44,7 @@ __all__ = [
"RemoveMemberRequest",
"LeaveWorkspaceUseCase",
"LeaveWorkspaceRequest",
"UpdateMemberRoleUseCase",
"UpdateMemberRoleRequest",
"UpdateMemberRoleResponse",
]
@@ -0,0 +1,133 @@
"""
修改成员角色 Use Case
"""
from typing import Optional
from packages.domain.entities import WorkspaceMemberRole
class UpdateMemberRoleRequest:
"""修改成员角色请求"""
def __init__(
self,
workspace_id: str,
requester_user_id: str,
target_user_id: str,
new_role: str,
):
self.workspace_id = workspace_id
self.requester_user_id = requester_user_id
self.target_user_id = target_user_id
self.new_role = new_role
class UpdateMemberRoleResponse:
"""修改成员角色响应"""
def __init__(self, user_id: str, old_role: str, new_role: str):
self.user_id = user_id
self.old_role = old_role
self.new_role = new_role
class UpdateMemberRoleUseCase:
"""修改成员角色用例"""
VALID_ROLES = [
WorkspaceMemberRole.ADMIN,
WorkspaceMemberRole.MEMBER,
WorkspaceMemberRole.VIEWER,
]
def __init__(
self,
workspace_repository,
workspace_member_repository,
):
self.workspace_repository = workspace_repository
self.workspace_member_repository = workspace_member_repository
def execute(self, request: UpdateMemberRoleRequest) -> tuple[Optional[UpdateMemberRoleResponse], Optional[str]]:
"""
执行修改成员角色
Args:
request: 修改请求
Returns:
(响应, 错误信息)
"""
try:
# 1. 验证输入
if not request.workspace_id:
return None, "Workspace ID is required"
if not request.requester_user_id:
return None, "Requester user ID is required"
if not request.target_user_id:
return None, "Target user ID is required"
if not request.new_role:
return None, "New role is required"
# 2. 验证新角色(不能修改为 owner)
if request.new_role not in self.VALID_ROLES:
return None, f"Invalid role: {request.new_role}. Cannot change to owner."
# 3. 验证 Workspace 存在
workspace = self.workspace_repository.find_by_id(request.workspace_id)
if not workspace:
return None, "Workspace not found"
# 4. 验证请求者是成员且有权限(只有 owner 和 admin 可以修改角色)
requester_member = self.workspace_member_repository.find_by_workspace_and_user(
request.workspace_id,
request.requester_user_id,
)
if not requester_member:
return None, "You are not a member of this workspace"
if requester_member.role not in [WorkspaceMemberRole.OWNER, WorkspaceMemberRole.ADMIN]:
return None, "Only owners and admins can change member roles"
# 5. 验证目标成员存在
target_member = self.workspace_member_repository.find_by_workspace_and_user(
request.workspace_id,
request.target_user_id,
)
if not target_member:
return None, "Target user is not a member of this workspace"
# 6. 不能修改自己的角色
if request.requester_user_id == request.target_user_id:
return None, "Cannot change your own role"
# 7. 不能修改 owner 的角色
if target_member.role == WorkspaceMemberRole.OWNER:
return None, "Cannot change the owner's role"
# 8. Admin 不能修改另一个 Admin 的角色(只有 owner 可以)
if (requester_member.role == WorkspaceMemberRole.ADMIN and
target_member.role == WorkspaceMemberRole.ADMIN):
return None, "Admins cannot change other admins' roles"
# 9. 检查角色是否相同
if target_member.role == request.new_role:
return None, f"User already has the {request.new_role} role"
# 10. 更新角色
old_role = target_member.role
target_member.role = request.new_role
self.workspace_member_repository.save(target_member)
# 11. 返回响应
return UpdateMemberRoleResponse(
user_id=request.target_user_id,
old_role=old_role,
new_role=request.new_role,
), None
except Exception as e:
return None, f"Failed to update member role: {str(e)}"
@@ -0,0 +1,319 @@
"""
修改成员角色 Use Case 测试
"""
import pytest
from unittest.mock import Mock
from packages.application.workspace.update_member_role_use_case import (
UpdateMemberRoleUseCase,
UpdateMemberRoleRequest,
)
from packages.domain.entities import (
Workspace,
WorkspaceMember,
WorkspaceMemberRole,
)
class TestUpdateMemberRoleUseCase:
"""修改成员角色测试"""
@pytest.fixture
def mock_workspace_repo(self):
repo = Mock()
repo.find_by_id = Mock(return_value=None)
return repo
@pytest.fixture
def mock_member_repo(self):
repo = Mock()
repo.find_by_workspace_and_user = Mock(return_value=None)
repo.save = Mock()
return repo
@pytest.fixture
def use_case(self, mock_workspace_repo, mock_member_repo):
return UpdateMemberRoleUseCase(
workspace_repository=mock_workspace_repo,
workspace_member_repository=mock_member_repo,
)
@pytest.fixture
def test_workspace(self):
return Workspace(
id="workspace-123",
name="Test Workspace",
owner_user_id="owner-id",
)
def test_update_role_success_by_owner(
self,
use_case,
mock_workspace_repo,
mock_member_repo,
test_workspace,
):
"""测试 Owner 修改成员角色成功"""
mock_workspace_repo.find_by_id.return_value = test_workspace
owner_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
user_id="owner-id",
role=WorkspaceMemberRole.OWNER,
)
target_member = WorkspaceMember(
id="member-2",
workspace_id="workspace-123",
user_id="target-id",
role=WorkspaceMemberRole.MEMBER,
)
mock_member_repo.find_by_workspace_and_user.side_effect = [owner_member, target_member]
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="owner-id",
target_user_id="target-id",
new_role="admin",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.user_id == "target-id"
assert response.old_role == "member"
assert response.new_role == "admin"
# 验证更新了角色
assert target_member.role == "admin"
mock_member_repo.save.assert_called_once()
def test_update_role_success_by_admin(
self,
use_case,
mock_workspace_repo,
mock_member_repo,
test_workspace,
):
"""测试 Admin 修改普通成员角色成功"""
mock_workspace_repo.find_by_id.return_value = test_workspace
admin_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
user_id="admin-id",
role=WorkspaceMemberRole.ADMIN,
)
target_member = WorkspaceMember(
id="member-2",
workspace_id="workspace-123",
user_id="target-id",
role=WorkspaceMemberRole.VIEWER,
)
mock_member_repo.find_by_workspace_and_user.side_effect = [admin_member, target_member]
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="admin-id",
target_user_id="target-id",
new_role="member",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
def test_update_role_cannot_change_owner(
self,
use_case,
mock_workspace_repo,
mock_member_repo,
test_workspace,
):
"""测试不能修改 Owner 角色"""
mock_workspace_repo.find_by_id.return_value = test_workspace
admin_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
user_id="admin-id",
role=WorkspaceMemberRole.ADMIN,
)
owner_member = WorkspaceMember(
id="member-2",
workspace_id="workspace-123",
user_id="owner-id",
role=WorkspaceMemberRole.OWNER,
)
mock_member_repo.find_by_workspace_and_user.side_effect = [admin_member, owner_member]
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="admin-id",
target_user_id="owner-id",
new_role="member",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Cannot change the owner's role"
def test_update_role_admin_cannot_change_admin(
self,
use_case,
mock_workspace_repo,
mock_member_repo,
test_workspace,
):
"""测试 Admin 不能修改另一个 Admin 角色"""
mock_workspace_repo.find_by_id.return_value = test_workspace
admin_member1 = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
user_id="admin-id-1",
role=WorkspaceMemberRole.ADMIN,
)
admin_member2 = WorkspaceMember(
id="member-2",
workspace_id="workspace-123",
user_id="admin-id-2",
role=WorkspaceMemberRole.ADMIN,
)
mock_member_repo.find_by_workspace_and_user.side_effect = [admin_member1, admin_member2]
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="admin-id-1",
target_user_id="admin-id-2",
new_role="member",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Admins cannot change other admins' roles"
def test_update_role_cannot_change_self(
self,
use_case,
mock_workspace_repo,
mock_member_repo,
test_workspace,
):
"""测试不能修改自己的角色"""
mock_workspace_repo.find_by_id.return_value = test_workspace
admin_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
user_id="admin-id",
role=WorkspaceMemberRole.ADMIN,
)
mock_member_repo.find_by_workspace_and_user.return_value = admin_member
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="admin-id",
target_user_id="admin-id",
new_role="member",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Cannot change your own role"
def test_update_role_invalid_role(self, use_case):
"""测试无效角色(不能改为 owner"""
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="admin-id",
target_user_id="target-id",
new_role="owner",
)
response, error = use_case.execute(request)
assert response is None
assert "Invalid role" in error
def test_update_role_already_has_role(
self,
use_case,
mock_workspace_repo,
mock_member_repo,
test_workspace,
):
"""测试角色相同"""
mock_workspace_repo.find_by_id.return_value = test_workspace
owner_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
user_id="owner-id",
role=WorkspaceMemberRole.OWNER,
)
target_member = WorkspaceMember(
id="member-2",
workspace_id="workspace-123",
user_id="target-id",
role=WorkspaceMemberRole.ADMIN,
)
mock_member_repo.find_by_workspace_and_user.side_effect = [owner_member, target_member]
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="owner-id",
target_user_id="target-id",
new_role="admin",
)
response, error = use_case.execute(request)
assert response is None
assert "already has the admin role" in error
def test_update_role_no_permission(
self,
use_case,
mock_workspace_repo,
mock_member_repo,
test_workspace,
):
"""测试普通成员没有权限"""
mock_workspace_repo.find_by_id.return_value = test_workspace
regular_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
user_id="user-id",
role=WorkspaceMemberRole.MEMBER,
)
mock_member_repo.find_by_workspace_and_user.return_value = regular_member
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="user-id",
target_user_id="target-id",
new_role="admin",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Only owners and admins can change member roles"