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
This commit is contained in:
Xiaoxia AI
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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)}"
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user