From 8ff42a1bf9fcfc6a6423c96d0a7b9250321a7bbf Mon Sep 17 00:00:00 2001 From: Xiaoxia AI Date: Wed, 17 Jun 2026 05:10:06 +0800 Subject: [PATCH] feat(workspace): add remove member and leave workspace use cases - Implement RemoveMemberUseCase with role-based permission - Owner/Admin can remove members, but Admin cannot remove other Admins - Cannot remove workspace owner or remove yourself - Implement LeaveWorkspaceUseCase for self-removal - Owner cannot leave (must transfer ownership or delete workspace first) - Add 9 comprehensive unit tests (all passed) Phase 4 Task 15/68 completed --- packages/application/workspace/__init__.py | 10 + .../workspace/remove_member_use_case.py | 165 ++++++++ tests/unit/test_remove_member_use_case.py | 367 ++++++++++++++++++ 3 files changed, 542 insertions(+) create mode 100644 packages/application/workspace/remove_member_use_case.py create mode 100644 tests/unit/test_remove_member_use_case.py diff --git a/packages/application/workspace/__init__.py b/packages/application/workspace/__init__.py index 0c710b9d0..7fe6cca3a 100644 --- a/packages/application/workspace/__init__.py +++ b/packages/application/workspace/__init__.py @@ -16,6 +16,12 @@ from packages.application.workspace.accept_invitation_use_case import ( DeclineInvitationUseCase, DeclineInvitationRequest, ) +from packages.application.workspace.remove_member_use_case import ( + RemoveMemberUseCase, + RemoveMemberRequest, + LeaveWorkspaceUseCase, + LeaveWorkspaceRequest, +) __all__ = [ "CreateWorkspaceUseCase", @@ -29,4 +35,8 @@ __all__ = [ "AcceptInvitationResponse", "DeclineInvitationUseCase", "DeclineInvitationRequest", + "RemoveMemberUseCase", + "RemoveMemberRequest", + "LeaveWorkspaceUseCase", + "LeaveWorkspaceRequest", ] diff --git a/packages/application/workspace/remove_member_use_case.py b/packages/application/workspace/remove_member_use_case.py new file mode 100644 index 000000000..d23d09192 --- /dev/null +++ b/packages/application/workspace/remove_member_use_case.py @@ -0,0 +1,165 @@ +""" +移除成员 Use Case +""" +from typing import Optional + +from packages.domain.entities import WorkspaceMemberRole + + +class RemoveMemberRequest: + """移除成员请求""" + + def __init__( + self, + workspace_id: str, + requester_user_id: str, + target_user_id: str, + ): + self.workspace_id = workspace_id + self.requester_user_id = requester_user_id + self.target_user_id = target_user_id + + +class RemoveMemberUseCase: + """移除成员用例""" + + def __init__( + self, + workspace_repository, + workspace_member_repository, + ): + self.workspace_repository = workspace_repository + self.workspace_member_repository = workspace_member_repository + + def execute(self, request: RemoveMemberRequest) -> tuple[bool, Optional[str]]: + """ + 执行移除成员 + + Args: + request: 移除请求 + + Returns: + (是否成功, 错误信息) + """ + try: + # 1. 验证输入 + if not request.workspace_id: + return False, "Workspace ID is required" + + if not request.requester_user_id: + return False, "Requester user ID is required" + + if not request.target_user_id: + return False, "Target user ID is required" + + # 2. 验证 Workspace 存在 + workspace = self.workspace_repository.find_by_id(request.workspace_id) + if not workspace: + return False, "Workspace not found" + + # 3. 验证请求者是成员且有权限 + requester_member = self.workspace_member_repository.find_by_workspace_and_user( + request.workspace_id, + request.requester_user_id, + ) + if not requester_member: + return False, "You are not a member of this workspace" + + if requester_member.role not in [WorkspaceMemberRole.OWNER, WorkspaceMemberRole.ADMIN]: + return False, "Only owners and admins can remove members" + + # 4. 验证目标成员存在 + target_member = self.workspace_member_repository.find_by_workspace_and_user( + request.workspace_id, + request.target_user_id, + ) + if not target_member: + return False, "Target user is not a member of this workspace" + + # 5. 不能移除自己（应该用离开 workspace 的功能） + if request.requester_user_id == request.target_user_id: + return False, "Cannot remove yourself. Use leave workspace instead." + + # 6. 不能移除 owner + if target_member.role == WorkspaceMemberRole.OWNER: + return False, "Cannot remove the workspace owner" + + # 7. Admin 不能移除另一个 Admin（只有 owner 可以） + if (requester_member.role == WorkspaceMemberRole.ADMIN and + target_member.role == WorkspaceMemberRole.ADMIN): + return False, "Admins cannot remove other admins" + + # 8. 删除成员记录 + success = self.workspace_member_repository.delete(target_member.id) + if not success: + return False, "Failed to remove member" + + return True, None + + except Exception as e: + return False, f"Failed to remove member: {str(e)}" + + +class LeaveWorkspaceRequest: + """离开 Workspace 请求""" + + def __init__(self, workspace_id: str, user_id: str): + self.workspace_id = workspace_id + self.user_id = user_id + + +class LeaveWorkspaceUseCase: + """离开 Workspace 用例""" + + def __init__( + self, + workspace_repository, + workspace_member_repository, + ): + self.workspace_repository = workspace_repository + self.workspace_member_repository = workspace_member_repository + + def execute(self, request: LeaveWorkspaceRequest) -> tuple[bool, Optional[str]]: + """ + 执行离开 Workspace + + Args: + request: 离开请求 + + Returns: + (是否成功, 错误信息) + """ + try: + # 1. 验证输入 + if not request.workspace_id: + return False, "Workspace ID is required" + + if not request.user_id: + return False, "User ID is required" + + # 2. 验证 Workspace 存在 + workspace = self.workspace_repository.find_by_id(request.workspace_id) + if not workspace: + return False, "Workspace not found" + + # 3. 验证用户是成员 + member = self.workspace_member_repository.find_by_workspace_and_user( + request.workspace_id, + request.user_id, + ) + if not member: + return False, "You are not a member of this workspace" + + # 4. Owner 不能离开（需要先转移 ownership 或删除 workspace） + if member.role == WorkspaceMemberRole.OWNER: + return False, "Owner cannot leave workspace. Transfer ownership or delete workspace first." + + # 5. 删除成员记录 + success = self.workspace_member_repository.delete(member.id) + if not success: + return False, "Failed to leave workspace" + + return True, None + + except Exception as e: + return False, f"Failed to leave workspace: {str(e)}" diff --git a/tests/unit/test_remove_member_use_case.py b/tests/unit/test_remove_member_use_case.py new file mode 100644 index 000000000..fed2cb084 --- /dev/null +++ b/tests/unit/test_remove_member_use_case.py @@ -0,0 +1,367 @@ +""" +移除成员 Use Case 测试 +""" +import pytest +from unittest.mock import Mock +from packages.application.workspace.remove_member_use_case import ( + RemoveMemberUseCase, + RemoveMemberRequest, + LeaveWorkspaceUseCase, + LeaveWorkspaceRequest, +) +from packages.domain.entities import ( + Workspace, + WorkspaceMember, + WorkspaceMemberRole, +) + + +class TestRemoveMemberUseCase: + """移除成员测试""" + + @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.delete = Mock(return_value=True) + return repo + + @pytest.fixture + def use_case(self, mock_workspace_repo, mock_member_repo): + return RemoveMemberUseCase( + 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_remove_member_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 = RemoveMemberRequest( + workspace_id="workspace-123", + requester_user_id="owner-id", + target_user_id="target-id", + ) + + success, error = use_case.execute(request) + + assert success is True + assert error is None + mock_member_repo.delete.assert_called_once_with("member-2") + + def test_remove_member_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 = RemoveMemberRequest( + workspace_id="workspace-123", + requester_user_id="admin-id", + target_user_id="target-id", + ) + + success, error = use_case.execute(request) + + assert success is True + assert error is None + + def test_remove_member_cannot_remove_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 = RemoveMemberRequest( + workspace_id="workspace-123", + requester_user_id="admin-id", + target_user_id="owner-id", + ) + + success, error = use_case.execute(request) + + assert success is False + assert error == "Cannot remove the workspace owner" + + def test_remove_member_admin_cannot_remove_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 = RemoveMemberRequest( + workspace_id="workspace-123", + requester_user_id="admin-id-1", + target_user_id="admin-id-2", + ) + + success, error = use_case.execute(request) + + assert success is False + assert error == "Admins cannot remove other admins" + + def test_remove_member_cannot_remove_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 = RemoveMemberRequest( + workspace_id="workspace-123", + requester_user_id="admin-id", + target_user_id="admin-id", + ) + + success, error = use_case.execute(request) + + assert success is False + assert "Cannot remove yourself" in error + + def test_remove_member_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 = RemoveMemberRequest( + workspace_id="workspace-123", + requester_user_id="user-id", + target_user_id="target-id", + ) + + success, error = use_case.execute(request) + + assert success is False + assert error == "Only owners and admins can remove members" + + +class TestLeaveWorkspaceUseCase: + """离开 Workspace 测试""" + + @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.delete = Mock(return_value=True) + return repo + + @pytest.fixture + def use_case(self, mock_workspace_repo, mock_member_repo): + return LeaveWorkspaceUseCase( + 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_leave_workspace_success( + self, + use_case, + mock_workspace_repo, + mock_member_repo, + test_workspace, + ): + """测试离开工作空间成功""" + mock_workspace_repo.find_by_id.return_value = test_workspace + + 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 = member + + request = LeaveWorkspaceRequest( + workspace_id="workspace-123", + user_id="user-id", + ) + + success, error = use_case.execute(request) + + assert success is True + assert error is None + mock_member_repo.delete.assert_called_once_with("member-1") + + def test_leave_workspace_owner_cannot_leave( + 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, + ) + mock_member_repo.find_by_workspace_and_user.return_value = owner_member + + request = LeaveWorkspaceRequest( + workspace_id="workspace-123", + user_id="owner-id", + ) + + success, error = use_case.execute(request) + + assert success is False + assert "Owner cannot leave workspace" in error + + def test_leave_workspace_not_member( + self, + use_case, + mock_workspace_repo, + mock_member_repo, + test_workspace, + ): + """测试不是成员""" + mock_workspace_repo.find_by_id.return_value = test_workspace + mock_member_repo.find_by_workspace_and_user.return_value = None + + request = LeaveWorkspaceRequest( + workspace_id="workspace-123", + user_id="user-id", + ) + + success, error = use_case.execute(request) + + assert success is False + assert error == "You are not a member of this workspace"