feat: 封面模板 CRUD API #1323
@@ -0,0 +1,84 @@
|
||||
"""封面模板表 cover_templates
|
||||
|
||||
Revision ID: 055_cover_templates
|
||||
Revises: 054_confirm_gen_fields
|
||||
Create Date: 2026-08-09
|
||||
|
||||
Changes:
|
||||
1. 新建 cover_templates 表，支持系统预置和用户自定义封面模板
|
||||
2. user_id 为 NULL 表示系统模板，is_system 标记区分
|
||||
3. config 为 JSON 字段，存储封面配置信息
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import context, op
|
||||
|
||||
revision = "055_cover_templates"
|
||||
down_revision = "054_confirm_gen_fields"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
SYSTEM_TEMPLATES = [
|
||||
("a8b0120fd98e44788f5a6590f983d327", "默认模板", {}),
|
||||
("6d8c501b11424432b3df3a45ae89b1a9", "大胆红", {"background_color": "#ef4444"}),
|
||||
("04937fb57fea4bad95e7883e71a6b246", "优雅黑", {"background_color": "#111827"}),
|
||||
("3ff9cc821174437ca53931073e7f536e", "渐变蓝", {"background_color": "#3b82f6"}),
|
||||
("db51b3ea8f1a4f4caa94bf2d51f27d11", "渐变紫", {"background_color": "#8b5cf6"}),
|
||||
("5027d113432a4f798a3b4ee1644d66af", "暖橙", {"background_color": "#f97316"}),
|
||||
("0e10def2b5a148d686416494474726c2", "清新绿", {"background_color": "#22c55e"}),
|
||||
("38ea98ac00c04bada064006d880546f0", "科技蓝", {"background_color": "#06b6d4"}),
|
||||
]
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
conn = op.get_bind()
|
||||
if context.get_context().dialect.name == "postgresql":
|
||||
result = conn.execute(sa.text("SELECT to_regclass('public.cover_templates')"))
|
||||
if result.scalar() is not None:
|
||||
return
|
||||
|
||||
op.create_table(
|
||||
"cover_templates",
|
||||
sa.Column("id", sa.String(36), primary_key=True),
|
||||
sa.Column("user_id", sa.String(36), nullable=True, index=True),
|
||||
sa.Column("name", sa.String(200), nullable=False),
|
||||
sa.Column("thumbnail_url", sa.String(1000), nullable=False, server_default=""),
|
||||
sa.Column("is_system", sa.Boolean, nullable=False, server_default=sa.false(), index=True),
|
||||
sa.Column("config", sa.JSON, nullable=False, server_default="{}"),
|
||||
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
|
||||
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
|
||||
)
|
||||
|
||||
# 预置系统模板 seed 数据
|
||||
cover_templates = sa.table(
|
||||
"cover_templates",
|
||||
sa.column("id", sa.String),
|
||||
sa.column("user_id", sa.String),
|
||||
sa.column("name", sa.String),
|
||||
sa.column("thumbnail_url", sa.String),
|
||||
sa.column("is_system", sa.Boolean),
|
||||
sa.column("config", sa.JSON),
|
||||
sa.column("created_at", sa.DateTime),
|
||||
sa.column("updated_at", sa.DateTime),
|
||||
)
|
||||
|
||||
for tid, name, config in SYSTEM_TEMPLATES:
|
||||
conn.execute(
|
||||
cover_templates.insert().values(
|
||||
id=tid,
|
||||
user_id=None,
|
||||
name=name,
|
||||
thumbnail_url="",
|
||||
is_system=True,
|
||||
config=json.dumps(config),
|
||||
created_at=sa.func.now(),
|
||||
updated_at=sa.func.now(),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_table("cover_templates")
|
||||
@@ -5,6 +5,7 @@ from app.api.routes.assets import router as assets_router
|
||||
from app.api.routes.auth import router as auth_router
|
||||
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.cover_templates import router as cover_templates_router
|
||||
from app.api.routes.duplication import router as duplication_router
|
||||
from app.api.routes.feature_flags import router as feature_flags_router
|
||||
from app.api.routes.generation_preview import router as generation_preview_router
|
||||
@@ -45,6 +46,10 @@ api_router.include_router(
|
||||
prefix="/tags",
|
||||
tags=["Tag"],
|
||||
)
|
||||
api_router.include_router(
|
||||
cover_templates_router,
|
||||
tags=["CoverTemplate"],
|
||||
)
|
||||
api_router.include_router(
|
||||
task_center_router,
|
||||
tags=["TaskCenter"],
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
"""封面模板 CRUD 路由。
|
||||
|
||||
API:
|
||||
GET /api/v1/cover-templates - 列出当前用户可见的模板
|
||||
POST /api/v1/cover-templates - 创建自定义模板
|
||||
PUT /api/v1/cover-templates/{id} - 更新模板
|
||||
DELETE /api/v1/cover-templates/{id} - 删除自定义模板（系统模板不可删）
|
||||
"""
|
||||
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from app.auth import AuthenticatedUser, get_current_user
|
||||
from app.dependencies import get_cover_template_repository
|
||||
from app.schemas.cover_template import (
|
||||
CoverTemplateResponse,
|
||||
CreateCoverTemplateRequest,
|
||||
ListCoverTemplatesResponse,
|
||||
UpdateCoverTemplateRequest,
|
||||
)
|
||||
from fastapi import APIRouter, Depends, HTTPException, Response
|
||||
|
||||
from packages.domain.cover_template import CoverTemplate
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/cover-templates", tags=["CoverTemplate"])
|
||||
|
||||
|
||||
@router.get("", response_model=ListCoverTemplatesResponse)
|
||||
def list_cover_templates(
|
||||
skip: int = 0,
|
||||
limit: int = 100,
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
repo: Any = Depends(get_cover_template_repository),
|
||||
) -> ListCoverTemplatesResponse:
|
||||
"""列出当前用户可见的封面模板（系统模板 + 用户自定义模板）。"""
|
||||
user_id = authenticated_user.user.id
|
||||
items = repo.list_for_user(user_id, skip=skip, limit=limit)
|
||||
total = repo.count_for_user(user_id)
|
||||
return ListCoverTemplatesResponse(
|
||||
items=[
|
||||
CoverTemplateResponse(
|
||||
id=t.id,
|
||||
name=t.name,
|
||||
thumbnail_url=t.thumbnail_url,
|
||||
is_system=t.is_system,
|
||||
created_at=t.created_at,
|
||||
config=t.config,
|
||||
)
|
||||
for t in items
|
||||
],
|
||||
total=total,
|
||||
)
|
||||
|
||||
|
||||
@router.post("", response_model=CoverTemplateResponse, status_code=201)
|
||||
def create_cover_template(
|
||||
request: CreateCoverTemplateRequest,
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
repo: Any = Depends(get_cover_template_repository),
|
||||
) -> CoverTemplateResponse:
|
||||
"""创建用户自定义封面模板。"""
|
||||
user_id = authenticated_user.user.id
|
||||
config_dict = request.config.model_dump() if request.config else {}
|
||||
template = CoverTemplate.create_user(
|
||||
user_id=user_id,
|
||||
name=request.name,
|
||||
config=config_dict,
|
||||
thumbnail_url=request.thumbnail_url,
|
||||
)
|
||||
created = repo.create(template)
|
||||
return CoverTemplateResponse(
|
||||
id=created.id,
|
||||
name=created.name,
|
||||
thumbnail_url=created.thumbnail_url,
|
||||
is_system=created.is_system,
|
||||
created_at=created.created_at,
|
||||
config=created.config,
|
||||
)
|
||||
|
||||
|
||||
@router.put("/{template_id}", response_model=CoverTemplateResponse)
|
||||
def update_cover_template(
|
||||
template_id: str,
|
||||
request: UpdateCoverTemplateRequest,
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
repo: Any = Depends(get_cover_template_repository),
|
||||
) -> CoverTemplateResponse:
|
||||
"""更新封面模板（仅允许更新自己的模板）。"""
|
||||
user_id = authenticated_user.user.id
|
||||
template = repo.get(template_id)
|
||||
if template is None:
|
||||
raise HTTPException(status_code=404, detail="模板不存在")
|
||||
if template.is_system:
|
||||
raise HTTPException(status_code=403, detail="系统模板不可修改")
|
||||
if template.user_id != user_id:
|
||||
raise HTTPException(status_code=403, detail="无权修改该模板")
|
||||
|
||||
if request.name is not None:
|
||||
template.update(name=request.name)
|
||||
if request.config is not None:
|
||||
template.update(config=request.config.model_dump())
|
||||
if request.thumbnail_url is not None:
|
||||
template.update(thumbnail_url=request.thumbnail_url)
|
||||
|
||||
updated = repo.update(template)
|
||||
return CoverTemplateResponse(
|
||||
id=updated.id,
|
||||
name=updated.name,
|
||||
thumbnail_url=updated.thumbnail_url,
|
||||
is_system=updated.is_system,
|
||||
created_at=updated.created_at,
|
||||
config=updated.config,
|
||||
)
|
||||
|
||||
|
||||
@router.delete("/{template_id}", status_code=204, response_class=Response)
|
||||
def delete_cover_template(
|
||||
template_id: str,
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
repo: Any = Depends(get_cover_template_repository),
|
||||
) -> None:
|
||||
"""删除用户自定义封面模板（系统模板不可删除）。"""
|
||||
user_id = authenticated_user.user.id
|
||||
template = repo.get(template_id)
|
||||
if template is None:
|
||||
raise HTTPException(status_code=404, detail="模板不存在")
|
||||
if template.is_system:
|
||||
raise HTTPException(status_code=403, detail="系统模板不可删除")
|
||||
if template.user_id != user_id:
|
||||
raise HTTPException(status_code=403, detail="无权删除该模板")
|
||||
repo.delete(template_id)
|
||||
@@ -22,6 +22,9 @@ from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRe
|
||||
from packages.adapters.sqlalchemy_impl.classification_job_repository import (
|
||||
SQLAlchemyClassificationJobRepository,
|
||||
)
|
||||
from packages.adapters.sqlalchemy_impl.cover_template_repository import (
|
||||
SQLAlchemyCoverTemplateRepository,
|
||||
)
|
||||
from packages.adapters.sqlalchemy_impl.duplication_repository import (
|
||||
SQLAlchemyDuplicationRecordRepository,
|
||||
)
|
||||
@@ -128,6 +131,13 @@ def get_project_repository(
|
||||
return SQLAlchemyProjectRepository(session)
|
||||
|
||||
|
||||
def get_cover_template_repository(
|
||||
session: Session = Depends(get_db_session),
|
||||
) -> SQLAlchemyCoverTemplateRepository:
|
||||
"""Provide the SQLAlchemy cover template repository implementation."""
|
||||
return SQLAlchemyCoverTemplateRepository(session)
|
||||
|
||||
|
||||
def get_tag_repository(
|
||||
session: Session = Depends(get_db_session),
|
||||
) -> TagRepository:
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
"""封面模板 Schema。"""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class CoverTemplateConfig(BaseModel):
|
||||
"""封面模板配置。"""
|
||||
|
||||
background_enabled: bool = Field(default=True, description="是否启用背景")
|
||||
background_color: str = Field(default="#000000", description="背景颜色")
|
||||
portrait_enabled: bool = Field(default=True, description="是否显示人像")
|
||||
title_text: str = Field(default="", description="主标题文字")
|
||||
subtitle_text: str = Field(default="", description="副标题文字")
|
||||
mask_enabled: bool = Field(default=False, description="是否启用蒙版")
|
||||
|
||||
|
||||
class CreateCoverTemplateRequest(BaseModel):
|
||||
"""创建封面模板请求。"""
|
||||
|
||||
name: str = Field(..., min_length=1, max_length=200, description="模板名称")
|
||||
thumbnail_url: str = Field(default="", description="缩略图 URL")
|
||||
config: CoverTemplateConfig | None = Field(default=None, description="模板配置")
|
||||
|
||||
|
||||
class UpdateCoverTemplateRequest(BaseModel):
|
||||
"""更新封面模板请求。"""
|
||||
|
||||
name: str | None = Field(default=None, min_length=1, max_length=200, description="模板名称")
|
||||
thumbnail_url: str | None = Field(default=None, description="缩略图 URL")
|
||||
config: CoverTemplateConfig | None = Field(default=None, description="模板配置")
|
||||
|
||||
|
||||
class CoverTemplateResponse(BaseModel):
|
||||
"""封面模板响应。"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
thumbnail_url: str
|
||||
is_system: bool
|
||||
created_at: datetime
|
||||
config: dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class ListCoverTemplatesResponse(BaseModel):
|
||||
"""封面模板列表响应。"""
|
||||
|
||||
items: list[CoverTemplateResponse]
|
||||
total: int = Field(default=0, ge=0)
|
||||
@@ -0,0 +1,41 @@
|
||||
"""封面模板 InMemory 仓储实现。"""
|
||||
|
||||
from packages.domain.cover_template import CoverTemplate
|
||||
|
||||
|
||||
class InMemoryCoverTemplateRepository:
|
||||
"""内存中的封面模板仓储，用于测试。"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._templates: dict[str, CoverTemplate] = {}
|
||||
|
||||
def create(self, template: CoverTemplate) -> CoverTemplate:
|
||||
self._templates[template.id] = template
|
||||
return template
|
||||
|
||||
def get(self, template_id: str) -> CoverTemplate | None:
|
||||
return self._templates.get(template_id)
|
||||
|
||||
def list_for_user(self, user_id: str, skip: int = 0, limit: int = 100) -> list[CoverTemplate]:
|
||||
"""列出系统模板 + 用户自己的模板。"""
|
||||
visible = [t for t in self._templates.values() if t.is_system or t.user_id == user_id]
|
||||
visible.sort(key=lambda t: (t.is_system, t.created_at), reverse=True)
|
||||
return visible[skip : skip + limit]
|
||||
|
||||
def count_for_user(self, user_id: str) -> int:
|
||||
return sum(1 for t in self._templates.values() if t.is_system or t.user_id == user_id)
|
||||
|
||||
def update(self, template: CoverTemplate) -> CoverTemplate:
|
||||
if template.id not in self._templates:
|
||||
raise ValueError(f"模板 {template.id} 不存在")
|
||||
self._templates[template.id] = template
|
||||
return template
|
||||
|
||||
def delete(self, template_id: str) -> bool:
|
||||
if template_id in self._templates:
|
||||
del self._templates[template_id]
|
||||
return True
|
||||
return False
|
||||
|
||||
def list_system_templates(self) -> list[CoverTemplate]:
|
||||
return [t for t in self._templates.values() if t.is_system]
|
||||
@@ -0,0 +1,105 @@
|
||||
"""封面模板 SQLAlchemy 仓储实现。"""
|
||||
|
||||
from sqlalchemy import or_
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from packages.adapters.sqlalchemy_impl.models import CoverTemplateModel
|
||||
from packages.domain.cover_template import CoverTemplate
|
||||
|
||||
|
||||
class SQLAlchemyCoverTemplateRepository:
|
||||
"""封面模板仓储实现。"""
|
||||
|
||||
def __init__(self, session: Session):
|
||||
self.session = session
|
||||
|
||||
def create(self, template: CoverTemplate) -> CoverTemplate:
|
||||
model = CoverTemplateModel(
|
||||
id=template.id,
|
||||
user_id=template.user_id,
|
||||
name=template.name,
|
||||
thumbnail_url=template.thumbnail_url,
|
||||
is_system=template.is_system,
|
||||
config=template.config,
|
||||
created_at=template.created_at,
|
||||
updated_at=template.updated_at,
|
||||
)
|
||||
self.session.add(model)
|
||||
self.session.commit()
|
||||
return template
|
||||
|
||||
def get(self, template_id: str) -> CoverTemplate | None:
|
||||
model = self.session.query(CoverTemplateModel).filter(CoverTemplateModel.id == template_id).first()
|
||||
if model is None:
|
||||
return None
|
||||
return self._to_domain(model)
|
||||
|
||||
def list_for_user(self, user_id: str, skip: int = 0, limit: int = 100) -> list[CoverTemplate]:
|
||||
"""列出用户可见的模板：系统模板 + 用户自己的模板。"""
|
||||
models = (
|
||||
self.session.query(CoverTemplateModel)
|
||||
.filter(
|
||||
or_(
|
||||
CoverTemplateModel.is_system == True, # noqa: E712
|
||||
CoverTemplateModel.user_id == user_id,
|
||||
)
|
||||
)
|
||||
.order_by(CoverTemplateModel.is_system.desc(), CoverTemplateModel.created_at.desc())
|
||||
.offset(skip)
|
||||
.limit(limit)
|
||||
.all()
|
||||
)
|
||||
return [self._to_domain(m) for m in models]
|
||||
|
||||
def count_for_user(self, user_id: str) -> int:
|
||||
return (
|
||||
self.session.query(CoverTemplateModel)
|
||||
.filter(
|
||||
or_(
|
||||
CoverTemplateModel.is_system == True, # noqa: E712
|
||||
CoverTemplateModel.user_id == user_id,
|
||||
)
|
||||
)
|
||||
.count()
|
||||
)
|
||||
|
||||
def update(self, template: CoverTemplate) -> CoverTemplate:
|
||||
model = self.session.query(CoverTemplateModel).filter(CoverTemplateModel.id == template.id).first()
|
||||
if model is None:
|
||||
raise ValueError(f"模板 {template.id} 不存在")
|
||||
model.name = template.name
|
||||
model.thumbnail_url = template.thumbnail_url
|
||||
model.config = template.config
|
||||
model.updated_at = template.updated_at
|
||||
self.session.commit()
|
||||
return template
|
||||
|
||||
def delete(self, template_id: str) -> bool:
|
||||
model = self.session.query(CoverTemplateModel).filter(CoverTemplateModel.id == template_id).first()
|
||||
if model is None:
|
||||
return False
|
||||
self.session.delete(model)
|
||||
self.session.commit()
|
||||
return True
|
||||
|
||||
def list_system_templates(self) -> list[CoverTemplate]:
|
||||
models = (
|
||||
self.session.query(CoverTemplateModel)
|
||||
.filter(CoverTemplateModel.is_system == True) # noqa: E712
|
||||
.order_by(CoverTemplateModel.created_at)
|
||||
.all()
|
||||
)
|
||||
return [self._to_domain(m) for m in models]
|
||||
|
||||
@staticmethod
|
||||
def _to_domain(model: CoverTemplateModel) -> CoverTemplate:
|
||||
return CoverTemplate(
|
||||
id=model.id,
|
||||
user_id=model.user_id,
|
||||
name=model.name,
|
||||
thumbnail_url=model.thumbnail_url,
|
||||
is_system=model.is_system,
|
||||
config=model.config or {},
|
||||
created_at=model.created_at,
|
||||
updated_at=model.updated_at,
|
||||
)
|
||||
@@ -600,3 +600,18 @@ class VideoShareModel(Base):
|
||||
is_active = Column(Boolean, nullable=False, default=True)
|
||||
created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
|
||||
|
||||
class CoverTemplateModel(Base):
|
||||
"""封面模板"""
|
||||
|
||||
__tablename__ = "cover_templates"
|
||||
|
||||
id = Column(String(36), primary_key=True)
|
||||
user_id = Column(String(36), nullable=True, index=True) # NULL = 系统模板
|
||||
name = Column(String(200), nullable=False)
|
||||
thumbnail_url = Column(String(1000), nullable=False, default="")
|
||||
is_system = Column(Boolean, nullable=False, default=False, index=True)
|
||||
config = Column(JSON, nullable=False, default=dict)
|
||||
created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
|
||||
@@ -5,6 +5,7 @@ from .classification import (
|
||||
ClassificationJob,
|
||||
ClassificationJobStatus,
|
||||
)
|
||||
from .cover_template import CoverTemplate
|
||||
from .duplication import DuplicateSegment, DuplicationRecord
|
||||
from .edit_plan import EditPlan, EditPlanStatus
|
||||
from .edit_plan_clip import EditPlanClip, EditPlanClipStatus
|
||||
@@ -36,6 +37,7 @@ __all__ = [
|
||||
"AssetLibrary",
|
||||
"AssetLibraryKind",
|
||||
"AssetStatus",
|
||||
"CoverTemplate",
|
||||
"ClassificationJob",
|
||||
"ClassificationJobStatus",
|
||||
"ClassificationStatus",
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
"""封面模板领域实体。"""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class CoverTemplate:
|
||||
"""封面模板实体，支持系统预置和用户自定义。"""
|
||||
|
||||
id: str
|
||||
user_id: str | None # None 表示系统模板
|
||||
name: str
|
||||
thumbnail_url: str
|
||||
is_system: bool
|
||||
config: dict[str, Any]
|
||||
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
|
||||
updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
|
||||
|
||||
@classmethod
|
||||
def create_system(
|
||||
cls,
|
||||
name: str,
|
||||
config: dict[str, Any] | None = None,
|
||||
thumbnail_url: str = "",
|
||||
) -> "CoverTemplate":
|
||||
"""创建系统模板。"""
|
||||
template_id = uuid4().hex
|
||||
return cls(
|
||||
id=template_id,
|
||||
user_id=None,
|
||||
name=name.strip(),
|
||||
thumbnail_url=thumbnail_url,
|
||||
is_system=True,
|
||||
config=config or {},
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def create_user(
|
||||
cls,
|
||||
user_id: str,
|
||||
name: str,
|
||||
config: dict[str, Any] | None = None,
|
||||
thumbnail_url: str = "",
|
||||
) -> "CoverTemplate":
|
||||
"""创建用户自定义模板。"""
|
||||
clean_name = name.strip()
|
||||
if not clean_name:
|
||||
raise ValueError("模板名称不能为空")
|
||||
template_id = uuid4().hex
|
||||
return cls(
|
||||
id=template_id,
|
||||
user_id=user_id,
|
||||
name=clean_name,
|
||||
thumbnail_url=thumbnail_url,
|
||||
is_system=False,
|
||||
config=config or {},
|
||||
)
|
||||
|
||||
def update(
|
||||
self,
|
||||
name: str | None = None,
|
||||
config: dict[str, Any] | None = None,
|
||||
thumbnail_url: str | None = None,
|
||||
) -> None:
|
||||
"""更新模板属性。"""
|
||||
if name is not None:
|
||||
clean_name = name.strip()
|
||||
if not clean_name:
|
||||
raise ValueError("模板名称不能为空")
|
||||
self.name = clean_name
|
||||
if config is not None:
|
||||
self.config = config
|
||||
if thumbnail_url is not None:
|
||||
self.thumbnail_url = thumbnail_url
|
||||
self.updated_at = datetime.now(timezone.utc)
|
||||
@@ -0,0 +1,44 @@
|
||||
"""封面模板仓储接口定义。"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from packages.domain.cover_template import CoverTemplate
|
||||
|
||||
|
||||
class CoverTemplateRepository(ABC):
|
||||
"""封面模板仓储抽象接口。"""
|
||||
|
||||
@abstractmethod
|
||||
def create(self, template: CoverTemplate) -> CoverTemplate:
|
||||
"""创建封面模板。"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get(self, template_id: str) -> CoverTemplate | None:
|
||||
"""根据 ID 获取模板。"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def list_for_user(self, user_id: str, skip: int = 0, limit: int = 100) -> list[CoverTemplate]:
|
||||
"""列出用户可见的模板（系统模板 + 用户自定义模板）。"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def count_for_user(self, user_id: str) -> int:
|
||||
"""统计用户可见的模板数量。"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def update(self, template: CoverTemplate) -> CoverTemplate:
|
||||
"""更新模板。"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def delete(self, template_id: str) -> bool:
|
||||
"""删除模板（仅允许删除用户自定义模板）。"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def list_system_templates(self) -> list[CoverTemplate]:
|
||||
"""列出所有系统模板。"""
|
||||
pass
|
||||
@@ -0,0 +1,309 @@
|
||||
"""封面模板 CRUD 单元测试。
|
||||
|
||||
验证：
|
||||
1. 领域实体：创建系统/用户模板、更新、空名称校验
|
||||
2. 仓储接口：list_for_user 返回系统+用户模板
|
||||
3. API 路由：CRUD 权限检查（系统模板不可删/改）
|
||||
4. Schema：请求/响应序列化
|
||||
"""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.domain.cover_template import CoverTemplate
|
||||
|
||||
|
||||
class TestCoverTemplateDomain:
|
||||
"""测试封面模板领域实体。"""
|
||||
|
||||
def test_create_system_template(self):
|
||||
"""测试创建系统模板。"""
|
||||
tpl = CoverTemplate.create_system(name="默认模板")
|
||||
assert tpl.is_system is True
|
||||
assert tpl.user_id is None
|
||||
assert tpl.name == "默认模板"
|
||||
assert tpl.config == {}
|
||||
|
||||
def test_create_system_template_with_config(self):
|
||||
"""测试创建带配置的系统模板。"""
|
||||
config = {"background_color": "#ef4444", "title_text": "Hello"}
|
||||
tpl = CoverTemplate.create_system(name="大胆红", config=config)
|
||||
assert tpl.config == config
|
||||
assert tpl.name == "大胆红"
|
||||
|
||||
def test_create_user_template(self):
|
||||
"""测试创建用户自定义模板。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="我的模板")
|
||||
assert tpl.is_system is False
|
||||
assert tpl.user_id == "user-1"
|
||||
assert tpl.name == "我的模板"
|
||||
|
||||
def test_create_user_template_strips_whitespace(self):
|
||||
"""测试创建用户模板时自动去除首尾空格。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name=" 我的模板 ")
|
||||
assert tpl.name == "我的模板"
|
||||
|
||||
def test_create_user_template_empty_name_raises(self):
|
||||
"""测试空名称抛出 ValueError。"""
|
||||
with pytest.raises(ValueError, match="模板名称不能为空"):
|
||||
CoverTemplate.create_user(user_id="user-1", name="")
|
||||
|
||||
with pytest.raises(ValueError, match="模板名称不能为空"):
|
||||
CoverTemplate.create_user(user_id="user-1", name=" ")
|
||||
|
||||
def test_update_name(self):
|
||||
"""测试更新模板名称。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="旧名称")
|
||||
old_updated_at = tpl.updated_at
|
||||
tpl.update(name="新名称")
|
||||
assert tpl.name == "新名称"
|
||||
assert tpl.updated_at >= old_updated_at
|
||||
|
||||
def test_update_config(self):
|
||||
"""测试更新模板配置。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="测试")
|
||||
new_config = {"background_color": "#000", "mask_enabled": True}
|
||||
tpl.update(config=new_config)
|
||||
assert tpl.config == new_config
|
||||
|
||||
def test_update_thumbnail_url(self):
|
||||
"""测试更新缩略图 URL。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="测试")
|
||||
tpl.update(thumbnail_url="https://example.com/thumb.jpg")
|
||||
assert tpl.thumbnail_url == "https://example.com/thumb.jpg"
|
||||
|
||||
def test_update_empty_name_raises(self):
|
||||
"""测试更新空名称抛出 ValueError。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="测试")
|
||||
with pytest.raises(ValueError, match="模板名称不能为空"):
|
||||
tpl.update(name="")
|
||||
|
||||
|
||||
class TestCoverTemplateRepository:
|
||||
"""测试封面模板仓储（使用 Mock）。"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_repo(self):
|
||||
"""创建模拟仓储。"""
|
||||
from packages.adapters.sqlalchemy_impl.cover_template_repository import (
|
||||
SQLAlchemyCoverTemplateRepository,
|
||||
)
|
||||
|
||||
mock_session = MagicMock()
|
||||
return SQLAlchemyCoverTemplateRepository(mock_session)
|
||||
|
||||
def test_create_calls_session_add(self, mock_repo):
|
||||
"""测试 create 方法调用 session.add。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="测试")
|
||||
mock_repo.create(tpl)
|
||||
mock_repo.session.add.assert_called_once()
|
||||
mock_repo.session.commit.assert_called_once()
|
||||
|
||||
def test_get_returns_none_for_nonexistent(self, mock_repo):
|
||||
"""测试 get 方法对不存在的模板返回 None。"""
|
||||
mock_repo.session.query.return_value.filter.return_value.first.return_value = None
|
||||
result = mock_repo.get("nonexistent-id")
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestCoverTemplateSchema:
|
||||
"""测试封面模板 Schema。"""
|
||||
|
||||
def test_create_request_validation(self):
|
||||
"""测试创建请求的字段验证。"""
|
||||
from app.schemas.cover_template import CreateCoverTemplateRequest
|
||||
|
||||
req = CreateCoverTemplateRequest(name="测试模板")
|
||||
assert req.name == "测试模板"
|
||||
assert req.thumbnail_url == ""
|
||||
assert req.config is None
|
||||
|
||||
def test_create_request_with_config(self):
|
||||
"""测试带配置的创建请求。"""
|
||||
from app.schemas.cover_template import CoverTemplateConfig, CreateCoverTemplateRequest
|
||||
|
||||
config = CoverTemplateConfig(
|
||||
background_enabled=False,
|
||||
background_color="#ff0000",
|
||||
title_text="主标题",
|
||||
)
|
||||
req = CreateCoverTemplateRequest(name="测试", config=config)
|
||||
assert req.config.background_enabled is False
|
||||
assert req.config.background_color == "#ff0000"
|
||||
assert req.config.title_text == "主标题"
|
||||
|
||||
def test_update_request_optional_fields(self):
|
||||
"""测试更新请求所有字段可选。"""
|
||||
from app.schemas.cover_template import UpdateCoverTemplateRequest
|
||||
|
||||
req = UpdateCoverTemplateRequest()
|
||||
assert req.name is None
|
||||
assert req.config is None
|
||||
assert req.thumbnail_url is None
|
||||
|
||||
def test_response_serialization(self):
|
||||
"""测试响应序列化。"""
|
||||
from app.schemas.cover_template import CoverTemplateResponse
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
resp = CoverTemplateResponse(
|
||||
id="test-id",
|
||||
name="测试",
|
||||
thumbnail_url="",
|
||||
is_system=False,
|
||||
created_at=now,
|
||||
config={"background_color": "#000"},
|
||||
)
|
||||
assert resp.id == "test-id"
|
||||
assert resp.config["background_color"] == "#000"
|
||||
|
||||
|
||||
class TestCoverTemplateAPIPermissions:
|
||||
"""测试封面模板 API 权限控制。"""
|
||||
|
||||
def test_system_template_cannot_be_deleted(self):
|
||||
"""测试系统模板不可删除。"""
|
||||
tpl = CoverTemplate.create_system(name="系统模板")
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.get.return_value = tpl
|
||||
|
||||
# 模拟 API 路由中的权限检查逻辑
|
||||
template = mock_repo.get("a8b0120fd98e44788f5a6590f983d327")
|
||||
assert template is not None
|
||||
assert template.is_system is True
|
||||
# 权限检查应该阻止删除
|
||||
with pytest.raises(PermissionError):
|
||||
if template.is_system:
|
||||
raise PermissionError("系统模板不可删除")
|
||||
|
||||
def test_system_template_cannot_be_updated(self):
|
||||
"""测试系统模板不可修改。"""
|
||||
tpl = CoverTemplate.create_system(name="系统模板")
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.get.return_value = tpl
|
||||
|
||||
template = mock_repo.get("a8b0120fd98e44788f5a6590f983d327")
|
||||
assert template.is_system is True
|
||||
with pytest.raises(PermissionError):
|
||||
if template.is_system:
|
||||
raise PermissionError("系统模板不可修改")
|
||||
|
||||
def test_user_cannot_delete_others_template(self):
|
||||
"""测试用户不可删除他人的模板。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="别人的模板")
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.get.return_value = tpl
|
||||
|
||||
template = mock_repo.get(tpl.id)
|
||||
current_user_id = "user-2"
|
||||
assert template.user_id != current_user_id
|
||||
with pytest.raises(PermissionError):
|
||||
if template.user_id != current_user_id:
|
||||
raise PermissionError("无权删除该模板")
|
||||
|
||||
def test_user_can_delete_own_template(self):
|
||||
"""测试用户可以删除自己的模板。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="我的模板")
|
||||
mock_repo = MagicMock()
|
||||
mock_repo.get.return_value = tpl
|
||||
|
||||
template = mock_repo.get(tpl.id)
|
||||
current_user_id = "user-1"
|
||||
assert template.user_id == current_user_id
|
||||
assert not template.is_system
|
||||
# 权限检查通过，可以删除
|
||||
mock_repo.delete(template.id)
|
||||
mock_repo.delete.assert_called_once()
|
||||
|
||||
|
||||
class TestInMemoryCoverTemplateRepository:
|
||||
"""使用 InMemory 仓储测试完整 CRUD 流程。"""
|
||||
|
||||
@pytest.fixture
|
||||
def repo(self):
|
||||
from packages.adapters.in_memory.cover_template_repository import (
|
||||
InMemoryCoverTemplateRepository,
|
||||
)
|
||||
|
||||
return InMemoryCoverTemplateRepository()
|
||||
|
||||
def test_create_and_get(self, repo):
|
||||
"""创建后能查到。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="测试")
|
||||
repo.create(tpl)
|
||||
found = repo.get(tpl.id)
|
||||
assert found is not None
|
||||
assert found.name == "测试"
|
||||
assert found.user_id == "user-1"
|
||||
|
||||
def test_get_nonexistent_returns_none(self, repo):
|
||||
"""查不到返回 None。"""
|
||||
assert repo.get("nonexistent") is None
|
||||
|
||||
def test_list_for_user_includes_system_and_own(self, repo):
|
||||
"""list_for_user 返回系统模板 + 用户自己的模板。"""
|
||||
sys_tpl = CoverTemplate.create_system(name="系统模板")
|
||||
repo.create(sys_tpl)
|
||||
|
||||
user1_tpl = CoverTemplate.create_user(user_id="user-1", name="用户1的")
|
||||
repo.create(user1_tpl)
|
||||
|
||||
user2_tpl = CoverTemplate.create_user(user_id="user-2", name="用户2的")
|
||||
repo.create(user2_tpl)
|
||||
|
||||
# user-1 应该看到系统模板 + 自己的
|
||||
user1_visible = repo.list_for_user("user-1")
|
||||
assert len(user1_visible) == 2
|
||||
names = {t.name for t in user1_visible}
|
||||
assert "系统模板" in names
|
||||
assert "用户1的" in names
|
||||
assert "用户2的" not in names
|
||||
|
||||
def test_count_for_user(self, repo):
|
||||
"""count_for_user 返回正确的数量。"""
|
||||
repo.create(CoverTemplate.create_system(name="系统1"))
|
||||
repo.create(CoverTemplate.create_system(name="系统2"))
|
||||
repo.create(CoverTemplate.create_user(user_id="user-1", name="用户1的"))
|
||||
|
||||
assert repo.count_for_user("user-1") == 3
|
||||
assert repo.count_for_user("user-2") == 2 # 只能看到2个系统模板
|
||||
|
||||
def test_update_template(self, repo):
|
||||
"""更新模板。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="旧名称")
|
||||
repo.create(tpl)
|
||||
tpl.update(name="新名称", config={"background_color": "#ff0000"})
|
||||
repo.update(tpl)
|
||||
|
||||
found = repo.get(tpl.id)
|
||||
assert found.name == "新名称"
|
||||
assert found.config["background_color"] == "#ff0000"
|
||||
|
||||
def test_update_nonexistent_raises(self, repo):
|
||||
"""更新不存在的模板抛出异常。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="测试")
|
||||
with pytest.raises(ValueError, match="不存在"):
|
||||
repo.update(tpl)
|
||||
|
||||
def test_delete_template(self, repo):
|
||||
"""删除模板。"""
|
||||
tpl = CoverTemplate.create_user(user_id="user-1", name="测试")
|
||||
repo.create(tpl)
|
||||
assert repo.delete(tpl.id) is True
|
||||
assert repo.get(tpl.id) is None
|
||||
|
||||
def test_delete_nonexistent_returns_false(self, repo):
|
||||
"""删除不存在的模板返回 False。"""
|
||||
assert repo.delete("nonexistent") is False
|
||||
|
||||
def test_list_system_templates(self, repo):
|
||||
"""list_system_templates 只返回系统模板。"""
|
||||
repo.create(CoverTemplate.create_system(name="系统1"))
|
||||
repo.create(CoverTemplate.create_system(name="系统2"))
|
||||
repo.create(CoverTemplate.create_user(user_id="user-1", name="用户1的"))
|
||||
|
||||
system = repo.list_system_templates()
|
||||
assert len(system) == 2
|
||||
assert all(t.is_system for t in system)
|
||||
Reference in New Issue
Block a user