xiaoxia
f4b4f1fc4f
CI/CD Pipeline / Validate Code Quality And Tests (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production Runtime Images (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
feat: ASR自动字幕能力
328 lines
10 KiB
Python
Executable File
328 lines
10 KiB
Python
Executable File
"""ASR 自动字幕集成测试 — 验证渲染管道接入 ASR 的完整链路。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import tempfile
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
from video_processing.unified_render_service import UnifiedRenderService
|
||
|
||
from packages.adapters.asr.mock_asr_service import MockASRService
|
||
from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
|
||
|
||
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class FakeClip:
|
||
"""模拟 EditPlanClip。"""
|
||
|
||
id: str
|
||
plan_id: str = "plan_001"
|
||
asset_id: str = "asset_001"
|
||
clip_type: str = "video"
|
||
start_time: float = 0.0
|
||
duration: float = 10.0
|
||
layer: int = 0
|
||
role: str = "main"
|
||
config: dict = None
|
||
|
||
|
||
@dataclass
|
||
class FakePlan:
|
||
"""模拟 EditPlan。"""
|
||
|
||
id: str = "plan_001"
|
||
config: dict = None
|
||
|
||
|
||
@pytest.fixture
|
||
def work_dir():
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
yield Path(tmpdir)
|
||
|
||
|
||
@pytest.fixture
|
||
def test_video_path():
|
||
"""用 ffmpeg 生成一个5秒的测试视频（带音频）。"""
|
||
import subprocess
|
||
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
video_path = Path(tmpdir) / "test.mp4"
|
||
cmd = [
|
||
"ffmpeg",
|
||
"-y",
|
||
"-f",
|
||
"lavfi",
|
||
"-i",
|
||
"testsrc=duration=5:size=320x240:rate=30",
|
||
"-f",
|
||
"lavfi",
|
||
"-i",
|
||
"sine=frequency=440:duration=5",
|
||
"-c:v",
|
||
"libx264",
|
||
"-preset",
|
||
"ultrafast",
|
||
"-c:a",
|
||
"aac",
|
||
"-shortest",
|
||
str(video_path),
|
||
]
|
||
result = subprocess.run(cmd, capture_output=True, timeout=30)
|
||
if result.returncode != 0:
|
||
pytest.skip(f"ffmpeg 不可用或生成测试视频失败: {result.stderr[:200]}")
|
||
yield video_path
|
||
|
||
|
||
# ── 测试：ASR 服务接入 ────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestASRServiceIntegration:
|
||
def test_asr_service_in_init(self):
|
||
"""验证 asr_service 参数正确传递。"""
|
||
plan = FakePlan(config={})
|
||
service = UnifiedRenderService(
|
||
plan=plan,
|
||
clips=[],
|
||
asset_path_map={},
|
||
work_dir=Path("/tmp"),
|
||
asr_service=MockASRService(),
|
||
)
|
||
assert service.asr_service is not None
|
||
assert isinstance(service.asr_service, MockASRService)
|
||
|
||
def test_no_asr_service_default(self):
|
||
"""验证不传 asr_service 时默认 None。"""
|
||
plan = FakePlan(config={})
|
||
service = UnifiedRenderService(
|
||
plan=plan,
|
||
clips=[],
|
||
asset_path_map={},
|
||
work_dir=Path("/tmp"),
|
||
)
|
||
assert service.asr_service is None
|
||
|
||
def test_maybe_generate_ass_auto_subtitle_with_asr(self, work_dir, test_video_path):
|
||
"""验证 ASR 自动字幕模式：有 asr_service + auto_generated=true 时生成 ASS。"""
|
||
plan = FakePlan(
|
||
config={
|
||
"subtitle": {
|
||
"enabled": True,
|
||
"auto_generated": True,
|
||
"position": "bottom",
|
||
}
|
||
}
|
||
)
|
||
clips = [
|
||
FakeClip(id="clip_1", asset_id="asset_1", duration=5.0),
|
||
]
|
||
asset_map = {"asset_1": test_video_path}
|
||
|
||
service = UnifiedRenderService(
|
||
plan=plan,
|
||
clips=clips,
|
||
asset_path_map=asset_map,
|
||
work_dir=work_dir,
|
||
output_width=320,
|
||
output_height=240,
|
||
asr_service=MockASRService(mock_text="这是ASR自动生成的测试字幕。用来验证渲染管道是否正常接入。"),
|
||
)
|
||
|
||
ass_path = service._maybe_generate_ass(5.0)
|
||
|
||
assert ass_path is not None
|
||
assert ass_path.exists()
|
||
content = ass_path.read_text(encoding="utf-8")
|
||
assert "[Events]" in content
|
||
assert "Dialogue:" in content
|
||
assert "ASR" in content
|
||
|
||
def test_maybe_generate_ass_no_asr_service_skip_auto(self, work_dir):
|
||
"""验证没有 asr_service 时，即使 auto_generated=true 也不生成 ASR 字幕。"""
|
||
plan = FakePlan(
|
||
config={
|
||
"subtitle": {
|
||
"enabled": True,
|
||
"auto_generated": True,
|
||
"text": "",
|
||
}
|
||
}
|
||
)
|
||
service = UnifiedRenderService(
|
||
plan=plan,
|
||
clips=[],
|
||
asset_path_map={},
|
||
work_dir=work_dir,
|
||
asr_service=None, # 没有 ASR 服务
|
||
)
|
||
|
||
ass_path = service._maybe_generate_ass(5.0)
|
||
# 没有 ASR 服务 + 没有静态字幕文本 → 返回 None
|
||
assert ass_path is None
|
||
|
||
def test_maybe_generate_ass_auto_mode_ignores_text(self, work_dir):
|
||
"""验证 ASR 模式下即使有 text 字段也走 ASR（ASR无结果则无字幕）。"""
|
||
plan = FakePlan(
|
||
config={
|
||
"subtitle": {
|
||
"enabled": True,
|
||
"auto_generated": True,
|
||
"text": "静态字幕文本", # ASR模式下忽略此字段
|
||
}
|
||
}
|
||
)
|
||
mock_asr = MockASRService()
|
||
mock_asr.transcribe = MagicMock(side_effect=mock_asr.transcribe)
|
||
|
||
service = UnifiedRenderService(
|
||
plan=plan,
|
||
clips=[],
|
||
asset_path_map={},
|
||
work_dir=work_dir,
|
||
asr_service=mock_asr,
|
||
)
|
||
|
||
ass_path = service._maybe_generate_ass(5.0)
|
||
# ASR模式下无素材 → 无结果 → 返回None（不fallback到静态text）
|
||
assert ass_path is None
|
||
|
||
def test_maybe_generate_ass_title_still_works(self, work_dir):
|
||
"""验证 ASR 模式下不影响 title 的处理（两者独立）。"""
|
||
plan = FakePlan(
|
||
config={
|
||
"title": {
|
||
"enabled": True,
|
||
"text": "视频标题",
|
||
"position": "top",
|
||
},
|
||
"subtitle": {
|
||
"enabled": False, # 字幕关闭
|
||
"auto_generated": True,
|
||
},
|
||
}
|
||
)
|
||
mock_asr = MockASRService()
|
||
|
||
service = UnifiedRenderService(
|
||
plan=plan,
|
||
clips=[],
|
||
asset_path_map={},
|
||
work_dir=work_dir,
|
||
asr_service=mock_asr,
|
||
)
|
||
|
||
ass_path = service._maybe_generate_ass(5.0)
|
||
assert ass_path is not None
|
||
content = ass_path.read_text(encoding="utf-8")
|
||
assert "视频标题" in content
|
||
|
||
def test_asr_failure_does_not_block(self, work_dir, test_video_path):
|
||
"""验证 ASR 失败时不阻断主流程，降级为无字幕。"""
|
||
plan = FakePlan(
|
||
config={
|
||
"subtitle": {
|
||
"enabled": True,
|
||
"auto_generated": True,
|
||
}
|
||
}
|
||
)
|
||
clips = [FakeClip(id="clip_1", asset_id="asset_1", duration=5.0)]
|
||
asset_map = {"asset_1": test_video_path}
|
||
|
||
# ASR 服务总是抛异常
|
||
bad_asr = MockASRService()
|
||
bad_asr.transcribe = MagicMock(side_effect=RuntimeError("ASR service down"))
|
||
|
||
service = UnifiedRenderService(
|
||
plan=plan,
|
||
clips=clips,
|
||
asset_path_map=asset_map,
|
||
work_dir=work_dir,
|
||
output_width=320,
|
||
output_height=240,
|
||
asr_service=bad_asr,
|
||
)
|
||
|
||
# 应该不抛异常，返回 None（降级）
|
||
ass_path = service._maybe_generate_ass(5.0)
|
||
assert ass_path is None # ASR 失败 → 无字幕
|
||
|
||
def test_auto_subtitle_disabled(self, work_dir):
|
||
"""验证 subtitle.enabled=false 时即使 auto_generated=true 也不生成。"""
|
||
plan = FakePlan(
|
||
config={
|
||
"subtitle": {
|
||
"enabled": False,
|
||
"auto_generated": True,
|
||
}
|
||
}
|
||
)
|
||
mock_asr = MockASRService()
|
||
mock_asr.transcribe = MagicMock()
|
||
|
||
service = UnifiedRenderService(
|
||
plan=plan,
|
||
clips=[],
|
||
asset_path_map={},
|
||
work_dir=work_dir,
|
||
asr_service=mock_asr,
|
||
)
|
||
|
||
ass_path = service._maybe_generate_ass(5.0)
|
||
assert ass_path is None
|
||
mock_asr.transcribe.assert_not_called()
|
||
|
||
|
||
# ── 测试：SubtitleConfig 扩展 ────────────────────────────────────────────────
|
||
|
||
|
||
class TestSubtitleConfigExtension:
|
||
def test_config_has_auto_generated_field(self):
|
||
"""验证 SubtitleConfig 有 auto_generated 字段。"""
|
||
from packages.domain.config_schemas import SubtitleConfig
|
||
|
||
config = SubtitleConfig()
|
||
assert hasattr(config, "auto_generated")
|
||
assert config.auto_generated is False # 默认关闭
|
||
|
||
def test_config_default_values(self):
|
||
"""验证新增字段的默认值。"""
|
||
from packages.domain.config_schemas import SubtitleConfig
|
||
|
||
config = SubtitleConfig()
|
||
assert config.auto_generated is False
|
||
assert config.language == ""
|
||
assert config.max_chars_per_line == 20
|
||
assert config.min_chars_per_segment == 8
|
||
|
||
def test_config_custom_values(self):
|
||
"""验证可以自定义 ASR 相关字段。"""
|
||
from packages.domain.config_schemas import SubtitleConfig
|
||
|
||
config = SubtitleConfig(
|
||
auto_generated=True,
|
||
language="zh",
|
||
max_chars_per_line=15,
|
||
min_chars_per_segment=5,
|
||
)
|
||
assert config.auto_generated is True
|
||
assert config.language == "zh"
|
||
assert config.max_chars_per_line == 15
|
||
assert config.min_chars_per_segment == 5
|
||
|
||
def test_config_validation_max_chars(self):
|
||
"""验证 max_chars_per_line 的范围校验。"""
|
||
from pydantic import ValidationError
|
||
|
||
from packages.domain.config_schemas import SubtitleConfig
|
||
|
||
with pytest.raises(ValidationError):
|
||
SubtitleConfig(max_chars_per_line=5) # 小于8
|
||
|
||
with pytest.raises(ValidationError):
|
||
SubtitleConfig(max_chars_per_line=50) # 大于40
|