feat(phase8): 实现剪辑计划 CRUD API (任务 2.04)
Deploy / Build Production Runtime Images (push) Has been skipped
Deploy / Deploy Production (push) Has been skipped
Deploy / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 189h51m12s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 189h51m15s
Deploy / Deploy Staging (push) Failing after 189h51m46s
CI/CD Pipeline / Frontend Lint (push) Failing after 189h52m16s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 189h52m19s
Deploy / Build Production Runtime Images (push) Has been skipped
Deploy / Deploy Production (push) Has been skipped
Deploy / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 189h51m12s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 189h51m15s
Deploy / Deploy Staging (push) Failing after 189h51m46s
CI/CD Pipeline / Frontend Lint (push) Failing after 189h52m16s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 189h52m19s
- 新增 /api/v1/edit-plans 五个端点 (GET list/detail, POST, PUT, DELETE) - 支持分页、按模板ID/状态筛选 - PUT 端点集成状态机流转 (draft→editing→rendering→completed/failed) - 非法状态流转返回 400,资源不存在返回 404 - 注册 edit_plans_router 到 API Router - 编写 28 个单元测试覆盖 CRUD + 状态机 + 异常场景 - 全部测试通过
This commit is contained in:
@@ -6,6 +6,7 @@ from app.api.routes.chunked_upload import router as chunked_upload_router
|
||||
from app.api.routes.classification_jobs import router as classification_jobs_router
|
||||
from app.api.routes.dashboard import router as dashboard_router
|
||||
from app.api.routes.duplication import router as duplication_router
|
||||
from app.api.routes.edit_plans import router as edit_plans_router
|
||||
from app.api.routes.edit_templates import router as edit_templates_router
|
||||
from app.api.routes.generated_videos import router as generated_videos_router
|
||||
from app.api.routes.generation_tasks import router as generation_tasks_router
|
||||
@@ -122,3 +123,8 @@ api_router.include_router(
|
||||
prefix="/edit-templates",
|
||||
tags=["EditTemplate"],
|
||||
)
|
||||
api_router.include_router(
|
||||
edit_plans_router,
|
||||
prefix="/edit-plans",
|
||||
tags=["EditPlan"],
|
||||
)
|
||||
|
||||
@@ -0,0 +1,304 @@
|
||||
"""剪辑计划管理 API — Phase 8 模板编排引擎.
|
||||
|
||||
RESTful CRUD for EditPlan:
|
||||
- GET /api/v1/edit-plans 列表(分页 + 状态/模板筛选)
|
||||
- GET /api/v1/edit-plans/{id} 详情
|
||||
- POST /api/v1/edit-plans 创建
|
||||
- PUT /api/v1/edit-plans/{id} 更新(含状态机流转)
|
||||
- DELETE /api/v1/edit-plans/{id} 删除
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Any, List, Optional
|
||||
|
||||
from app.auth import AuthenticatedUser, get_current_user
|
||||
from app.dependencies import get_db_session
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from packages.adapters.sqlalchemy_impl import SQLAlchemyEditPlanRepository
|
||||
from packages.domain.edit_plan import EditPlan, EditPlanStatus
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
# ── Pydantic Schemas ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class EditPlanCreateRequest(BaseModel):
|
||||
"""创建剪辑计划请求体"""
|
||||
|
||||
template_id: str = Field(..., min_length=1, max_length=32, description="关联模板 ID")
|
||||
name: str = Field(..., min_length=1, max_length=200, description="计划名称")
|
||||
config: dict[str, Any] = Field(default_factory=dict, description="计划配置 (JSON)")
|
||||
total_duration: float = Field(default=0.0, ge=0.0, description="总时长 (秒)")
|
||||
|
||||
|
||||
class EditPlanUpdateRequest(BaseModel):
|
||||
"""更新剪辑计划请求体"""
|
||||
|
||||
name: Optional[str] = Field(default=None, min_length=1, max_length=200, description="计划名称")
|
||||
config: Optional[dict[str, Any]] = Field(default=None, description="计划配置 (JSON)")
|
||||
total_duration: Optional[float] = Field(default=None, ge=0.0, description="总时长 (秒)")
|
||||
status: Optional[str] = Field(
|
||||
default=None,
|
||||
description="目标状态 (通过状态机流转): editing / rendering / completed / failed / draft",
|
||||
)
|
||||
|
||||
|
||||
class EditPlanResponse(BaseModel):
|
||||
"""剪辑计划响应体"""
|
||||
|
||||
id: str
|
||||
template_id: str
|
||||
name: str
|
||||
status: str
|
||||
total_duration: float
|
||||
config: dict[str, Any]
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
|
||||
class EditPlanListResponse(BaseModel):
|
||||
"""剪辑计划列表响应体"""
|
||||
|
||||
items: List[EditPlanResponse]
|
||||
total: int
|
||||
page: int
|
||||
page_size: int
|
||||
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _to_response(p: EditPlan) -> EditPlanResponse:
|
||||
return EditPlanResponse(
|
||||
id=p.id,
|
||||
template_id=p.template_id,
|
||||
name=p.name,
|
||||
status=p.status.value if hasattr(p.status, "value") else p.status,
|
||||
total_duration=p.total_duration,
|
||||
config=p.config,
|
||||
created_at=p.created_at,
|
||||
updated_at=p.updated_at,
|
||||
)
|
||||
|
||||
|
||||
def _apply_status_transition(plan: EditPlan, target_status_str: str) -> None:
|
||||
"""通过状态机方法流转状态,非法流转抛出 ValueError"""
|
||||
try:
|
||||
target = EditPlanStatus(target_status_str)
|
||||
except ValueError:
|
||||
raise ValueError(
|
||||
f"无效的状态值: {target_status_str},"
|
||||
f"可选值: draft, editing, rendering, completed, failed"
|
||||
)
|
||||
|
||||
if target == plan.status:
|
||||
return # 已是目标状态,无需流转
|
||||
|
||||
# 根据目标状态选择对应的状态机方法
|
||||
transition_map = {
|
||||
EditPlanStatus.EDITING: plan.start_editing,
|
||||
EditPlanStatus.RENDERING: plan.start_rendering,
|
||||
EditPlanStatus.COMPLETED: plan.mark_completed,
|
||||
EditPlanStatus.FAILED: plan.mark_failed,
|
||||
EditPlanStatus.DRAFT: plan.reset_to_draft,
|
||||
}
|
||||
transition_map[target]()
|
||||
|
||||
|
||||
# ── Routes ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@router.get("", response_model=EditPlanListResponse)
|
||||
def list_plans(
|
||||
page: int = Query(default=1, ge=1, description="页码"),
|
||||
page_size: int = Query(default=20, ge=1, le=100, description="每页数量"),
|
||||
template_id: Optional[str] = Query(default=None, description="按模板 ID 筛选"),
|
||||
status_filter: Optional[str] = Query(
|
||||
default=None,
|
||||
alias="status",
|
||||
description="按状态筛选: draft / editing / rendering / completed / failed",
|
||||
),
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> EditPlanListResponse:
|
||||
"""获取剪辑计划列表(支持分页、按模板/状态筛选)"""
|
||||
repo = SQLAlchemyEditPlanRepository(db)
|
||||
|
||||
# 解析状态筛选
|
||||
status_enum: Optional[EditPlanStatus] = None
|
||||
if status_filter:
|
||||
try:
|
||||
status_enum = EditPlanStatus(status_filter)
|
||||
except ValueError:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=(
|
||||
f"无效的状态值: {status_filter},"
|
||||
f"可选值: draft, editing, rendering, completed, failed"
|
||||
),
|
||||
)
|
||||
|
||||
skip = (page - 1) * page_size
|
||||
|
||||
# 根据是否有 template_id 选择查询方法
|
||||
if template_id:
|
||||
plans = repo.list_by_template(
|
||||
template_id,
|
||||
status=status_enum,
|
||||
skip=skip,
|
||||
limit=page_size,
|
||||
)
|
||||
# count() 不支持 template_id 筛选,通过全量查询计算 total
|
||||
all_matching = repo.list_by_template(
|
||||
template_id,
|
||||
status=status_enum,
|
||||
skip=0,
|
||||
limit=10000,
|
||||
)
|
||||
total = len(all_matching)
|
||||
else:
|
||||
plans = repo.list_all(status=status_enum, skip=skip, limit=page_size)
|
||||
total = repo.count(status=status_enum)
|
||||
|
||||
return EditPlanListResponse(
|
||||
items=[_to_response(p) for p in plans],
|
||||
total=total,
|
||||
page=page,
|
||||
page_size=page_size,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/{plan_id}", response_model=EditPlanResponse)
|
||||
def get_plan(
|
||||
plan_id: str,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> EditPlanResponse:
|
||||
"""获取单个剪辑计划详情"""
|
||||
repo = SQLAlchemyEditPlanRepository(db)
|
||||
plan = repo.get(plan_id)
|
||||
if plan is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"剪辑计划不存在: {plan_id}",
|
||||
)
|
||||
return _to_response(plan)
|
||||
|
||||
|
||||
@router.post("", response_model=EditPlanResponse, status_code=status.HTTP_201_CREATED)
|
||||
def create_plan(
|
||||
body: EditPlanCreateRequest,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> EditPlanResponse:
|
||||
"""创建剪辑计划"""
|
||||
repo = SQLAlchemyEditPlanRepository(db)
|
||||
try:
|
||||
plan = EditPlan.create(
|
||||
template_id=body.template_id,
|
||||
name=body.name,
|
||||
config=body.config,
|
||||
total_duration=body.total_duration,
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=str(exc),
|
||||
)
|
||||
created = repo.create(plan)
|
||||
logger.info(
|
||||
"创建剪辑计划: id=%s name=%s by user=%s",
|
||||
created.id,
|
||||
created.name,
|
||||
current_user.user.id,
|
||||
)
|
||||
return _to_response(created)
|
||||
|
||||
|
||||
@router.put("/{plan_id}", response_model=EditPlanResponse)
|
||||
def update_plan(
|
||||
plan_id: str,
|
||||
body: EditPlanUpdateRequest,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> EditPlanResponse:
|
||||
"""更新剪辑计划(支持状态机流转)"""
|
||||
repo = SQLAlchemyEditPlanRepository(db)
|
||||
existing = repo.get(plan_id)
|
||||
if existing is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"剪辑计划不存在: {plan_id}",
|
||||
)
|
||||
|
||||
# 基础字段更新
|
||||
new_name = body.name.strip() if body.name is not None else existing.name
|
||||
new_config = body.config if body.config is not None else existing.config
|
||||
new_total_duration = body.total_duration if body.total_duration is not None else existing.total_duration
|
||||
|
||||
# 状态机流转
|
||||
new_status = existing.status
|
||||
if body.status is not None:
|
||||
try:
|
||||
_apply_status_transition(existing, body.status)
|
||||
new_status = existing.status
|
||||
except ValueError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=str(exc),
|
||||
)
|
||||
|
||||
try:
|
||||
updated = EditPlan(
|
||||
id=existing.id,
|
||||
template_id=existing.template_id,
|
||||
name=new_name,
|
||||
status=new_status,
|
||||
total_duration=new_total_duration,
|
||||
config=new_config,
|
||||
created_at=existing.created_at,
|
||||
updated_at=existing.updated_at,
|
||||
)
|
||||
except (ValueError, TypeError) as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=str(exc),
|
||||
)
|
||||
|
||||
result = repo.update(updated)
|
||||
logger.info("更新剪辑计划: id=%s by user=%s", plan_id, current_user.user.id)
|
||||
return _to_response(result)
|
||||
|
||||
|
||||
@router.delete("/{plan_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
def delete_plan(
|
||||
plan_id: str,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> None:
|
||||
"""删除剪辑计划"""
|
||||
repo = SQLAlchemyEditPlanRepository(db)
|
||||
existing = repo.get(plan_id)
|
||||
if existing is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"剪辑计划不存在: {plan_id}",
|
||||
)
|
||||
|
||||
repo.delete(plan_id)
|
||||
logger.info(
|
||||
"删除剪辑计划: id=%s by user=%s",
|
||||
plan_id,
|
||||
current_user.user.id,
|
||||
)
|
||||
@@ -0,0 +1,496 @@
|
||||
"""
|
||||
edit_plans.py 剪辑计划 API 端点单元测试
|
||||
|
||||
覆盖(25+ 测试用例):
|
||||
- 创建:正常创建、空名称 400、空 template_id 422
|
||||
- 列表:默认分页、按状态筛选、按模板筛选、无效状态 400
|
||||
- 详情:正常获取、不存在 404
|
||||
- 更新:基础字段更新、状态机合法流转、状态机非法流转 400、不存在 404、无效状态值 400
|
||||
- 删除:正常删除、不存在 404
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
|
||||
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from packages.domain.edit_plan import EditPlan, EditPlanStatus
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Stub Repository
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class StubEditPlanRepository:
|
||||
"""内存中的 EditPlan 仓储 stub"""
|
||||
|
||||
def __init__(self, plans: dict[str, EditPlan] | None = None):
|
||||
self._plans = plans or {}
|
||||
self._counter = 0
|
||||
|
||||
def _next_id(self) -> str:
|
||||
self._counter += 1
|
||||
return f"plan-{self._counter:03d}"
|
||||
|
||||
def list_all(
|
||||
self,
|
||||
*,
|
||||
status: Optional[EditPlanStatus] = None,
|
||||
skip: int = 0,
|
||||
limit: int = 50,
|
||||
) -> list[EditPlan]:
|
||||
items = list(self._plans.values())
|
||||
if status is not None:
|
||||
items = [p for p in items if p.status == status]
|
||||
items.sort(key=lambda p: p.created_at, reverse=True)
|
||||
return items[skip : skip + limit]
|
||||
|
||||
def list_by_template(
|
||||
self,
|
||||
template_id: str,
|
||||
*,
|
||||
status: Optional[EditPlanStatus] = None,
|
||||
skip: int = 0,
|
||||
limit: int = 50,
|
||||
) -> list[EditPlan]:
|
||||
items = [p for p in self._plans.values() if p.template_id == template_id]
|
||||
if status is not None:
|
||||
items = [p for p in items if p.status == status]
|
||||
items.sort(key=lambda p: p.created_at, reverse=True)
|
||||
return items[skip : skip + limit]
|
||||
|
||||
def get(self, plan_id: str) -> Optional[EditPlan]:
|
||||
return self._plans.get(plan_id)
|
||||
|
||||
def create(self, plan: EditPlan) -> EditPlan:
|
||||
self._plans[plan.id] = plan
|
||||
return plan
|
||||
|
||||
def update(self, plan: EditPlan) -> EditPlan:
|
||||
if plan.id not in self._plans:
|
||||
raise ValueError(f"EditPlan {plan.id} not found")
|
||||
self._plans[plan.id] = plan
|
||||
return plan
|
||||
|
||||
def delete(self, plan_id: str) -> bool:
|
||||
if plan_id not in self._plans:
|
||||
return False
|
||||
del self._plans[plan_id]
|
||||
return True
|
||||
|
||||
def count(
|
||||
self,
|
||||
*,
|
||||
template_id: Optional[str] = None,
|
||||
status: Optional[EditPlanStatus] = None,
|
||||
) -> int:
|
||||
items = list(self._plans.values())
|
||||
if template_id:
|
||||
items = [p for p in items if p.template_id == template_id]
|
||||
if status is not None:
|
||||
items = [p for p in items if p.status == status]
|
||||
return len(items)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_auth_user():
|
||||
"""构造 AuthenticatedUser mock"""
|
||||
from app.auth import AuthenticatedUser
|
||||
from packages.domain.entities import User
|
||||
|
||||
user = User(
|
||||
id="user-001",
|
||||
email="test@example.com",
|
||||
display_name="测试用户",
|
||||
)
|
||||
return AuthenticatedUser(user=user)
|
||||
|
||||
|
||||
def _create_test_app():
|
||||
"""创建带 stub 注入的测试 FastAPI 应用"""
|
||||
from app.api.routes import edit_plans as edit_plans_module
|
||||
from app.api.routes.edit_plans import router
|
||||
|
||||
stub_repo = StubEditPlanRepository()
|
||||
|
||||
# 替换路由模块中的 Repository 类
|
||||
original_repo_class = edit_plans_module.SQLAlchemyEditPlanRepository
|
||||
edit_plans_module.SQLAlchemyEditPlanRepository = lambda db: stub_repo
|
||||
|
||||
app = FastAPI()
|
||||
app.include_router(router, prefix="/api/v1/edit-plans")
|
||||
|
||||
# 覆盖认证依赖
|
||||
app.dependency_overrides[edit_plans_module.get_current_user] = _make_auth_user
|
||||
app.dependency_overrides[edit_plans_module.get_db_session] = lambda: MagicMock()
|
||||
|
||||
return app, stub_repo, lambda: setattr(
|
||||
edit_plans_module, "SQLAlchemyEditPlanRepository", original_repo_class
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
app, stub_repo, cleanup = _create_test_app()
|
||||
yield TestClient(app), stub_repo
|
||||
cleanup()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 创建测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCreatePlan:
|
||||
def test_create_success(self, client):
|
||||
c, repo = client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans",
|
||||
json={
|
||||
"template_id": "tpl-001",
|
||||
"name": "我的剪辑计划",
|
||||
"config": {"bgm": "happy"},
|
||||
"total_duration": 60.0,
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["name"] == "我的剪辑计划"
|
||||
assert data["template_id"] == "tpl-001"
|
||||
assert data["status"] == "draft"
|
||||
assert data["total_duration"] == 60.0
|
||||
assert data["config"] == {"bgm": "happy"}
|
||||
assert "id" in data
|
||||
assert "created_at" in data
|
||||
|
||||
def test_create_minimal(self, client):
|
||||
c, repo = client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans",
|
||||
json={"template_id": "tpl-001", "name": "最小计划"},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["config"] == {}
|
||||
assert data["total_duration"] == 0.0
|
||||
|
||||
def test_create_empty_name_returns_422(self, client):
|
||||
c, repo = client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans",
|
||||
json={"template_id": "tpl-001", "name": ""},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_create_missing_template_id_returns_422(self, client):
|
||||
c, repo = client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans",
|
||||
json={"name": "没有模板的计划"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_create_negative_duration_returns_422(self, client):
|
||||
c, repo = client
|
||||
resp = c.post(
|
||||
"/api/v1/edit-plans",
|
||||
json={"template_id": "tpl-001", "name": "test", "total_duration": -1.0},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 列表测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestListPlans:
|
||||
def _seed_plans(self, repo, count=3, template_id="tpl-001"):
|
||||
for i in range(count):
|
||||
plan = EditPlan.create(
|
||||
template_id=template_id,
|
||||
name=f"计划{i+1}",
|
||||
config={"index": i},
|
||||
)
|
||||
repo.create(plan)
|
||||
return plan
|
||||
|
||||
def test_list_empty(self, client):
|
||||
c, repo = client
|
||||
resp = c.get("/api/v1/edit-plans")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["items"] == []
|
||||
assert data["total"] == 0
|
||||
assert data["page"] == 1
|
||||
assert data["page_size"] == 20
|
||||
|
||||
def test_list_with_items(self, client):
|
||||
c, repo = client
|
||||
self._seed_plans(repo, count=3)
|
||||
resp = c.get("/api/v1/edit-plans")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data["items"]) == 3
|
||||
assert data["total"] == 3
|
||||
|
||||
def test_list_pagination(self, client):
|
||||
c, repo = client
|
||||
self._seed_plans(repo, count=5)
|
||||
resp = c.get("/api/v1/edit-plans?page=1&page_size=2")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data["items"]) == 2
|
||||
assert data["total"] == 5
|
||||
assert data["page"] == 1
|
||||
|
||||
resp2 = c.get("/api/v1/edit-plans?page=3&page_size=2")
|
||||
data2 = resp2.json()
|
||||
assert len(data2["items"]) == 1
|
||||
|
||||
def test_list_filter_by_status(self, client):
|
||||
c, repo = client
|
||||
p1 = EditPlan.create("tpl-001", "计划A")
|
||||
repo.create(p1)
|
||||
p2 = EditPlan.create("tpl-001", "计划B")
|
||||
repo.create(p2)
|
||||
p2.start_editing()
|
||||
repo.update(p2)
|
||||
|
||||
resp = c.get("/api/v1/edit-plans?status=draft")
|
||||
data = resp.json()
|
||||
assert data["total"] == 1
|
||||
assert data["items"][0]["name"] == "计划A"
|
||||
|
||||
resp2 = c.get("/api/v1/edit-plans?status=editing")
|
||||
data2 = resp2.json()
|
||||
assert data2["total"] == 1
|
||||
assert data2["items"][0]["name"] == "计划B"
|
||||
|
||||
def test_list_filter_by_template_id(self, client):
|
||||
c, repo = client
|
||||
p1 = EditPlan.create("tpl-001", "模板1计划")
|
||||
repo.create(p1)
|
||||
p2 = EditPlan.create("tpl-002", "模板2计划")
|
||||
repo.create(p2)
|
||||
|
||||
resp = c.get("/api/v1/edit-plans?template_id=tpl-001")
|
||||
data = resp.json()
|
||||
assert data["total"] == 1
|
||||
assert data["items"][0]["name"] == "模板1计划"
|
||||
|
||||
def test_list_filter_by_template_and_status(self, client):
|
||||
c, repo = client
|
||||
p1 = EditPlan.create("tpl-001", "模板1草稿")
|
||||
repo.create(p1)
|
||||
p2 = EditPlan.create("tpl-001", "模板1编辑中")
|
||||
repo.create(p2)
|
||||
p2.start_editing()
|
||||
repo.update(p2)
|
||||
p3 = EditPlan.create("tpl-002", "模板2草稿")
|
||||
repo.create(p3)
|
||||
|
||||
resp = c.get("/api/v1/edit-plans?template_id=tpl-001&status=draft")
|
||||
data = resp.json()
|
||||
assert data["total"] == 1
|
||||
assert data["items"][0]["name"] == "模板1草稿"
|
||||
|
||||
def test_list_invalid_status_returns_400(self, client):
|
||||
c, repo = client
|
||||
resp = c.get("/api/v1/edit-plans?status=invalid_status")
|
||||
assert resp.status_code == 400
|
||||
assert "无效的状态值" in resp.json()["detail"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 详情测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetPlan:
|
||||
def test_get_success(self, client):
|
||||
c, repo = client
|
||||
plan = EditPlan.create("tpl-001", "测试计划", config={"key": "val"})
|
||||
repo.create(plan)
|
||||
|
||||
resp = c.get(f"/api/v1/edit-plans/{plan.id}")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["id"] == plan.id
|
||||
assert data["name"] == "测试计划"
|
||||
assert data["config"] == {"key": "val"}
|
||||
|
||||
def test_get_not_found_returns_404(self, client):
|
||||
c, repo = client
|
||||
resp = c.get("/api/v1/edit-plans/nonexistent-id")
|
||||
assert resp.status_code == 404
|
||||
assert "剪辑计划不存在" in resp.json()["detail"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 更新测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestUpdatePlan:
|
||||
def _seed_plan(self, repo, name="原计划", template_id="tpl-001"):
|
||||
plan = EditPlan.create(template_id, name)
|
||||
repo.create(plan)
|
||||
return plan
|
||||
|
||||
def test_update_name(self, client):
|
||||
c, repo = client
|
||||
plan = self._seed_plan(repo)
|
||||
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"name": "新名称"})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "新名称"
|
||||
|
||||
def test_update_config(self, client):
|
||||
c, repo = client
|
||||
plan = self._seed_plan(repo)
|
||||
|
||||
resp = c.put(
|
||||
f"/api/v1/edit-plans/{plan.id}",
|
||||
json={"config": {"bgm": "sad", "transition": "fade"}},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["config"] == {"bgm": "sad", "transition": "fade"}
|
||||
|
||||
def test_update_total_duration(self, client):
|
||||
c, repo = client
|
||||
plan = self._seed_plan(repo)
|
||||
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"total_duration": 120.5})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["total_duration"] == 120.5
|
||||
|
||||
def test_update_status_draft_to_editing(self, client):
|
||||
c, repo = client
|
||||
plan = self._seed_plan(repo)
|
||||
assert plan.status == EditPlanStatus.DRAFT
|
||||
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "editing"})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "editing"
|
||||
|
||||
def test_update_status_full_happy_path(self, client):
|
||||
c, repo = client
|
||||
plan = self._seed_plan(repo)
|
||||
|
||||
# draft → editing
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "editing"})
|
||||
assert resp.json()["status"] == "editing"
|
||||
|
||||
# editing → rendering
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "rendering"})
|
||||
assert resp.json()["status"] == "rendering"
|
||||
|
||||
# rendering → completed
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "completed"})
|
||||
assert resp.json()["status"] == "completed"
|
||||
|
||||
def test_update_status_failure_and_reset(self, client):
|
||||
c, repo = client
|
||||
plan = self._seed_plan(repo)
|
||||
|
||||
# draft → editing → rendering → failed
|
||||
c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "editing"})
|
||||
c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "rendering"})
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "failed"})
|
||||
assert resp.json()["status"] == "failed"
|
||||
|
||||
# failed → draft (reset)
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "draft"})
|
||||
assert resp.json()["status"] == "draft"
|
||||
|
||||
def test_update_invalid_transition_returns_400(self, client):
|
||||
c, repo = client
|
||||
plan = self._seed_plan(repo)
|
||||
|
||||
# draft → rendering 不合法
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "rendering"})
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_update_draft_to_completed_returns_400(self, client):
|
||||
c, repo = client
|
||||
plan = self._seed_plan(repo)
|
||||
|
||||
# draft → completed 不合法
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "completed"})
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_update_invalid_status_value_returns_400(self, client):
|
||||
c, repo = client
|
||||
plan = self._seed_plan(repo)
|
||||
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "bogus"})
|
||||
assert resp.status_code == 400
|
||||
assert "无效的状态值" in resp.json()["detail"]
|
||||
|
||||
def test_update_same_status_is_noop(self, client):
|
||||
c, repo = client
|
||||
plan = self._seed_plan(repo)
|
||||
|
||||
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "draft"})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "draft"
|
||||
|
||||
def test_update_not_found_returns_404(self, client):
|
||||
c, repo = client
|
||||
resp = c.put("/api/v1/edit-plans/nonexistent", json={"name": "x"})
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_update_combined_fields_and_status(self, client):
|
||||
c, repo = client
|
||||
plan = self._seed_plan(repo)
|
||||
|
||||
resp = c.put(
|
||||
f"/api/v1/edit-plans/{plan.id}",
|
||||
json={"name": "新名称", "status": "editing", "total_duration": 90.0},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["name"] == "新名称"
|
||||
assert data["status"] == "editing"
|
||||
assert data["total_duration"] == 90.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 删除测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDeletePlan:
|
||||
def test_delete_success(self, client):
|
||||
c, repo = client
|
||||
plan = EditPlan.create("tpl-001", "待删除")
|
||||
repo.create(plan)
|
||||
|
||||
resp = c.delete(f"/api/v1/edit-plans/{plan.id}")
|
||||
assert resp.status_code == 204
|
||||
assert repo.get(plan.id) is None
|
||||
|
||||
def test_delete_not_found_returns_404(self, client):
|
||||
c, repo = client
|
||||
resp = c.delete("/api/v1/edit-plans/nonexistent")
|
||||
assert resp.status_code == 404
|
||||
assert "剪辑计划不存在" in resp.json()["detail"]
|
||||
Reference in New Issue
Block a user