feat(workspace): add list members 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 5s

- Implement ListMembersUseCase to get all workspace members
- Return member info with user details (username, email, display_name)
- Show role, inviter, and join time for each member
- Only workspace members can view member list
- Add 6 comprehensive unit tests (all passed)

Phase 4 Task 18/68 completed
This commit is contained in:
Xiaoxia AI
2026-06-17 07:06:33 +08:00
parent 6f3c881322
commit a040f33d6a
3 changed files with 377 additions and 0 deletions
@@ -36,6 +36,12 @@ from packages.application.workspace.list_workspaces_use_case import (
WorkspaceInfo,
WorkspaceDetailInfo,
)
from packages.application.workspace.list_members_use_case import (
ListMembersUseCase,
ListMembersRequest,
ListMembersResponse,
MemberInfo,
)
__all__ = [
"CreateWorkspaceUseCase",
@@ -63,4 +69,8 @@ __all__ = [
"GetWorkspaceDetailRequest",
"WorkspaceInfo",
"WorkspaceDetailInfo",
"ListMembersUseCase",
"ListMembersRequest",
"ListMembersResponse",
"MemberInfo",
]
@@ -0,0 +1,117 @@
"""
获取成员列表 Use Case
"""
from typing import Optional, List
from datetime import datetime
class MemberInfo:
"""成员信息"""
def __init__(
self,
member_id: str,
user_id: str,
username: str,
email: str,
display_name: str,
role: str,
invited_by: Optional[str],
joined_at: datetime,
):
self.member_id = member_id
self.user_id = user_id
self.username = username
self.email = email
self.display_name = display_name
self.role = role
self.invited_by = invited_by
self.joined_at = joined_at
class ListMembersRequest:
"""获取成员列表请求"""
def __init__(self, workspace_id: str, requester_user_id: str):
self.workspace_id = workspace_id
self.requester_user_id = requester_user_id
class ListMembersResponse:
"""获取成员列表响应"""
def __init__(self, members: List[MemberInfo]):
self.members = members
class ListMembersUseCase:
"""获取成员列表用例"""
def __init__(
self,
workspace_repository,
workspace_member_repository,
user_repository,
):
self.workspace_repository = workspace_repository
self.workspace_member_repository = workspace_member_repository
self.user_repository = user_repository
def execute(self, request: ListMembersRequest) -> tuple[Optional[ListMembersResponse], 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"
# 2. 验证工作空间存在
workspace = self.workspace_repository.find_by_id(request.workspace_id)
if not workspace:
return None, "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 None, "You are not a member of this workspace"
# 4. 获取所有成员
members = self.workspace_member_repository.find_by_workspace(request.workspace_id)
# 5. 获取每个成员的用户信息
member_infos = []
for member in members:
user = self.user_repository.find_by_id(member.user_id)
if not user:
continue # 跳过不存在的用户
member_info = MemberInfo(
member_id=member.id,
user_id=user.id,
username=user.username,
email=user.email,
display_name=user.display_name,
role=member.role,
invited_by=member.invited_by,
joined_at=member.joined_at,
)
member_infos.append(member_info)
# 6. 返回响应
return ListMembersResponse(members=member_infos), None
except Exception as e:
return None, f"Failed to list members: {str(e)}"
+250
View File
@@ -0,0 +1,250 @@
"""
获取成员列表 Use Case 测试
"""
import pytest
from unittest.mock import Mock
from datetime import datetime, timezone
from packages.application.workspace.list_members_use_case import (
ListMembersUseCase,
ListMembersRequest,
)
from packages.domain.entities import (
Workspace,
WorkspaceMember,
WorkspaceMemberRole,
User,
)
class TestListMembersUseCase:
"""获取成员列表测试"""
@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.find_by_workspace = Mock(return_value=[])
return repo
@pytest.fixture
def mock_user_repo(self):
repo = Mock()
repo.find_by_id = Mock(return_value=None)
return repo
@pytest.fixture
def use_case(self, mock_workspace_repo, mock_member_repo, mock_user_repo):
return ListMembersUseCase(
workspace_repository=mock_workspace_repo,
workspace_member_repository=mock_member_repo,
user_repository=mock_user_repo,
)
@pytest.fixture
def test_workspace(self):
return Workspace(
id="workspace-123",
name="Test Workspace",
owner_user_id="owner-id",
)
def test_list_members_success(
self,
use_case,
mock_workspace_repo,
mock_member_repo,
mock_user_repo,
test_workspace,
):
"""测试获取成员列表成功"""
mock_workspace_repo.find_by_id.return_value = test_workspace
# 请求者是 Admin
requester_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 = requester_member
# 3 个成员
member1 = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
user_id="owner-id",
role=WorkspaceMemberRole.OWNER,
invited_by=None,
)
member2 = WorkspaceMember(
id="member-2",
workspace_id="workspace-123",
user_id="admin-id",
role=WorkspaceMemberRole.ADMIN,
invited_by="owner-id",
)
member3 = WorkspaceMember(
id="member-3",
workspace_id="workspace-123",
user_id="user-id",
role=WorkspaceMemberRole.MEMBER,
invited_by="admin-id",
)
mock_member_repo.find_by_workspace.return_value = [member1, member2, member3]
# 用户信息
user1 = User(
id="owner-id",
email="owner@test.com",
username="owner",
display_name="Owner User",
)
user2 = User(
id="admin-id",
email="admin@test.com",
username="admin",
display_name="Admin User",
)
user3 = User(
id="user-id",
email="user@test.com",
username="user",
display_name="Regular User",
)
mock_user_repo.find_by_id.side_effect = [user1, user2, user3]
request = ListMembersRequest(
workspace_id="workspace-123",
requester_user_id="admin-id",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert len(response.members) == 3
# 验证第一个成员(Owner
m1 = response.members[0]
assert m1.user_id == "owner-id"
assert m1.username == "owner"
assert m1.email == "owner@test.com"
assert m1.role == "owner"
assert m1.invited_by is None
# 验证第二个成员(Admin
m2 = response.members[1]
assert m2.user_id == "admin-id"
assert m2.role == "admin"
assert m2.invited_by == "owner-id"
# 验证第三个成员(Member
m3 = response.members[2]
assert m3.user_id == "user-id"
assert m3.role == "member"
assert m3.invited_by == "admin-id"
def test_list_members_workspace_not_found(
self,
use_case,
mock_workspace_repo,
):
"""测试工作空间不存在"""
mock_workspace_repo.find_by_id.return_value = None
request = ListMembersRequest(
workspace_id="nonexistent",
requester_user_id="user-id",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Workspace not found"
def test_list_members_requester_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 = ListMembersRequest(
workspace_id="workspace-123",
requester_user_id="outsider-id",
)
response, error = use_case.execute(request)
assert response is None
assert error == "You are not a member of this workspace"
def test_list_members_empty_workspace(
self,
use_case,
mock_workspace_repo,
mock_member_repo,
test_workspace,
):
"""测试空工作空间(理论上不应该发生)"""
mock_workspace_repo.find_by_id.return_value = test_workspace
requester_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 = requester_member
mock_member_repo.find_by_workspace.return_value = []
request = ListMembersRequest(
workspace_id="workspace-123",
requester_user_id="user-id",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert len(response.members) == 0
def test_list_members_missing_workspace_id(self, use_case):
"""测试缺少工作空间 ID"""
request = ListMembersRequest(
workspace_id="",
requester_user_id="user-id",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Workspace ID is required"
def test_list_members_missing_requester_id(self, use_case):
"""测试缺少请求者 ID"""
request = ListMembersRequest(
workspace_id="workspace-123",
requester_user_id="",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Requester user ID is required"