Compare commits

...

2 Commits

Author SHA1 Message Date
xiaoxia 2014eaf745 fix: #1286 can_generate 增加最后防线自动修复,解决「没有可渲染的就绪片段」报错 (#1288)
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 35s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 51s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m0s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m47s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 2m2s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m50s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m14s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 2m2s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 4m40s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m38s
AI Code Review / AI Code Review (pull_request) Successful in 7m12s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 7m55s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Successful in 6s
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 52s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Waiting to run
2026-08-07 22:49:07 +08:00
SaaS Frontend Agent 38794923d7 fix(ci): 修复 staging 测试 shell 引号嵌套 bug,测试从未实际运行
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 41s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 51s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m2s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m23s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m46s
AI Code Review / AI Code Review (pull_request) Successful in 2m42s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 4m22s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 4m56s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 5m22s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 5m21s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 4m9s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 9m30s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Failing after 19s
根因:
1. E2E Tests: sh -lc "...$i..." 双引号导致 $i 被外层 bash 展开为 unbound variable
2. API Tests: sh -lc '...\"...\"...' 单引号内 \" 是字面反斜杠+引号,dash 报 Syntax error
3. Production E2E 同样存在引号问题

修复:
- 提取 docker run 逻辑为 scripts/ci/run_staging_tests.sh
- 容器内用 bash -c '...'(单引号保护,$i 不会被外层展开)
- 用 bash 替代 sh(dash),避免引号解析差异
- YAML 中简化为 bash scripts/ci/run_staging_tests.sh e2e|api
- Production E2E 同样修复为 bash -c '...'
2026-08-07 22:36:55 +08:00
6 changed files with 486 additions and 45 deletions
+4 -39
View File
@@ -1086,25 +1086,7 @@ jobs:
- name: Run Playwright E2E on staging
shell: bash
run: |
set -eu
# DooD模式下不能用-v挂载(宿主机路径与CI容器路径不一致)
# 改用 docker create + docker cp 方式把代码拷进容器
CONTAINER_NAME="staging-e2e-$$"
# 强制清理可能残留的同名容器(上一次异常退出时未清理)
docker rm -f "$CONTAINER_NAME" 2>/dev/null || true
docker create --name "$CONTAINER_NAME" --ipc=host \
-e E2E_BASE_URL=https://staging.xiaoxiajianji.com \
-e E2E_API_BASE=https://staging-api.xiaoxiajianji.com/api/v1 \
-e E2E_BROWSER_CHANNEL=chromium \
-e PLAYWRIGHT_HEADLESS=1 \
-w /workspace/apps/web \
git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
sh -lc "for i in 1 2 3; do npm ci --registry=https://registry.npmmirror.com && break; echo \"npm ci failed (attempt $i), retrying in 15s...\"; sleep 15; done && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts"
docker cp apps "$CONTAINER_NAME:/workspace/"
docker start -a "$CONTAINER_NAME"
EXIT_CODE=$(docker wait "$CONTAINER_NAME")
docker rm "$CONTAINER_NAME" 2>/dev/null || true
exit $EXIT_CODE
bash scripts/ci/run_staging_tests.sh e2e
- name: Job duration summary
if: always()
@@ -1151,23 +1133,7 @@ jobs:
- name: Run API integration tests on staging
shell: bash
run: |
set -eu
# DooD模式下不能用-v挂载(宿主机路径与CI容器路径不一致)
# 改用 docker create + docker cp 方式把代码拷进容器
CONTAINER_NAME="staging-api-tests-$$"
# 强制清理可能残留的同名容器(上一次异常退出时未清理)
docker rm -f "$CONTAINER_NAME" 2>/dev/null || true
docker create --name "$CONTAINER_NAME" \
-e E2E_BASE_URL=https://staging.xiaoxiajianji.com \
-e E2E_API_BASE=https://staging-api.xiaoxiajianji.com/api/v1 \
-w /workspace/apps/web \
git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
sh -lc 'for i in 1 2 3; do npm ci --registry=https://registry.npmmirror.com && break; echo \"npm ci failed (attempt $i), retrying in 15s...\"; sleep 15; done && npx playwright test --reporter=line e2e/test_auth.spec.ts e2e/test_asset.spec.ts e2e/test_project.spec.ts'
docker cp apps "$CONTAINER_NAME:/workspace/"
docker start -a "$CONTAINER_NAME"
EXIT_CODE=$(docker wait "$CONTAINER_NAME")
docker rm "$CONTAINER_NAME" 2>/dev/null || true
exit $EXIT_CODE
bash scripts/ci/run_staging_tests.sh api
- name: Job duration summary
if: always()
@@ -1494,9 +1460,8 @@ jobs:
shell: sh
run: bash scripts/ci/step_timer_start.sh
- name: Run production browser E2E
shell: sh
shell: bash
run: |
set -eu
docker run --rm --ipc=host \
-e E2E_BASE_URL=https://saas.xiaoxiajianji.com \
-e E2E_API_BASE=https://api.xiaoxiajianji.com/api/v1 \
@@ -1505,7 +1470,7 @@ jobs:
-v "$PWD:/workspace" \
-w /workspace/apps/web \
git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
sh -lc 'for i in 1 2 3; do npm ci --registry=https://registry.npmmirror.com && break; echo \"npm ci failed (attempt $i), retrying in 15s...\"; sleep 15; done && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts e2e/core-titles.spec.ts'
bash -c 'for i in 1 2 3; do npm ci --registry=https://registry.npmmirror.com && break; echo "npm ci attempt $i failed, retrying..."; sleep 15; done && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts e2e/core-titles.spec.ts'
- name: Job duration summary
if: always()
@@ -98,6 +98,15 @@ def _auto_fallback_assign_assets(
clips_without_asset = [c for c in all_clips if not c.asset_id]
config_asset_ids = (plan_check.config or {}).get("asset_ids", [])
logger.info(
"模板编辑器自动兜底3 诊断: plan=%s total_clips=%d "
"clips_without_asset=%d config_asset_ids=%r",
plan_id,
len(all_clips),
len(clips_without_asset),
config_asset_ids[:5] if config_asset_ids else [],
)
if clips_without_asset and config_asset_ids:
logger.info(
"模板编辑器自动兜底3: plan=%s%d 个无素材片段分配 %d 个指定素材",
@@ -105,11 +114,42 @@ def _auto_fallback_assign_assets(
len(clips_without_asset),
len(config_asset_ids),
)
assigned = 0
for i, clip in enumerate(clips_without_asset):
asset_idx = i % len(config_asset_ids)
svc.assign_asset(clip.id, config_asset_ids[asset_idx])
logger.info("模板编辑器自动兜底3: plan=%s 素材分配完成", plan_id)
clips_without_asset = []
try:
svc.assign_asset(clip.id, config_asset_ids[asset_idx])
assigned += 1
except Exception as exc:
logger.error(
"模板编辑器自动兜底3: plan=%s clip=%s 分配素材 %s 失败: %s",
plan_id,
clip.id,
config_asset_ids[asset_idx],
exc,
)
logger.info(
"模板编辑器自动兜底3: plan=%s 素材分配完成 assigned=%d/%d",
plan_id,
assigned,
len(clips_without_asset),
)
# 重新检查剩余无素材片段
all_clips_after = svc.list_clips(plan_id)
clips_without_asset = [c for c in all_clips_after if not c.asset_id]
if clips_without_asset:
logger.warning(
"模板编辑器自动兜底3: plan=%s 仍有 %d 个片段无素材",
plan_id,
len(clips_without_asset),
)
elif not clips_without_asset:
logger.info("模板编辑器自动兜底3: plan=%s 所有片段已有素材,跳过", plan_id)
elif not config_asset_ids:
logger.info(
"模板编辑器自动兜底3: plan=%s config.asset_ids 为空,跳过分配",
plan_id,
)
return clips_without_asset
@@ -74,7 +74,7 @@ def generate_editor_draft(
plan_svc, plan_id, plan_check, clips_without_asset, asset_library_repo, asset_repo
)
# 检查是否可生成
# 检查是否可生成(含最后防线自动修复 + 诊断日志)
try:
can_gen, reason = plan_svc.can_generate(plan_id)
except ValueError as exc:
+57 -2
View File
@@ -571,6 +571,10 @@ class EditPlanService:
def can_generate(self, plan_id: str) -> tuple[bool, str]:
"""检查是否可以触发渲染
包含最后一道防线的自动修复:
- 如果 clips 存在但都没有 asset_id,且 config.asset_ids 非空,
直接在内部执行素材分配,不再依赖前置 fallback 链路。
Returns:
tuple: (can_generate, reason)
"""
@@ -586,8 +590,59 @@ class EditPlanService:
return False, "请先添加片段后再生成视频"
# 检查是否至少有一个片段分配了素材
if not any(c.asset_id for c in clips):
return False, "没有可渲染的就绪片段,请确保已选择素材"
has_asset = any(c.asset_id for c in clips)
config_asset_ids_count = len((plan.config or {}).get("asset_ids", []))
clips_with_asset_count = sum(1 for c in clips if c.asset_id)
logger.info(
"can_generate 诊断: plan=%s status=%s total_clips=%d " "clips_with_asset=%d config_asset_ids_count=%d",
plan_id,
plan.status,
len(clips),
clips_with_asset_count,
config_asset_ids_count,
)
if not has_asset:
# ── 最后防线:自动从 config.asset_ids 分配素材 ──
config_asset_ids = (plan.config or {}).get("asset_ids", [])
if config_asset_ids:
logger.warning(
"can_generate 最后防线触发: plan=%s clips=%d 均无素材," "从 config.asset_ids(%d个) 自动分配",
plan_id,
len(clips),
len(config_asset_ids),
)
clips_without_asset = [c for c in clips if not c.asset_id]
assigned_count = 0
for i, clip in enumerate(clips_without_asset):
asset_idx = i % len(config_asset_ids)
try:
self.assign_asset(clip.id, config_asset_ids[asset_idx])
assigned_count += 1
except Exception as exc:
logger.warning(
"can_generate 最后防线: plan=%s clip=%s 分配素材 %s 失败: %s",
plan_id,
clip.id,
config_asset_ids[asset_idx],
exc,
)
logger.info(
"can_generate 最后防线: plan=%s 已为 %d/%d 个片段分配素材",
plan_id,
assigned_count,
len(clips_without_asset),
)
# 重新加载 clips 验证分配结果
clips = self._clip_repo.list_by_plan(plan_id)
if not any(c.asset_id for c in clips):
return False, "没有可渲染的就绪片段,自动修复后仍未分配素材"
else:
logger.warning(
"can_generate 失败: plan=%s clips=%d 均无素材," "且 config.asset_ids 为空,无法自动修复",
plan_id,
len(clips),
)
return False, "没有可渲染的就绪片段,请确保已选择素材"
return True, ""
+50
View File
@@ -0,0 +1,50 @@
#!/usr/bin/env bash
# scripts/ci/run_staging_tests.sh
# 在 staging 环境中运行 Playwright 测试
# 用法: bash scripts/ci/run_staging_tests.sh <mode>
# mode: e2e | api
#
# 解决 .gitea/workflows/ci-pipeline.yml 中多层引号嵌套问题:
# - 外层 YAML → bash → docker create → 容器内 sh/bash → 字符串解析
# - 提取为脚本后,只有两层:bash → 容器内 bash(单引号保护)
set -eu
MODE="${1:-e2e}"
CONTAINER_NAME="staging-${MODE}-$$"
# 强制清理可能残留的同名容器
docker rm -f "$CONTAINER_NAME" 2>/dev/null || true
if [ "$MODE" = "e2e" ]; then
docker create --name "$CONTAINER_NAME" --ipc=host \
-e E2E_BASE_URL=https://staging.xiaoxiajianji.com \
-e E2E_API_BASE=https://staging-api.xiaoxiajianji.com/api/v1 \
-e E2E_BROWSER_CHANNEL=chromium \
-e PLAYWRIGHT_HEADLESS=1 \
-w /workspace/apps/web \
git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
bash -c 'for i in 1 2 3; do npm ci --registry=https://registry.npmmirror.com && break; echo "npm ci attempt $i failed, retrying in 15s..."; sleep 15; done && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts'
elif [ "$MODE" = "api" ]; then
docker create --name "$CONTAINER_NAME" \
-e E2E_BASE_URL=https://staging.xiaoxiajianji.com \
-e E2E_API_BASE=https://staging-api.xiaoxiajianji.com/api/v1 \
-w /workspace/apps/web \
git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
bash -c 'for i in 1 2 3; do npm ci --registry=https://registry.npmmirror.com && break; echo "npm ci attempt $i failed, retrying in 15s..."; sleep 15; done && npx playwright test --reporter=line e2e/test_auth.spec.ts e2e/test_asset.spec.ts e2e/test_project.spec.ts'
else
echo "ERROR: Unknown mode '$MODE'. Use 'e2e' or 'api'." >&2
exit 1
fi
# 把代码拷进容器
docker cp apps "$CONTAINER_NAME:/workspace/"
# 启动并等待
docker start -a "$CONTAINER_NAME"
EXIT_CODE=$(docker wait "$CONTAINER_NAME")
# 清理容器
docker rm "$CONTAINER_NAME" 2>/dev/null || true
exit "$EXIT_CODE"
@@ -0,0 +1,331 @@
"""
#1286 can_generate 最后防线自动修复 — 单元测试
覆盖场景:
1. clips 无素材 + config.asset_ids 非空 → 自动分配成功 → can_generate 返回 True
2. clips 无素材 + config.asset_ids 为空 → 无法修复 → can_generate 返回 False
3. clips 已有素材 → 正常通过,不触发自动修复
4. 自动修复后素材数量与 clips 数量一致(循环分配验证)
5. 自动修复不影响预览生成流程(预览不依赖 EditPlan clips
6. config 为 None 时安全降级
7. 无 clips 时仍返回 False
8. 部分 clips 有素材时正常通过(不触发自动修复)
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from typing import Any, List, Optional
from unittest.mock import MagicMock
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from packages.domain.edit_plan import EditPlan, EditPlanStatus
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
# ---------------------------------------------------------------------------
# Stub Repositories
# ---------------------------------------------------------------------------
class StubEditPlanRepository:
def __init__(self) -> None:
self._plans: dict[str, EditPlan] = {}
self._counter = 0
def _next_id(self) -> str:
self._counter += 1
return f"plan-auto-{self._counter:03d}"
def get(self, plan_id: str) -> Optional[EditPlan]:
return self._plans.get(plan_id)
def create(self, plan: EditPlan) -> EditPlan:
if not plan.id:
plan = EditPlan(
id=self._next_id(),
template_id=plan.template_id,
name=plan.name,
status=plan.status,
total_duration=plan.total_duration,
source_edit_plan_id=plan.source_edit_plan_id,
project_id=plan.project_id,
created_by_user_id=plan.created_by_user_id,
config=plan.config,
)
self._plans[plan.id] = plan
return plan
def update(self, plan: EditPlan) -> EditPlan:
self._plans[plan.id] = plan
return plan
def delete(self, plan_id: str) -> bool:
return self._plans.pop(plan_id, None) is not None
def list_all(self, **kwargs):
return list(self._plans.values())
class StubEditPlanClipRepository:
def __init__(self) -> None:
self._clips: dict[str, EditPlanClip] = {}
self._counter = 0
def _next_id(self) -> str:
self._counter += 1
return f"clip-auto-{self._counter:03d}"
def get(self, clip_id: str) -> Optional[EditPlanClip]:
return self._clips.get(clip_id)
def create(self, clip: EditPlanClip) -> EditPlanClip:
if not clip.id:
clip = EditPlanClip(
id=self._next_id(),
plan_id=clip.plan_id,
clip_type=clip.clip_type,
order=clip.order,
duration=clip.duration,
status=clip.status,
asset_id=clip.asset_id,
text_content=clip.text_content,
config=clip.config,
template_clip_config_id=clip.template_clip_config_id,
transition_effect=clip.transition_effect,
)
self._clips[clip.id] = clip
return clip
def update(self, clip: EditPlanClip) -> EditPlanClip:
self._clips[clip.id] = clip
return clip
def delete(self, clip_id: str) -> bool:
return self._clips.pop(clip_id, None) is not None
def list_by_plan(
self,
plan_id: str,
*,
status: Optional[EditPlanClipStatus] = None,
skip: int = 0,
limit: int = 100,
) -> List[EditPlanClip]:
clips = [c for c in self._clips.values() if c.plan_id == plan_id]
if status is not None:
clips = [c for c in clips if c.status == status]
return sorted(clips, key=lambda c: c.order)[skip : skip + limit]
def delete_by_plan(self, plan_id: str) -> int:
to_del = [cid for cid, c in self._clips.items() if c.plan_id == plan_id]
for cid in to_del:
del self._clips[cid]
return len(to_del)
def count_by_plan(self, plan_id: str) -> int:
return sum(1 for c in self._clips.values() if c.plan_id == plan_id)
class StubGenerationTaskRepository:
def __init__(self):
pass
def _make_service():
from app.services.edit_plan_service import EditPlanService
db = MagicMock()
svc = EditPlanService(db)
svc._plan_repo = StubEditPlanRepository()
svc._clip_repo = StubEditPlanClipRepository()
svc._generation_task_repo = StubGenerationTaskRepository()
return svc
# ===========================================================================
# 测试用例
# ===========================================================================
class TestCanGenerateAutoRepair:
"""#1286 can_generate 最后防线自动修复"""
def test_auto_repair_with_config_asset_ids(self):
"""clips 无素材 + config.asset_ids 非空 → 自动分配成功 → can_generate True"""
svc = _make_service()
p = svc.create_plan(
"tpl-001",
"测试",
config={"asset_ids": ["asset-001", "asset-002"]},
)
svc.transition_status(p.id, EditPlanStatus.EDITING)
svc.create_clip(p.id, "intro", 0)
svc.create_clip(p.id, "main", 1)
# clips 无素材
clips = svc.list_clips(p.id)
assert all(not c.asset_id for c in clips)
# can_generate 应触发自动修复
can, reason = svc.can_generate(p.id)
assert can is True
assert reason == ""
# 验证 clips 已被分配素材
clips_after = svc.list_clips(p.id)
assert all(c.asset_id is not None for c in clips_after)
def test_auto_repair_fails_without_config_asset_ids(self):
"""clips 无素材 + config.asset_ids 为空 → 无法修复 → False"""
svc = _make_service()
p = svc.create_plan("tpl-001", "测试", config={})
svc.transition_status(p.id, EditPlanStatus.EDITING)
svc.create_clip(p.id, "intro", 0)
can, reason = svc.can_generate(p.id)
assert can is False
assert "没有可渲染" in reason or "素材" in reason
def test_no_repair_when_clips_have_assets(self):
"""clips 已有素材 → 正常通过,不触发自动修复"""
svc = _make_service()
p = svc.create_plan(
"tpl-001",
"测试",
config={"asset_ids": ["asset-001", "asset-002"]},
)
svc.transition_status(p.id, EditPlanStatus.EDITING)
clip = svc.create_clip(p.id, "intro", 0)
svc.assign_asset(clip.id, "asset-001")
can, reason = svc.can_generate(p.id)
assert can is True
# 确认 clip 的 asset_id 没有被改变
clips = svc.list_clips(p.id)
assert clips[0].asset_id == "asset-001"
def test_auto_repair_circular_assignment(self):
"""素材少于 clips 时循环分配(取模)"""
svc = _make_service()
p = svc.create_plan(
"tpl-001",
"测试",
config={"asset_ids": ["asset-A"]},
)
svc.transition_status(p.id, EditPlanStatus.EDITING)
svc.create_clip(p.id, "intro", 0)
svc.create_clip(p.id, "main", 1)
svc.create_clip(p.id, "outro", 2)
can, reason = svc.can_generate(p.id)
assert can is True
clips = svc.list_clips(p.id)
# 所有 3 个 clips 都应被分配了同一个 asset-A
assert all(c.asset_id == "asset-A" for c in clips)
def test_auto_repair_does_not_affect_preview_flow(self):
"""预览生成走 generation_preview.py,不依赖 EditPlan clips 的 can_generate"""
svc = _make_service()
p = svc.create_plan(
"tpl-001",
"测试",
config={"asset_ids": ["asset-001"]},
)
# 不切到 editing 状态,模拟预览场景
svc.create_clip(p.id, "intro", 0)
can, reason = svc.can_generate(p.id)
assert can is False
assert "编辑" in reason or "模板" in reason
def test_auto_repair_empty_config(self):
"""config 为 None 时也不报错"""
svc = _make_service()
p = svc.create_plan("tpl-001", "测试")
svc.transition_status(p.id, EditPlanStatus.EDITING)
svc.create_clip(p.id, "intro", 0)
can, reason = svc.can_generate(p.id)
assert can is False
assert "没有可渲染" in reason or "素材" in reason
def test_auto_repair_no_clips_still_fails(self):
"""没有 clips 时仍然返回 False(不进入自动修复分支)"""
svc = _make_service()
p = svc.create_plan(
"tpl-001",
"测试",
config={"asset_ids": ["asset-001"]},
)
svc.transition_status(p.id, EditPlanStatus.EDITING)
can, reason = svc.can_generate(p.id)
assert can is False
assert "请先添加片段" in reason
def test_auto_repair_partial_assets_still_passes(self):
"""部分 clips 有素材、部分没有 → 至少有一个有素材 → 通过(原有逻辑)"""
svc = _make_service()
p = svc.create_plan(
"tpl-001",
"测试",
config={"asset_ids": ["asset-001"]},
)
svc.transition_status(p.id, EditPlanStatus.EDITING)
clip1 = svc.create_clip(p.id, "intro", 0)
svc.create_clip(p.id, "main", 1)
svc.assign_asset(clip1.id, "asset-001")
# 至少一个 clip 有素材 → 通过(不触发自动修复)
can, reason = svc.can_generate(p.id)
assert can is True
def test_auto_repair_multiple_assets_distributed(self):
"""多个素材按顺序分配给多个 clips"""
svc = _make_service()
p = svc.create_plan(
"tpl-001",
"测试",
config={"asset_ids": ["asset-A", "asset-B", "asset-C"]},
)
svc.transition_status(p.id, EditPlanStatus.EDITING)
svc.create_clip(p.id, "intro", 0)
svc.create_clip(p.id, "main", 1)
svc.create_clip(p.id, "outro", 2)
can, reason = svc.can_generate(p.id)
assert can is True
clips = svc.list_clips(p.id)
assert clips[0].asset_id == "asset-A"
assert clips[1].asset_id == "asset-B"
assert clips[2].asset_id == "asset-C"
def test_auto_repair_idempotent_on_second_call(self):
"""第二次调用 can_generate 不会重复分配(已有素材则跳过自动修复)"""
svc = _make_service()
p = svc.create_plan(
"tpl-001",
"测试",
config={"asset_ids": ["asset-001"]},
)
svc.transition_status(p.id, EditPlanStatus.EDITING)
svc.create_clip(p.id, "intro", 0)
# 第一次调用触发自动修复
can1, _ = svc.can_generate(p.id)
assert can1 is True
# 第二次调用应该直接通过,不再触发修复
can2, reason2 = svc.can_generate(p.id)
assert can2 is True
assert reason2 == ""