feat(#1894): 废弃标题库整合到文案库 - scripts 模型新增标题字段 (#1927)
CI/CD Pipeline / Check push changed paths (push) Successful in 4s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 1m46s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m51s
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 15m40s
CI/CD Pipeline / Frontend Unit Tests (push) Failing after 1m50s
CI/CD Pipeline / Build Staging API Image (push) Successful in 18m32s
CI/CD Pipeline / Integration Tests (push) Successful in 3m4s
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Successful in 3m25s
CI/CD Pipeline / Validate - Style (push) Successful in 3m35s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 44s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 1m38s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 2m53s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 3m4s
CI/CD Pipeline / Validate - Security (push) Successful in 8m10s
CI/CD Pipeline / Unit Tests (push) Successful in 9m9s
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Failing after 9h23m26s
CI/CD Pipeline / PR Build Worker Image (push) Failing after 9h48m20s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 9h22m39s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 9h47m32s
CI/CD Pipeline / PR Build Web Image (push) Failing after 9h47m31s
CI/CD Pipeline / PR Build API Image (push) Failing after 9h47m32s
CI/CD Pipeline / Retag skipped Staging API Image (push) Failing after 9h28m53s
CI/CD Pipeline / CI Gate (push) Failing after 9h22m39s
CI/CD Pipeline / Canary Release to Production (push) Failing after 9h22m37s
CI/CD Pipeline / Build Production Web Image (push) Failing after 9h22m39s
CI/CD Pipeline / Build Production API Image (push) Failing after 9h22m39s
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Failing after 9h28m53s
CI/CD Pipeline / Frontend Lint (push) Failing after 9h31m49s
CI/CD Pipeline / Retag skipped Staging Web Image (push) Failing after 10h4m41s
CI/CD Pipeline / Check push changed paths (push) Successful in 4s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 1m46s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 1m51s
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 15m40s
CI/CD Pipeline / Frontend Unit Tests (push) Failing after 1m50s
CI/CD Pipeline / Build Staging API Image (push) Successful in 18m32s
CI/CD Pipeline / Integration Tests (push) Successful in 3m4s
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Successful in 3m25s
CI/CD Pipeline / Validate - Style (push) Successful in 3m35s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 44s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 1m38s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 2m53s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 3m4s
CI/CD Pipeline / Validate - Security (push) Successful in 8m10s
CI/CD Pipeline / Unit Tests (push) Successful in 9m9s
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Failing after 9h23m26s
CI/CD Pipeline / PR Build Worker Image (push) Failing after 9h48m20s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 9h22m39s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 9h47m32s
CI/CD Pipeline / PR Build Web Image (push) Failing after 9h47m31s
CI/CD Pipeline / PR Build API Image (push) Failing after 9h47m32s
CI/CD Pipeline / Retag skipped Staging API Image (push) Failing after 9h28m53s
CI/CD Pipeline / CI Gate (push) Failing after 9h22m39s
CI/CD Pipeline / Canary Release to Production (push) Failing after 9h22m37s
CI/CD Pipeline / Build Production Web Image (push) Failing after 9h22m39s
CI/CD Pipeline / Build Production API Image (push) Failing after 9h22m39s
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Failing after 9h28m53s
CI/CD Pipeline / Frontend Lint (push) Failing after 9h31m49s
CI/CD Pipeline / Retag skipped Staging Web Image (push) Failing after 10h4m41s
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
This commit was merged in pull request #1927.
This commit is contained in:
+1
-1
@@ -1,2 +1,2 @@
|
||||
CI trigger file - safe to delete
|
||||
updated!
|
||||
retrigger at 2026-09-15 20:31:24 UTC
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
"""#1894: merge title_libraries into scripts — add title_text/title_category/title_config
|
||||
|
||||
Revision ID: 077_merge_title_libs
|
||||
Revises: 076_membership_points
|
||||
Create Date: 2026-09-15
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import context, op
|
||||
|
||||
revision = "077_merge_title_libs"
|
||||
down_revision = "076_membership_points"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
with op.batch_alter_table("scripts") as batch:
|
||||
batch.add_column(
|
||||
sa.Column("title_text", sa.String(500), nullable=False, server_default=""),
|
||||
)
|
||||
batch.add_column(
|
||||
sa.Column("title_category", sa.String(50), nullable=False, server_default=""),
|
||||
)
|
||||
batch.add_column(
|
||||
sa.Column("title_config", sa.JSON, nullable=False, server_default="{}"),
|
||||
)
|
||||
|
||||
if context.get_context().dialect.name == "postgresql":
|
||||
conn = op.get_bind()
|
||||
result = conn.execute(sa.text("SELECT to_regclass('public.title_libraries')"))
|
||||
if result.scalar() is not None:
|
||||
conn.execute(sa.text("""
|
||||
INSERT INTO scripts
|
||||
(id, user_id, title, content, segments, tags,
|
||||
title_text, title_category, title_config,
|
||||
created_at, updated_at)
|
||||
SELECT
|
||||
gen_random_uuid()::TEXT,
|
||||
tl.user_id,
|
||||
COALESCE(tl.name, '迁移标题'),
|
||||
COALESCE(tl.text, ''),
|
||||
'[]'::JSONB,
|
||||
COALESCE(tl.tags, '[]'::JSONB),
|
||||
COALESCE(tl.text, ''),
|
||||
COALESCE(tl.category, ''),
|
||||
COALESCE(tl."metadata", '{}'::JSONB),
|
||||
tl.created_at,
|
||||
tl.updated_at
|
||||
FROM title_libraries tl
|
||||
WHERE tl.is_active = true
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM scripts s
|
||||
WHERE s.user_id = tl.user_id
|
||||
AND s.title_text = COALESCE(tl.text, '')
|
||||
AND s.title_category = COALESCE(tl.category, '')
|
||||
AND s.created_at = tl.created_at
|
||||
)
|
||||
"""))
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
with op.batch_alter_table("scripts") as batch:
|
||||
batch.drop_column("title_config")
|
||||
batch.drop_column("title_category")
|
||||
batch.drop_column("title_text")
|
||||
@@ -36,6 +36,9 @@ def _to_response(script) -> ScriptResponse:
|
||||
for s in segments
|
||||
],
|
||||
tags=script.tags or [],
|
||||
title_text=getattr(script, "title_text", "") or "",
|
||||
title_category=getattr(script, "title_category", "") or "",
|
||||
title_config=getattr(script, "title_config", None) or {},
|
||||
created_at=script.created_at,
|
||||
updated_at=script.updated_at,
|
||||
)
|
||||
@@ -70,6 +73,9 @@ def create_script(
|
||||
content=request.content,
|
||||
segments=[s.model_dump() for s in request.segments],
|
||||
tags=request.tags,
|
||||
title_text=request.title_text or "",
|
||||
title_category=request.title_category or "",
|
||||
title_config=request.title_config or {},
|
||||
)
|
||||
return _to_response(script)
|
||||
|
||||
@@ -104,6 +110,9 @@ def update_script(
|
||||
content=request.content,
|
||||
segments=[s.model_dump() for s in request.segments] if request.segments is not None else None,
|
||||
tags=request.tags,
|
||||
title_text=request.title_text,
|
||||
title_category=request.title_category,
|
||||
title_config=request.title_config,
|
||||
)
|
||||
except ScriptNotFoundError as exc:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Script not found") from exc
|
||||
|
||||
@@ -1,7 +1,13 @@
|
||||
"""Title library CRUD routes."""
|
||||
"""Title library CRUD routes.
|
||||
|
||||
.. deprecated::
|
||||
标题库 API 已废弃(#1894),标题配置已整合到 scripts 模型。
|
||||
所有接口保留向后兼容,但返回 Warning header 并记录日志。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from app.api.routes._helpers import get_user_plan
|
||||
@@ -35,6 +41,21 @@ from packages.application.title_library.use_cases import (
|
||||
from packages.ports.user_repository import UserRepository
|
||||
|
||||
router = APIRouter()
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_DEPRECATION_WARNING = (
|
||||
'299 - "Title library API is deprecated; migrate to scripts.title_text/'
|
||||
'title_category/title_config (issue #1894)"'
|
||||
)
|
||||
|
||||
|
||||
def _deprecation_headers() -> dict:
|
||||
"""返回 deprecation Warning header (ASCII-only, RFC 7234 §5.5)."""
|
||||
return {"Warning": _DEPRECATION_WARNING, "Deprecation": "true"}
|
||||
|
||||
|
||||
def _log_deprecation(endpoint: str) -> None:
|
||||
logger.warning("[Deprecated] title_library API 调用: %s — %s", endpoint, _DEPRECATION_WARNING)
|
||||
|
||||
|
||||
def _get_title_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyTitleLibraryRepository:
|
||||
@@ -59,12 +80,17 @@ def _to_response(item) -> TitleLibraryItemResponse:
|
||||
|
||||
@router.get("", response_model=ListTitleLibraryResponse)
|
||||
def list_titles(
|
||||
response: Response,
|
||||
category: Optional[str] = Query(None),
|
||||
skip: int = Query(0, ge=0),
|
||||
limit: int = Query(50, ge=1, le=200),
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
|
||||
) -> ListTitleLibraryResponse:
|
||||
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代."""
|
||||
_log_deprecation("list_titles")
|
||||
for k, v in _deprecation_headers().items():
|
||||
response.headers[k] = v
|
||||
user_id = authenticated_user.user.id
|
||||
use_case = ListTitleLibraryUseCase(title_repository)
|
||||
items = use_case.execute(user_id, category=category, skip=skip, limit=limit)
|
||||
@@ -77,6 +103,7 @@ def list_titles(
|
||||
|
||||
@router.post("/pick", response_model=TitleLibraryItemResponse)
|
||||
def pick_title(
|
||||
response: Response,
|
||||
category: Optional[str] = Query(None, description="按分类筛选,不填则从全部标题中选"),
|
||||
exclude_ids: Optional[str] = Query(
|
||||
None,
|
||||
@@ -85,10 +112,15 @@ def pick_title(
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
|
||||
) -> TitleLibraryItemResponse:
|
||||
"""智能选择一个标题。
|
||||
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代.
|
||||
|
||||
智能选择一个标题。
|
||||
|
||||
策略:优先使用次数少的,从最少的前5个中随机选一个,兼顾公平和多样性。
|
||||
"""
|
||||
_log_deprecation("pick_title")
|
||||
for k, v in _deprecation_headers().items():
|
||||
response.headers[k] = v
|
||||
user_id = authenticated_user.user.id
|
||||
exclude_list: list[str] = []
|
||||
if exclude_ids:
|
||||
@@ -113,9 +145,14 @@ def pick_title(
|
||||
@router.get("/{title_id}", response_model=TitleLibraryItemResponse)
|
||||
def get_title(
|
||||
title_id: str,
|
||||
response: Response,
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
|
||||
) -> TitleLibraryItemResponse:
|
||||
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代."""
|
||||
_log_deprecation("get_title")
|
||||
for k, v in _deprecation_headers().items():
|
||||
response.headers[k] = v
|
||||
user_id = authenticated_user.user.id
|
||||
use_case = GetTitleLibraryUseCase(title_repository)
|
||||
item = use_case.execute(title_id, user_id)
|
||||
@@ -126,11 +163,16 @@ def get_title(
|
||||
|
||||
@router.post("", response_model=TitleLibraryItemResponse, status_code=status.HTTP_201_CREATED)
|
||||
def create_title(
|
||||
response: Response,
|
||||
request: CreateTitleLibraryRequest,
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
|
||||
user_repository: UserRepository = Depends(get_user_repository),
|
||||
) -> TitleLibraryItemResponse:
|
||||
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代."""
|
||||
_log_deprecation("create_title")
|
||||
for k, v in _deprecation_headers().items():
|
||||
response.headers[k] = v
|
||||
user_id = authenticated_user.user.id
|
||||
plan_name = get_user_plan(user_id, user_repository)
|
||||
command = CreateTitleLibraryCommand(
|
||||
@@ -155,10 +197,15 @@ def create_title(
|
||||
@router.put("/{title_id}", response_model=TitleLibraryItemResponse)
|
||||
def update_title(
|
||||
title_id: str,
|
||||
response: Response,
|
||||
request: UpdateTitleLibraryRequest,
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
|
||||
) -> TitleLibraryItemResponse:
|
||||
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代."""
|
||||
_log_deprecation("update_title")
|
||||
for k, v in _deprecation_headers().items():
|
||||
response.headers[k] = v
|
||||
user_id = authenticated_user.user.id
|
||||
command = UpdateTitleLibraryCommand(
|
||||
title_id=title_id,
|
||||
@@ -180,9 +227,14 @@ def update_title(
|
||||
@router.delete("/{title_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, response_class=Response)
|
||||
def delete_title(
|
||||
title_id: str,
|
||||
response: Response,
|
||||
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
||||
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
|
||||
) -> Response:
|
||||
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代."""
|
||||
_log_deprecation("delete_title")
|
||||
for k, v in _deprecation_headers().items():
|
||||
response.headers[k] = v
|
||||
user_id = authenticated_user.user.id
|
||||
use_case = DeleteTitleLibraryUseCase(title_repository)
|
||||
deleted = use_case.execute(title_id, user_id)
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
@@ -22,6 +22,9 @@ class ScriptResponse(BaseModel):
|
||||
content: str
|
||||
segments: list[ScriptSegment] = Field(default_factory=list)
|
||||
tags: list[str] = Field(default_factory=list)
|
||||
title_text: str = ""
|
||||
title_category: str = ""
|
||||
title_config: Dict[str, Any] = Field(default_factory=dict)
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
@@ -36,6 +39,9 @@ class CreateScriptRequest(BaseModel):
|
||||
content: str = ""
|
||||
segments: list[ScriptSegment] = Field(default_factory=list)
|
||||
tags: list[str] = Field(default_factory=list)
|
||||
title_text: str = ""
|
||||
title_category: str = ""
|
||||
title_config: Optional[Dict[str, Any]] = None
|
||||
|
||||
|
||||
class UpdateScriptRequest(BaseModel):
|
||||
@@ -43,3 +49,6 @@ class UpdateScriptRequest(BaseModel):
|
||||
content: Optional[str] = None
|
||||
segments: Optional[list[ScriptSegment]] = None
|
||||
tags: Optional[list[str]] = None
|
||||
title_text: Optional[str] = None
|
||||
title_category: Optional[str] = None
|
||||
title_config: Optional[Dict[str, Any]] = None
|
||||
|
||||
@@ -51,6 +51,9 @@ class ScriptService:
|
||||
content: str = "",
|
||||
segments: list | None = None,
|
||||
tags: list | None = None,
|
||||
title_text: str = "",
|
||||
title_category: str = "",
|
||||
title_config: dict | None = None,
|
||||
) -> ScriptModel:
|
||||
script = ScriptModel(
|
||||
id=str(uuid.uuid4()),
|
||||
@@ -59,6 +62,9 @@ class ScriptService:
|
||||
content=content,
|
||||
segments=segments if segments is not None else [],
|
||||
tags=tags if tags is not None else [],
|
||||
title_text=title_text or "",
|
||||
title_category=title_category or "",
|
||||
title_config=title_config if title_config is not None else {},
|
||||
)
|
||||
self.db.add(script)
|
||||
self.db.commit()
|
||||
@@ -83,6 +89,9 @@ class ScriptService:
|
||||
content: Optional[str] = None,
|
||||
segments: Optional[list] = None,
|
||||
tags: Optional[list] = None,
|
||||
title_text: Optional[str] = None,
|
||||
title_category: Optional[str] = None,
|
||||
title_config: Optional[dict] = None,
|
||||
) -> ScriptModel:
|
||||
script = self.get_script(script_id, user_id)
|
||||
if title is not None:
|
||||
@@ -93,11 +102,27 @@ class ScriptService:
|
||||
script.segments = segments
|
||||
if tags is not None:
|
||||
script.tags = tags
|
||||
if title_text is not None:
|
||||
script.title_text = title_text
|
||||
if title_category is not None:
|
||||
script.title_category = title_category
|
||||
if title_config is not None:
|
||||
script.title_config = title_config
|
||||
script.updated_at = datetime.now(UTC)
|
||||
self.db.commit()
|
||||
self.db.refresh(script)
|
||||
return script
|
||||
|
||||
# ── title config ─────────────────────────────────────────────────────
|
||||
|
||||
def get_title_config_for_script(self, script_id: str, user_id: str) -> dict:
|
||||
"""从 script 读取标题配置,返回可直接用于渲染的 title_config dict."""
|
||||
script = self.get_script(script_id, user_id)
|
||||
config = dict(script.title_config or {})
|
||||
if not config.get("text") and script.title_text:
|
||||
config["text"] = script.title_text
|
||||
return config
|
||||
|
||||
# ── delete ────────────────────────────────────────────────────────────
|
||||
|
||||
def delete_script(self, script_id: str, user_id: str) -> bool:
|
||||
|
||||
@@ -671,6 +671,10 @@ class ScriptModel(Base):
|
||||
content = Column(Text, nullable=False, default="")
|
||||
segments = Column(JSON, nullable=False, default=list)
|
||||
tags = Column(JSON, nullable=False, default=list)
|
||||
# #1894: 废弃标题库整合到文案库 — 标题配置字段
|
||||
title_text = Column(String(500), nullable=False, default="")
|
||||
title_category = Column(String(50), nullable=False, default="")
|
||||
title_config = Column(JSON, nullable=False, default=dict)
|
||||
created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
|
||||
updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
|
||||
|
||||
|
||||
@@ -0,0 +1,334 @@
|
||||
"""#1894 废弃标题库整合到文案库 — 集成测试.
|
||||
|
||||
覆盖:
|
||||
- ScriptModel 新字段 (title_text / title_category / title_config)
|
||||
- ScriptService CRUD 新字段支持
|
||||
- ScriptService.get_title_config_for_script 方法
|
||||
- Scripts API 路由的新字段传递
|
||||
- title_libraries API deprecated Warning header
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import uuid
|
||||
from datetime import UTC, datetime, timezone
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
# 确保 apps/api 在 sys.path 中(conftest 已加 root,但 apps/api 也需要)
|
||||
_APPS_API = str(Path(__file__).resolve().parents[2] / "apps" / "api")
|
||||
if _APPS_API not in sys.path:
|
||||
sys.path.insert(0, _APPS_API)
|
||||
os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing")
|
||||
|
||||
from main import app # noqa: E402
|
||||
|
||||
# ── helpers ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _make_script(**overrides):
|
||||
"""构造一个模拟 ScriptModel 对象."""
|
||||
defaults = dict(
|
||||
id=str(uuid.uuid4()),
|
||||
user_id="user-001",
|
||||
title="测试文案",
|
||||
content="这是内容",
|
||||
segments=[],
|
||||
tags=["测试"],
|
||||
title_text="开场大标题",
|
||||
title_category="片头",
|
||||
title_config={
|
||||
"text": "开场大标题",
|
||||
"font": "思源黑体",
|
||||
"font_size": 48,
|
||||
"font_color": "#FFFFFF",
|
||||
"position": "top",
|
||||
},
|
||||
created_at=datetime(2026, 9, 1, tzinfo=timezone.utc),
|
||||
updated_at=datetime(2026, 9, 1, tzinfo=timezone.utc),
|
||||
)
|
||||
defaults.update(overrides)
|
||||
return MagicMock(**defaults)
|
||||
|
||||
|
||||
# ── TestScriptModelNewFields ─────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestScriptModelNewFields:
|
||||
"""验证 ScriptModel 新增字段的定义."""
|
||||
|
||||
def test_model_has_title_text_column(self):
|
||||
from packages.adapters.sqlalchemy_impl.models import ScriptModel
|
||||
|
||||
assert hasattr(ScriptModel, "title_text")
|
||||
col = ScriptModel.__table__.columns["title_text"]
|
||||
assert col is not None
|
||||
assert str(col.type) == "VARCHAR(500)"
|
||||
|
||||
def test_model_has_title_category_column(self):
|
||||
from packages.adapters.sqlalchemy_impl.models import ScriptModel
|
||||
|
||||
assert hasattr(ScriptModel, "title_category")
|
||||
col = ScriptModel.__table__.columns["title_category"]
|
||||
assert col is not None
|
||||
assert str(col.type) == "VARCHAR(50)"
|
||||
|
||||
def test_model_has_title_config_column(self):
|
||||
from packages.adapters.sqlalchemy_impl.models import ScriptModel
|
||||
|
||||
assert hasattr(ScriptModel, "title_config")
|
||||
col = ScriptModel.__table__.columns["title_config"]
|
||||
assert col is not None
|
||||
|
||||
def test_model_defaults(self):
|
||||
"""新字段默认值为空字符串/空 dict."""
|
||||
from packages.adapters.sqlalchemy_impl.models import ScriptModel
|
||||
|
||||
s = ScriptModel(id="x", user_id="u", title="t")
|
||||
# 检查 default 值
|
||||
assert ScriptModel.__table__.columns["title_text"].default.arg == ""
|
||||
assert ScriptModel.__table__.columns["title_category"].default.arg == ""
|
||||
|
||||
|
||||
# ── TestScriptServiceTitleConfig ─────────────────────────────────────────
|
||||
|
||||
|
||||
class TestScriptServiceTitleConfig:
|
||||
"""验证 ScriptService 新方法 get_title_config_for_script."""
|
||||
|
||||
def test_get_title_config_returns_script_config(self):
|
||||
from app.services.script_service import ScriptService
|
||||
|
||||
db = MagicMock()
|
||||
mock_script = _make_script(
|
||||
title_text="从文案读取",
|
||||
title_config={"text": "从文案读取", "font": "Arial", "font_size": 36},
|
||||
)
|
||||
db.query.return_value.filter.return_value.first.return_value = mock_script
|
||||
|
||||
svc = ScriptService(db)
|
||||
result = svc.get_title_config_for_script("script-1", "user-001")
|
||||
|
||||
assert result["text"] == "从文案读取"
|
||||
assert result["font"] == "Arial"
|
||||
assert result["font_size"] == 36
|
||||
|
||||
def test_get_title_config_fills_text_from_title_text(self):
|
||||
"""title_config 为空时,用 title_text 填充 text 字段."""
|
||||
from app.services.script_service import ScriptService
|
||||
|
||||
db = MagicMock()
|
||||
mock_script = _make_script(
|
||||
title_text="纯文本标题",
|
||||
title_config={},
|
||||
)
|
||||
db.query.return_value.filter.return_value.first.return_value = mock_script
|
||||
|
||||
svc = ScriptService(db)
|
||||
result = svc.get_title_config_for_script("script-2", "user-001")
|
||||
|
||||
assert result["text"] == "纯文本标题"
|
||||
|
||||
def test_get_title_config_raises_on_not_found(self):
|
||||
from app.services.script_service import ScriptNotFoundError, ScriptService
|
||||
|
||||
db = MagicMock()
|
||||
db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
svc = ScriptService(db)
|
||||
with pytest.raises(ScriptNotFoundError):
|
||||
svc.get_title_config_for_script("nonexistent", "user-001")
|
||||
|
||||
def test_get_title_config_validates_user_ownership(self):
|
||||
"""script 不属于当前用户时应抛异常."""
|
||||
from app.services.script_service import ScriptNotFoundError, ScriptService
|
||||
|
||||
db = MagicMock()
|
||||
db.query.return_value.filter.return_value.first.return_value = None # 不同用户查不到
|
||||
|
||||
svc = ScriptService(db)
|
||||
with pytest.raises(ScriptNotFoundError):
|
||||
svc.get_title_config_for_script("script-other-user", "user-001")
|
||||
|
||||
|
||||
# ── TestScriptServiceCreateWithNewFields ─────────────────────────────────
|
||||
|
||||
|
||||
class TestScriptServiceCreateWithNewFields:
|
||||
"""验证 create_script 和 update_script 支持新字段."""
|
||||
|
||||
def test_create_script_with_title_fields(self):
|
||||
from app.services.script_service import ScriptService
|
||||
|
||||
db = MagicMock()
|
||||
svc = ScriptService(db)
|
||||
|
||||
script = svc.create_script(
|
||||
user_id="user-001",
|
||||
title="新文案",
|
||||
content="内容",
|
||||
title_text="标题文字",
|
||||
title_category="片尾",
|
||||
title_config={"text": "标题文字", "font_size": 24},
|
||||
)
|
||||
|
||||
db.add.assert_called_once()
|
||||
db.commit.assert_called_once()
|
||||
assert script.title_text == "标题文字"
|
||||
assert script.title_category == "片尾"
|
||||
assert script.title_config == {"text": "标题文字", "font_size": 24}
|
||||
|
||||
def test_update_script_title_fields(self):
|
||||
from app.services.script_service import ScriptService
|
||||
|
||||
db = MagicMock()
|
||||
existing = _make_script(title_text="旧标题", title_category="旧分类", title_config={"old": True})
|
||||
db.query.return_value.filter.return_value.first.return_value = existing
|
||||
|
||||
svc = ScriptService(db)
|
||||
updated = svc.update_script(
|
||||
script_id=existing.id,
|
||||
user_id="user-001",
|
||||
title_text="新标题",
|
||||
title_category="新分类",
|
||||
title_config={"new": True},
|
||||
)
|
||||
|
||||
assert updated.title_text == "新标题"
|
||||
assert updated.title_category == "新分类"
|
||||
assert updated.title_config == {"new": True}
|
||||
|
||||
|
||||
# ── TestScriptsRoutesNewFields ───────────────────────────────────────────
|
||||
|
||||
|
||||
def _make_mock_auth_user(user_id="user-001"):
|
||||
"""创建 mock 认证用户."""
|
||||
return MagicMock(user=MagicMock(id=user_id))
|
||||
|
||||
|
||||
class TestScriptsRoutesNewFields:
|
||||
"""验证 scripts API 路由正确处理新字段 — 使用 dependency_overrides 绕过真实 DB/Auth."""
|
||||
|
||||
def setup_method(self):
|
||||
from app.api.routes.scripts import _get_service, get_current_user
|
||||
|
||||
self._mock_svc = MagicMock()
|
||||
self._mock_user = _make_mock_auth_user()
|
||||
|
||||
def _override_svc():
|
||||
return self._mock_svc
|
||||
|
||||
def _override_user():
|
||||
return self._mock_user
|
||||
|
||||
app.dependency_overrides[_get_service] = _override_svc
|
||||
app.dependency_overrides[get_current_user] = _override_user
|
||||
self.client = TestClient(app)
|
||||
|
||||
def teardown_method(self):
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
def test_create_script_passes_title_fields(self):
|
||||
mock_script = _make_script(
|
||||
title_text="测试标题",
|
||||
title_category="片头",
|
||||
title_config={"text": "测试标题", "font_size": 48},
|
||||
)
|
||||
self._mock_svc.create_script.return_value = mock_script
|
||||
|
||||
resp = self.client.post(
|
||||
"/api/v1/scripts",
|
||||
json={
|
||||
"title": "新文案",
|
||||
"content": "内容",
|
||||
"title_text": "测试标题",
|
||||
"title_category": "片头",
|
||||
"title_config": {"text": "测试标题", "font_size": 48},
|
||||
},
|
||||
)
|
||||
|
||||
assert resp.status_code == 201, resp.text
|
||||
call_kwargs = self._mock_svc.create_script.call_args[1]
|
||||
assert call_kwargs["title_text"] == "测试标题"
|
||||
assert call_kwargs["title_category"] == "片头"
|
||||
assert call_kwargs["title_config"] == {"text": "测试标题", "font_size": 48}
|
||||
|
||||
def test_get_script_response_includes_title_fields(self):
|
||||
mock_script = _make_script(
|
||||
title_text="响应标题",
|
||||
title_category="片尾",
|
||||
title_config={"text": "响应标题", "position": "bottom"},
|
||||
)
|
||||
self._mock_svc.get_script.return_value = mock_script
|
||||
|
||||
resp = self.client.get("/api/v1/scripts/script-123")
|
||||
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["title_text"] == "响应标题"
|
||||
assert data["title_category"] == "片尾"
|
||||
assert data["title_config"]["position"] == "bottom"
|
||||
|
||||
|
||||
# ── TestTitleLibraryDeprecated ───────────────────────────────────────────
|
||||
|
||||
|
||||
class TestTitleLibraryDeprecated:
|
||||
"""验证 title_libraries API 返回 deprecated Warning header — 使用 dependency_overrides 绕过真实 DB/Auth."""
|
||||
|
||||
def setup_method(self):
|
||||
from app.api.routes.titles import _get_title_repository, get_current_user
|
||||
from app.dependencies import get_user_repository
|
||||
|
||||
self._mock_repo = MagicMock()
|
||||
self._mock_user_repo = MagicMock()
|
||||
self._mock_user = _make_mock_auth_user()
|
||||
|
||||
app.dependency_overrides[_get_title_repository] = lambda: self._mock_repo
|
||||
app.dependency_overrides[get_user_repository] = lambda: self._mock_user_repo
|
||||
app.dependency_overrides[get_current_user] = lambda: self._mock_user
|
||||
self.client = TestClient(app)
|
||||
|
||||
def teardown_method(self):
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
def test_list_titles_has_warning_header(self):
|
||||
# list_titles 调 use_case + repo, 注入真实用例但 mock 掉 repo 的 list/count
|
||||
self._mock_repo.list_by_user.return_value = []
|
||||
self._mock_repo.count_by_user.return_value = 0
|
||||
|
||||
resp = self.client.get("/api/v1/titles")
|
||||
|
||||
assert resp.status_code == 200, resp.text
|
||||
headers_lower = {k.lower(): v for k, v in resp.headers.items()}
|
||||
assert "warning" in headers_lower or "deprecation" in headers_lower
|
||||
assert "1894" in resp.headers.get("Warning", "") or "1894" in resp.headers.get("warning", "")
|
||||
|
||||
def test_get_title_has_warning_header(self):
|
||||
from packages.domain.title_library import TitleLibraryItem
|
||||
|
||||
mock_item = TitleLibraryItem(
|
||||
id="t1",
|
||||
user_id="user-001",
|
||||
name="测试",
|
||||
text="标题文字",
|
||||
category="通用",
|
||||
description="",
|
||||
tags=[],
|
||||
usage_count=0,
|
||||
is_active=True,
|
||||
created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
updated_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
)
|
||||
self._mock_repo.get.return_value = mock_item
|
||||
|
||||
resp = self.client.get("/api/v1/titles/t1")
|
||||
assert resp.status_code == 200, resp.text
|
||||
warning_header = resp.headers.get("Warning", "") or resp.headers.get("warning", "")
|
||||
assert "1894" in warning_header or "deprecated" in warning_header.lower()
|
||||
@@ -70,12 +70,21 @@ class TestUpdateScriptRequest:
|
||||
assert r.content is None
|
||||
assert r.segments is None
|
||||
assert r.tags is None
|
||||
assert r.title_text is None
|
||||
assert r.title_category is None
|
||||
assert r.title_config is None
|
||||
|
||||
def test_partial_update(self):
|
||||
r = UpdateScriptRequest(title="新标题")
|
||||
assert r.title == "新标题"
|
||||
assert r.content is None
|
||||
|
||||
def test_partial_update_title_fields(self):
|
||||
r = UpdateScriptRequest(title_text="新标题文本", title_category="娱乐")
|
||||
assert r.title_text == "新标题文本"
|
||||
assert r.title_category == "娱乐"
|
||||
assert r.title is None
|
||||
|
||||
|
||||
class TestScriptResponse:
|
||||
def test_response_construction(self):
|
||||
@@ -87,11 +96,28 @@ class TestScriptResponse:
|
||||
content="内容",
|
||||
segments=[ScriptSegment(text="段1")],
|
||||
tags=["t1"],
|
||||
title_text="标题文案",
|
||||
title_category="科技",
|
||||
title_config={"font": "思源黑体", "size": 48},
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
assert r.id == "s1"
|
||||
assert len(r.segments) == 1
|
||||
assert r.title_text == "标题文案"
|
||||
assert r.title_category == "科技"
|
||||
assert r.title_config["font"] == "思源黑体"
|
||||
|
||||
def test_response_defaults(self):
|
||||
"""新字段有默认值,不传也能构造."""
|
||||
now = datetime(2026, 9, 8, 12, 0, 0, tzinfo=UTC)
|
||||
r = ScriptResponse(
|
||||
id="s1", user_id="u1", title="标题", content="",
|
||||
segments=[], tags=[], created_at=now, updated_at=now,
|
||||
)
|
||||
assert r.title_text == ""
|
||||
assert r.title_category == ""
|
||||
assert r.title_config == {}
|
||||
|
||||
|
||||
class TestScriptListResponse:
|
||||
@@ -141,6 +167,9 @@ class TestRouteHandlers:
|
||||
mock_script.content = "内容"
|
||||
mock_script.segments = [{"text": "段1", "duration": None}]
|
||||
mock_script.tags = []
|
||||
mock_script.title_text = ""
|
||||
mock_script.title_category = ""
|
||||
mock_script.title_config = {}
|
||||
mock_script.created_at = datetime(2026, 9, 8, tzinfo=UTC)
|
||||
mock_script.updated_at = datetime(2026, 9, 8, tzinfo=UTC)
|
||||
svc.create_script.return_value = mock_script
|
||||
@@ -163,6 +192,9 @@ class TestRouteHandlers:
|
||||
mock_script.content = ""
|
||||
mock_script.segments = []
|
||||
mock_script.tags = []
|
||||
mock_script.title_text = ""
|
||||
mock_script.title_category = ""
|
||||
mock_script.title_config = {}
|
||||
mock_script.created_at = datetime(2026, 9, 8, tzinfo=UTC)
|
||||
mock_script.updated_at = datetime(2026, 9, 8, tzinfo=UTC)
|
||||
svc.list_scripts.return_value = ([mock_script], 1)
|
||||
@@ -183,6 +215,9 @@ class TestRouteHandlers:
|
||||
mock_script.content = ""
|
||||
mock_script.segments = []
|
||||
mock_script.tags = []
|
||||
mock_script.title_text = ""
|
||||
mock_script.title_category = ""
|
||||
mock_script.title_config = {}
|
||||
mock_script.created_at = datetime(2026, 9, 8, tzinfo=UTC)
|
||||
mock_script.updated_at = datetime(2026, 9, 8, tzinfo=UTC)
|
||||
svc.get_script.return_value = mock_script
|
||||
@@ -215,6 +250,9 @@ class TestRouteHandlers:
|
||||
mock_script.content = "原内容"
|
||||
mock_script.segments = []
|
||||
mock_script.tags = []
|
||||
mock_script.title_text = ""
|
||||
mock_script.title_category = ""
|
||||
mock_script.title_config = {}
|
||||
mock_script.created_at = datetime(2026, 9, 8, tzinfo=UTC)
|
||||
mock_script.updated_at = datetime(2026, 9, 8, tzinfo=UTC)
|
||||
svc.update_script.return_value = mock_script
|
||||
@@ -260,3 +298,4 @@ class TestRouteHandlers:
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
delete_script("bad", authenticated_user=auth, svc=svc)
|
||||
assert exc_info.value.status_code == 404
|
||||
|
||||
|
||||
Reference in New Issue
Block a user