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

style: normalize python formatting gates

Browse Source
This commit is contained in:
Xiaoxia AI
2026-06-21 06:52:19 +08:00
parent 0809a079c5
commit bfbaddbd9a
129 changed files with 3024 additions and 2485 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_api.py
+108 -72
View File
@@ -1,6 +1,7 @@
"""
API 集成测试
"""
import pytest
from fastapi.testclient import TestClient
@@ -11,130 +12,165 @@ client = TestClient(app)
class TestAuthAPI:
"""认证 API 集成测试"""
def test_register_success(self):
"""测试注册成功"""
response = client.post("/api/v1/auth/register", json={
"email": "test@example.com",
"password": "SecurePass123",
"username": "testuser",
"display_name": "Test User",
})
response = client.post(
"/api/v1/auth/register",
json={
"email": "test@example.com",
"password": "SecurePass123",
"username": "testuser",
"display_name": "Test User",
},
)
assert response.status_code == 201
data = response.json()
assert data["email"] == "test@example.com"
assert data["username"] == "testuser"
assert "user_id" in data
def test_register_duplicate_email(self):
"""测试重复邮箱注册"""
# 先注册一个用户
client.post("/api/v1/auth/register", json={
"email": "duplicate@example.com",
"password": "SecurePass123",
"username": "user1",
"display_name": "User 1",
})
client.post(
"/api/v1/auth/register",
json={
"email": "duplicate@example.com",
"password": "SecurePass123",
"username": "user1",
"display_name": "User 1",
},
)
# 尝试用相同邮箱再次注册
response = client.post("/api/v1/auth/register", json={
"email": "duplicate@example.com",
"password": "SecurePass123",
"username": "user2",
"display_name": "User 2",
})
response = client.post(
"/api/v1/auth/register",
json={
"email": "duplicate@example.com",
"password": "SecurePass123",
"username": "user2",
"display_name": "User 2",
},
)
assert response.status_code == 400
assert "already registered" in response.json()["detail"].lower()
def test_login_success(self):
"""测试登录成功"""
# 先注册
client.post("/api/v1/auth/register", json={
"email": "login@example.com",
"password": "SecurePass123",
"username": "loginuser",
"display_name": "Login User",
})
client.post(
"/api/v1/auth/register",
json={
"email": "login@example.com",
"password": "SecurePass123",
"username": "loginuser",
"display_name": "Login User",
},
)
# 登录
response = client.post("/api/v1/auth/login", json={
"email": "login@example.com",
"password": "SecurePass123",
})
response = client.post(
"/api/v1/auth/login",
json={
"email": "login@example.com",
"password": "SecurePass123",
},
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert "refresh_token" in data
assert data["token_type"] == "bearer"
def test_login_wrong_password(self):
"""测试密码错误"""
response = client.post("/api/v1/auth/login", json={
"email": "login@example.com",
"password": "WrongPassword123",
})
response = client.post(
"/api/v1/auth/login",
json={
"email": "login@example.com",
"password": "WrongPassword123",
},
)
assert response.status_code == 401
class TestWorkspaceAPI:
"""工作空间 API 集成测试"""
def setup_method(self):
"""每个测试前的准备"""
# 注册并登录，获取 token
client.post("/api/v1/auth/register", json={
"email": "workspace@example.com",
"password": "SecurePass123",
"username": "workspaceuser",
"display_name": "Workspace User",
})
response = client.post("/api/v1/auth/login", json={
"email": "workspace@example.com",
"password": "SecurePass123",
})
client.post(
"/api/v1/auth/register",
json={
"email": "workspace@example.com",
"password": "SecurePass123",
"username": "workspaceuser",
"display_name": "Workspace User",
},
)
response = client.post(
"/api/v1/auth/login",
json={
"email": "workspace@example.com",
"password": "SecurePass123",
},
)
self.token = response.json()["access_token"]
self.headers = {"Authorization": f"Bearer {self.token}"}
def test_create_workspace(self):
"""测试创建工作空间"""
response = client.post("/api/v1/workspaces", json={
"name": "My Workspace",
"subscription_plan": "free",
}, headers=self.headers)
response = client.post(
"/api/v1/workspaces",
json={
"name": "My Workspace",
"subscription_plan": "free",
},
headers=self.headers,
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "My Workspace"
assert data["subscription_plan"] == "free"
assert data["max_projects"] == 3
def test_list_workspaces(self):
"""测试获取工作空间列表"""
# 创建工作空间
client.post("/api/v1/workspaces", json={
"name": "Workspace 1",
}, headers=self.headers)
client.post(
"/api/v1/workspaces",
json={
"name": "Workspace 1",
},
headers=self.headers,
)
# 获取列表
response = client.get("/api/v1/workspaces", headers=self.headers)
assert response.status_code == 200
data = response.json()
assert len(data["workspaces"]) > 0
assert data["workspaces"][0]["name"] == "Workspace 1"
def test_create_workspace_unauthorized(self):
"""测试未登录创建工作空间"""
response = client.post("/api/v1/workspaces", json={
"name": "Unauthorized Workspace",
})
response = client.post(
"/api/v1/workspaces",
json={
"name": "Unauthorized Workspace",
},
)
assert response.status_code == 403 # FastAPI HTTPBearer 返回 403
tests/integration/test_asset_tags.py
+12 -12
View File
@@ -13,10 +13,10 @@ def test_add_tag_to_asset():
storage_key="uploads/abc/video.mp4",
mime_type="video/mp4",
)
asset.add_tag("风景")
asset.add_tag("自然")
assert len(asset.tags) == 2
assert "风景" in asset.tags
assert "自然" in asset.tags
@@ -32,10 +32,10 @@ def test_add_duplicate_tag_should_ignore():
storage_key="uploads/abc/video.mp4",
mime_type="video/mp4",
)
asset.add_tag("风景")
asset.add_tag("风景") # 重复
assert len(asset.tags) == 1
assert asset.tags.count("风景") == 1
@@ -50,10 +50,10 @@ def test_add_empty_tag_should_fail():
storage_key="uploads/abc/video.mp4",
mime_type="video/mp4",
)
with pytest.raises(ValueError, match="标签不能为空"):
asset.add_tag("")
with pytest.raises(ValueError, match="标签不能为空"):
asset.add_tag(" ") # 仅空格
@@ -68,12 +68,12 @@ def test_remove_tag_from_asset():
storage_key="uploads/abc/video.mp4",
mime_type="video/mp4",
)
asset.add_tag("风景")
asset.add_tag("自然")
asset.remove_tag("风景")
assert len(asset.tags) == 1
assert "风景" not in asset.tags
assert "自然" in asset.tags
@@ -89,11 +89,11 @@ def test_remove_nonexistent_tag_should_be_idempotent():
storage_key="uploads/abc/video.mp4",
mime_type="video/mp4",
)
asset.add_tag("风景")
# 删除不存在的标签，不应报错
asset.remove_tag("不存在的标签")
assert len(asset.tags) == 1
assert "风景" in asset.tags
tests/integration/test_classification_pipeline.py
+15 -12
View File
@@ -1,6 +1,9 @@
from packages.application import SubmitClassificationJobCommand, SubmitClassificationJobUseCase
from packages.adapters.in_memory import InMemoryClassificationJobRepository
from packages.domain import ClassificationJobStatus, AssetClassification
from packages.application import (
SubmitClassificationJobCommand,
SubmitClassificationJobUseCase,
)
from packages.domain import AssetClassification, ClassificationJobStatus
def simulate_classify_asset(job_id: str, job_repo: InMemoryClassificationJobRepository) -> dict:
@@ -8,24 +11,24 @@ def simulate_classify_asset(job_id: str, job_repo: InMemoryClassificationJobRepo
job = job_repo.get(job_id)
if job is None:
return {"status": "failed", "error": "job not found"}
try:
# Update job status to PROCESSING
job.status = ClassificationJobStatus.PROCESSING
job_repo.update(job)
# Mock classification
asset_id_hash = sum(ord(c) for c in job.asset_id)
classifications = list(AssetClassification)
classification = classifications[asset_id_hash % len(classifications)]
confidence = 0.85
# Update job status to COMPLETED
job.status = ClassificationJobStatus.COMPLETED
job.classification = classification.value
job.confidence = confidence
job_repo.update(job)
return {
"status": "completed",
"job_id": job.id,
@@ -37,7 +40,7 @@ def simulate_classify_asset(job_id: str, job_repo: InMemoryClassificationJobRepo
job.status = ClassificationJobStatus.FAILED
job.error_message = str(e)
job_repo.update(job)
return {
"status": "failed",
"job_id": job.id,
@@ -48,7 +51,7 @@ def simulate_classify_asset(job_id: str, job_repo: InMemoryClassificationJobRepo
def test_classification_pipeline():
"""Test the full classification pipeline: submit job -> worker processes -> result."""
job_repo = InMemoryClassificationJobRepository()
# Submit classification job
use_case = SubmitClassificationJobUseCase(job_repo)
job = use_case.execute(
@@ -58,18 +61,18 @@ def test_classification_pipeline():
asset_id="asset-123",
)
)
assert job.status == ClassificationJobStatus.PENDING
assert job.classification == ""
assert job.confidence == 0.0
# Simulate worker task execution
result = simulate_classify_asset(job.id, job_repo)
assert result["status"] == "completed"
assert "classification" in result
assert "confidence" in result
# Verify job was updated
updated_job = job_repo.get(job.id)
assert updated_job is not None
tests/integration/test_generation_pipeline.py
+16 -3
View File
@@ -1,6 +1,10 @@
from datetime import datetime, timezone
from packages.application import CreateGenerationTaskCommand, CreateGenerationTaskUseCase, GetGeneratedVideoDownloadUrlUseCase
from packages.application import (
CreateGenerationTaskCommand,
CreateGenerationTaskUseCase,
GetGeneratedVideoDownloadUrlUseCase,
)
from packages.domain import GeneratedVideo, GenerationTaskStatus
@@ -41,7 +45,11 @@ class DummyGeneratedVideoRepository:
return [video for video in self.items.values() if video.generation_task_id == generation_task_id]
def simulate_generate_video(task_id: str, task_repo: DummyGenerationTaskRepository, video_repo: DummyGeneratedVideoRepository) -> dict:
def simulate_generate_video(
task_id: str,
task_repo: DummyGenerationTaskRepository,
video_repo: DummyGeneratedVideoRepository,
) -> dict:
task = task_repo.get(task_id)
if task is None:
return {"status": "failed", "error": "task not found"}
@@ -75,7 +83,12 @@ def simulate_generate_video(task_id: str, task_repo: DummyGenerationTaskReposito
task.completed_at = datetime.now(timezone.utc)
task_repo.update(task)
return {"status": "completed", "task_id": task.id, "video_id": video.id, "file_url": file_url}
return {
"status": "completed",
"task_id": task.id,
"video_id": video.id,
"file_url": file_url,
}
def test_create_generation_task_smoke():
tests/integration/test_ingest_pipeline.py
+22 -15
View File
@@ -1,9 +1,16 @@
from packages.adapters.in_memory import (
InMemoryAssetRepository,
InMemoryIngestJobRepository,
)
from packages.application import SubmitIngestJobCommand, SubmitIngestJobUseCase
from packages.adapters.in_memory import InMemoryAssetRepository, InMemoryIngestJobRepository
from packages.domain import Asset, IngestJob, IngestJobStatus
def simulate_ingest_asset(job_id: str, job_repo: InMemoryIngestJobRepository, asset_repo: InMemoryAssetRepository) -> dict:
def simulate_ingest_asset(
job_id: str,
job_repo: InMemoryIngestJobRepository,
asset_repo: InMemoryAssetRepository,
) -> dict:
"""
Simulate ingest asset logic without Celery.
This is the core business logic that would run inside the worker task.
@@ -11,12 +18,12 @@ def simulate_ingest_asset(job_id: str, job_repo: InMemoryIngestJobRepository, as
job = job_repo.get(job_id)
if job is None:
return {"status": "failed", "error": "job not found"}
try:
# Update job status to PROCESSING
job.status = IngestJobStatus.PROCESSING
job_repo.update(job)
# Mock metadata extraction
mime_type = "video/mp4" if job.storage_key.endswith(".mp4") else "image/jpeg"
metadata = {
@@ -25,10 +32,10 @@ def simulate_ingest_asset(job_id: str, job_repo: InMemoryIngestJobRepository, as
"height": 1080,
"size_bytes": 1024000,
}
# Extract filename from storage_key
filename = job.storage_key.split("/")[-1]
# Create Asset
asset = Asset.create(
workspace_id=job.workspace_id,
@@ -40,12 +47,12 @@ def simulate_ingest_asset(job_id: str, job_repo: InMemoryIngestJobRepository, as
metadata=metadata,
)
asset_repo.create(asset)
# Update job status to COMPLETED
job.status = IngestJobStatus.COMPLETED
job.result_asset_id = asset.id
job_repo.update(job)
return {
"status": "completed",
"job_id": job.id,
@@ -56,7 +63,7 @@ def simulate_ingest_asset(job_id: str, job_repo: InMemoryIngestJobRepository, as
job.status = IngestJobStatus.FAILED
job.error_message = str(e)
job_repo.update(job)
return {
"status": "failed",
"job_id": job.id,
@@ -68,7 +75,7 @@ def test_ingest_asset_pipeline():
"""Test the full ingest pipeline: submit job -> worker processes -> asset created."""
job_repo = InMemoryIngestJobRepository()
asset_repo = InMemoryAssetRepository()
# Submit ingest job
use_case = SubmitIngestJobUseCase(job_repo)
job = use_case.execute(
@@ -79,22 +86,22 @@ def test_ingest_asset_pipeline():
storage_key="uploads/test-video.mp4",
)
)
assert job.status == IngestJobStatus.PENDING
assert job.result_asset_id == ""
# Simulate worker task execution
result = simulate_ingest_asset(job.id, job_repo, asset_repo)
assert result["status"] == "completed"
assert "asset_id" in result
# Verify job was updated
updated_job = job_repo.get(job.id)
assert updated_job is not None
assert updated_job.status == IngestJobStatus.COMPLETED
assert updated_job.result_asset_id != ""
# Verify asset was created
assets = asset_repo.list_by_library("lib-1")
assert len(assets) == 1
tests/integration/test_project_management.py
+37 -36
View File
@@ -1,4 +1,5 @@
"""项目管理功能集成测试"""
import pytest
from packages.adapters.in_memory.project_management_repositories import (
@@ -23,7 +24,7 @@ def test_create_task():
"""测试创建任务"""
repo = InMemoryTaskRepository()
use_case = CreateTaskUseCase(repo)
task = use_case.execute(
project_id="proj_1",
workspace_id="ws_1",
@@ -31,7 +32,7 @@ def test_create_task():
description="实现用户登录功能",
priority=TaskPriority.HIGH,
)
assert task.id is not None
assert task.name == "开发登录功能"
assert task.status == TaskStatus.PENDING
@@ -43,7 +44,7 @@ def test_list_tasks():
"""测试获取任务列表"""
repo = InMemoryTaskRepository()
create_use_case = CreateTaskUseCase(repo)
# 创建两个任务
create_use_case.execute(
project_id="proj_1",
@@ -55,11 +56,11 @@ def test_list_tasks():
workspace_id="ws_1",
name="任务2",
)
# 查询任务列表
list_use_case = ListProjectTasksUseCase(repo)
tasks = list_use_case.execute("proj_1")
assert len(tasks) == 2
assert tasks[0].name == "任务1"
assert tasks[1].name == "任务2"
@@ -70,17 +71,17 @@ def test_update_task_status():
repo = InMemoryTaskRepository()
create_use_case = CreateTaskUseCase(repo)
update_use_case = UpdateTaskStatusUseCase(repo)
# 创建任务
task = create_use_case.execute(
project_id="proj_1",
workspace_id="ws_1",
name="测试任务",
)
# 更新状态为进行中
updated_task = update_use_case.execute(task.id, TaskStatus.IN_PROGRESS)
assert updated_task.status == TaskStatus.IN_PROGRESS
assert updated_task.actual_start_date is not None
@@ -90,23 +91,23 @@ def test_update_task_progress():
repo = InMemoryTaskRepository()
create_use_case = CreateTaskUseCase(repo)
progress_use_case = UpdateTaskProgressUseCase(repo)
# 创建任务
task = create_use_case.execute(
project_id="proj_1",
workspace_id="ws_1",
name="测试任务",
)
# 更新进度到 50%
updated_task = progress_use_case.execute(task.id, 50.0)
assert updated_task.progress == 50.0
assert updated_task.status == TaskStatus.IN_PROGRESS
# 更新进度到 100%
completed_task = progress_use_case.execute(task.id, 100.0)
assert completed_task.progress == 100.0
assert completed_task.status == TaskStatus.COMPLETED
assert completed_task.actual_end_date is not None
@@ -116,14 +117,14 @@ def test_create_milestone():
"""测试创建里程碑"""
repo = InMemoryMilestoneRepository()
use_case = CreateMilestoneUseCase(repo)
milestone = use_case.execute(
project_id="proj_1",
workspace_id="ws_1",
name="V1.0 发布",
description="第一个正式版本",
)
assert milestone.id is not None
assert milestone.name == "V1.0 发布"
assert milestone.completed is False
@@ -135,7 +136,7 @@ def test_create_and_resolve_issue():
create_use_case = CreateTaskIssueUseCase(repo)
resolve_use_case = ResolveTaskIssueUseCase(repo)
list_use_case = ListTaskIssuesUseCase(repo)
# 创建问题
issue = create_use_case.execute(
task_id="task_1",
@@ -144,17 +145,17 @@ def test_create_and_resolve_issue():
title="接口报错",
description="调用登录接口返回 500",
)
assert issue.id is not None
assert issue.title == "接口报错"
assert issue.resolved is False
# 解决问题
resolved_issue = resolve_use_case.execute(issue.id)
assert resolved_issue.resolved is True
assert resolved_issue.resolved_at is not None
# 查询任务问题列表
issues = list_use_case.execute("task_1")
assert len(issues) == 1
@@ -165,14 +166,14 @@ def test_task_hierarchy():
"""测试任务层级关系"""
repo = InMemoryTaskRepository()
create_use_case = CreateTaskUseCase(repo)
# 创建父任务
parent_task = create_use_case.execute(
project_id="proj_1",
workspace_id="ws_1",
name="开发用户模块",
)
# 创建子任务
child_task_1 = create_use_case.execute(
project_id="proj_1",
@@ -180,17 +181,17 @@ def test_task_hierarchy():
name="登录功能",
parent_task_id=parent_task.id,
)
child_task_2 = create_use_case.execute(
project_id="proj_1",
workspace_id="ws_1",
name="注册功能",
parent_task_id=parent_task.id,
)
# 查询子任务
children = repo.list_by_parent(parent_task.id)
assert len(children) == 2
assert children[0].parent_task_id == parent_task.id
assert children[1].parent_task_id == parent_task.id
@@ -199,11 +200,11 @@ def test_task_hierarchy():
def test_get_task_detail():
"""测试获取任务详情"""
from packages.application.get_task_detail_use_case import GetTaskDetailUseCase
repo = InMemoryTaskRepository()
create_use_case = CreateTaskUseCase(repo)
get_use_case = GetTaskDetailUseCase(repo)
# 创建任务
task = create_use_case.execute(
project_id="proj_1",
@@ -211,14 +212,14 @@ def test_get_task_detail():
name="测试任务",
description="这是一个测试任务",
)
# 获取详情
retrieved_task = get_use_case.execute(task.id)
assert retrieved_task.id == task.id
assert retrieved_task.name == "测试任务"
assert retrieved_task.description == "这是一个测试任务"
# 测试不存在的任务
try:
get_use_case.execute("nonexistent_id")
@@ -230,11 +231,11 @@ def test_get_task_detail():
def test_update_task():
"""测试任务基本信息更新"""
from packages.application.update_task_use_case import UpdateTaskUseCase
repo = InMemoryTaskRepository()
create_use_case = CreateTaskUseCase(repo)
update_use_case = UpdateTaskUseCase(repo)
# 创建任务
task = create_use_case.execute(
project_id="proj_1",
@@ -243,7 +244,7 @@ def test_update_task():
description="原始描述",
priority="low",
)
# 更新任务
updated_task = update_use_case.execute(
task_id=task.id,
@@ -251,17 +252,17 @@ def test_update_task():
description="更新后的描述",
priority="high",
)
assert updated_task.name == "更新后的任务"
assert updated_task.description == "更新后的描述"
assert updated_task.priority == "high"
# 部分更新
partial_updated = update_use_case.execute(
task_id=task.id,
name="又更新了",
)
assert partial_updated.name == "又更新了"
assert partial_updated.description == "更新后的描述" # 保持不变
assert partial_updated.priority == "high" # 保持不变
tests/integration/test_projects.py
+6 -6
View File
@@ -1,3 +1,9 @@
from packages.adapters.in_memory import (
InMemoryAssetLibraryRepository,
InMemoryAssetRepository,
InMemoryIngestJobRepository,
InMemoryProjectRepository,
)
from packages.application import (
CreateAssetCommand,
CreateAssetLibraryCommand,
@@ -11,12 +17,6 @@ from packages.application import (
SubmitIngestJobCommand,
SubmitIngestJobUseCase,
)
from packages.adapters.in_memory import (
InMemoryAssetLibraryRepository,
InMemoryAssetRepository,
InMemoryIngestJobRepository,
InMemoryProjectRepository,
)
from packages.domain import AssetLibraryKind, IngestJobStatus
tests/integration/test_sqlalchemy_repositories.py
+7 -5
View File
@@ -2,7 +2,9 @@ from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from packages.adapters.sqlalchemy_impl.models import Base
from packages.adapters.sqlalchemy_impl.project_repository import SQLAlchemyProjectRepository
from packages.adapters.sqlalchemy_impl.project_repository import (
SQLAlchemyProjectRepository,
)
from packages.application import CreateProjectCommand, CreateProjectUseCase
@@ -13,12 +15,12 @@ def test_sqlalchemy_project_repository():
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)
session: Session = SessionLocal()
try:
# Create repository and use case
repository = SQLAlchemyProjectRepository(session)
use_case = CreateProjectUseCase(repository)
# Create project
project = use_case.execute(
CreateProjectCommand(
@@ -27,10 +29,10 @@ def test_sqlalchemy_project_repository():
description="Test description",
)
)
assert project.name == "Test Project"
assert project.workspace_id == "ws-1"
# List projects
projects = repository.list_by_workspace("ws-1")
assert len(projects) == 1
tests/integration/test_upload_pipeline.py
+13 -9
View File
@@ -1,5 +1,8 @@
from packages.adapters.in_memory import (
InMemoryAssetRepository,
InMemoryIngestJobRepository,
)
from packages.application import SubmitIngestJobCommand, SubmitIngestJobUseCase
from packages.adapters.in_memory import InMemoryAssetRepository, InMemoryIngestJobRepository
from packages.domain import IngestJobStatus
@@ -13,11 +16,12 @@ def simulate_upload_and_ingest(
) -> dict:
"""Simulate full upload → ingest pipeline."""
from uuid import uuid4
from tests.integration.test_ingest_pipeline import simulate_ingest_asset
# Mock storage: generate storage_key
storage_key = f"uploads/{uuid4().hex[:8]}/{filename}"
# Submit ingest job
use_case = SubmitIngestJobUseCase(job_repo)
job = use_case.execute(
@@ -28,10 +32,10 @@ def simulate_upload_and_ingest(
storage_key=storage_key,
)
)
# Simulate worker task
result = simulate_ingest_asset(job.id, job_repo, asset_repo)
return {
"storage_key": storage_key,
"job_id": job.id,
@@ -43,7 +47,7 @@ def test_upload_to_asset_full_pipeline():
"""Test full pipeline: upload → storage → ingest job → worker → asset created."""
job_repo = InMemoryIngestJobRepository()
asset_repo = InMemoryAssetRepository()
# Simulate upload
result = simulate_upload_and_ingest(
workspace_id="ws-1",
@@ -53,17 +57,17 @@ def test_upload_to_asset_full_pipeline():
job_repo=job_repo,
asset_repo=asset_repo,
)
assert "storage_key" in result
assert result["storage_key"].endswith("demo-video.mp4")
assert result["worker_result"]["status"] == "completed"
# Verify job was completed
job = job_repo.get(result["job_id"])
assert job is not None
assert job.status == IngestJobStatus.COMPLETED
assert job.result_asset_id != ""
# Verify asset was created
assets = asset_repo.list_by_library("lib-1")
assert len(assets) == 1
tests/unit/test_accept_invitation_use_case.py
+66 -57
View File
@@ -1,62 +1,71 @@
"""
接受/拒绝邀请 Use Case 测试
"""
import pytest
from unittest.mock import Mock
from datetime import datetime, timedelta, timezone
from unittest.mock import Mock
import pytest
from packages.application.workspace.accept_invitation_use_case import (
AcceptInvitationUseCase,
AcceptInvitationRequest,
DeclineInvitationUseCase,
AcceptInvitationUseCase,
DeclineInvitationRequest,
DeclineInvitationUseCase,
)
from packages.domain.entities import (
InvitationStatus,
User,
Workspace,
WorkspaceInvitation,
WorkspaceMember,
User,
InvitationStatus,
)
class TestAcceptInvitationUseCase:
"""接受邀请测试"""
@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.save = Mock()
return repo
@pytest.fixture
def mock_invitation_repo(self):
repo = Mock()
repo.find_by_token = Mock(return_value=None)
repo.save = Mock()
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_invitation_repo, mock_user_repo):
def use_case(
self,
mock_workspace_repo,
mock_member_repo,
mock_invitation_repo,
mock_user_repo,
):
return AcceptInvitationUseCase(
workspace_repository=mock_workspace_repo,
workspace_member_repository=mock_member_repo,
workspace_invitation_repository=mock_invitation_repo,
user_repository=mock_user_repo,
)
@pytest.fixture
def test_workspace(self):
return Workspace(
@@ -64,7 +73,7 @@ class TestAcceptInvitationUseCase:
name="Test Workspace",
owner_user_id="owner-id",
)
@pytest.fixture
def test_user(self):
return User(
@@ -73,7 +82,7 @@ class TestAcceptInvitationUseCase:
username="invitee",
display_name="Invitee User",
)
@pytest.fixture
def valid_invitation(self):
return WorkspaceInvitation(
@@ -86,7 +95,7 @@ class TestAcceptInvitationUseCase:
status=InvitationStatus.PENDING,
expires_at=datetime.now(timezone.utc) + timedelta(days=7),
)
def test_accept_invitation_success(
self,
use_case,
@@ -103,45 +112,45 @@ class TestAcceptInvitationUseCase:
mock_user_repo.find_by_id.return_value = test_user
mock_workspace_repo.find_by_id.return_value = test_workspace
mock_member_repo.find_by_workspace_and_user.return_value = None
request = AcceptInvitationRequest(
invitation_token="valid-token",
user_id="user-123",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.workspace_id == "workspace-123"
assert response.workspace_name == "Test Workspace"
assert response.role == "member"
# 验证创建了成员记录
mock_member_repo.save.assert_called_once()
member = mock_member_repo.save.call_args[0][0]
assert member.user_id == "user-123"
assert member.role == "member"
assert member.invited_by == "inviter-id"
# 验证更新了邀请状态
assert valid_invitation.status == InvitationStatus.ACCEPTED
assert valid_invitation.accepted_at is not None
def test_accept_invitation_invalid_token(self, use_case, mock_invitation_repo):
"""测试无效令牌"""
mock_invitation_repo.find_by_token.return_value = None
request = AcceptInvitationRequest(
invitation_token="invalid-token",
user_id="user-123",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Invalid invitation token"
def test_accept_invitation_already_accepted(
self,
use_case,
@@ -151,17 +160,17 @@ class TestAcceptInvitationUseCase:
"""测试邀请已被接受"""
valid_invitation.status = InvitationStatus.ACCEPTED
mock_invitation_repo.find_by_token.return_value = valid_invitation
request = AcceptInvitationRequest(
invitation_token="valid-token",
user_id="user-123",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Invitation has already been accepted"
def test_accept_invitation_expired(
self,
use_case,
@@ -171,18 +180,18 @@ class TestAcceptInvitationUseCase:
"""测试邀请已过期"""
valid_invitation.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
mock_invitation_repo.find_by_token.return_value = valid_invitation
request = AcceptInvitationRequest(
invitation_token="valid-token",
user_id="user-123",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Invitation has expired"
assert valid_invitation.status == InvitationStatus.EXPIRED
def test_accept_invitation_email_mismatch(
self,
use_case,
@@ -192,7 +201,7 @@ class TestAcceptInvitationUseCase:
):
"""测试邮箱不匹配"""
mock_invitation_repo.find_by_token.return_value = valid_invitation
different_user = User(
id="user-123",
email="different@test.com",
@@ -200,17 +209,17 @@ class TestAcceptInvitationUseCase:
display_name="Different User",
)
mock_user_repo.find_by_id.return_value = different_user
request = AcceptInvitationRequest(
invitation_token="valid-token",
user_id="user-123",
)
response, error = use_case.execute(request)
assert response is None
assert error == "This invitation is for a different email address"
def test_accept_invitation_already_member(
self,
use_case,
@@ -226,7 +235,7 @@ class TestAcceptInvitationUseCase:
mock_invitation_repo.find_by_token.return_value = valid_invitation
mock_user_repo.find_by_id.return_value = test_user
mock_workspace_repo.find_by_id.return_value = test_workspace
existing_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -234,41 +243,41 @@ class TestAcceptInvitationUseCase:
role="admin",
)
mock_member_repo.find_by_workspace_and_user.return_value = existing_member
request = AcceptInvitationRequest(
invitation_token="valid-token",
user_id="user-123",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.role == "admin" # 返回现有角色
# 不创建新成员记录
mock_member_repo.save.assert_not_called()
# 但仍标记邀请为已接受
assert valid_invitation.status == InvitationStatus.ACCEPTED
class TestDeclineInvitationUseCase:
"""拒绝邀请测试"""
@pytest.fixture
def mock_invitation_repo(self):
repo = Mock()
repo.find_by_token = Mock(return_value=None)
repo.save = Mock()
return repo
@pytest.fixture
def use_case(self, mock_invitation_repo):
return DeclineInvitationUseCase(
workspace_invitation_repository=mock_invitation_repo,
)
@pytest.fixture
def valid_invitation(self):
return WorkspaceInvitation(
@@ -281,35 +290,35 @@ class TestDeclineInvitationUseCase:
status=InvitationStatus.PENDING,
expires_at=datetime.now(timezone.utc) + timedelta(days=7),
)
def test_decline_invitation_success(self, use_case, mock_invitation_repo, valid_invitation):
"""测试拒绝邀请成功"""
mock_invitation_repo.find_by_token.return_value = valid_invitation
request = DeclineInvitationRequest(invitation_token="valid-token")
success, error = use_case.execute(request)
assert success is True
assert error is None
assert valid_invitation.status == InvitationStatus.DECLINED
def test_decline_invitation_invalid_token(self, use_case, mock_invitation_repo):
"""测试无效令牌"""
mock_invitation_repo.find_by_token.return_value = None
request = DeclineInvitationRequest(invitation_token="invalid-token")
success, error = use_case.execute(request)
assert success is False
assert error == "Invalid invitation token"
def test_decline_invitation_already_accepted(self, use_case, mock_invitation_repo, valid_invitation):
"""测试邀请已被接受"""
valid_invitation.status = InvitationStatus.ACCEPTED
mock_invitation_repo.find_by_token.return_value = valid_invitation
request = DeclineInvitationRequest(invitation_token="valid-token")
success, error = use_case.execute(request)
assert success is False
assert error == "Invitation has already been accepted"
tests/unit/test_auth_simple.py
+1
View File
@@ -18,6 +18,7 @@ spec.loader.exec_module(auth_simple)
_create_access_token = auth_simple._create_access_token
_verify_password_with_legacy_upgrade = auth_simple._verify_password_with_legacy_upgrade
from app.config import settings
from packages.adapters.sqlalchemy_impl.models import UserModel
from packages.domain.auth import password_hasher
tests/unit/test_create_workspace_use_case.py
+47 -38
View File
@@ -1,36 +1,39 @@
"""
创建 Workspace Use Case 测试
"""
import pytest
from unittest.mock import Mock
import pytest
from packages.application.workspace import (
CreateWorkspaceUseCase,
CreateWorkspaceRequest,
CreateWorkspaceUseCase,
)
from packages.domain.entities import User
class TestCreateWorkspaceUseCase:
"""创建工作空间测试"""
@pytest.fixture
def mock_workspace_repo(self):
repo = Mock()
repo.save = Mock()
return repo
@pytest.fixture
def mock_member_repo(self):
repo = Mock()
repo.save = Mock()
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 CreateWorkspaceUseCase(
@@ -38,7 +41,7 @@ class TestCreateWorkspaceUseCase:
workspace_member_repository=mock_member_repo,
user_repository=mock_user_repo,
)
@pytest.fixture
def test_user(self):
return User(
@@ -47,123 +50,129 @@ class TestCreateWorkspaceUseCase:
username="testuser",
display_name="Test User",
)
def test_create_workspace_success_free_plan(self, use_case, mock_workspace_repo, mock_member_repo, mock_user_repo, test_user):
def test_create_workspace_success_free_plan(
self, use_case, mock_workspace_repo, mock_member_repo, mock_user_repo, test_user
):
"""测试创建免费工作空间"""
mock_user_repo.find_by_id.return_value = test_user
request = CreateWorkspaceRequest(
name="My Workspace",
owner_user_id="user-123",
subscription_plan="free",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.name == "My Workspace"
assert response.subscription_plan == "free"
assert response.max_projects == 3
assert response.max_storage_gb == 10
# 验证保存了 workspace
mock_workspace_repo.save.assert_called_once()
workspace = mock_workspace_repo.save.call_args[0][0]
assert workspace.name == "My Workspace"
assert workspace.owner_user_id == "user-123"
# 验证创建了 owner 成员
mock_member_repo.save.assert_called_once()
member = mock_member_repo.save.call_args[0][0]
assert member.user_id == "user-123"
assert member.role == "owner"
def test_create_workspace_pro_plan(self, use_case, mock_workspace_repo, mock_member_repo, mock_user_repo, test_user):
def test_create_workspace_pro_plan(
self, use_case, mock_workspace_repo, mock_member_repo, mock_user_repo, test_user
):
"""测试创建 Pro 工作空间"""
mock_user_repo.find_by_id.return_value = test_user
request = CreateWorkspaceRequest(
name="Pro Workspace",
owner_user_id="user-123",
subscription_plan="pro",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.subscription_plan == "pro"
assert response.max_projects == 999999 # unlimited
assert response.max_storage_gb == 100
def test_create_workspace_enterprise_plan(self, use_case, mock_workspace_repo, mock_member_repo, mock_user_repo, test_user):
def test_create_workspace_enterprise_plan(
self, use_case, mock_workspace_repo, mock_member_repo, mock_user_repo, test_user
):
"""测试创建 Enterprise 工作空间"""
mock_user_repo.find_by_id.return_value = test_user
request = CreateWorkspaceRequest(
name="Enterprise Workspace",
owner_user_id="user-123",
subscription_plan="enterprise",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.subscription_plan == "enterprise"
assert response.max_projects == 999999
assert response.max_storage_gb == 1000
def test_create_workspace_missing_name(self, use_case):
"""测试缺少名称"""
request = CreateWorkspaceRequest(
name="",
owner_user_id="user-123",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Workspace name is required"
def test_create_workspace_name_too_long(self, use_case):
"""测试名称过长"""
request = CreateWorkspaceRequest(
name="A" * 101,
owner_user_id="user-123",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Workspace name is too long (max 100 characters)"
def test_create_workspace_user_not_found(self, use_case, mock_user_repo):
"""测试用户不存在"""
mock_user_repo.find_by_id.return_value = None
request = CreateWorkspaceRequest(
name="My Workspace",
owner_user_id="nonexistent-user",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Owner user not found"
def test_create_workspace_invalid_plan(self, use_case, mock_user_repo, test_user):
"""测试无效的订阅计划"""
mock_user_repo.find_by_id.return_value = test_user
request = CreateWorkspaceRequest(
name="My Workspace",
owner_user_id="user-123",
subscription_plan="invalid_plan",
)
response, error = use_case.execute(request)
assert response is None
assert "Invalid subscription plan" in error
tests/unit/test_email_service.py
+47 -44
View File
@@ -1,14 +1,17 @@
"""
邮件服务测试
"""
from unittest.mock import MagicMock, Mock, patch
import pytest
from unittest.mock import Mock, patch, MagicMock
from packages.domain.auth.email_service import EmailService, EmailConfig
from packages.domain.auth.email_service import EmailConfig, EmailService
class TestEmailService:
"""邮件服务测试"""
@pytest.fixture
def email_config(self):
"""创建测试邮件配置"""
@@ -21,41 +24,41 @@ class TestEmailService:
from_name="Test Service",
use_tls=True,
)
@pytest.fixture
def email_service(self, email_config):
"""创建邮件服务实例"""
return EmailService(config=email_config)
@patch('smtplib.SMTP')
@patch("smtplib.SMTP")
def test_send_email_success(self, mock_smtp, email_service):
"""测试发送邮件成功"""
# Mock SMTP 服务器
mock_server = MagicMock()
mock_smtp.return_value.__enter__.return_value = mock_server
success, error = email_service.send_email(
to_email="user@example.com",
subject="Test Email",
html_body="<h1>Test</h1>",
text_body="Test",
)
assert success is True
assert error is None
# 验证 SMTP 调用
mock_smtp.assert_called_once_with("smtp.test.com", 587)
mock_server.starttls.assert_called_once()
mock_server.login.assert_called_once_with("test@test.com", "test-password")
mock_server.sendmail.assert_called_once()
@patch('smtplib.SMTP')
@patch("smtplib.SMTP")
def test_send_email_with_cc_bcc(self, mock_smtp, email_service):
"""测试发送邮件带抄送和密送"""
mock_server = MagicMock()
mock_smtp.return_value.__enter__.return_value = mock_server
success, error = email_service.send_email(
to_email="user@example.com",
subject="Test Email",
@@ -63,9 +66,9 @@ class TestEmailService:
cc=["cc1@example.com", "cc2@example.com"],
bcc=["bcc@example.com"],
)
assert success is True
# 验证收件人列表包含所有人
call_args = mock_server.sendmail.call_args
recipients = call_args[0][1]
@@ -73,73 +76,73 @@ class TestEmailService:
assert "cc1@example.com" in recipients
assert "cc2@example.com" in recipients
assert "bcc@example.com" in recipients
@patch('smtplib.SMTP')
@patch("smtplib.SMTP")
def test_send_email_smtp_error(self, mock_smtp, email_service):
"""测试 SMTP 错误处理"""
# Mock SMTP 抛出异常
mock_smtp.side_effect = Exception("SMTP connection failed")
success, error = email_service.send_email(
to_email="user@example.com",
subject="Test",
html_body="<h1>Test</h1>",
)
assert success is False
assert error is not None
assert "SMTP connection failed" in error
@patch('smtplib.SMTP')
@patch("smtplib.SMTP")
def test_send_verification_email(self, mock_smtp, email_service):
"""测试发送验证邮件"""
mock_server = MagicMock()
mock_smtp.return_value.__enter__.return_value = mock_server
success, error = email_service.send_verification_email(
to_email="user@example.com",
username="TestUser",
verification_url="https://example.com/verify?token=abc123",
)
assert success is True
assert error is None
# 验证发送了邮件
mock_server.sendmail.assert_called_once()
call_args = mock_server.sendmail.call_args
# 验证收件人
assert call_args[0][1] == ["user@example.com"]
@patch('smtplib.SMTP')
@patch("smtplib.SMTP")
def test_send_password_reset_email(self, mock_smtp, email_service):
"""测试发送密码重置邮件"""
mock_server = MagicMock()
mock_smtp.return_value.__enter__.return_value = mock_server
success, error = email_service.send_password_reset_email(
to_email="user@example.com",
username="TestUser",
reset_url="https://example.com/reset?token=xyz789",
)
assert success is True
assert error is None
# 验证发送了邮件
mock_server.sendmail.assert_called_once()
call_args = mock_server.sendmail.call_args
# 验证收件人
assert call_args[0][1] == ["user@example.com"]
@patch('smtplib.SMTP')
@patch("smtplib.SMTP")
def test_send_workspace_invitation_email(self, mock_smtp, email_service):
"""测试发送工作空间邀请邮件"""
mock_server = MagicMock()
mock_smtp.return_value.__enter__.return_value = mock_server
success, error = email_service.send_workspace_invitation_email(
to_email="user@example.com",
inviter_name="Alice",
@@ -147,18 +150,18 @@ class TestEmailService:
role="admin",
invitation_url="https://example.com/invite?token=inv123",
)
assert success is True
assert error is None
# 验证发送了邮件
mock_server.sendmail.assert_called_once()
call_args = mock_server.sendmail.call_args
# 验证收件人
assert call_args[0][1] == ["user@example.com"]
@patch('smtplib.SMTP')
@patch("smtplib.SMTP")
def test_email_without_tls(self, mock_smtp):
"""测试不使用 TLS 发送邮件"""
config = EmailConfig(
@@ -170,27 +173,27 @@ class TestEmailService:
use_tls=False,
)
service = EmailService(config=config)
mock_server = MagicMock()
mock_smtp.return_value.__enter__.return_value = mock_server
success, error = service.send_email(
to_email="user@example.com",
subject="Test",
html_body="Test",
)
assert success is True
# 验证不调用 starttls
mock_server.starttls.assert_not_called()
# 验证不调用 login（没有用户名密码）
mock_server.login.assert_not_called()
def test_default_config(self):
"""测试默认配置"""
service = EmailService()
assert service.config.smtp_host == "smtp.gmail.com"
assert service.config.smtp_port == 587
assert service.config.use_tls is True
tests/unit/test_invite_member_use_case.py
+54 -45
View File
@@ -1,52 +1,61 @@
"""
邀请成员 Use Case 测试
"""
import pytest
from unittest.mock import Mock
from datetime import datetime, timezone
from unittest.mock import Mock
import pytest
from packages.application.workspace.invite_member_use_case import (
InviteMemberUseCase,
InviteMemberRequest,
InviteMemberUseCase,
)
from packages.domain.entities import (
User,
Workspace,
WorkspaceMember,
WorkspaceMemberRole,
User,
)
class TestInviteMemberUseCase:
"""邀请成员测试"""
@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)
return repo
@pytest.fixture
def mock_invitation_repo(self):
repo = Mock()
repo.find_pending_by_workspace_and_email = Mock(return_value=None)
repo.save = Mock()
return repo
@pytest.fixture
def mock_user_repo(self):
repo = Mock()
repo.find_by_email = Mock(return_value=None)
repo.find_by_id = Mock(return_value=None)
return repo
@pytest.fixture
def use_case(self, mock_workspace_repo, mock_member_repo, mock_invitation_repo, mock_user_repo):
def use_case(
self,
mock_workspace_repo,
mock_member_repo,
mock_invitation_repo,
mock_user_repo,
):
email_service = Mock()
email_service.send_workspace_invitation_email.return_value = (True, None)
return InviteMemberUseCase(
@@ -58,7 +67,7 @@ class TestInviteMemberUseCase:
invitation_expire_days=7,
email_service=email_service,
)
@pytest.fixture
def test_workspace(self):
return Workspace(
@@ -66,7 +75,7 @@ class TestInviteMemberUseCase:
name="Test Workspace",
owner_user_id="owner-id",
)
@pytest.fixture
def test_inviter(self):
return User(
@@ -75,7 +84,7 @@ class TestInviteMemberUseCase:
username="inviter",
display_name="Inviter User",
)
@pytest.fixture
def owner_member(self):
return WorkspaceMember(
@@ -84,7 +93,7 @@ class TestInviteMemberUseCase:
user_id="inviter-id",
role=WorkspaceMemberRole.OWNER,
)
@pytest.fixture
def admin_member(self):
return WorkspaceMember(
@@ -93,7 +102,7 @@ class TestInviteMemberUseCase:
user_id="inviter-id",
role=WorkspaceMemberRole.ADMIN,
)
def test_invite_member_success_by_owner(
self,
use_case,
@@ -109,31 +118,31 @@ class TestInviteMemberUseCase:
mock_workspace_repo.find_by_id.return_value = test_workspace
mock_member_repo.find_by_workspace_and_user.return_value = owner_member
mock_user_repo.find_by_id.return_value = test_inviter
request = InviteMemberRequest(
workspace_id="workspace-123",
inviter_user_id="inviter-id",
invitee_email="newuser@test.com",
role="member",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.invitee_email == "newuser@test.com"
assert response.role == "member"
assert response.expires_at is not None
# 验证保存了邀请
mock_invitation_repo.save.assert_called_once()
invitation = mock_invitation_repo.save.call_args[0][0]
assert invitation.invitee_email == "newuser@test.com"
assert invitation.status == "pending"
# 验证发送了邮件
use_case.email_service.send_workspace_invitation_email.assert_called_once()
def test_invite_member_success_by_admin(
self,
use_case,
@@ -149,35 +158,35 @@ class TestInviteMemberUseCase:
mock_workspace_repo.find_by_id.return_value = test_workspace
mock_member_repo.find_by_workspace_and_user.return_value = admin_member
mock_user_repo.find_by_id.return_value = test_inviter
request = InviteMemberRequest(
workspace_id="workspace-123",
inviter_user_id="inviter-id",
invitee_email="newuser@test.com",
role="viewer",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
def test_invite_member_workspace_not_found(self, use_case, mock_workspace_repo):
"""测试 Workspace 不存在"""
mock_workspace_repo.find_by_id.return_value = None
request = InviteMemberRequest(
workspace_id="nonexistent",
inviter_user_id="inviter-id",
invitee_email="newuser@test.com",
role="member",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Workspace not found"
def test_invite_member_inviter_not_member(
self,
use_case,
@@ -188,19 +197,19 @@ class TestInviteMemberUseCase:
"""测试邀请人不是成员"""
mock_workspace_repo.find_by_id.return_value = test_workspace
mock_member_repo.find_by_workspace_and_user.return_value = None
request = InviteMemberRequest(
workspace_id="workspace-123",
inviter_user_id="nonmember-id",
invitee_email="newuser@test.com",
role="member",
)
response, error = use_case.execute(request)
assert response is None
assert error == "You are not a member of this workspace"
def test_invite_member_inviter_no_permission(
self,
use_case,
@@ -210,7 +219,7 @@ class TestInviteMemberUseCase:
):
"""测试邀请人没有权限（只是普通成员）"""
mock_workspace_repo.find_by_id.return_value = test_workspace
regular_member = WorkspaceMember(
id="member-3",
workspace_id="workspace-123",
@@ -218,19 +227,19 @@ class TestInviteMemberUseCase:
role=WorkspaceMemberRole.MEMBER,
)
mock_member_repo.find_by_workspace_and_user.return_value = regular_member
request = InviteMemberRequest(
workspace_id="workspace-123",
inviter_user_id="inviter-id",
invitee_email="newuser@test.com",
role="member",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Only owners and admins can invite members"
def test_invite_member_already_member(
self,
use_case,
@@ -251,7 +260,7 @@ class TestInviteMemberUseCase:
role=WorkspaceMemberRole.MEMBER,
),
]
existing_user = User(
id="existing-user-id",
email="existing@test.com",
@@ -259,19 +268,19 @@ class TestInviteMemberUseCase:
display_name="Existing User",
)
mock_user_repo.find_by_email.return_value = existing_user
request = InviteMemberRequest(
workspace_id="workspace-123",
inviter_user_id="inviter-id",
invitee_email="existing@test.com",
role="member",
)
response, error = use_case.execute(request)
assert response is None
assert error == "User is already a member of this workspace"
def test_invite_member_invalid_role(self, use_case):
"""测试无效角色（不能邀请为 owner）"""
request = InviteMemberRequest(
@@ -280,8 +289,8 @@ class TestInviteMemberUseCase:
invitee_email="newuser@test.com",
role="owner",
)
response, error = use_case.execute(request)
assert response is None
assert "Invalid role" in error
tests/unit/test_jwt_service.py
+40 -72
View File
@@ -1,135 +1,107 @@
"""
JWT 工具类测试
"""
import pytest
from datetime import datetime, timedelta
import pytest
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
from packages.domain.auth.jwt_service import (
JWTService,
JWTConfig,
TokenType,
)
from packages.domain.auth.jwt_service import JWTConfig, JWTService, TokenType
class TestJWTService:
"""JWT 服务测试"""
@pytest.fixture
def jwt_service(self):
"""创建 JWT 服务实例"""
config = JWTConfig()
config.SECRET_KEY = "test-secret-key-for-testing"
return JWTService(config)
def test_create_access_token(self, jwt_service):
"""测试创建 access_token"""
token = jwt_service.create_access_token(
user_id="user-123",
workspace_id="workspace-456",
role="admin"
)
token = jwt_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="admin")
assert isinstance(token, str)
assert len(token) > 0
# 验证 Token 内容
payload = jwt_service.verify_access_token(token)
assert payload["sub"] == "user-123"
assert payload["workspace_id"] == "workspace-456"
assert payload["role"] == "admin"
assert payload["type"] == TokenType.ACCESS
def test_create_refresh_token(self, jwt_service):
"""测试创建 refresh_token"""
token = jwt_service.create_refresh_token(
user_id="user-123",
session_id="session-789"
)
token = jwt_service.create_refresh_token(user_id="user-123", session_id="session-789")
assert isinstance(token, str)
assert len(token) > 0
# 验证 Token 内容
payload = jwt_service.verify_refresh_token(token)
assert payload["sub"] == "user-123"
assert payload["session_id"] == "session-789"
assert payload["type"] == TokenType.REFRESH
def test_verify_valid_access_token(self, jwt_service):
"""测试验证有效的 access_token"""
token = jwt_service.create_access_token(
user_id="user-123",
workspace_id="workspace-456",
role="member"
)
token = jwt_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="member")
payload = jwt_service.verify_access_token(token)
assert payload["sub"] == "user-123"
assert payload["workspace_id"] == "workspace-456"
assert payload["role"] == "member"
def test_verify_expired_token(self, jwt_service):
"""测试验证过期的 Token"""
# 创建一个已过期的配置（使用相同的 SECRET_KEY）
config = JWTConfig()
config.SECRET_KEY = "test-secret-key-for-testing" # 与 fixture 相同
config.ACCESS_TOKEN_EXPIRE_MINUTES = -1 # 负数，立即过期
expired_service = JWTService(config)
token = expired_service.create_access_token(
user_id="user-123",
workspace_id="workspace-456",
role="admin"
)
token = expired_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="admin")
# 验证应该抛出过期异常
with pytest.raises(ExpiredSignatureError):
jwt_service.verify_access_token(token)
def test_verify_invalid_token(self, jwt_service):
"""测试验证无效的 Token"""
invalid_token = "invalid.token.string"
with pytest.raises(InvalidTokenError):
jwt_service.verify_access_token(invalid_token)
def test_verify_wrong_token_type(self, jwt_service):
"""测试验证错误类型的 Token"""
# 创建 refresh_token
refresh_token = jwt_service.create_refresh_token(
user_id="user-123",
session_id="session-789"
)
refresh_token = jwt_service.create_refresh_token(user_id="user-123", session_id="session-789")
# 用 verify_access_token 验证应该失败
with pytest.raises(ValueError, match="Token type must be 'access'"):
jwt_service.verify_access_token(refresh_token)
# 反过来也一样
access_token = jwt_service.create_access_token(
user_id="user-123",
workspace_id="workspace-456",
role="admin"
)
access_token = jwt_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="admin")
with pytest.raises(ValueError, match="Token type must be 'refresh'"):
jwt_service.verify_refresh_token(access_token)
def test_verify_tampered_token(self, jwt_service):
"""测试验证被篡改的 Token"""
token = jwt_service.create_access_token(
user_id="user-123",
workspace_id="workspace-456",
role="admin"
)
token = jwt_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="admin")
# 篡改 Token（修改最后几个字符）
tampered_token = token[:-5] + "XXXXX"
with pytest.raises(InvalidTokenError):
jwt_service.verify_access_token(tampered_token)
def test_additional_claims(self, jwt_service):
"""测试额外的声明"""
token = jwt_service.create_access_token(
@@ -138,27 +110,23 @@ class TestJWTService:
role="admin",
additional_claims={
"email": "user@example.com",
"display_name": "Test User"
}
"display_name": "Test User",
},
)
payload = jwt_service.verify_access_token(token)
assert payload["email"] == "user@example.com"
assert payload["display_name"] == "Test User"
def test_decode_unsafe(self, jwt_service):
"""测试不安全解码（不验证签名）"""
token = jwt_service.create_access_token(
user_id="user-123",
workspace_id="workspace-456",
role="admin"
)
token = jwt_service.create_access_token(user_id="user-123", workspace_id="workspace-456", role="admin")
# 不验证签名地解码
payload = jwt_service.decode_token_unsafe(token)
assert payload is not None
assert payload["sub"] == "user-123"
# 无效 Token 应该返回 None
invalid_payload = jwt_service.decode_token_unsafe("invalid.token")
assert invalid_payload is None
tests/unit/test_list_members_use_case.py
+47 -44
View File
@@ -1,43 +1,46 @@
"""
获取成员列表 Use Case 测试
"""
import pytest
from unittest.mock import Mock
from datetime import datetime, timezone
from unittest.mock import Mock
import pytest
from packages.application.workspace.list_members_use_case import (
ListMembersUseCase,
ListMembersRequest,
ListMembersUseCase,
)
from packages.domain.entities import (
User,
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(
@@ -45,7 +48,7 @@ class TestListMembersUseCase:
workspace_member_repository=mock_member_repo,
user_repository=mock_user_repo,
)
@pytest.fixture
def test_workspace(self):
return Workspace(
@@ -53,7 +56,7 @@ class TestListMembersUseCase:
name="Test Workspace",
owner_user_id="owner-id",
)
def test_list_members_success(
self,
use_case,
@@ -64,7 +67,7 @@ class TestListMembersUseCase:
):
"""测试获取成员列表成功"""
mock_workspace_repo.find_by_id.return_value = test_workspace
# 请求者是 Admin
requester_member = WorkspaceMember(
id="member-1",
@@ -73,7 +76,7 @@ class TestListMembersUseCase:
role=WorkspaceMemberRole.ADMIN,
)
mock_member_repo.find_by_workspace_and_user.return_value = requester_member
# 3 个成员
member1 = WorkspaceMember(
id="member-1",
@@ -82,7 +85,7 @@ class TestListMembersUseCase:
role=WorkspaceMemberRole.OWNER,
invited_by=None,
)
member2 = WorkspaceMember(
id="member-2",
workspace_id="workspace-123",
@@ -90,7 +93,7 @@ class TestListMembersUseCase:
role=WorkspaceMemberRole.ADMIN,
invited_by="owner-id",
)
member3 = WorkspaceMember(
id="member-3",
workspace_id="workspace-123",
@@ -98,9 +101,9 @@ class TestListMembersUseCase:
role=WorkspaceMemberRole.MEMBER,
invited_by="admin-id",
)
mock_member_repo.find_by_workspace.return_value = [member1, member2, member3]
# 用户信息
user1 = User(
id="owner-id",
@@ -108,34 +111,34 @@ class TestListMembersUseCase:
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"
@@ -143,19 +146,19 @@ class TestListMembersUseCase:
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,
@@ -163,17 +166,17 @@ class TestListMembersUseCase:
):
"""测试工作空间不存在"""
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,
@@ -184,17 +187,17 @@ class TestListMembersUseCase:
"""测试请求者不是成员"""
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,
@@ -204,7 +207,7 @@ class TestListMembersUseCase:
):
"""测试空工作空间（理论上不应该发生）"""
mock_workspace_repo.find_by_id.return_value = test_workspace
requester_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -213,38 +216,38 @@ class TestListMembersUseCase:
)
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"
tests/unit/test_list_workspaces_use_case.py
+50 -51
View File
@@ -1,45 +1,44 @@
"""
获取工作空间列表和详情 Use Case 测试
"""
import pytest
from unittest.mock import Mock
from datetime import datetime, timezone
from unittest.mock import Mock
import pytest
from packages.application.workspace.list_workspaces_use_case import (
ListWorkspacesUseCase,
ListWorkspacesRequest,
GetWorkspaceDetailUseCase,
GetWorkspaceDetailRequest,
GetWorkspaceDetailUseCase,
ListWorkspacesRequest,
ListWorkspacesUseCase,
)
from packages.domain.entities import (
Workspace,
WorkspaceMember,
WorkspaceMemberRole,
)
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,
@@ -54,40 +53,40 @@ class TestListWorkspacesUseCase:
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"
@@ -95,7 +94,7 @@ class TestListWorkspacesUseCase:
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"
@@ -103,7 +102,7 @@ class TestListWorkspacesUseCase:
assert ws2.user_role == "member"
assert ws2.member_count == 5
assert ws2.subscription_plan == "pro"
def test_list_workspaces_no_memberships(
self,
use_case,
@@ -111,46 +110,46 @@ class TestListWorkspacesUseCase:
):
"""测试用户没有加入任何工作空间"""
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(
@@ -163,7 +162,7 @@ class TestGetWorkspaceDetailUseCase:
max_storage_gb=100,
used_storage_gb=25.5,
)
def test_get_workspace_detail_success(
self,
use_case,
@@ -173,7 +172,7 @@ class TestGetWorkspaceDetailUseCase:
):
"""测试获取工作空间详情成功"""
mock_workspace_repo.find_by_id.return_value = test_workspace
membership = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -182,14 +181,14 @@ class TestGetWorkspaceDetailUseCase:
)
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"
@@ -201,7 +200,7 @@ class TestGetWorkspaceDetailUseCase:
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,
@@ -209,17 +208,17 @@ class TestGetWorkspaceDetailUseCase:
):
"""测试工作空间不存在"""
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,
@@ -230,37 +229,37 @@ class TestGetWorkspaceDetailUseCase:
"""测试用户不是成员"""
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"
tests/unit/test_login_use_case.py
+50 -47
View File
@@ -1,35 +1,38 @@
"""
用户登录 Use Case 测试
"""
import pytest
from unittest.mock import Mock
from datetime import datetime, timezone
from unittest.mock import Mock
import pytest
from packages.application.auth import (
LoginUseCase,
LoginRequest,
LogoutUseCase,
LoginUseCase,
LogoutRequest,
LogoutUseCase,
)
from packages.domain.entities import User
from packages.domain.auth import password_hasher
from packages.domain.entities import User
class TestLoginUseCase:
"""登录用例测试"""
@pytest.fixture
def mock_user_repo(self):
repo = Mock()
repo.find_by_email = Mock(return_value=None)
repo.save = Mock()
return repo
@pytest.fixture
def use_case(self, mock_user_repo):
session_store = Mock()
session_store.save_session.return_value = True
return LoginUseCase(user_repository=mock_user_repo, session_store=session_store)
@pytest.fixture
def test_user(self):
"""创建测试用户"""
@@ -42,20 +45,20 @@ class TestLoginUseCase:
password_hash=password_hash,
email_verified=True,
)
def test_login_success(self, use_case, mock_user_repo, test_user):
"""测试登录成功"""
mock_user_repo.find_by_email.return_value = test_user
request = LoginRequest(
email="test@example.com",
password="SecurePass123",
device_info="Chrome/Windows",
ip_address="192.168.1.1",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.user_id == "user-123"
@@ -64,124 +67,124 @@ class TestLoginUseCase:
assert response.access_token != ""
assert response.refresh_token != ""
assert response.expires_in > 0
# 验证保存了 session
use_case.session_store.save_session.assert_called_once()
# 验证更新了最后登录信息
mock_user_repo.save.assert_called_once()
def test_login_invalid_email(self, use_case, mock_user_repo):
"""测试邮箱不存在"""
mock_user_repo.find_by_email.return_value = None
request = LoginRequest(
email="nonexistent@example.com",
password="SecurePass123",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Invalid email or password"
def test_login_wrong_password(self, use_case, mock_user_repo, test_user):
"""测试密码错误"""
mock_user_repo.find_by_email.return_value = test_user
request = LoginRequest(
email="test@example.com",
password="WrongPassword123",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Invalid email or password"
def test_login_missing_email(self, use_case):
"""测试缺少邮箱"""
request = LoginRequest(
email="",
password="SecurePass123",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Email is required"
def test_login_missing_password(self, use_case, mock_user_repo, test_user):
"""测试缺少密码"""
mock_user_repo.find_by_email.return_value = test_user
request = LoginRequest(
email="test@example.com",
password="",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Password is required"
class TestLogoutUseCase:
"""登出用例测试"""
@pytest.fixture
def use_case(self):
session_store = Mock()
return LogoutUseCase(session_store=session_store)
def test_logout_current_device(self, use_case):
"""测试登出当前设备"""
use_case.session_store.delete_session.return_value = True
request = LogoutRequest(
user_id="user-123",
session_id="session-abc",
logout_all_devices=False,
)
success, error = use_case.execute(request)
assert success is True
assert error is None
use_case.session_store.delete_session.assert_called_once_with("session-abc")
def test_logout_all_devices(self, use_case):
"""测试登出所有设备"""
use_case.session_store.delete_all_user_sessions.return_value = 3
request = LogoutRequest(
user_id="user-123",
logout_all_devices=True,
)
success, error = use_case.execute(request)
assert success is True
assert error is None
use_case.session_store.delete_all_user_sessions.assert_called_once_with("user-123")
def test_logout_session_not_found(self, use_case):
"""测试 session 不存在"""
use_case.session_store.delete_session.return_value = False
request = LogoutRequest(
user_id="user-123",
session_id="nonexistent",
logout_all_devices=False,
)
success, error = use_case.execute(request)
assert success is False
assert error == "Session not found"
def test_logout_missing_session_id(self, use_case):
"""测试缺少 session_id"""
request = LogoutRequest(
@@ -189,8 +192,8 @@ class TestLogoutUseCase:
session_id=None,
logout_all_devices=False,
)
success, error = use_case.execute(request)
assert success is False
assert error == "Session ID is required"
tests/unit/test_password_hasher.py
+34 -35
View File
@@ -1,92 +1,91 @@
"""
密码哈希工具测试
"""
import pytest
from packages.domain.auth.password_hasher import (
PasswordHasher,
PasswordValidator,
)
from packages.domain.auth.password_hasher import PasswordHasher, PasswordValidator
class TestPasswordHasher:
"""密码哈希测试"""
@pytest.fixture
def hasher(self):
"""创建密码哈希器"""
return PasswordHasher(rounds=4) # 测试用低 cost，加快速度
def test_hash_password(self, hasher):
"""测试密码哈希"""
password = "MySecurePassword123"
hashed = hasher.hash_password(password)
assert isinstance(hashed, str)
assert len(hashed) > 0
assert hashed != password # 哈希后不等于原文
assert hashed.startswith("$2b$") # bcrypt 格式
def test_hash_same_password_different_result(self, hasher):
"""测试相同密码每次哈希结果不同（因为 salt 不同）"""
password = "MySecurePassword123"
hash1 = hasher.hash_password(password)
hash2 = hasher.hash_password(password)
assert hash1 != hash2 # salt 不同，哈希不同
def test_verify_correct_password(self, hasher):
"""测试验证正确的密码"""
password = "MySecurePassword123"
hashed = hasher.hash_password(password)
assert hasher.verify_password(password, hashed) is True
def test_verify_incorrect_password(self, hasher):
"""测试验证错误的密码"""
password = "MySecurePassword123"
hashed = hasher.hash_password(password)
assert hasher.verify_password("WrongPassword", hashed) is False
def test_verify_empty_password(self, hasher):
"""测试空密码验证"""
hashed = hasher.hash_password("test")
assert hasher.verify_password("", hashed) is False
def test_verify_empty_hash(self, hasher):
"""测试空哈希验证"""
assert hasher.verify_password("test", "") is False
def test_verify_invalid_hash(self, hasher):
"""测试无效的哈希"""
assert hasher.verify_password("test", "invalid-hash") is False
def test_hash_empty_password(self, hasher):
"""测试哈希空密码应该失败"""
with pytest.raises(ValueError, match="Password cannot be empty"):
hasher.hash_password("")
def test_invalid_rounds(self):
"""测试无效的 rounds 参数"""
with pytest.raises(ValueError, match="rounds must be between 4 and 31"):
PasswordHasher(rounds=2)
with pytest.raises(ValueError, match="rounds must be between 4 and 31"):
PasswordHasher(rounds=50)
def test_unicode_password(self, hasher):
"""测试 Unicode 密码"""
password = "密码123!@#"
hashed = hasher.hash_password(password)
assert hasher.verify_password(password, hashed) is True
assert hasher.verify_password("错误密码", hashed) is False
class TestPasswordValidator:
"""密码验证器测试"""
@pytest.fixture
def validator(self):
"""创建密码验证器"""
@@ -97,37 +96,37 @@ class TestPasswordValidator:
require_digit=True,
require_special=False,
)
def test_valid_password(self, validator):
"""测试有效密码"""
valid, error = validator.validate("MyPassword123")
assert valid is True
assert error is None
def test_password_too_short(self, validator):
"""测试密码太短"""
valid, error = validator.validate("Pass1")
assert valid is False
assert "at least 8 characters" in error
def test_password_no_uppercase(self, validator):
"""测试没有大写字母"""
valid, error = validator.validate("mypassword123")
assert valid is False
assert "uppercase letter" in error
def test_password_no_lowercase(self, validator):
"""测试没有小写字母"""
valid, error = validator.validate("MYPASSWORD123")
assert valid is False
assert "lowercase letter" in error
def test_password_no_digit(self, validator):
"""测试没有数字"""
valid, error = validator.validate("MyPassword")
assert valid is False
assert "digit" in error
def test_password_with_special_chars(self):
"""测试要求特殊字符"""
validator = PasswordValidator(
@@ -137,23 +136,23 @@ class TestPasswordValidator:
require_digit=True,
require_special=True,
)
# 没有特殊字符
valid, error = validator.validate("MyPassword123")
assert valid is False
assert "special character" in error
# 有特殊字符
valid, error = validator.validate("MyPassword123!")
assert valid is True
assert error is None
def test_empty_password(self, validator):
"""测试空密码"""
valid, error = validator.validate("")
assert valid is False
assert "cannot be empty" in error
def test_custom_min_length(self):
"""测试自定义最小长度"""
validator = PasswordValidator(
@@ -163,11 +162,11 @@ class TestPasswordValidator:
require_digit=False,
require_special=False,
)
valid, error = validator.validate("short")
assert valid is False
assert "at least 12 characters" in error
valid, error = validator.validate("longenoughpassword")
assert valid is True
assert error is None
tests/unit/test_password_reset_use_case.py
+43 -40
View File
@@ -1,28 +1,31 @@
"""
密码重置 Use Case 测试
"""
import pytest
from unittest.mock import Mock
from datetime import datetime, timedelta, timezone
from unittest.mock import Mock
import pytest
from packages.application.auth.password_reset_use_case import (
RequestPasswordResetUseCase,
RequestPasswordResetRequest,
ResetPasswordUseCase,
RequestPasswordResetUseCase,
ResetPasswordRequest,
ResetPasswordUseCase,
)
from packages.domain.entities import User
class TestRequestPasswordResetUseCase:
"""请求密码重置测试"""
@pytest.fixture
def mock_user_repo(self):
repo = Mock()
repo.find_by_email = Mock(return_value=None)
repo.save = Mock()
return repo
@pytest.fixture
def use_case(self, mock_user_repo):
email_service = Mock()
@@ -33,7 +36,7 @@ class TestRequestPasswordResetUseCase:
token_expire_hours=1,
email_service=email_service,
)
@pytest.fixture
def test_user(self):
return User(
@@ -43,62 +46,62 @@ class TestRequestPasswordResetUseCase:
display_name="Test User",
password_hash="hash",
)
def test_request_reset_success(self, use_case, mock_user_repo, test_user):
"""测试请求重置成功"""
mock_user_repo.find_by_email.return_value = test_user
request = RequestPasswordResetRequest(email="test@example.com")
success, error = use_case.execute(request)
assert success is True
assert error is None
# 验证保存了用户
mock_user_repo.save.assert_called_once()
saved_user = mock_user_repo.save.call_args[0][0]
assert saved_user.password_reset_token is not None
assert saved_user.password_reset_expires_at is not None
# 验证发送了邮件
use_case.email_service.send_password_reset_email.assert_called_once()
def test_request_reset_user_not_exists(self, use_case, mock_user_repo):
"""测试用户不存在（仍返回成功，避免暴露）"""
mock_user_repo.find_by_email.return_value = None
request = RequestPasswordResetRequest(email="nonexistent@example.com")
success, error = use_case.execute(request)
assert success is True # 安全考虑，仍返回成功
assert error is None
# 不发送邮件
use_case.email_service.send_password_reset_email.assert_not_called()
def test_request_reset_missing_email(self, use_case):
"""测试缺少邮箱"""
request = RequestPasswordResetRequest(email="")
success, error = use_case.execute(request)
assert success is False
assert error == "Email is required"
class TestResetPasswordUseCase:
"""重置密码测试"""
@pytest.fixture
def mock_user_repo(self):
repo = Mock()
repo.find_by_password_reset_token = Mock(return_value=None)
repo.save = Mock()
return repo
@pytest.fixture
def use_case(self, mock_user_repo):
return ResetPasswordUseCase(user_repository=mock_user_repo)
@pytest.fixture
def test_user(self):
return User(
@@ -110,68 +113,68 @@ class TestResetPasswordUseCase:
password_reset_token="valid-token",
password_reset_expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
)
def test_reset_password_success(self, use_case, mock_user_repo, test_user):
"""测试重置密码成功"""
mock_user_repo.find_by_password_reset_token.return_value = test_user
request = ResetPasswordRequest(
token="valid-token",
new_password="NewSecurePass123",
)
success, error = use_case.execute(request)
assert success is True
assert error is None
# 验证密码已更新
assert test_user.password_hash != "old-hash"
assert test_user.password_reset_token is None
assert test_user.password_reset_expires_at is None
# 验证保存了用户
mock_user_repo.save.assert_called_once()
def test_reset_password_weak_password(self, use_case, mock_user_repo, test_user):
"""测试弱密码"""
mock_user_repo.find_by_password_reset_token.return_value = test_user
request = ResetPasswordRequest(
token="valid-token",
new_password="weak",
)
success, error = use_case.execute(request)
assert success is False
assert "at least 8 characters" in error
def test_reset_password_invalid_token(self, use_case, mock_user_repo):
"""测试无效令牌"""
mock_user_repo.find_by_password_reset_token.return_value = None
request = ResetPasswordRequest(
token="invalid-token",
new_password="NewSecurePass123",
)
success, error = use_case.execute(request)
assert success is False
assert error == "Invalid or expired reset token"
def test_reset_password_expired_token(self, use_case, mock_user_repo, test_user):
"""测试过期令牌"""
test_user.password_reset_expires_at = datetime.now(timezone.utc) - timedelta(hours=1)
mock_user_repo.find_by_password_reset_token.return_value = test_user
request = ResetPasswordRequest(
token="valid-token",
new_password="NewSecurePass123",
)
success, error = use_case.execute(request)
assert success is False
assert error == "Reset token has expired"
def test_reset_password_missing_token(self, use_case):
"""测试缺少令牌"""
request = ResetPasswordRequest(
@@ -179,19 +182,19 @@ class TestResetPasswordUseCase:
new_password="NewSecurePass123",
)
success, error = use_case.execute(request)
assert success is False
assert error == "Reset token is required"
def test_reset_password_missing_password(self, use_case, mock_user_repo, test_user):
"""测试缺少新密码"""
mock_user_repo.find_by_password_reset_token.return_value = test_user
request = ResetPasswordRequest(
token="valid-token",
new_password="",
)
success, error = use_case.execute(request)
assert success is False
assert error == "New password is required"
tests/unit/test_permissions.py
+45 -49
View File
@@ -1,32 +1,28 @@
"""
权限验证辅助函数测试
"""
import pytest
from unittest.mock import Mock
from packages.domain.permissions import (
PermissionChecker,
Permission,
has_permission,
)
from packages.domain.entities import (
WorkspaceMember,
WorkspaceMemberRole,
)
import pytest
from packages.domain.entities import WorkspaceMember, WorkspaceMemberRole
from packages.domain.permissions import Permission, PermissionChecker, has_permission
class TestPermissionChecker:
"""权限检查器测试"""
@pytest.fixture
def mock_member_repo(self):
repo = Mock()
repo.find_by_workspace_and_user = Mock(return_value=None)
return repo
@pytest.fixture
def checker(self, mock_member_repo):
return PermissionChecker(workspace_member_repository=mock_member_repo)
def test_check_workspace_access_has_access(self, checker, mock_member_repo):
"""测试有访问权限"""
member = WorkspaceMember(
@@ -36,21 +32,21 @@ class TestPermissionChecker:
role=WorkspaceMemberRole.MEMBER,
)
mock_member_repo.find_by_workspace_and_user.return_value = member
has_access, role = checker.check_workspace_access("workspace-123", "user-123")
assert has_access is True
assert role == "member"
def test_check_workspace_access_no_access(self, checker, mock_member_repo):
"""测试无访问权限"""
mock_member_repo.find_by_workspace_and_user.return_value = None
has_access, role = checker.check_workspace_access("workspace-123", "user-123")
assert has_access is False
assert role is None
def test_check_is_owner_true(self, checker, mock_member_repo):
"""测试是 Owner"""
member = WorkspaceMember(
@@ -60,11 +56,11 @@ class TestPermissionChecker:
role=WorkspaceMemberRole.OWNER,
)
mock_member_repo.find_by_workspace_and_user.return_value = member
is_owner = checker.check_is_owner("workspace-123", "user-123")
assert is_owner is True
def test_check_is_owner_false(self, checker, mock_member_repo):
"""测试不是 Owner"""
member = WorkspaceMember(
@@ -74,11 +70,11 @@ class TestPermissionChecker:
role=WorkspaceMemberRole.ADMIN,
)
mock_member_repo.find_by_workspace_and_user.return_value = member
is_owner = checker.check_is_owner("workspace-123", "user-123")
assert is_owner is False
def test_check_is_admin_or_owner_admin(self, checker, mock_member_repo):
"""测试是 Admin"""
member = WorkspaceMember(
@@ -88,11 +84,11 @@ class TestPermissionChecker:
role=WorkspaceMemberRole.ADMIN,
)
mock_member_repo.find_by_workspace_and_user.return_value = member
is_admin_or_owner = checker.check_is_admin_or_owner("workspace-123", "user-123")
assert is_admin_or_owner is True
def test_check_is_admin_or_owner_owner(self, checker, mock_member_repo):
"""测试是 Owner"""
member = WorkspaceMember(
@@ -102,11 +98,11 @@ class TestPermissionChecker:
role=WorkspaceMemberRole.OWNER,
)
mock_member_repo.find_by_workspace_and_user.return_value = member
is_admin_or_owner = checker.check_is_admin_or_owner("workspace-123", "user-123")
assert is_admin_or_owner is True
def test_check_is_admin_or_owner_member(self, checker, mock_member_repo):
"""测试是普通成员"""
member = WorkspaceMember(
@@ -116,11 +112,11 @@ class TestPermissionChecker:
role=WorkspaceMemberRole.MEMBER,
)
mock_member_repo.find_by_workspace_and_user.return_value = member
is_admin_or_owner = checker.check_is_admin_or_owner("workspace-123", "user-123")
assert is_admin_or_owner is False
def test_check_can_create_project_member(self, checker, mock_member_repo):
"""测试 Member 可以创建项目"""
member = WorkspaceMember(
@@ -130,11 +126,11 @@ class TestPermissionChecker:
role=WorkspaceMemberRole.MEMBER,
)
mock_member_repo.find_by_workspace_and_user.return_value = member
can_create = checker.check_can_create_project("workspace-123", "user-123")
assert can_create is True
def test_check_can_create_project_viewer(self, checker, mock_member_repo):
"""测试 Viewer 不能创建项目"""
member = WorkspaceMember(
@@ -144,11 +140,11 @@ class TestPermissionChecker:
role=WorkspaceMemberRole.VIEWER,
)
mock_member_repo.find_by_workspace_and_user.return_value = member
can_create = checker.check_can_create_project("workspace-123", "user-123")
assert can_create is False
def test_check_can_delete_project_member(self, checker, mock_member_repo):
"""测试 Member 不能删除项目"""
member = WorkspaceMember(
@@ -158,11 +154,11 @@ class TestPermissionChecker:
role=WorkspaceMemberRole.MEMBER,
)
mock_member_repo.find_by_workspace_and_user.return_value = member
can_delete = checker.check_can_delete_project("workspace-123", "user-123")
assert can_delete is False
def test_check_can_delete_project_admin(self, checker, mock_member_repo):
"""测试 Admin 可以删除项目"""
member = WorkspaceMember(
@@ -172,33 +168,33 @@ class TestPermissionChecker:
role=WorkspaceMemberRole.ADMIN,
)
mock_member_repo.find_by_workspace_and_user.return_value = member
can_delete = checker.check_can_delete_project("workspace-123", "user-123")
assert can_delete is True
class TestPermissionFunctions:
"""权限函数测试"""
def test_has_permission_owner(self):
"""测试 Owner 权限"""
assert has_permission(WorkspaceMemberRole.OWNER, Permission.WORKSPACE_DELETE) is True
assert has_permission(WorkspaceMemberRole.OWNER, Permission.MEMBER_REMOVE) is True
assert has_permission(WorkspaceMemberRole.OWNER, Permission.PROJECT_CREATE) is True
def test_has_permission_admin(self):
"""测试 Admin 权限"""
assert has_permission(WorkspaceMemberRole.ADMIN, Permission.WORKSPACE_EDIT) is True
assert has_permission(WorkspaceMemberRole.ADMIN, Permission.MEMBER_REMOVE) is True
assert has_permission(WorkspaceMemberRole.ADMIN, Permission.WORKSPACE_DELETE) is False
def test_has_permission_member(self):
"""测试 Member 权限"""
assert has_permission(WorkspaceMemberRole.MEMBER, Permission.PROJECT_CREATE) is True
assert has_permission(WorkspaceMemberRole.MEMBER, Permission.PROJECT_DELETE) is False
assert has_permission(WorkspaceMemberRole.MEMBER, Permission.MEMBER_INVITE) is False
def test_has_permission_viewer(self):
"""测试 Viewer 权限"""
assert has_permission(WorkspaceMemberRole.VIEWER, Permission.WORKSPACE_VIEW) is True
tests/unit/test_quota.py
+43 -44
View File
@@ -1,39 +1,38 @@
"""
配额检查服务测试
"""
import pytest
from unittest.mock import Mock
from packages.domain.quota import (
QuotaChecker,
QuotaWarningLevel,
get_warning_level,
)
import pytest
from packages.domain.entities import Workspace
from packages.domain.quota import QuotaChecker, QuotaWarningLevel, get_warning_level
class TestQuotaChecker:
"""配额检查器测试"""
@pytest.fixture
def mock_workspace_repo(self):
repo = Mock()
repo.find_by_id = Mock(return_value=None)
repo.save = Mock()
return repo
@pytest.fixture
def mock_project_repo(self):
repo = Mock()
repo.count_by_workspace = Mock(return_value=0)
return repo
@pytest.fixture
def checker(self, mock_workspace_repo, mock_project_repo):
return QuotaChecker(
workspace_repository=mock_workspace_repo,
project_repository=mock_project_repo,
)
def test_check_can_create_project_within_limit(
self,
checker,
@@ -50,12 +49,12 @@ class TestQuotaChecker:
)
mock_workspace_repo.find_by_id.return_value = workspace
mock_project_repo.count_by_workspace.return_value = 2
can_create, error = checker.check_can_create_project("workspace-123")
assert can_create is True
assert error is None
def test_check_can_create_project_at_limit(
self,
checker,
@@ -72,12 +71,12 @@ class TestQuotaChecker:
)
mock_workspace_repo.find_by_id.return_value = workspace
mock_project_repo.count_by_workspace.return_value = 3
can_create, error = checker.check_can_create_project("workspace-123")
assert can_create is False
assert "Project limit reached" in error
def test_check_can_create_project_unlimited(
self,
checker,
@@ -93,13 +92,13 @@ class TestQuotaChecker:
max_projects=999999,
)
mock_workspace_repo.find_by_id.return_value = workspace
mock_project_repo.count_by_workspace.return_value=1000
mock_project_repo.count_by_workspace.return_value = 1000
can_create, error = checker.check_can_create_project("workspace-123")
assert can_create is True
assert error is None
def test_check_storage_available_within_limit(
self,
checker,
@@ -115,12 +114,12 @@ class TestQuotaChecker:
used_storage_gb=5.0,
)
mock_workspace_repo.find_by_id.return_value = workspace
can_store, error = checker.check_storage_available("workspace-123", 3.0)
assert can_store is True
assert error is None
def test_check_storage_available_exceeded(
self,
checker,
@@ -136,12 +135,12 @@ class TestQuotaChecker:
used_storage_gb=8.0,
)
mock_workspace_repo.find_by_id.return_value = workspace
can_store, error = checker.check_storage_available("workspace-123", 3.0)
assert can_store is False
assert "Storage limit exceeded" in error
def test_get_quota_status(
self,
checker,
@@ -160,9 +159,9 @@ class TestQuotaChecker:
)
mock_workspace_repo.find_by_id.return_value = workspace
mock_project_repo.count_by_workspace.return_value = 2
status = checker.get_quota_status("workspace-123")
assert status is not None
assert status["workspace_id"] == "workspace-123"
assert status["subscription_plan"] == "free"
@@ -173,7 +172,7 @@ class TestQuotaChecker:
assert status["storage"]["limit_gb"] == 10
assert status["storage"]["remaining_gb"] == 2.5
assert status["storage"]["usage_percent"] == 75.0
def test_get_quota_status_unlimited(
self,
checker,
@@ -192,12 +191,12 @@ class TestQuotaChecker:
)
mock_workspace_repo.find_by_id.return_value = workspace
mock_project_repo.count_by_workspace.return_value = 1000
status = checker.get_quota_status("workspace-123")
assert status["projects"]["unlimited"] is True
assert status["projects"]["usage_percent"] == 0
def test_update_storage_usage_increase(
self,
checker,
@@ -212,14 +211,14 @@ class TestQuotaChecker:
used_storage_gb=5.0,
)
mock_workspace_repo.find_by_id.return_value = workspace
success, error = checker.update_storage_usage("workspace-123", 2.5)
assert success is True
assert error is None
assert workspace.used_storage_gb == 7.5
mock_workspace_repo.save.assert_called_once()
def test_update_storage_usage_decrease(
self,
checker,
@@ -234,12 +233,12 @@ class TestQuotaChecker:
used_storage_gb=5.0,
)
mock_workspace_repo.find_by_id.return_value = workspace
success, error = checker.update_storage_usage("workspace-123", -2.0)
assert success is True
assert workspace.used_storage_gb == 3.0
def test_update_storage_usage_prevent_negative(
self,
checker,
@@ -254,33 +253,33 @@ class TestQuotaChecker:
used_storage_gb=2.0,
)
mock_workspace_repo.find_by_id.return_value = workspace
success, error = checker.update_storage_usage("workspace-123", -5.0)
assert success is True
assert workspace.used_storage_gb == 0.0
class TestWarningLevel:
"""警告级别测试"""
def test_get_warning_level_normal(self):
"""测试正常级别"""
assert get_warning_level(50.0) == QuotaWarningLevel.NORMAL
assert get_warning_level(79.9) == QuotaWarningLevel.NORMAL
def test_get_warning_level_warning(self):
"""测试警告级别"""
assert get_warning_level(80.0) == QuotaWarningLevel.WARNING
assert get_warning_level(85.0) == QuotaWarningLevel.WARNING
assert get_warning_level(89.9) == QuotaWarningLevel.WARNING
def test_get_warning_level_critical(self):
"""测试严重级别"""
assert get_warning_level(90.0) == QuotaWarningLevel.CRITICAL
assert get_warning_level(95.0) == QuotaWarningLevel.CRITICAL
assert get_warning_level(99.9) == QuotaWarningLevel.CRITICAL
def test_get_warning_level_exceeded(self):
"""测试超出级别"""
assert get_warning_level(100.0) == QuotaWarningLevel.EXCEEDED
tests/unit/test_register_user_use_case.py
+47 -41
View File
@@ -1,20 +1,23 @@
"""
用户注册 Use Case 测试
"""
import pytest
from unittest.mock import Mock
import pytest
from packages.application.auth import (
RegisterUserUseCase,
RegisterUserRequest,
VerifyEmailUseCase,
RegisterUserUseCase,
VerifyEmailRequest,
VerifyEmailUseCase,
)
from packages.domain.entities import User
class TestRegisterUserUseCase:
"""注册用例测试"""
@pytest.fixture
def mock_user_repo(self):
"""Mock 用户仓储"""
@@ -24,7 +27,7 @@ class TestRegisterUserUseCase:
repo.find_by_verification_token = Mock(return_value=None)
repo.save = Mock()
return repo
@pytest.fixture
def use_case(self, mock_user_repo):
"""创建注册用例"""
@@ -35,26 +38,26 @@ class TestRegisterUserUseCase:
base_url="https://test.com",
email_service=email_service,
)
def test_register_user_success(self, use_case, mock_user_repo):
"""测试注册成功"""
request = RegisterUserRequest(
email="test@example.com",
password="SecurePass123",
username="testuser",
display_name="Test User",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.email == "test@example.com"
assert response.username == "testuser"
assert response.display_name == "Test User"
assert response.email_verification_sent is True
# 验证保存了用户
mock_user_repo.save.assert_called_once()
saved_user = mock_user_repo.save.call_args[0][0]
@@ -62,7 +65,7 @@ class TestRegisterUserUseCase:
assert saved_user.password_hash != ""
assert saved_user.email_verified is False
assert saved_user.email_verification_token is not None
def test_register_user_weak_password(self, use_case):
"""测试弱密码"""
request = RegisterUserRequest(
@@ -71,13 +74,13 @@ class TestRegisterUserUseCase:
username="testuser",
display_name="Test User",
)
response, error = use_case.execute(request)
assert response is None
assert error is not None
assert "at least 8 characters" in error
def test_register_user_email_exists(self, use_case, mock_user_repo):
"""测试邮箱已存在"""
# Mock 返回已存在的用户
@@ -88,19 +91,19 @@ class TestRegisterUserUseCase:
display_name="Existing",
)
mock_user_repo.find_by_email.return_value = existing_user
request = RegisterUserRequest(
email="test@example.com",
password="SecurePass123",
username="testuser",
display_name="Test User",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Email already registered"
def test_register_user_username_taken(self, use_case, mock_user_repo):
"""测试用户名已被占用"""
existing_user = User(
@@ -110,19 +113,19 @@ class TestRegisterUserUseCase:
display_name="Other",
)
mock_user_repo.find_by_username.return_value = existing_user
request = RegisterUserRequest(
email="test@example.com",
password="SecurePass123",
username="testuser",
display_name="Test User",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Username already taken"
def test_register_user_missing_email(self, use_case):
"""测试缺少邮箱"""
request = RegisterUserRequest(
@@ -131,25 +134,28 @@ class TestRegisterUserUseCase:
username="testuser",
display_name="Test User",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Email is required"
def test_register_user_email_send_failure(self, use_case, mock_user_repo):
"""测试邮件发送失败（用户仍然创建）"""
use_case.email_service.send_verification_email.return_value = (False, "SMTP error")
use_case.email_service.send_verification_email.return_value = (
False,
"SMTP error",
)
request = RegisterUserRequest(
email="test@example.com",
password="SecurePass123",
username="testuser",
display_name="Test User",
)
response, error = use_case.execute(request)
assert error is None # 用户创建成功
assert response is not None
assert response.email_verification_sent is False # 但邮件发送失败
@@ -157,18 +163,18 @@ class TestRegisterUserUseCase:
class TestVerifyEmailUseCase:
"""邮箱验证用例测试"""
@pytest.fixture
def mock_user_repo(self):
repo = Mock()
repo.find_by_verification_token = Mock(return_value=None)
repo.save = Mock()
return repo
@pytest.fixture
def use_case(self, mock_user_repo):
return VerifyEmailUseCase(user_repository=mock_user_repo)
def test_verify_email_success(self, use_case, mock_user_repo):
"""测试验证成功"""
user = User(
@@ -180,28 +186,28 @@ class TestVerifyEmailUseCase:
email_verification_token="valid-token",
)
mock_user_repo.find_by_verification_token.return_value = user
request = VerifyEmailRequest(token="valid-token")
success, error = use_case.execute(request)
assert success is True
assert error is None
# 验证用户状态已更新
assert user.email_verified is True
assert user.email_verification_token is None
mock_user_repo.save.assert_called_once()
def test_verify_email_invalid_token(self, use_case, mock_user_repo):
"""测试无效令牌"""
mock_user_repo.find_by_verification_token.return_value = None
request = VerifyEmailRequest(token="invalid-token")
success, error = use_case.execute(request)
assert success is False
assert error == "Invalid or expired verification token"
def test_verify_email_already_verified(self, use_case, mock_user_repo):
"""测试已验证的邮箱"""
user = User(
@@ -213,9 +219,9 @@ class TestVerifyEmailUseCase:
email_verification_token="old-token",
)
mock_user_repo.find_by_verification_token.return_value = user
request = VerifyEmailRequest(token="old-token")
success, error = use_case.execute(request)
assert success is True # 已验证也返回成功
assert error is None
tests/unit/test_remove_member_use_case.py
+86 -75
View File
@@ -1,44 +1,43 @@
"""
移除成员 Use Case 测试
"""
import pytest
from unittest.mock import Mock
import pytest
from packages.application.workspace.remove_member_use_case import (
RemoveMemberUseCase,
RemoveMemberRequest,
LeaveWorkspaceUseCase,
LeaveWorkspaceRequest,
LeaveWorkspaceUseCase,
RemoveMemberRequest,
RemoveMemberUseCase,
)
from packages.domain.entities import (
Workspace,
WorkspaceMember,
WorkspaceMemberRole,
)
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(
@@ -46,7 +45,7 @@ class TestRemoveMemberUseCase:
name="Test Workspace",
owner_user_id="owner-id",
)
def test_remove_member_success_by_owner(
self,
use_case,
@@ -56,35 +55,38 @@ class TestRemoveMemberUseCase:
):
"""测试 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]
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,
@@ -94,34 +96,37 @@ class TestRemoveMemberUseCase:
):
"""测试 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]
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,
@@ -131,34 +136,37 @@ class TestRemoveMemberUseCase:
):
"""测试不能移除 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]
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,
@@ -168,34 +176,37 @@ class TestRemoveMemberUseCase:
):
"""测试 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]
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,
@@ -205,27 +216,27 @@ class TestRemoveMemberUseCase:
):
"""测试不能移除自己"""
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,
@@ -235,51 +246,51 @@ class TestRemoveMemberUseCase:
):
"""测试普通成员没有权限移除"""
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(
@@ -287,7 +298,7 @@ class TestLeaveWorkspaceUseCase:
name="Test Workspace",
owner_user_id="owner-id",
)
def test_leave_workspace_success(
self,
use_case,
@@ -297,7 +308,7 @@ class TestLeaveWorkspaceUseCase:
):
"""测试离开工作空间成功"""
mock_workspace_repo.find_by_id.return_value = test_workspace
member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -305,18 +316,18 @@ class TestLeaveWorkspaceUseCase:
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,
@@ -326,7 +337,7 @@ class TestLeaveWorkspaceUseCase:
):
"""测试 Owner 不能离开"""
mock_workspace_repo.find_by_id.return_value = test_workspace
owner_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -334,17 +345,17 @@ class TestLeaveWorkspaceUseCase:
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,
@@ -355,13 +366,13 @@ class TestLeaveWorkspaceUseCase:
"""测试不是成员"""
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"
tests/unit/test_session_store.py
+53 -51
View File
@@ -1,17 +1,19 @@
"""
Redis Session 存储测试
"""
import pytest
import json
from datetime import datetime
from unittest.mock import Mock, MagicMock
from unittest.mock import MagicMock, Mock
import pytest
from packages.domain.auth.session_store import SessionStore
class TestSessionStore:
"""Session 存储测试"""
@pytest.fixture
def mock_redis(self):
"""创建 Mock Redis 客户端"""
@@ -19,46 +21,46 @@ class TestSessionStore:
redis_mock.data = {} # 模拟内存存储
redis_mock.expires = {} # 模拟过期时间
redis_mock.sets = {} # 模拟集合
def setex(key, seconds, value):
redis_mock.data[key] = value
redis_mock.expires[key] = seconds
return True
def get(key):
return redis_mock.data.get(key)
def delete(key):
if key in redis_mock.data:
del redis_mock.data[key]
return 1
return 0
def exists(key):
return 1 if key in redis_mock.data else 0
def ttl(key):
return redis_mock.expires.get(key, -1)
def sadd(key, *values):
if key not in redis_mock.sets:
redis_mock.sets[key] = set()
redis_mock.sets[key].update(values)
return len(values)
def smembers(key):
return redis_mock.sets.get(key, set())
def srem(key, *values):
if key in redis_mock.sets:
redis_mock.sets[key].discard(*values)
return len(values)
return 0
def expire(key, seconds):
redis_mock.expires[key] = seconds
return True
redis_mock.setex = setex
redis_mock.get = get
redis_mock.delete = delete
@@ -68,14 +70,14 @@ class TestSessionStore:
redis_mock.smembers = smembers
redis_mock.srem = srem
redis_mock.expire = expire
return redis_mock
@pytest.fixture
def session_store(self, mock_redis):
"""创建 Session 存储实例"""
return SessionStore(redis_client=mock_redis)
def test_save_session(self, session_store, mock_redis):
"""测试保存 Session"""
result = session_store.save_session(
@@ -86,27 +88,27 @@ class TestSessionStore:
ip_address="192.168.1.1",
expires_in_seconds=3600,
)
assert result is True
# 验证数据已保存
session_key = "session:session-123"
assert session_key in mock_redis.data
session_data = json.loads(mock_redis.data[session_key])
assert session_data["session_id"] == "session-123"
assert session_data["user_id"] == "user-456"
assert session_data["device_info"] == "Chrome/Windows"
assert session_data["ip_address"] == "192.168.1.1"
# 验证 refresh_token 已保存
refresh_token_key = "refresh_token:session-123"
assert mock_redis.data[refresh_token_key] == "refresh-token-abc"
# 验证用户 Session 集合已更新
user_sessions_key = "user_sessions:user-456"
assert "session-123" in mock_redis.sets[user_sessions_key]
def test_get_session(self, session_store, mock_redis):
"""测试获取 Session"""
# 先保存
@@ -117,20 +119,20 @@ class TestSessionStore:
device_info="Chrome",
ip_address="127.0.0.1",
)
# 获取
session = session_store.get_session("session-123")
assert session is not None
assert session["session_id"] == "session-123"
assert session["user_id"] == "user-456"
assert session["device_info"] == "Chrome"
def test_get_nonexistent_session(self, session_store):
"""测试获取不存在的 Session"""
session = session_store.get_session("nonexistent")
assert session is None
def test_get_refresh_token(self, session_store):
"""测试获取 refresh_token"""
session_store.save_session(
@@ -140,10 +142,10 @@ class TestSessionStore:
device_info="Chrome",
ip_address="127.0.0.1",
)
token = session_store.get_refresh_token("session-123")
assert token == "my-refresh-token"
def test_update_last_active(self, session_store, mock_redis):
"""测试更新最后活跃时间"""
session_store.save_session(
@@ -153,19 +155,19 @@ class TestSessionStore:
device_info="Chrome",
ip_address="127.0.0.1",
)
# 获取原始时间
session1 = session_store.get_session("session-123")
original_time = session1["last_active_at"]
# 更新
result = session_store.update_last_active("session-123")
assert result is True
# 验证时间已更新
session2 = session_store.get_session("session-123")
assert session2["last_active_at"] >= original_time
def test_delete_session(self, session_store, mock_redis):
"""测试删除 Session"""
session_store.save_session(
@@ -175,22 +177,22 @@ class TestSessionStore:
device_info="Chrome",
ip_address="127.0.0.1",
)
# 删除
result = session_store.delete_session("session-123")
assert result is True
# 验证已删除
session = session_store.get_session("session-123")
assert session is None
token = session_store.get_refresh_token("session-123")
assert token is None
# 验证从用户集合中移除
user_sessions_key = "user_sessions:user-456"
assert "session-123" not in mock_redis.sets.get(user_sessions_key, set())
def test_get_user_sessions(self, session_store):
"""测试获取用户的所有 Session"""
# 创建多个 Session
@@ -201,7 +203,7 @@ class TestSessionStore:
device_info="Chrome",
ip_address="192.168.1.1",
)
session_store.save_session(
session_id="session-2",
user_id="user-456",
@@ -209,15 +211,15 @@ class TestSessionStore:
device_info="Firefox",
ip_address="192.168.1.2",
)
# 获取
sessions = session_store.get_user_sessions("user-456")
assert len(sessions) == 2
session_ids = [s["session_id"] for s in sessions]
assert "session-1" in session_ids
assert "session-2" in session_ids
def test_delete_all_user_sessions(self, session_store):
"""测试删除用户的所有 Session"""
# 创建多个 Session
@@ -228,7 +230,7 @@ class TestSessionStore:
device_info="Chrome",
ip_address="127.0.0.1",
)
session_store.save_session(
session_id="session-2",
user_id="user-456",
@@ -236,19 +238,19 @@ class TestSessionStore:
device_info="Firefox",
ip_address="127.0.0.1",
)
# 删除所有
count = session_store.delete_all_user_sessions("user-456")
assert count == 2
# 验证已删除
sessions = session_store.get_user_sessions("user-456")
assert len(sessions) == 0
def test_session_exists(self, session_store):
"""测试检查 Session 是否存在"""
assert session_store.session_exists("nonexistent") is False
session_store.save_session(
session_id="session-123",
user_id="user-456",
@@ -256,9 +258,9 @@ class TestSessionStore:
device_info="Chrome",
ip_address="127.0.0.1",
)
assert session_store.session_exists("session-123") is True
def test_multiple_users(self, session_store):
"""测试多用户隔离"""
# 用户 1 的 Session
@@ -269,7 +271,7 @@ class TestSessionStore:
device_info="Chrome",
ip_address="127.0.0.1",
)
# 用户 2 的 Session
session_store.save_session(
session_id="session-user2",
@@ -278,12 +280,12 @@ class TestSessionStore:
device_info="Firefox",
ip_address="127.0.0.1",
)
# 验证隔离
user1_sessions = session_store.get_user_sessions("user-1")
assert len(user1_sessions) == 1
assert user1_sessions[0]["session_id"] == "session-user1"
user2_sessions = session_store.get_user_sessions("user-2")
assert len(user2_sessions) == 1
assert user2_sessions[0]["session_id"] == "session-user2"
tests/unit/test_subscription_use_case.py
+56 -57
View File
@@ -1,44 +1,43 @@
"""
Subscription 管理 Use Case 测试
"""
import pytest
from unittest.mock import Mock
import pytest
from packages.application.workspace.subscription_use_case import (
UpgradeSubscriptionUseCase,
UpgradeSubscriptionRequest,
CancelSubscriptionUseCase,
CancelSubscriptionRequest,
CancelSubscriptionUseCase,
UpgradeSubscriptionRequest,
UpgradeSubscriptionUseCase,
)
from packages.domain.entities import (
Workspace,
WorkspaceMember,
WorkspaceMemberRole,
)
from packages.domain.entities import Workspace, WorkspaceMember, WorkspaceMemberRole
class TestUpgradeSubscriptionUseCase:
"""升级订阅测试"""
@pytest.fixture
def mock_workspace_repo(self):
repo = Mock()
repo.find_by_id = Mock(return_value=None)
repo.save = Mock()
return repo
@pytest.fixture
def mock_member_repo(self):
repo = Mock()
repo.find_by_workspace_and_user = Mock(return_value=None)
return repo
@pytest.fixture
def use_case(self, mock_workspace_repo, mock_member_repo):
return UpgradeSubscriptionUseCase(
workspace_repository=mock_workspace_repo,
workspace_member_repository=mock_member_repo,
)
def test_upgrade_from_free_to_pro(
self,
use_case,
@@ -53,7 +52,7 @@ class TestUpgradeSubscriptionUseCase:
subscription_plan="free",
)
mock_workspace_repo.find_by_id.return_value = workspace
owner_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -61,27 +60,27 @@ class TestUpgradeSubscriptionUseCase:
role=WorkspaceMemberRole.OWNER,
)
mock_member_repo.find_by_workspace_and_user.return_value = owner_member
request = UpgradeSubscriptionRequest(
workspace_id="workspace-123",
requester_user_id="owner-id",
new_plan="pro",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.old_plan == "free"
assert response.new_plan == "pro"
assert response.max_projects == 999999
assert response.max_storage_gb == 100
# 验证更新了 workspace
assert workspace.subscription_plan == "pro"
assert workspace.max_projects == 999999
assert workspace.subscription_expires_at is not None
def test_upgrade_from_pro_to_enterprise(
self,
use_case,
@@ -98,7 +97,7 @@ class TestUpgradeSubscriptionUseCase:
max_storage_gb=100,
)
mock_workspace_repo.find_by_id.return_value = workspace
owner_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -106,21 +105,21 @@ class TestUpgradeSubscriptionUseCase:
role=WorkspaceMemberRole.OWNER,
)
mock_member_repo.find_by_workspace_and_user.return_value = owner_member
request = UpgradeSubscriptionRequest(
workspace_id="workspace-123",
requester_user_id="owner-id",
new_plan="enterprise",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.old_plan == "pro"
assert response.new_plan == "enterprise"
assert response.max_storage_gb == 1000
def test_upgrade_cannot_downgrade(
self,
use_case,
@@ -135,7 +134,7 @@ class TestUpgradeSubscriptionUseCase:
subscription_plan="pro",
)
mock_workspace_repo.find_by_id.return_value = workspace
owner_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -143,18 +142,18 @@ class TestUpgradeSubscriptionUseCase:
role=WorkspaceMemberRole.OWNER,
)
mock_member_repo.find_by_workspace_and_user.return_value = owner_member
request = UpgradeSubscriptionRequest(
workspace_id="workspace-123",
requester_user_id="owner-id",
new_plan="free",
)
response, error = use_case.execute(request)
assert response is None
assert "Cannot downgrade" in error
def test_upgrade_already_on_plan(
self,
use_case,
@@ -169,7 +168,7 @@ class TestUpgradeSubscriptionUseCase:
subscription_plan="pro",
)
mock_workspace_repo.find_by_id.return_value = workspace
owner_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -177,18 +176,18 @@ class TestUpgradeSubscriptionUseCase:
role=WorkspaceMemberRole.OWNER,
)
mock_member_repo.find_by_workspace_and_user.return_value = owner_member
request = UpgradeSubscriptionRequest(
workspace_id="workspace-123",
requester_user_id="owner-id",
new_plan="pro",
)
response, error = use_case.execute(request)
assert response is None
assert "already on pro plan" in error
def test_upgrade_only_owner_can_upgrade(
self,
use_case,
@@ -203,7 +202,7 @@ class TestUpgradeSubscriptionUseCase:
subscription_plan="free",
)
mock_workspace_repo.find_by_id.return_value = workspace
admin_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -211,42 +210,42 @@ class TestUpgradeSubscriptionUseCase:
role=WorkspaceMemberRole.ADMIN,
)
mock_member_repo.find_by_workspace_and_user.return_value = admin_member
request = UpgradeSubscriptionRequest(
workspace_id="workspace-123",
requester_user_id="admin-id",
new_plan="pro",
)
response, error = use_case.execute(request)
assert response is None
assert "Only workspace owner" in error
class TestCancelSubscriptionUseCase:
"""取消订阅测试"""
@pytest.fixture
def mock_workspace_repo(self):
repo = Mock()
repo.find_by_id = Mock(return_value=None)
repo.save = Mock()
return repo
@pytest.fixture
def mock_member_repo(self):
repo = Mock()
repo.find_by_workspace_and_user = Mock(return_value=None)
return repo
@pytest.fixture
def use_case(self, mock_workspace_repo, mock_member_repo):
return CancelSubscriptionUseCase(
workspace_repository=mock_workspace_repo,
workspace_member_repository=mock_member_repo,
)
def test_cancel_subscription_success(
self,
use_case,
@@ -263,7 +262,7 @@ class TestCancelSubscriptionUseCase:
max_storage_gb=100,
)
mock_workspace_repo.find_by_id.return_value = workspace
owner_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -271,23 +270,23 @@ class TestCancelSubscriptionUseCase:
role=WorkspaceMemberRole.OWNER,
)
mock_member_repo.find_by_workspace_and_user.return_value = owner_member
request = CancelSubscriptionRequest(
workspace_id="workspace-123",
requester_user_id="owner-id",
)
success, error = use_case.execute(request)
assert success is True
assert error is None
# 验证降级到 free
assert workspace.subscription_plan == "free"
assert workspace.max_projects == 3
assert workspace.max_storage_gb == 10
assert workspace.subscription_expires_at is None
def test_cancel_already_free(
self,
use_case,
@@ -302,7 +301,7 @@ class TestCancelSubscriptionUseCase:
subscription_plan="free",
)
mock_workspace_repo.find_by_id.return_value = workspace
owner_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -310,17 +309,17 @@ class TestCancelSubscriptionUseCase:
role=WorkspaceMemberRole.OWNER,
)
mock_member_repo.find_by_workspace_and_user.return_value = owner_member
request = CancelSubscriptionRequest(
workspace_id="workspace-123",
requester_user_id="owner-id",
)
success, error = use_case.execute(request)
assert success is False
assert "already on free plan" in error
def test_cancel_only_owner_can_cancel(
self,
use_case,
@@ -335,7 +334,7 @@ class TestCancelSubscriptionUseCase:
subscription_plan="pro",
)
mock_workspace_repo.find_by_id.return_value = workspace
admin_member = WorkspaceMember(
id="member-1",
workspace_id="workspace-123",
@@ -343,13 +342,13 @@ class TestCancelSubscriptionUseCase:
role=WorkspaceMemberRole.ADMIN,
)
mock_member_repo.find_by_workspace_and_user.return_value = admin_member
request = CancelSubscriptionRequest(
workspace_id="workspace-123",
requester_user_id="admin-id",
)
success, error = use_case.execute(request)
assert success is False
assert "Only workspace owner" in error
tests/unit/test_update_member_role_use_case.py
+81 -67
View File
@@ -1,42 +1,41 @@
"""
修改成员角色 Use Case 测试
"""
import pytest
from unittest.mock import Mock
import pytest
from packages.application.workspace.update_member_role_use_case import (
UpdateMemberRoleUseCase,
UpdateMemberRoleRequest,
UpdateMemberRoleUseCase,
)
from packages.domain.entities import (
Workspace,
WorkspaceMember,
WorkspaceMemberRole,
)
from packages.domain.entities import Workspace, WorkspaceMember, WorkspaceMemberRole
class TestUpdateMemberRoleUseCase:
"""修改成员角色测试"""
@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.save = Mock()
return repo
@pytest.fixture
def use_case(self, mock_workspace_repo, mock_member_repo):
return UpdateMemberRoleUseCase(
workspace_repository=mock_workspace_repo,
workspace_member_repository=mock_member_repo,
)
@pytest.fixture
def test_workspace(self):
return Workspace(
@@ -44,7 +43,7 @@ class TestUpdateMemberRoleUseCase:
name="Test Workspace",
owner_user_id="owner-id",
)
def test_update_role_success_by_owner(
self,
use_case,
@@ -54,42 +53,45 @@ class TestUpdateMemberRoleUseCase:
):
"""测试 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]
mock_member_repo.find_by_workspace_and_user.side_effect = [
owner_member,
target_member,
]
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="owner-id",
target_user_id="target-id",
new_role="admin",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
assert response.user_id == "target-id"
assert response.old_role == "member"
assert response.new_role == "admin"
# 验证更新了角色
assert target_member.role == "admin"
mock_member_repo.save.assert_called_once()
def test_update_role_success_by_admin(
self,
use_case,
@@ -99,35 +101,38 @@ class TestUpdateMemberRoleUseCase:
):
"""测试 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]
mock_member_repo.find_by_workspace_and_user.side_effect = [
admin_member,
target_member,
]
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="admin-id",
target_user_id="target-id",
new_role="member",
)
response, error = use_case.execute(request)
assert error is None
assert response is not None
def test_update_role_cannot_change_owner(
self,
use_case,
@@ -137,35 +142,38 @@ class TestUpdateMemberRoleUseCase:
):
"""测试不能修改 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]
mock_member_repo.find_by_workspace_and_user.side_effect = [
admin_member,
owner_member,
]
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="admin-id",
target_user_id="owner-id",
new_role="member",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Cannot change the owner's role"
def test_update_role_admin_cannot_change_admin(
self,
use_case,
@@ -175,35 +183,38 @@ class TestUpdateMemberRoleUseCase:
):
"""测试 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]
mock_member_repo.find_by_workspace_and_user.side_effect = [
admin_member1,
admin_member2,
]
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="admin-id-1",
target_user_id="admin-id-2",
new_role="member",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Admins cannot change other admins' roles"
def test_update_role_cannot_change_self(
self,
use_case,
@@ -213,28 +224,28 @@ class TestUpdateMemberRoleUseCase:
):
"""测试不能修改自己的角色"""
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 = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="admin-id",
target_user_id="admin-id",
new_role="member",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Cannot change your own role"
def test_update_role_invalid_role(self, use_case):
"""测试无效角色（不能改为 owner）"""
request = UpdateMemberRoleRequest(
@@ -243,12 +254,12 @@ class TestUpdateMemberRoleUseCase:
target_user_id="target-id",
new_role="owner",
)
response, error = use_case.execute(request)
assert response is None
assert "Invalid role" in error
def test_update_role_already_has_role(
self,
use_case,
@@ -258,35 +269,38 @@ class TestUpdateMemberRoleUseCase:
):
"""测试角色相同"""
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.ADMIN,
)
mock_member_repo.find_by_workspace_and_user.side_effect = [owner_member, target_member]
mock_member_repo.find_by_workspace_and_user.side_effect = [
owner_member,
target_member,
]
request = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="owner-id",
target_user_id="target-id",
new_role="admin",
)
response, error = use_case.execute(request)
assert response is None
assert "already has the admin role" in error
def test_update_role_no_permission(
self,
use_case,
@@ -296,24 +310,24 @@ class TestUpdateMemberRoleUseCase:
):
"""测试普通成员没有权限"""
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 = UpdateMemberRoleRequest(
workspace_id="workspace-123",
requester_user_id="user-id",
target_user_id="target-id",
new_role="admin",
)
response, error = use_case.execute(request)
assert response is None
assert error == "Only owners and admins can change member roles"