From ee3935abdcd457e33f52fdf2e0e5c51a0bce34ab Mon Sep 17 00:00:00 2001 From: Xiaoxia AI Date: Wed, 17 Jun 2026 07:04:55 +0800 Subject: [PATCH] feat(workspace): add list workspaces and get detail use cases - Implement ListWorkspacesUseCase to get user's all workspaces - Return workspace info with user's role and member count - Implement GetWorkspaceDetailUseCase with permission check - Show detailed subscription and storage info - Verify user is member before showing details - Add 8 comprehensive unit tests (all passed) Phase 4 Task 17/68 completed --- packages/application/workspace/__init__.py | 16 ++ .../workspace/list_workspaces_use_case.py | 217 ++++++++++++++ tests/unit/test_list_workspaces_use_case.py | 266 ++++++++++++++++++ 3 files changed, 499 insertions(+) create mode 100644 packages/application/workspace/list_workspaces_use_case.py create mode 100644 tests/unit/test_list_workspaces_use_case.py diff --git a/packages/application/workspace/__init__.py b/packages/application/workspace/__init__.py index 89e476880..f63a37a2b 100644 --- a/packages/application/workspace/__init__.py +++ b/packages/application/workspace/__init__.py @@ -27,6 +27,15 @@ from packages.application.workspace.update_member_role_use_case import ( UpdateMemberRoleRequest, UpdateMemberRoleResponse, ) +from packages.application.workspace.list_workspaces_use_case import ( + ListWorkspacesUseCase, + ListWorkspacesRequest, + ListWorkspacesResponse, + GetWorkspaceDetailUseCase, + GetWorkspaceDetailRequest, + WorkspaceInfo, + WorkspaceDetailInfo, +) __all__ = [ "CreateWorkspaceUseCase", @@ -47,4 +56,11 @@ __all__ = [ "UpdateMemberRoleUseCase", "UpdateMemberRoleRequest", "UpdateMemberRoleResponse", + "ListWorkspacesUseCase", + "ListWorkspacesRequest", + "ListWorkspacesResponse", + "GetWorkspaceDetailUseCase", + "GetWorkspaceDetailRequest", + "WorkspaceInfo", + "WorkspaceDetailInfo", ] diff --git a/packages/application/workspace/list_workspaces_use_case.py b/packages/application/workspace/list_workspaces_use_case.py new file mode 100644 index 000000000..048f89117 --- /dev/null +++ b/packages/application/workspace/list_workspaces_use_case.py @@ -0,0 +1,217 @@ +""" +获取工作空间列表和详情 Use Case +""" +from typing import Optional, List +from datetime import datetime + + +class WorkspaceInfo: + """工作空间信息""" + + def __init__( + self, + workspace_id: str, + name: str, + owner_user_id: str, + subscription_plan: str, + subscription_status: str, + max_projects: int, + max_storage_gb: int, + used_storage_gb: float, + member_count: int, + user_role: str, + created_at: datetime, + ): + self.workspace_id = workspace_id + self.name = name + self.owner_user_id = owner_user_id + self.subscription_plan = subscription_plan + self.subscription_status = subscription_status + self.max_projects = max_projects + self.max_storage_gb = max_storage_gb + self.used_storage_gb = used_storage_gb + self.member_count = member_count + self.user_role = user_role + self.created_at = created_at + + +class ListWorkspacesRequest: + """获取工作空间列表请求""" + + def __init__(self, user_id: str): + self.user_id = user_id + + +class ListWorkspacesResponse: + """获取工作空间列表响应""" + + def __init__(self, workspaces: List[WorkspaceInfo]): + self.workspaces = workspaces + + +class ListWorkspacesUseCase: + """获取工作空间列表用例""" + + def __init__( + self, + workspace_repository, + workspace_member_repository, + ): + self.workspace_repository = workspace_repository + self.workspace_member_repository = workspace_member_repository + + def execute(self, request: ListWorkspacesRequest) -> tuple[Optional[ListWorkspacesResponse], Optional[str]]: + """ + 执行获取工作空间列表 + + Args: + request: 请求 + + Returns: + (响应, 错误信息) + """ + try: + # 1. 验证输入 + if not request.user_id: + return None, "User ID is required" + + # 2. 获取用户所有的成员记录 + memberships = self.workspace_member_repository.find_by_user(request.user_id) + + # 3. 获取每个工作空间的信息 + workspace_infos = [] + for membership in memberships: + workspace = self.workspace_repository.find_by_id(membership.workspace_id) + if not workspace: + continue + + # 获取成员数量 + member_count = self.workspace_member_repository.count_by_workspace(membership.workspace_id) + + workspace_info = WorkspaceInfo( + workspace_id=workspace.id, + name=workspace.name, + owner_user_id=workspace.owner_user_id, + subscription_plan=workspace.subscription_plan, + subscription_status=workspace.subscription_status, + max_projects=workspace.max_projects, + max_storage_gb=workspace.max_storage_gb, + used_storage_gb=workspace.used_storage_gb, + member_count=member_count, + user_role=membership.role, + created_at=workspace.created_at, + ) + workspace_infos.append(workspace_info) + + # 4. 返回响应 + return ListWorkspacesResponse(workspaces=workspace_infos), None + + except Exception as e: + return None, f"Failed to list workspaces: {str(e)}" + + +class GetWorkspaceDetailRequest: + """获取工作空间详情请求""" + + def __init__(self, workspace_id: str, user_id: str): + self.workspace_id = workspace_id + self.user_id = user_id + + +class WorkspaceDetailInfo: + """工作空间详情信息""" + + def __init__( + self, + workspace_id: str, + name: str, + owner_user_id: str, + subscription_plan: str, + subscription_status: str, + subscription_expires_at: Optional[datetime], + max_projects: int, + max_storage_gb: int, + used_storage_gb: float, + member_count: int, + user_role: str, + created_at: datetime, + ): + self.workspace_id = workspace_id + self.name = name + self.owner_user_id = owner_user_id + self.subscription_plan = subscription_plan + self.subscription_status = subscription_status + self.subscription_expires_at = subscription_expires_at + self.max_projects = max_projects + self.max_storage_gb = max_storage_gb + self.used_storage_gb = used_storage_gb + self.member_count = member_count + self.user_role = user_role + self.created_at = created_at + + +class GetWorkspaceDetailUseCase: + """获取工作空间详情用例""" + + def __init__( + self, + workspace_repository, + workspace_member_repository, + ): + self.workspace_repository = workspace_repository + self.workspace_member_repository = workspace_member_repository + + def execute(self, request: GetWorkspaceDetailRequest) -> tuple[Optional[WorkspaceDetailInfo], Optional[str]]: + """ + 执行获取工作空间详情 + + Args: + request: 请求 + + Returns: + (详情信息, 错误信息) + """ + try: + # 1. 验证输入 + if not request.workspace_id: + return None, "Workspace ID is required" + + if not request.user_id: + return None, "User ID is required" + + # 2. 验证工作空间存在 + workspace = self.workspace_repository.find_by_id(request.workspace_id) + if not workspace: + return None, "Workspace not found" + + # 3. 验证用户是成员 + membership = self.workspace_member_repository.find_by_workspace_and_user( + request.workspace_id, + request.user_id, + ) + if not membership: + return None, "You are not a member of this workspace" + + # 4. 获取成员数量 + member_count = self.workspace_member_repository.count_by_workspace(request.workspace_id) + + # 5. 构建详情信息 + detail_info = WorkspaceDetailInfo( + workspace_id=workspace.id, + name=workspace.name, + owner_user_id=workspace.owner_user_id, + subscription_plan=workspace.subscription_plan, + subscription_status=workspace.subscription_status, + subscription_expires_at=workspace.subscription_expires_at, + max_projects=workspace.max_projects, + max_storage_gb=workspace.max_storage_gb, + used_storage_gb=workspace.used_storage_gb, + member_count=member_count, + user_role=membership.role, + created_at=workspace.created_at, + ) + + return detail_info, None + + except Exception as e: + return None, f"Failed to get workspace detail: {str(e)}" diff --git a/tests/unit/test_list_workspaces_use_case.py b/tests/unit/test_list_workspaces_use_case.py new file mode 100644 index 000000000..ccb2090e3 --- /dev/null +++ b/tests/unit/test_list_workspaces_use_case.py @@ -0,0 +1,266 @@ +""" +获取工作空间列表和详情 Use Case 测试 +""" +import pytest +from unittest.mock import Mock +from datetime import datetime, timezone +from packages.application.workspace.list_workspaces_use_case import ( + ListWorkspacesUseCase, + ListWorkspacesRequest, + GetWorkspaceDetailUseCase, + GetWorkspaceDetailRequest, +) +from packages.domain.entities import ( + Workspace, + WorkspaceMember, + WorkspaceMemberRole, +) + + +class TestListWorkspacesUseCase: + """获取工作空间列表测试""" + + @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_user = Mock(return_value=[]) + repo.count_by_workspace = Mock(return_value=0) + return repo + + @pytest.fixture + def use_case(self, mock_workspace_repo, mock_member_repo): + return ListWorkspacesUseCase( + workspace_repository=mock_workspace_repo, + workspace_member_repository=mock_member_repo, + ) + + def test_list_workspaces_success( + self, + use_case, + mock_workspace_repo, + mock_member_repo, + ): + """测试获取工作空间列表成功""" + # 用户是 2 个工作空间的成员 + membership1 = WorkspaceMember( + id="member-1", + workspace_id="workspace-1", + user_id="user-123", + role=WorkspaceMemberRole.OWNER, + ) + + membership2 = WorkspaceMember( + id="member-2", + workspace_id="workspace-2", + user_id="user-123", + role=WorkspaceMemberRole.MEMBER, + ) + + mock_member_repo.find_by_user.return_value = [membership1, membership2] + + workspace1 = Workspace( + id="workspace-1", + name="My Workspace", + owner_user_id="user-123", + subscription_plan="free", + ) + + workspace2 = Workspace( + id="workspace-2", + name="Team Workspace", + owner_user_id="other-user", + subscription_plan="pro", + ) + + mock_workspace_repo.find_by_id.side_effect = [workspace1, workspace2] + mock_member_repo.count_by_workspace.side_effect = [1, 5] + + request = ListWorkspacesRequest(user_id="user-123") + response, error = use_case.execute(request) + + assert error is None + assert response is not None + assert len(response.workspaces) == 2 + + # 验证第一个工作空间 + ws1 = response.workspaces[0] + assert ws1.workspace_id == "workspace-1" + assert ws1.name == "My Workspace" + assert ws1.user_role == "owner" + assert ws1.member_count == 1 + assert ws1.subscription_plan == "free" + + # 验证第二个工作空间 + ws2 = response.workspaces[1] + assert ws2.workspace_id == "workspace-2" + assert ws2.name == "Team Workspace" + assert ws2.user_role == "member" + assert ws2.member_count == 5 + assert ws2.subscription_plan == "pro" + + def test_list_workspaces_no_memberships( + self, + use_case, + mock_member_repo, + ): + """测试用户没有加入任何工作空间""" + mock_member_repo.find_by_user.return_value = [] + + request = ListWorkspacesRequest(user_id="user-123") + response, error = use_case.execute(request) + + assert error is None + assert response is not None + assert len(response.workspaces) == 0 + + def test_list_workspaces_missing_user_id(self, use_case): + """测试缺少用户 ID""" + request = ListWorkspacesRequest(user_id="") + response, error = use_case.execute(request) + + assert response is None + assert error == "User ID is required" + + +class TestGetWorkspaceDetailUseCase: + """获取工作空间详情测试""" + + @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.count_by_workspace = Mock(return_value=0) + return repo + + @pytest.fixture + def use_case(self, mock_workspace_repo, mock_member_repo): + return GetWorkspaceDetailUseCase( + 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", + subscription_plan="pro", + subscription_status="active", + max_projects=999999, + max_storage_gb=100, + used_storage_gb=25.5, + ) + + def test_get_workspace_detail_success( + self, + use_case, + mock_workspace_repo, + mock_member_repo, + test_workspace, + ): + """测试获取工作空间详情成功""" + mock_workspace_repo.find_by_id.return_value = test_workspace + + membership = WorkspaceMember( + id="member-1", + workspace_id="workspace-123", + user_id="user-123", + role=WorkspaceMemberRole.ADMIN, + ) + mock_member_repo.find_by_workspace_and_user.return_value = membership + mock_member_repo.count_by_workspace.return_value = 8 + + request = GetWorkspaceDetailRequest( + workspace_id="workspace-123", + user_id="user-123", + ) + + detail, error = use_case.execute(request) + + assert error is None + assert detail is not None + assert detail.workspace_id == "workspace-123" + assert detail.name == "Test Workspace" + assert detail.subscription_plan == "pro" + assert detail.subscription_status == "active" + assert detail.max_projects == 999999 + assert detail.max_storage_gb == 100 + assert detail.used_storage_gb == 25.5 + assert detail.member_count == 8 + assert detail.user_role == "admin" + + def test_get_workspace_detail_not_found( + self, + use_case, + mock_workspace_repo, + ): + """测试工作空间不存在""" + mock_workspace_repo.find_by_id.return_value = None + + request = GetWorkspaceDetailRequest( + workspace_id="nonexistent", + user_id="user-123", + ) + + detail, error = use_case.execute(request) + + assert detail is None + assert error == "Workspace not found" + + def test_get_workspace_detail_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 = GetWorkspaceDetailRequest( + workspace_id="workspace-123", + user_id="user-123", + ) + + detail, error = use_case.execute(request) + + assert detail is None + assert error == "You are not a member of this workspace" + + def test_get_workspace_detail_missing_workspace_id(self, use_case): + """测试缺少工作空间 ID""" + request = GetWorkspaceDetailRequest( + workspace_id="", + user_id="user-123", + ) + + detail, error = use_case.execute(request) + + assert detail is None + assert error == "Workspace ID is required" + + def test_get_workspace_detail_missing_user_id(self, use_case): + """测试缺少用户 ID""" + request = GetWorkspaceDetailRequest( + workspace_id="workspace-123", + user_id="", + ) + + detail, error = use_case.execute(request) + + assert detail is None + assert error == "User ID is required"