zycandClaude Fable 5
216a711291
后端: - 新增 celery_health 生成前置闸——无 worker 在线时图片/视频生成入口 一律 503,防"提交到 ARK 后无人轮询、结果悬空+额度冻结"的数据丢失 - 拼接导出:帧率跟随源众数、字幕逐句重映射、原声/BGM 混音修复 - 资金核算审计脚本 + genesis 账目回填 + 滞留预留清理命令 - 接入 yunqi provider 与豆包 TTS 模型(catalog/migrations/bootstrap 命令) 前端:全站页面更新(pipeline/library/products/projects/team/account 等), 新增共享 pager 分页组件 QA:刷新 function-audit 全量输出,新增 full-qa 报告 文档:BP 产品介绍资料、design/CLAUDE.md Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
317 lines
15 KiB
Python
317 lines
15 KiB
Python
from django.test import TestCase
|
|
from unittest.mock import patch
|
|
|
|
from rest_framework.test import APIClient
|
|
|
|
from apps.accounts.models import Team, TeamMember, User
|
|
from apps.ai.models import ModelConfig, ModelProvider
|
|
from apps.assets.models import Asset
|
|
from apps.billing.models import CreditAccount, CreditLedger
|
|
from apps.products.models import Product
|
|
from apps.projects.models import (
|
|
Project,
|
|
ProjectStage,
|
|
ScriptVersion,
|
|
SubtitleTrack,
|
|
Timeline,
|
|
TimelineClip,
|
|
VideoSegment,
|
|
VideoSegmentVersion,
|
|
)
|
|
|
|
|
|
class ProjectApiTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username="owner", password="pass")
|
|
self.team = Team.objects.create(name="E2E Team", owner=self.user)
|
|
TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
|
|
CreditAccount.objects.create(team=self.team, balance="100.0000")
|
|
self.product = Product.objects.create(team=self.team, created_by=self.user, title="Test Product")
|
|
self.provider = ModelProvider.objects.create(
|
|
name="volcengine",
|
|
display_name="Volcano",
|
|
base_url="https://ark.cn-beijing.volces.com/api/v3",
|
|
)
|
|
self.model = ModelConfig.objects.create(
|
|
provider=self.provider,
|
|
name="doubao-seed-2-0-pro-260215",
|
|
display_name="Doubao",
|
|
capability=ModelConfig.Capability.TEXT,
|
|
endpoint="chat/completions",
|
|
unit_price="2.0000",
|
|
)
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
|
|
def test_create_project_initializes_pipeline(self):
|
|
response = self.client.post(
|
|
"/api/projects/",
|
|
{"name": "Launch Video", "product": str(self.product.id)},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
project = Project.objects.get(id=response.data["id"])
|
|
self.assertEqual(project.team, self.team)
|
|
self.assertEqual(project.created_by, self.user)
|
|
self.assertEqual(project.stages.count(), 5)
|
|
self.assertEqual(project.video_segments.count(), 4)
|
|
self.assertEqual(
|
|
list(project.stages.values_list("stage", flat=True)),
|
|
[
|
|
ProjectStage.Stage.SCRIPT,
|
|
ProjectStage.Stage.BASE_ASSETS,
|
|
ProjectStage.Stage.STORYBOARD,
|
|
ProjectStage.Stage.VIDEO,
|
|
ProjectStage.Stage.EXPORT,
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
list(project.video_segments.values_list("target_duration_seconds", flat=True)),
|
|
[15, 15, 15, 15],
|
|
)
|
|
self.assertEqual(
|
|
list(project.video_segments.values_list("sort_order", flat=True)),
|
|
[0, 1, 2, 3],
|
|
)
|
|
self.assertTrue(VideoSegment.objects.filter(project=project).exists())
|
|
|
|
@patch("apps.ai.services.VolcanoArkProvider")
|
|
def test_generate_script_creates_script_segments_and_charges_credit(self, provider_cls):
|
|
provider = provider_cls.return_value
|
|
provider.chat_completion.return_value = {
|
|
"choices": [
|
|
{
|
|
"message": {
|
|
"content": "1. 开场吸引\n2. 展示卖点\n3. 使用场景\n4. 促单转化",
|
|
}
|
|
}
|
|
]
|
|
}
|
|
provider.extract_text.return_value = "1. 开场吸引\n2. 展示卖点\n3. 使用场景\n4. 促单转化"
|
|
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="Launch Video")
|
|
for stage in [
|
|
ProjectStage.Stage.SCRIPT,
|
|
ProjectStage.Stage.BASE_ASSETS,
|
|
ProjectStage.Stage.STORYBOARD,
|
|
ProjectStage.Stage.VIDEO,
|
|
ProjectStage.Stage.EXPORT,
|
|
]:
|
|
ProjectStage.objects.create(project=project, stage=stage)
|
|
|
|
response = self.client.post(
|
|
f"/api/projects/{project.id}/generate-script/",
|
|
{"prompt": "突出高转化"},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
script = ScriptVersion.objects.get(project=project)
|
|
self.assertEqual(script.segments.count(), 4)
|
|
self.assertEqual(CreditLedger.objects.filter(team=self.team, ledger_type=CreditLedger.Type.CHARGE).count(), 1)
|
|
|
|
def test_adopt_video_version_remaps_timeline_draft(self):
|
|
"""切换采用版本后,时间线草稿里引用本场旧版本资产的片段必须跟随换成新资产(剪辑台/导出都读草稿)。"""
|
|
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P")
|
|
make_asset = lambda name: Asset.objects.create(
|
|
team=self.team, created_by=self.user, name=name,
|
|
asset_type=Asset.Type.VIDEO, source=Asset.Source.AI_GENERATED, category=Asset.Category.VIDEO_CLIP,
|
|
)
|
|
old_asset, new_asset = make_asset("v1"), make_asset("v2")
|
|
segment = VideoSegment.objects.create(project=project, sort_order=0)
|
|
old_ver = VideoSegmentVersion.objects.create(video_segment=segment, asset=old_asset, is_adopted=True)
|
|
new_ver = VideoSegmentVersion.objects.create(video_segment=segment, asset=new_asset)
|
|
segment.adopted_version = old_ver
|
|
segment.save(update_fields=["adopted_version"])
|
|
timeline = Timeline.objects.create(project=project)
|
|
clip = TimelineClip.objects.create(
|
|
timeline=timeline, asset=old_asset, sort_order=0,
|
|
duration_ms=15000, trim_start_ms=1000, trim_end_ms=9000,
|
|
)
|
|
|
|
response = self.client.post(
|
|
f"/api/projects/{project.id}/adopt-video-version/",
|
|
{"video_segment_id": str(segment.id), "version_id": str(new_ver.id)},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
clip.refresh_from_db()
|
|
self.assertEqual(clip.asset_id, new_asset.id)
|
|
# 旧素材上的裁剪点对新素材无意义,必须复位
|
|
self.assertEqual(clip.trim_start_ms, 0)
|
|
self.assertIsNone(clip.trim_end_ms)
|
|
|
|
def _make_audio_model(self):
|
|
return ModelConfig.objects.create(
|
|
provider=self.provider, name="doubao-voice-bigtts", display_name="豆包语音合成",
|
|
capability="audio", endpoint="api/v1/tts", unit_price="2.0000",
|
|
)
|
|
|
|
@patch("apps.ai.services.TosStorage")
|
|
@patch("apps.ai.services.VolcanoTtsProvider")
|
|
def test_generate_voiceover_creates_assets_and_charges_once(self, provider_cls, storage_cls):
|
|
"""每镜旁白 TTS:产出音频资产、timeline.metadata 落映射、一次调用只计一次费。"""
|
|
self._make_audio_model()
|
|
provider = provider_cls.return_value
|
|
provider.configured = True
|
|
provider.synthesize.return_value = (b"fake-mp3-bytes", 4200)
|
|
stored = storage_cls.return_value.upload_fileobj.return_value
|
|
stored.object_key, stored.bucket, stored.content_type, stored.size_bytes = "k.mp3", "b", "audio/mpeg", 14
|
|
|
|
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="VO")
|
|
response = self.client.post(
|
|
f"/api/projects/{project.id}/generate-voiceover/",
|
|
{"items": [{"index": 0, "text": "第一镜旁白"}, {"index": 1, "text": "第二镜旁白"}]},
|
|
format="json",
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 201)
|
|
voiceover = response.json()["voiceover"]
|
|
self.assertTrue(voiceover["enabled"])
|
|
self.assertEqual(len(voiceover["items"]), 2)
|
|
self.assertEqual(provider.synthesize.call_count, 2)
|
|
from apps.assets.models import Asset as AssetModel
|
|
self.assertEqual(AssetModel.objects.filter(team=self.team, asset_type="audio").count(), 2)
|
|
self.assertEqual(
|
|
CreditLedger.objects.filter(team=self.team, ledger_type=CreditLedger.Type.CHARGE).count(), 1
|
|
)
|
|
project.refresh_from_db()
|
|
self.assertEqual(len(project.timeline.metadata["voiceover"]["items"]), 2)
|
|
|
|
@patch("apps.ai.services.VolcanoTtsProvider")
|
|
def test_generate_voiceover_unconfigured_returns_400_and_no_charge(self, provider_cls):
|
|
"""语音凭证未配置:400 + 人话提示,不建任务不扣费。"""
|
|
self._make_audio_model()
|
|
provider_cls.return_value.configured = False
|
|
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="VO2")
|
|
response = self.client.post(
|
|
f"/api/projects/{project.id}/generate-voiceover/",
|
|
{"items": [{"index": 0, "text": "旁白"}]},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("语音合成未配置", response.json()["detail"])
|
|
self.assertEqual(CreditLedger.objects.filter(team=self.team, ledger_type=CreditLedger.Type.CHARGE).count(), 0)
|
|
|
|
def test_export_command_mixes_voiceover_above_bgm(self):
|
|
"""导出命令:人声按片段起点 adelay 延迟、先与原声 amix、再与 BGM amix。"""
|
|
from apps.projects.services.export import _build_export_command
|
|
|
|
specs = [{"ts": 0.0, "te": 15.0, "dur": 15.0}, {"ts": 0.0, "te": 15.0, "dur": 15.0}]
|
|
cmd = _build_export_command(
|
|
n=2, specs=specs, starts=[0.0, 15.0], total=30.0, transition="none",
|
|
sub_overlays=[], bgm_name="bgm.mp3", bgm_volume=0.6, has_audio=[False, False],
|
|
voice_overlays=[("vo0.mp3", 0.0, 15.0), ("vo1.mp3", 15.0, 15.0)],
|
|
)
|
|
self.assertIn("vo0.mp3", cmd)
|
|
self.assertIn("vo1.mp3", cmd)
|
|
graph = cmd[cmd.index("-filter_complex") + 1]
|
|
self.assertIn("adelay=15000|15000", graph)
|
|
self.assertIn("[avoice][vo0][vo1]amix=inputs=3", graph)
|
|
self.assertIn("[anarr][abgm]amix=inputs=2", graph)
|
|
|
|
def test_save_timeline_updates_voiceover_offsets(self):
|
|
"""拖动字幕块后保存:按 asset 回写句内起点 offset_ms,未拖动的句不受影响。"""
|
|
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="VOFF")
|
|
Timeline.objects.create(project=project, metadata={"voiceover": {"enabled": True, "items": [
|
|
{"index": 0, "cue": 0, "asset": "a1", "text": "x", "duration_ms": 2000, "offset_ms": 0},
|
|
{"index": 0, "cue": 1, "asset": "a2", "text": "y", "duration_ms": 2000, "offset_ms": 2000},
|
|
]}})
|
|
response = self.client.post(
|
|
f"/api/projects/{project.id}/save-timeline/",
|
|
{"voiceover": {"items": [{"asset": "a2", "offset_ms": 9000}]}},
|
|
format="json",
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
project.timeline.refresh_from_db()
|
|
items = project.timeline.metadata["voiceover"]["items"]
|
|
self.assertEqual(items[0]["offset_ms"], 0)
|
|
self.assertEqual(items[1]["offset_ms"], 9000)
|
|
|
|
def test_subtitle_cues_use_end_ms_and_follow_voiceover(self):
|
|
"""字幕烧入时序:cue 自带 end_ms 时按它收尾(不再拖到片段尾)——这是配音对齐的关键;
|
|
无保存字幕的回退路径,有配音时逐句窗口也要压进语音真实时长。"""
|
|
from apps.projects.services.export import _subtitle_cues
|
|
|
|
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="SUB")
|
|
timeline = Timeline.objects.create(
|
|
project=project,
|
|
metadata={"voiceover": {"enabled": True, "items": [{"index": 0, "asset": "x", "text": "t", "duration_ms": 4000}]}},
|
|
)
|
|
SubtitleTrack.objects.create(timeline=timeline, enabled=True, content=[
|
|
{"start_ms": 0, "end_ms": 2000, "text": "第一句"},
|
|
{"start_ms": 2000, "end_ms": 4000, "text": "第二句"},
|
|
])
|
|
specs = [{"ts": 0.0, "te": 15.0, "dur": 15.0}]
|
|
cues = _subtitle_cues(timeline, project, specs, [0.0], 15.0)
|
|
self.assertEqual(len(cues), 2)
|
|
self.assertEqual((cues[0][0], cues[0][1]), (0.0, 2.0))
|
|
self.assertEqual(cues[1][1], 4.0) # 最后一句在语音念完处收尾,而非拖到 15s
|
|
|
|
def test_save_timeline_voiceover_toggle_and_clear(self):
|
|
"""保存草稿可关/开配音、可移除映射(不动已生成的音频资产)。"""
|
|
project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="VO3")
|
|
Timeline.objects.create(
|
|
project=project,
|
|
metadata={"voiceover": {"enabled": True, "voice_type": "v", "items": [{"index": 0, "asset": "a", "text": "t"}]}},
|
|
)
|
|
response = self.client.post(
|
|
f"/api/projects/{project.id}/save-timeline/", {"voiceover": {"enabled": False}}, format="json"
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
project.timeline.refresh_from_db()
|
|
self.assertFalse(project.timeline.metadata["voiceover"]["enabled"])
|
|
|
|
response = self.client.post(
|
|
f"/api/projects/{project.id}/save-timeline/", {"voiceover": {"clear": True}}, format="json"
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
project.timeline.refresh_from_db()
|
|
self.assertNotIn("voiceover", project.timeline.metadata)
|
|
|
|
|
|
class WorkerGateTests(TestCase):
|
|
"""无 Celery worker 时,图片/视频生成入口必须 503 拒绝(防结果无人取回=数据丢失)。"""
|
|
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username="gate", password="pass")
|
|
self.team = Team.objects.create(name="Gate Team", owner=self.user)
|
|
TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
|
|
CreditAccount.objects.create(team=self.team, balance="100.0000")
|
|
self.product = Product.objects.create(team=self.team, created_by=self.user, title="Gate Product")
|
|
self.project = Project.objects.create(team=self.team, created_by=self.user, name="Gate", product=self.product)
|
|
self.client = APIClient()
|
|
self.client.force_authenticate(self.user)
|
|
|
|
def _blocked(self):
|
|
return patch("apps.common.celery_health.celery_worker_available", return_value=False)
|
|
|
|
def test_generation_endpoints_blocked_without_worker(self):
|
|
cases = [
|
|
(f"/api/projects/{self.project.id}/generate-base-asset/", {"kind": "person"}),
|
|
(f"/api/projects/{self.project.id}/generate-storyboard/", {}),
|
|
(f"/api/projects/{self.project.id}/submit-video-segment/", {"video_segment_id": "x"}),
|
|
("/api/ai/generate-image/", {"prompt": "测试"}),
|
|
]
|
|
with self._blocked():
|
|
for url, payload in cases:
|
|
response = self.client.post(url, payload, format="json")
|
|
self.assertEqual(response.status_code, 503, msg=f"{url} 应被 worker 闸拦截")
|
|
self.assertIn("worker", response.json()["detail"].lower() if isinstance(response.json().get("detail"), str) else "")
|
|
|
|
def test_eager_mode_bypasses_gate(self):
|
|
from apps.common.celery_health import celery_worker_available
|
|
|
|
# 测试配置 CELERY_TASK_ALWAYS_EAGER=True:任务就地执行,无需 worker,闸放行
|
|
self.assertTrue(celery_worker_available())
|
|
|
|
def test_ping_failure_means_unavailable(self):
|
|
from apps.common import celery_health
|
|
|
|
celery_health._cache["expires"] = 0.0
|
|
with self.settings(CELERY_TASK_ALWAYS_EAGER=False):
|
|
with patch.object(celery_health, "_ping_workers", side_effect=ConnectionError("broker down")):
|
|
self.assertFalse(celery_health.celery_worker_available())
|
|
celery_health._cache["expires"] = 0.0
|