diff --git a/packages/application/workspace/__init__.py b/packages/application/workspace/__init__.py index 7fe6cca3a..89e476880 100644 --- a/packages/application/workspace/__init__.py +++ b/packages/application/workspace/__init__.py @@ -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", ] diff --git a/packages/application/workspace/update_member_role_use_case.py b/packages/application/workspace/update_member_role_use_case.py new file mode 100644 index 000000000..f7110b658 --- /dev/null +++ b/packages/application/workspace/update_member_role_use_case.py @@ -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)}" diff --git a/tests/unit/test_update_member_role_use_case.py b/tests/unit/test_update_member_role_use_case.py new file mode 100644 index 000000000..68673427d --- /dev/null +++ b/tests/unit/test_update_member_role_use_case.py @@ -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"