feat(phase8): 任务 2.03 — API 模板管理 CRUD #144
@@ -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_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
|
||||
from app.api.routes.health import router as health_check_router
|
||||
@@ -116,3 +117,8 @@ api_router.include_router(
|
||||
prefix="/dashboard",
|
||||
tags=["Dashboard"],
|
||||
)
|
||||
api_router.include_router(
|
||||
edit_templates_router,
|
||||
prefix="/edit-templates",
|
||||
tags=["EditTemplate"],
|
||||
)
|
||||
|
||||
@@ -0,0 +1,267 @@
|
||||
"""模板管理 API — Phase 8 模板编排引擎.
|
||||
|
||||
RESTful CRUD for EditTemplate:
|
||||
- GET /api/v1/edit-templates 列表(分页 + 类型筛选)
|
||||
- GET /api/v1/edit-templates/{id} 详情
|
||||
- POST /api/v1/edit-templates 创建(管理员)
|
||||
- PUT /api/v1/edit-templates/{id} 更新
|
||||
- DELETE /api/v1/edit-templates/{id} 删除(软删除 → inactive)
|
||||
"""
|
||||
|
||||
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 SQLAlchemyEditTemplateRepository
|
||||
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
# ── Pydantic Schemas ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class EditTemplateCreateRequest(BaseModel):
|
||||
"""创建模板请求体"""
|
||||
|
||||
name: str = Field(..., min_length=1, max_length=200, description="模板名称")
|
||||
description: str = Field(default="", max_length=2000, description="模板描述")
|
||||
template_type: str = Field(default="default", max_length=50, description="模板类型")
|
||||
config: dict[str, Any] = Field(default_factory=dict, description="模板配置 (JSON)")
|
||||
preview_url: str = Field(default="", max_length=500, description="预览地址")
|
||||
sort_weight: int = Field(default=0, ge=0, le=9999, description="排序权重")
|
||||
|
||||
|
||||
class EditTemplateUpdateRequest(BaseModel):
|
||||
"""更新模板请求体"""
|
||||
|
||||
name: Optional[str] = Field(default=None, min_length=1, max_length=200, description="模板名称")
|
||||
description: Optional[str] = Field(default=None, max_length=2000, description="模板描述")
|
||||
template_type: Optional[str] = Field(default=None, max_length=50, description="模板类型")
|
||||
config: Optional[dict[str, Any]] = Field(default=None, description="模板配置 (JSON)")
|
||||
preview_url: Optional[str] = Field(default=None, max_length=500, description="预览地址")
|
||||
sort_weight: Optional[int] = Field(default=None, ge=0, le=9999, description="排序权重")
|
||||
status: Optional[str] = Field(default=None, description="状态: active / inactive")
|
||||
|
||||
|
||||
class EditTemplateResponse(BaseModel):
|
||||
"""模板响应体"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
description: str
|
||||
template_type: str
|
||||
config: dict[str, Any]
|
||||
preview_url: str
|
||||
sort_weight: int
|
||||
status: str
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
|
||||
class EditTemplateListResponse(BaseModel):
|
||||
"""模板列表响应体"""
|
||||
|
||||
items: List[EditTemplateResponse]
|
||||
total: int
|
||||
page: int
|
||||
page_size: int
|
||||
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _to_response(t: EditTemplate) -> EditTemplateResponse:
|
||||
return EditTemplateResponse(
|
||||
id=t.id,
|
||||
name=t.name,
|
||||
description=t.description,
|
||||
template_type=t.template_type,
|
||||
config=t.config,
|
||||
preview_url=t.preview_url,
|
||||
sort_weight=t.sort_weight,
|
||||
status=t.status.value if hasattr(t.status, "value") else t.status,
|
||||
created_at=t.created_at,
|
||||
updated_at=t.updated_at,
|
||||
)
|
||||
|
||||
|
||||
# ── Routes ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@router.get("", response_model=EditTemplateListResponse)
|
||||
def list_templates(
|
||||
page: int = Query(default=1, ge=1, description="页码"),
|
||||
page_size: int = Query(default=20, ge=1, le=100, description="每页数量"),
|
||||
template_type: Optional[str] = Query(default=None, description="按类型筛选"),
|
||||
status_filter: Optional[str] = Query(
|
||||
default=None,
|
||||
alias="status",
|
||||
description="按状态筛选: active / inactive",
|
||||
),
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> EditTemplateListResponse:
|
||||
"""获取模板列表(支持分页、按类型/状态筛选)"""
|
||||
repo = SQLAlchemyEditTemplateRepository(db)
|
||||
|
||||
# 解析状态筛选
|
||||
status_enum: Optional[EditTemplateStatus] = None
|
||||
if status_filter:
|
||||
try:
|
||||
status_enum = EditTemplateStatus(status_filter)
|
||||
except ValueError:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"无效的状态值: {status_filter},可选值: active, inactive",
|
||||
)
|
||||
|
||||
skip = (page - 1) * page_size
|
||||
templates = repo.list_all(
|
||||
template_type=template_type,
|
||||
status=status_enum,
|
||||
skip=skip,
|
||||
limit=page_size,
|
||||
)
|
||||
total = repo.count(template_type=template_type, status=status_enum)
|
||||
|
||||
return EditTemplateListResponse(
|
||||
items=[_to_response(t) for t in templates],
|
||||
total=total,
|
||||
page=page,
|
||||
page_size=page_size,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/{template_id}", response_model=EditTemplateResponse)
|
||||
def get_template(
|
||||
template_id: str,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> EditTemplateResponse:
|
||||
"""获取单个模板详情"""
|
||||
repo = SQLAlchemyEditTemplateRepository(db)
|
||||
template = repo.get(template_id)
|
||||
if template is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"模板不存在: {template_id}",
|
||||
)
|
||||
return _to_response(template)
|
||||
|
||||
|
||||
@router.post("", response_model=EditTemplateResponse, status_code=status.HTTP_201_CREATED)
|
||||
def create_template(
|
||||
body: EditTemplateCreateRequest,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> EditTemplateResponse:
|
||||
"""创建模板(管理员)"""
|
||||
repo = SQLAlchemyEditTemplateRepository(db)
|
||||
try:
|
||||
template = EditTemplate.create(
|
||||
name=body.name,
|
||||
description=body.description,
|
||||
template_type=body.template_type,
|
||||
config=body.config,
|
||||
preview_url=body.preview_url,
|
||||
sort_weight=body.sort_weight,
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=str(exc),
|
||||
)
|
||||
created = repo.create(template)
|
||||
logger.info("创建模板: id=%s name=%s by user=%s", created.id, created.name, current_user.user.id)
|
||||
return _to_response(created)
|
||||
|
||||
|
||||
@router.put("/{template_id}", response_model=EditTemplateResponse)
|
||||
def update_template(
|
||||
template_id: str,
|
||||
body: EditTemplateUpdateRequest,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> EditTemplateResponse:
|
||||
"""更新模板"""
|
||||
repo = SQLAlchemyEditTemplateRepository(db)
|
||||
existing = repo.get(template_id)
|
||||
if existing is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"模板不存在: {template_id}",
|
||||
)
|
||||
|
||||
# 部分更新:仅覆盖 body 中提供的字段
|
||||
new_name = body.name if body.name is not None else existing.name
|
||||
new_description = body.description if body.description is not None else existing.description
|
||||
new_template_type = body.template_type if body.template_type is not None else existing.template_type
|
||||
new_config = body.config if body.config is not None else existing.config
|
||||
new_preview_url = body.preview_url if body.preview_url is not None else existing.preview_url
|
||||
new_sort_weight = body.sort_weight if body.sort_weight is not None else existing.sort_weight
|
||||
new_status = existing.status
|
||||
if body.status is not None:
|
||||
try:
|
||||
new_status = EditTemplateStatus(body.status)
|
||||
except ValueError:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"无效的状态值: {body.status},可选值: active, inactive",
|
||||
)
|
||||
|
||||
try:
|
||||
updated = EditTemplate(
|
||||
id=existing.id,
|
||||
name=new_name.strip() if new_name else existing.name,
|
||||
description=new_description.strip() if new_description is not None else existing.description,
|
||||
template_type=new_template_type.strip() if new_template_type else existing.template_type,
|
||||
config=new_config,
|
||||
preview_url=new_preview_url.strip() if new_preview_url is not None else existing.preview_url,
|
||||
sort_weight=new_sort_weight,
|
||||
status=new_status,
|
||||
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", template_id, current_user.user.id)
|
||||
return _to_response(result)
|
||||
|
||||
|
||||
@router.delete("/{template_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
def delete_template(
|
||||
template_id: str,
|
||||
db: Session = Depends(get_db_session),
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> None:
|
||||
"""删除模板(软删除 → 设为 inactive)"""
|
||||
repo = SQLAlchemyEditTemplateRepository(db)
|
||||
existing = repo.get(template_id)
|
||||
if existing is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"模板不存在: {template_id}",
|
||||
)
|
||||
|
||||
# 软删除:将状态设为 inactive
|
||||
existing.deactivate()
|
||||
repo.update(existing)
|
||||
logger.info("删除模板(软删除): id=%s by user=%s", template_id, current_user.user.id)
|
||||
@@ -119,9 +119,16 @@ class SQLAlchemyEditTemplateRepository:
|
||||
self.session.commit()
|
||||
return True
|
||||
|
||||
def count(self, *, status: Optional[EditTemplateStatus] = None) -> int:
|
||||
def count(
|
||||
self,
|
||||
*,
|
||||
template_type: Optional[str] = None,
|
||||
status: Optional[EditTemplateStatus] = None,
|
||||
) -> int:
|
||||
"""统计模板数量"""
|
||||
query = self.session.query(EditTemplateModel)
|
||||
if template_type:
|
||||
query = query.filter(EditTemplateModel.template_type == template_type)
|
||||
if status:
|
||||
query = query.filter(EditTemplateModel.status == status)
|
||||
return query.count()
|
||||
|
||||
@@ -0,0 +1,425 @@
|
||||
"""模板管理 API 单元测试 — Phase 8 任务 2.03.
|
||||
|
||||
覆盖 5 个端点:
|
||||
GET /api/v1/edit-templates — 列表(分页 + 筛选)
|
||||
GET /api/v1/edit-templates/{id} — 详情
|
||||
POST /api/v1/edit-templates — 创建
|
||||
PUT /api/v1/edit-templates/{id} — 更新
|
||||
DELETE /api/v1/edit-templates/{id} — 软删除
|
||||
|
||||
使用 FastAPI TestClient + Stub Repository + dependency_overrides.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, 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
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
|
||||
|
||||
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
|
||||
|
||||
|
||||
# ── Stub Repository ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class StubEditTemplateRepository:
|
||||
"""内存中模拟 EditTemplate 仓储"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._store: dict[str, EditTemplate] = {}
|
||||
|
||||
def list_all(
|
||||
self,
|
||||
*,
|
||||
template_type: Optional[str] = None,
|
||||
status: Optional[EditTemplateStatus] = None,
|
||||
skip: int = 0,
|
||||
limit: int = 50,
|
||||
) -> list[EditTemplate]:
|
||||
items = list(self._store.values())
|
||||
if template_type:
|
||||
items = [t for t in items if t.template_type == template_type]
|
||||
if status:
|
||||
items = [t for t in items if t.status == status]
|
||||
items.sort(key=lambda t: t.created_at, reverse=True)
|
||||
return items[skip : skip + limit]
|
||||
|
||||
def list_active(
|
||||
self,
|
||||
*,
|
||||
template_type: Optional[str] = None,
|
||||
skip: int = 0,
|
||||
limit: int = 50,
|
||||
) -> list[EditTemplate]:
|
||||
return self.list_all(
|
||||
template_type=template_type, status=EditTemplateStatus.ACTIVE, skip=skip, limit=limit
|
||||
)
|
||||
|
||||
def get(self, template_id: str) -> Optional[EditTemplate]:
|
||||
return self._store.get(template_id)
|
||||
|
||||
def create(self, template: EditTemplate) -> EditTemplate:
|
||||
self._store[template.id] = template
|
||||
return template
|
||||
|
||||
def update(self, template: EditTemplate) -> EditTemplate:
|
||||
if template.id not in self._store:
|
||||
raise ValueError(f"EditTemplate {template.id} not found")
|
||||
self._store[template.id] = template
|
||||
return template
|
||||
|
||||
def delete(self, template_id: str) -> bool:
|
||||
if template_id in self._store:
|
||||
del self._store[template_id]
|
||||
return True
|
||||
return False
|
||||
|
||||
def count(
|
||||
self,
|
||||
*,
|
||||
template_type: Optional[str] = None,
|
||||
status: Optional[EditTemplateStatus] = None,
|
||||
) -> int:
|
||||
items = list(self._store.values())
|
||||
if template_type:
|
||||
items = [t for t in items if t.template_type == template_type]
|
||||
if status:
|
||||
items = [t for t in items if t.status == status]
|
||||
return len(items)
|
||||
|
||||
|
||||
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakeUser:
|
||||
id: str = "user-001"
|
||||
email: str = "test@example.com"
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakeAuthenticatedUser:
|
||||
user: FakeUser = field(default_factory=FakeUser)
|
||||
session_id: str | None = None
|
||||
token_type: str | None = None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def stub_repo() -> StubEditTemplateRepository:
|
||||
return StubEditTemplateRepository()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app(stub_repo: StubEditTemplateRepository) -> FastAPI:
|
||||
"""构建测试 FastAPI 应用,注入 Stub Repository"""
|
||||
from app.api.routes.edit_templates import router
|
||||
from app.auth import get_current_user
|
||||
from app.dependencies import get_db_session
|
||||
import app.api.routes.edit_templates as route_module
|
||||
|
||||
# 替换路由模块中的 Repository 类
|
||||
original_repo_cls = route_module.SQLAlchemyEditTemplateRepository
|
||||
route_module.SQLAlchemyEditTemplateRepository = lambda session: stub_repo
|
||||
|
||||
test_app = FastAPI()
|
||||
test_app.include_router(router, prefix="/api/v1/edit-templates")
|
||||
|
||||
# 覆盖依赖
|
||||
def override_get_db_session():
|
||||
yield MagicMock()
|
||||
|
||||
def override_get_current_user():
|
||||
return FakeAuthenticatedUser()
|
||||
|
||||
test_app.dependency_overrides[get_db_session] = override_get_db_session
|
||||
test_app.dependency_overrides[get_current_user] = override_get_current_user
|
||||
|
||||
yield test_app
|
||||
|
||||
# 恢复
|
||||
route_module.SQLAlchemyEditTemplateRepository = original_repo_cls
|
||||
test_app.dependency_overrides.clear()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(app: FastAPI) -> TestClient:
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def _make_template(name: str = "测试模板", **kwargs: Any) -> EditTemplate:
|
||||
return EditTemplate.create(name=name, **kwargs)
|
||||
|
||||
|
||||
# ── GET /api/v1/edit-templates (列表) ─────────────────────────────────────────
|
||||
|
||||
|
||||
class TestListTemplates:
|
||||
def test_empty_list(self, client: TestClient) -> None:
|
||||
resp = client.get("/api/v1/edit-templates")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["items"] == []
|
||||
assert data["total"] == 0
|
||||
assert data["page"] == 1
|
||||
|
||||
def test_list_with_items(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
|
||||
for i in range(3):
|
||||
stub_repo.create(_make_template(f"模板{i}"))
|
||||
resp = client.get("/api/v1/edit-templates")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == 3
|
||||
assert len(data["items"]) == 3
|
||||
|
||||
def test_pagination(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
|
||||
for i in range(5):
|
||||
stub_repo.create(_make_template(f"模板{i}"))
|
||||
resp = client.get("/api/v1/edit-templates?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
|
||||
assert data["page_size"] == 2
|
||||
|
||||
def test_filter_by_type(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
|
||||
stub_repo.create(_make_template("Vlog模板", template_type="vlog"))
|
||||
stub_repo.create(_make_template("短视频模板", template_type="short"))
|
||||
stub_repo.create(_make_template("另一个Vlog", template_type="vlog"))
|
||||
resp = client.get("/api/v1/edit-templates?template_type=vlog")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == 2
|
||||
assert all(item["template_type"] == "vlog" for item in data["items"])
|
||||
|
||||
def test_filter_by_status(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
|
||||
t1 = _make_template("活跃模板")
|
||||
stub_repo.create(t1)
|
||||
t2 = _make_template("停用模板", status=EditTemplateStatus.INACTIVE)
|
||||
stub_repo.create(t2)
|
||||
resp = client.get("/api/v1/edit-templates?status=active")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["total"] == 1
|
||||
assert data["items"][0]["name"] == "活跃模板"
|
||||
|
||||
def test_invalid_status_filter(self, client: TestClient) -> None:
|
||||
resp = client.get("/api/v1/edit-templates?status=invalid")
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_invalid_page(self, client: TestClient) -> None:
|
||||
resp = client.get("/api/v1/edit-templates?page=0")
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
# ── GET /api/v1/edit-templates/{id} (详情) ────────────────────────────────────
|
||||
|
||||
|
||||
class TestGetTemplate:
|
||||
def test_get_existing(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
|
||||
t = _make_template("详情模板", description="这是描述", template_type="vlog")
|
||||
stub_repo.create(t)
|
||||
resp = client.get(f"/api/v1/edit-templates/{t.id}")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["id"] == t.id
|
||||
assert data["name"] == "详情模板"
|
||||
assert data["description"] == "这是描述"
|
||||
assert data["template_type"] == "vlog"
|
||||
assert data["status"] == "active"
|
||||
|
||||
def test_get_not_found(self, client: TestClient) -> None:
|
||||
resp = client.get("/api/v1/edit-templates/nonexistent-id")
|
||||
assert resp.status_code == 404
|
||||
assert "不存在" in resp.json()["detail"]
|
||||
|
||||
|
||||
# ── POST /api/v1/edit-templates (创建) ────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCreateTemplate:
|
||||
def test_create_basic(self, client: TestClient) -> None:
|
||||
resp = client.post("/api/v1/edit-templates", json={"name": "新模板"})
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["name"] == "新模板"
|
||||
assert data["description"] == ""
|
||||
assert data["template_type"] == "default"
|
||||
assert data["status"] == "active"
|
||||
assert data["sort_weight"] == 0
|
||||
assert "id" in data
|
||||
|
||||
def test_create_with_all_fields(self, client: TestClient) -> None:
|
||||
body = {
|
||||
"name": "完整模板",
|
||||
"description": "完整描述",
|
||||
"template_type": "vlog",
|
||||
"config": {"key": "value"},
|
||||
"preview_url": "https://example.com/preview.mp4",
|
||||
"sort_weight": 10,
|
||||
}
|
||||
resp = client.post("/api/v1/edit-templates", json=body)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["name"] == "完整模板"
|
||||
assert data["description"] == "完整描述"
|
||||
assert data["template_type"] == "vlog"
|
||||
assert data["config"] == {"key": "value"}
|
||||
assert data["preview_url"] == "https://example.com/preview.mp4"
|
||||
assert data["sort_weight"] == 10
|
||||
|
||||
def test_create_empty_name(self, client: TestClient) -> None:
|
||||
resp = client.post("/api/v1/edit-templates", json={"name": ""})
|
||||
assert resp.status_code == 422 # Pydantic min_length=1
|
||||
|
||||
def test_create_whitespace_name(self, client: TestClient) -> None:
|
||||
resp = client.post("/api/v1/edit-templates", json={"name": " "})
|
||||
assert resp.status_code == 400 # domain validation
|
||||
|
||||
def test_create_missing_name(self, client: TestClient) -> None:
|
||||
resp = client.post("/api/v1/edit-templates", json={})
|
||||
assert resp.status_code == 422
|
||||
|
||||
def test_create_negative_sort_weight(self, client: TestClient) -> None:
|
||||
resp = client.post("/api/v1/edit-templates", json={"name": "模板", "sort_weight": -1})
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
# ── PUT /api/v1/edit-templates/{id} (更新) ────────────────────────────────────
|
||||
|
||||
|
||||
class TestUpdateTemplate:
|
||||
def test_update_name(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
|
||||
t = _make_template("旧名称")
|
||||
stub_repo.create(t)
|
||||
resp = client.put(f"/api/v1/edit-templates/{t.id}", json={"name": "新名称"})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "新名称"
|
||||
|
||||
def test_update_multiple_fields(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
|
||||
t = _make_template("模板")
|
||||
stub_repo.create(t)
|
||||
body = {"name": "更新后", "description": "新描述", "sort_weight": 5}
|
||||
resp = client.put(f"/api/v1/edit-templates/{t.id}", json=body)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["name"] == "更新后"
|
||||
assert data["description"] == "新描述"
|
||||
assert data["sort_weight"] == 5
|
||||
|
||||
def test_update_status(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
|
||||
t = _make_template("模板")
|
||||
stub_repo.create(t)
|
||||
resp = client.put(f"/api/v1/edit-templates/{t.id}", json={"status": "inactive"})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "inactive"
|
||||
|
||||
def test_update_invalid_status(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
|
||||
t = _make_template("模板")
|
||||
stub_repo.create(t)
|
||||
resp = client.put(f"/api/v1/edit-templates/{t.id}", json={"status": "bogus"})
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_update_not_found(self, client: TestClient) -> None:
|
||||
resp = client.put("/api/v1/edit-templates/nonexistent", json={"name": "x"})
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_partial_update_preserves_others(
|
||||
self, client: TestClient, stub_repo: StubEditTemplateRepository
|
||||
) -> None:
|
||||
t = _make_template("原名", description="原描述", template_type="vlog")
|
||||
stub_repo.create(t)
|
||||
resp = client.put(f"/api/v1/edit-templates/{t.id}", json={"name": "新名"})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["name"] == "新名"
|
||||
assert data["description"] == "原描述"
|
||||
assert data["template_type"] == "vlog"
|
||||
|
||||
def test_update_empty_body(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
|
||||
t = _make_template("模板")
|
||||
stub_repo.create(t)
|
||||
resp = client.put(f"/api/v1/edit-templates/{t.id}", json={})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "模板"
|
||||
|
||||
|
||||
# ── DELETE /api/v1/edit-templates/{id} (软删除) ───────────────────────────────
|
||||
|
||||
|
||||
class TestDeleteTemplate:
|
||||
def test_soft_delete(self, client: TestClient, stub_repo: StubEditTemplateRepository) -> None:
|
||||
t = _make_template("待删除")
|
||||
stub_repo.create(t)
|
||||
resp = client.delete(f"/api/v1/edit-templates/{t.id}")
|
||||
assert resp.status_code == 204
|
||||
# 软删除后仍存在,但状态为 inactive
|
||||
updated = stub_repo.get(t.id)
|
||||
assert updated is not None
|
||||
assert updated.status == EditTemplateStatus.INACTIVE
|
||||
|
||||
def test_soft_delete_not_found(self, client: TestClient) -> None:
|
||||
resp = client.delete("/api/v1/edit-templates/nonexistent")
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_soft_delete_idempotent(
|
||||
self, client: TestClient, stub_repo: StubEditTemplateRepository
|
||||
) -> None:
|
||||
t = _make_template("模板")
|
||||
stub_repo.create(t)
|
||||
# 第一次删除
|
||||
resp1 = client.delete(f"/api/v1/edit-templates/{t.id}")
|
||||
assert resp1.status_code == 204
|
||||
# 第二次删除(已经是 inactive,但仍可再次设为 inactive)
|
||||
resp2 = client.delete(f"/api/v1/edit-templates/{t.id}")
|
||||
assert resp2.status_code == 204
|
||||
|
||||
def test_deleted_not_in_active_list(
|
||||
self, client: TestClient, stub_repo: StubEditTemplateRepository
|
||||
) -> None:
|
||||
t = _make_template("模板")
|
||||
stub_repo.create(t)
|
||||
client.delete(f"/api/v1/edit-templates/{t.id}")
|
||||
resp = client.get("/api/v1/edit-templates?status=active")
|
||||
data = resp.json()
|
||||
assert data["total"] == 0
|
||||
|
||||
|
||||
# ── Response Schema 验证 ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestResponseSchema:
|
||||
def test_response_has_all_fields(
|
||||
self, client: TestClient, stub_repo: StubEditTemplateRepository
|
||||
) -> None:
|
||||
t = _make_template("模板", description="描述", template_type="vlog")
|
||||
stub_repo.create(t)
|
||||
resp = client.get(f"/api/v1/edit-templates/{t.id}")
|
||||
data = resp.json()
|
||||
expected_keys = {
|
||||
"id", "name", "description", "template_type",
|
||||
"config", "preview_url", "sort_weight", "status",
|
||||
"created_at", "updated_at",
|
||||
}
|
||||
assert set(data.keys()) == expected_keys
|
||||
|
||||
def test_list_response_structure(self, client: TestClient) -> None:
|
||||
resp = client.get("/api/v1/edit-templates")
|
||||
data = resp.json()
|
||||
assert "items" in data
|
||||
assert "total" in data
|
||||
assert "page" in data
|
||||
assert "page_size" in data
|
||||
assert isinstance(data["items"], list)
|
||||
Reference in New Issue
Block a user