xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases 7 Wiki Activity

feat(workspace): add list workspaces and get detail use cases

Browse Source 
- 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
This commit is contained in:
Xiaoxia AI
2026-06-17 07:04:55 +08:00
parent a222c7f278
commit ee3935abdc
3 changed files with 499 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/application/workspace/__init__.py
+16
View File
@@ -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",
]
packages/application/workspace/list_workspaces_use_case.py
+217
View File
@@ -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)}"
tests/unit/test_list_workspaces_use_case.py
+266
View File
@@ -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"