CI Bot
7ff7ae6c7d
CI/CD Pipeline / Frontend Lint (push) Successful in 38s
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 1m35s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Failing after 2s
CI/CD Pipeline / Build Staging Worker Image (push) Failing after 2s
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Unit Tests (push) Successful in 1m41s
CI/CD Pipeline / Integration Tests (push) Successful in 1m11s
CI/CD Pipeline / Build Staging Web Image (push) Failing after 1m11s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (push) Has been skipped
521 lines
19 KiB
Python
Executable File
521 lines
19 KiB
Python
Executable File
"""P1 长文本分段合成单元测试。
|
|
|
|
覆盖：
|
|
- text_splitter.split_text 分段逻辑
|
|
- audio_merger.AudioMerger 合并逻辑
|
|
- workflow 分段合成路径（同步 / 异步 / 失败）
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
|
|
from packages.application.tts_job.audio_merger import AudioMergeError, AudioMerger
|
|
from packages.application.tts_job.text_splitter import split_text
|
|
from packages.application.tts_job.workflow import TTSWorkflowService
|
|
from packages.domain.tts_job import TTSJob, TTSJobStatus
|
|
|
|
# ── text_splitter ─────────────────────────────────────────────
|
|
|
|
|
|
class TestSplitText:
|
|
"""测试文本分段工具。"""
|
|
|
|
def test_short_text_no_split(self) -> None:
|
|
"""短文本不拆分。"""
|
|
assert split_text("你好世界", max_chars=500) == ["你好世界"]
|
|
|
|
def test_empty_text(self) -> None:
|
|
"""空文本返回空列表。"""
|
|
assert split_text("") == []
|
|
assert split_text(" ") == []
|
|
|
|
def test_exact_threshold(self) -> None:
|
|
"""恰好等于阈值不拆分。"""
|
|
text = "a" * 500
|
|
assert split_text(text, max_chars=500) == [text]
|
|
|
|
def test_split_at_sentence_boundary(self) -> None:
|
|
"""在句子边界处分段。"""
|
|
text = "第一句话。" * 60 # 300 chars
|
|
text += "第二句话。" * 60 # 300 chars → total 600
|
|
segments = split_text(text, max_chars=500)
|
|
assert len(segments) >= 2
|
|
for seg in segments:
|
|
assert len(seg) <= 500
|
|
|
|
def test_split_at_newline(self) -> None:
|
|
"""在换行符处分段。"""
|
|
text = "段落一\n" * 100 # 300 chars
|
|
text += "段落二\n" * 100 # 300 chars
|
|
segments = split_text(text, max_chars=500)
|
|
assert len(segments) >= 2
|
|
|
|
def test_long_sentence_hard_split(self) -> None:
|
|
"""超长句子硬切。"""
|
|
text = "a" * 1200
|
|
segments = split_text(text, max_chars=500)
|
|
assert len(segments) >= 3
|
|
for seg in segments:
|
|
assert len(seg) <= 500
|
|
|
|
def test_merge_short_segments(self) -> None:
|
|
"""短段合并减少 API 调用。"""
|
|
# 多个短句子应该被合并
|
|
text = "你好。" * 120 # 360 chars, each sentence 3 chars
|
|
segments = split_text(text, max_chars=500)
|
|
# 短段应该被合并，段数应该比较少
|
|
assert len(segments) < 120
|
|
|
|
def test_preserves_order(self) -> None:
|
|
"""分段保持原始顺序。"""
|
|
text = "第一段。第二段。第三段。" + "x" * 490
|
|
segments = split_text(text, max_chars=500)
|
|
# 第一个段应该以 "第一段" 开头
|
|
assert segments[0].startswith("第一段")
|
|
|
|
|
|
# ── audio_merger ─────────────────────────────────────────────
|
|
|
|
|
|
class TestAudioMerger:
|
|
"""测试 FFmpeg 音频合并器。"""
|
|
|
|
def test_empty_list_raises(self) -> None:
|
|
"""空列表抛出 AudioMergeError。"""
|
|
merger = AudioMerger()
|
|
with pytest.raises(AudioMergeError, match="没有可合并"):
|
|
merger.merge([])
|
|
|
|
def test_single_file_returns_bytes(self) -> None:
|
|
"""单文件直接返回内容。"""
|
|
with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
|
|
f.write(b"fake audio content")
|
|
f.flush()
|
|
path = f.name
|
|
|
|
try:
|
|
merger = AudioMerger()
|
|
data = merger.merge([path])
|
|
assert data == b"fake audio content"
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
@patch("packages.application.tts_job.audio_merger.run_ffmpeg")
|
|
def test_ffmpeg_called_correctly(self, mock_run_ffmpeg: MagicMock) -> None:
|
|
"""多文件调用 FFmpeg concat（通过 run_ffmpeg 统一入口）。"""
|
|
# run_ffmpeg 成功返回 (stdout, stderr)
|
|
mock_run_ffmpeg.return_value = ("", "")
|
|
|
|
# 创建临时文件
|
|
paths = []
|
|
for _ in range(3):
|
|
with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
|
|
f.write(b"audio")
|
|
paths.append(f.name)
|
|
|
|
try:
|
|
merger = AudioMerger()
|
|
# Mock open for reading the merged output
|
|
with patch("builtins.open", create=True) as mock_open:
|
|
mock_open.return_value.__enter__ = lambda s: s
|
|
mock_open.return_value.read = lambda: b"merged audio"
|
|
try:
|
|
merger.merge(paths, output_format="mp3")
|
|
except (FileNotFoundError, OSError):
|
|
pass # Expected since we're mocking
|
|
|
|
# 验证 run_ffmpeg 被调用
|
|
mock_run_ffmpeg.assert_called_once()
|
|
cmd = mock_run_ffmpeg.call_args[0][0]
|
|
from shared.ffmpeg_utils import FFMPEG_BIN
|
|
|
|
assert cmd[0] == FFMPEG_BIN
|
|
assert "-f" in cmd
|
|
assert "concat" in cmd
|
|
finally:
|
|
for p in paths:
|
|
os.unlink(p)
|
|
|
|
@patch("packages.application.tts_job.audio_merger.run_ffmpeg")
|
|
def test_ffmpeg_failure_raises(self, mock_run_ffmpeg: MagicMock) -> None:
|
|
"""FFmpeg 失败抛出 AudioMergeError（通过 run_ffmpeg 抛出 CalledProcessError）。"""
|
|
from subprocess import CalledProcessError
|
|
|
|
mock_run_ffmpeg.side_effect = CalledProcessError(returncode=1, cmd=["ffmpeg"], stderr="error details")
|
|
|
|
paths = []
|
|
for _ in range(2):
|
|
with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
|
|
f.write(b"audio")
|
|
paths.append(f.name)
|
|
|
|
try:
|
|
merger = AudioMerger()
|
|
with pytest.raises(AudioMergeError, match="FFmpeg 合并失败"):
|
|
merger.merge(paths)
|
|
mock_run_ffmpeg.assert_called_once()
|
|
finally:
|
|
for p in paths:
|
|
os.unlink(p)
|
|
|
|
|
|
# ── workflow segment methods ─────────────────────────────────
|
|
|
|
|
|
def _make_job(**kwargs) -> TTSJob:
|
|
defaults = {
|
|
"id": "test_job_seg",
|
|
"user_id": "user_001",
|
|
"input_text": "x" * 600, # > 500 threshold
|
|
"voice_id": "voice_001",
|
|
"voice_model": "",
|
|
"project_id": "",
|
|
"voice_clone_profile_id": "",
|
|
"status": TTSJobStatus.PENDING,
|
|
"output_audio_url": "",
|
|
"output_audio_key": "",
|
|
"duration": 0.0,
|
|
"file_size": 0,
|
|
"sample_rate": 22050,
|
|
"format": "mp3",
|
|
"error_message": "",
|
|
"retry_count": 0,
|
|
"max_retries": 3,
|
|
"metadata": {},
|
|
"started_at": None,
|
|
"completed_at": None,
|
|
"created_at": datetime.now(timezone.utc),
|
|
"updated_at": datetime.now(timezone.utc),
|
|
}
|
|
defaults.update(kwargs)
|
|
return TTSJob(**defaults)
|
|
|
|
|
|
def _make_workflow(
|
|
cosyvoice_service: MagicMock | None = None,
|
|
repo: MagicMock | None = None,
|
|
storage: MagicMock | None = None,
|
|
) -> TTSWorkflowService:
|
|
if cosyvoice_service is None:
|
|
cosyvoice_service = MagicMock(spec=CosyVoiceService)
|
|
if repo is None:
|
|
repo = MagicMock()
|
|
repo.get.return_value = _make_job()
|
|
repo.update.side_effect = lambda j: j
|
|
if storage is None:
|
|
storage = MagicMock()
|
|
storage.upload_file.return_value = "https://oss.example.com/merged.mp3"
|
|
return TTSWorkflowService(
|
|
repository=repo,
|
|
cosyvoice_service=cosyvoice_service,
|
|
storage_service=storage,
|
|
)
|
|
|
|
|
|
class TestStartSegmentSynthesis:
|
|
"""测试 _start_segment_synthesis 分段合成入口。"""
|
|
|
|
def test_short_text_no_segment(self) -> None:
|
|
"""短文本不触发分段。"""
|
|
service = MagicMock(spec=CosyVoiceService)
|
|
service.submit_synthesize_task.return_value = {
|
|
"task_id": "task_1",
|
|
"audio_url": "",
|
|
}
|
|
repo = MagicMock()
|
|
repo.get.return_value = _make_job(input_text="短文本")
|
|
repo.update.side_effect = lambda j: j
|
|
|
|
workflow = _make_workflow(cosyvoice_service=service, repo=repo)
|
|
job = workflow.start_synthesis("test_job_seg")
|
|
|
|
# 短文本走普通路径，不调用分段
|
|
assert job.status == TTSJobStatus.PROCESSING
|
|
|
|
@patch("packages.application.tts_job.workflow.safe_download_file")
|
|
def test_long_text_sync_segments(self, mock_download: MagicMock) -> None:
|
|
"""长文本同步分段：所有段立即返回 audio_url，直接合并。"""
|
|
# Mock 分段音频下载
|
|
mock_download.return_value = 1024 # 模拟文件大小
|
|
|
|
service = MagicMock(spec=CosyVoiceService)
|
|
# 每个分段都同步返回 audio_url
|
|
service.submit_synthesize_task.side_effect = [
|
|
{"task_id": "", "audio_url": "https://temp.com/seg1.mp3", "duration": 2.0, "file_size": 1000},
|
|
{"task_id": "", "audio_url": "https://temp.com/seg2.mp3", "duration": 3.0, "file_size": 1500},
|
|
]
|
|
|
|
storage = MagicMock()
|
|
storage.upload_file.return_value = "https://oss.example.com/merged.mp3"
|
|
|
|
repo = MagicMock()
|
|
repo.get.return_value = _make_job()
|
|
repo.update.side_effect = lambda j: j
|
|
|
|
workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)
|
|
|
|
# Mock AudioMerger 避免真实 FFmpeg 调用
|
|
with patch("packages.application.tts_job.workflow.AudioMerger") as MockMerger:
|
|
mock_merger = MagicMock()
|
|
mock_merger.merge.return_value = b"merged audio data"
|
|
MockMerger.return_value = mock_merger
|
|
|
|
job = workflow.start_synthesis("test_job_seg")
|
|
|
|
assert job.status == TTSJobStatus.COMPLETED
|
|
assert job.output_audio_url == "https://oss.example.com/merged.mp3"
|
|
assert job.duration == 5.0 # 2.0 + 3.0
|
|
|
|
def test_long_text_async_segments(self) -> None:
|
|
"""长文本异步分段：返回 task_id，存入 metadata。"""
|
|
service = MagicMock(spec=CosyVoiceService)
|
|
# 每个分段返回 task_id（异步）
|
|
service.submit_synthesize_task.side_effect = [
|
|
{"task_id": "seg_task_1", "audio_url": "", "duration": 0.0, "file_size": 0},
|
|
{"task_id": "seg_task_2", "audio_url": "", "duration": 0.0, "file_size": 0},
|
|
]
|
|
|
|
repo = MagicMock()
|
|
repo.get.return_value = _make_job()
|
|
repo.update.side_effect = lambda j: j
|
|
|
|
workflow = _make_workflow(cosyvoice_service=service, repo=repo)
|
|
job = workflow.start_synthesis("test_job_seg")
|
|
|
|
assert job.status == TTSJobStatus.PROCESSING
|
|
assert "segment_task_ids" in job.metadata
|
|
assert job.metadata["segment_task_ids"] == ["seg_task_1", "seg_task_2"]
|
|
|
|
def test_segment_submit_failure_marks_failed(self) -> None:
|
|
"""分段提交失败时标记 job 为 failed。"""
|
|
service = MagicMock(spec=CosyVoiceService)
|
|
service.submit_synthesize_task.side_effect = CosyVoiceError("API error")
|
|
|
|
repo = MagicMock()
|
|
repo.get.return_value = _make_job()
|
|
repo.update.side_effect = lambda j: j
|
|
|
|
workflow = _make_workflow(cosyvoice_service=service, repo=repo)
|
|
job = workflow.start_synthesis("test_job_seg")
|
|
|
|
assert job.status == TTSJobStatus.FAILED
|
|
|
|
|
|
class TestUploadMergedToOSS:
|
|
"""测试 _upload_merged_to_oss 辅助方法。"""
|
|
|
|
def test_success(self) -> None:
|
|
"""成功上传返回 URL 和 key。"""
|
|
storage = MagicMock()
|
|
storage.upload_file.return_value = "https://oss.example.com/merged.mp3"
|
|
|
|
workflow = _make_workflow(storage=storage)
|
|
url, key = workflow._upload_merged_to_oss(b"audio data", "user_001", "job_001", "mp3")
|
|
|
|
assert url == "https://oss.example.com/merged.mp3"
|
|
assert key == "tts-outputs/user_001/job_001.mp3"
|
|
storage.upload_file.assert_called_once()
|
|
call_args = storage.upload_file.call_args
|
|
assert call_args[1]["content_type"] == "audio/mpeg"
|
|
|
|
def test_failure_returns_empty(self) -> None:
|
|
"""上传失败返回空字符串。"""
|
|
storage = MagicMock()
|
|
storage.upload_file.side_effect = Exception("OSS error")
|
|
|
|
workflow = _make_workflow(storage=storage)
|
|
url, key = workflow._upload_merged_to_oss(b"audio data", "user_001", "job_001", "mp3")
|
|
|
|
assert url == ""
|
|
assert key == ""
|
|
|
|
|
|
class TestHandleSegmentFailure:
|
|
"""测试 _handle_segment_failure。"""
|
|
|
|
def test_marks_job_failed(self) -> None:
|
|
"""标记 job 为 failed 并更新。"""
|
|
repo = MagicMock()
|
|
job = _make_job(status=TTSJobStatus.PROCESSING)
|
|
repo.get.return_value = job
|
|
repo.update.side_effect = lambda j: j
|
|
|
|
workflow = _make_workflow(repo=repo)
|
|
workflow._handle_segment_failure(job, "分段 1 合成失败")
|
|
|
|
assert job.status == TTSJobStatus.FAILED
|
|
assert "分段 1 合成失败" in job.error_message
|
|
repo.update.assert_called_once()
|
|
|
|
|
|
class TestPollSegmentTasks:
|
|
"""测试 _poll_segment_tasks 分段缺失重新合成（适配同步接口）。"""
|
|
|
|
@patch("packages.application.tts_job.workflow.safe_download_file")
|
|
def test_all_segments_done(self, mock_download: MagicMock) -> None:
|
|
"""所有分段缺少 audio_url 时重新同步合成，合并后标记完成。"""
|
|
# Mock 下载分段音频
|
|
mock_download.return_value = 1024 # 模拟文件大小
|
|
|
|
service = MagicMock(spec=CosyVoiceService)
|
|
service.submit_synthesize_task.side_effect = [
|
|
{"audio_url": "https://temp.com/seg1.mp3", "duration": 2.0, "file_size": 100},
|
|
{"audio_url": "https://temp.com/seg2.mp3", "duration": 3.0, "file_size": 200},
|
|
]
|
|
|
|
storage = MagicMock()
|
|
storage.upload_file.return_value = "https://oss.example.com/merged.mp3"
|
|
|
|
repo = MagicMock()
|
|
job = _make_job(
|
|
status=TTSJobStatus.PROCESSING,
|
|
# 长文本触发分段，用于重新合成时切分
|
|
input_text="这是一段很长的测试文本。" * 30,
|
|
metadata={
|
|
"segment_task_ids": ["task_1", "task_2"],
|
|
"segment_audio_urls": ["", ""],
|
|
"segment_format": "mp3",
|
|
},
|
|
)
|
|
repo.get.return_value = job
|
|
repo.update.side_effect = lambda j: j
|
|
|
|
workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)
|
|
|
|
with patch("packages.application.tts_job.workflow.AudioMerger") as MockMerger:
|
|
mock_merger = MagicMock()
|
|
mock_merger.merge.return_value = b"merged data"
|
|
MockMerger.return_value = mock_merger
|
|
|
|
result = workflow._poll_segment_tasks(job)
|
|
|
|
assert result.status == TTSJobStatus.COMPLETED
|
|
# 两个缺失分段都重新合成了
|
|
assert service.submit_synthesize_task.call_count == 2
|
|
|
|
def test_segment_resynthesis_failure(self) -> None:
|
|
"""分段重新合成失败时标记 job failed。"""
|
|
service = MagicMock(spec=CosyVoiceService)
|
|
service.submit_synthesize_task.side_effect = CosyVoiceError("Synthesis failed")
|
|
|
|
repo = MagicMock()
|
|
job = _make_job(
|
|
status=TTSJobStatus.PROCESSING,
|
|
input_text="这是一段很长的测试文本。" * 30,
|
|
metadata={
|
|
"segment_task_ids": ["task_1"],
|
|
"segment_audio_urls": [""],
|
|
},
|
|
)
|
|
repo.get.return_value = job
|
|
repo.update.side_effect = lambda j: j
|
|
|
|
workflow = _make_workflow(cosyvoice_service=service, repo=repo)
|
|
result = workflow._poll_segment_tasks(job)
|
|
|
|
assert result.status == TTSJobStatus.FAILED
|
|
|
|
@patch("packages.application.tts_job.workflow.safe_download_file")
|
|
def test_partial_audio_urls_reuse_existing(self, mock_download: MagicMock) -> None:
|
|
"""部分分段已有 audio_url 时直接复用，缺失的重新合成。"""
|
|
mock_download.return_value = 1024 # 模拟文件大小
|
|
|
|
service = MagicMock(spec=CosyVoiceService)
|
|
# 只有 1 个分段需要重新合成
|
|
service.submit_synthesize_task.return_value = {
|
|
"audio_url": "https://temp.com/seg2.mp3",
|
|
"duration": 3.0,
|
|
"file_size": 200,
|
|
}
|
|
|
|
storage = MagicMock()
|
|
storage.upload_file.return_value = "https://oss.example.com/merged.mp3"
|
|
|
|
repo = MagicMock()
|
|
job = _make_job(
|
|
status=TTSJobStatus.PROCESSING,
|
|
input_text="这是一段很长的测试文本。" * 30,
|
|
metadata={
|
|
"segment_task_ids": ["task_1", "task_2"],
|
|
"segment_audio_urls": ["https://temp.com/seg1.mp3", ""],
|
|
"segment_format": "mp3",
|
|
},
|
|
)
|
|
repo.get.return_value = job
|
|
repo.update.side_effect = lambda j: j
|
|
|
|
workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)
|
|
|
|
with patch("packages.application.tts_job.workflow.AudioMerger") as MockMerger:
|
|
mock_merger = MagicMock()
|
|
mock_merger.merge.return_value = b"merged data"
|
|
MockMerger.return_value = mock_merger
|
|
|
|
result = workflow._poll_segment_tasks(job)
|
|
|
|
assert result.status == TTSJobStatus.COMPLETED
|
|
# 只有 1 个缺失分段被重新合成
|
|
assert service.submit_synthesize_task.call_count == 1
|
|
|
|
|
|
class TestPollAndProcessSynthesisSegmentDetection:
|
|
"""测试 poll_and_process_synthesis 正确识别分段任务。"""
|
|
|
|
def test_detects_segment_task(self) -> None:
|
|
"""metadata 中有 segment_task_ids 时走分段轮询路径。"""
|
|
service = MagicMock(spec=CosyVoiceService)
|
|
|
|
repo = MagicMock()
|
|
job = _make_job(
|
|
status=TTSJobStatus.PROCESSING,
|
|
metadata={
|
|
"segment_task_ids": ["task_1", "task_2"],
|
|
"segment_audio_urls": ["", ""],
|
|
},
|
|
)
|
|
repo.get.return_value = job
|
|
repo.update.side_effect = lambda j: j
|
|
|
|
workflow = _make_workflow(cosyvoice_service=service, repo=repo)
|
|
|
|
with patch.object(workflow, "_poll_segment_tasks") as mock_poll:
|
|
mock_poll.return_value = job
|
|
workflow.poll_and_process_synthesis("test_job_seg")
|
|
mock_poll.assert_called_once()
|
|
|
|
def test_normal_task_no_segment(self) -> None:
|
|
"""普通任务不走分段路径。"""
|
|
service = MagicMock(spec=CosyVoiceService)
|
|
service.poll_synthesize_task.return_value = {
|
|
"audio_url": "https://temp.com/audio.mp3",
|
|
"duration": 5.0,
|
|
"file_size": 5000,
|
|
}
|
|
|
|
repo = MagicMock()
|
|
job = _make_job(
|
|
status=TTSJobStatus.PROCESSING,
|
|
metadata={"cosyvoice_task_id": "task_normal"},
|
|
)
|
|
repo.get.return_value = job
|
|
repo.update.side_effect = lambda j: j
|
|
|
|
storage = MagicMock()
|
|
storage.upload_file.return_value = "https://oss.example.com/audio.mp3"
|
|
|
|
workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)
|
|
|
|
with patch("packages.application.tts_job.workflow.safe_download_file") as mock_download:
|
|
mock_download.return_value = 1024 # 模拟文件大小
|
|
|
|
result = workflow.poll_and_process_synthesis("test_job_seg")
|
|
|
|
assert result.status == TTSJobStatus.COMPLETED
|