CI Bot
745ea06659
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 29s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 1m10s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m47s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m17s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (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 / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
P0-1: asset_analyzer.py 缩进错误 - 第147行起23行缩进多了一层（改subprocess时弄乱） - 修正 streams/format_info/for循环 等缩进 P0-2: TTS单测因SSRF接入挂了8个 - 根因：测试mock httpx，但safe_download_*在发请求前做DNS解析 - 修复：将httpx mock替换为safe_download_bytes/safe_download_file mock - test_tts_oss_transfer.py: 4个失败修复，mock safe_download_bytes - test_tts_segment_synthesis.py: 3个失败修复，mock safe_download_file - test_tts_streaming.py: 1个失败修复，mock safe_download_bytes - 安全层单测保持独立，33个test_url_security.py全绿 76个相关测试全绿（9+24+10+33）
285 lines
11 KiB
Python
Executable File
285 lines
11 KiB
Python
Executable File
"""TTS 音频转存 OSS 单元测试。
|
||
|
||
验证 TTSWorkflowService 在合成完成后将 CosyVoice 临时音频转存到 OSS，
|
||
存储永久 URL 到 TTSJob.output_audio_url，OSS key 到 output_audio_key。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timezone
|
||
import unittest
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from packages.application.cosyvoice_service import CosyVoiceService
|
||
from packages.application.tts_job.workflow import TTSWorkflowService
|
||
from packages.domain.tts_job import TTSJob, TTSJobStatus
|
||
|
||
|
||
def _make_job(**kwargs) -> TTSJob:
|
||
defaults = {
|
||
"id": "test_job_001",
|
||
"user_id": "user_001",
|
||
"input_text": "测试文本",
|
||
"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/tts-outputs/user_001/test_job_001.mp3"
|
||
return TTSWorkflowService(
|
||
repository=repo,
|
||
cosyvoice_service=cosyvoice_service,
|
||
storage_service=storage,
|
||
)
|
||
|
||
|
||
class TestTransferAudioToOSS:
|
||
"""测试 _transfer_audio_to_oss 方法。"""
|
||
|
||
@patch("packages.application.tts_job.workflow.safe_download_bytes")
|
||
def test_success_download_and_upload(self, mock_download: MagicMock) -> None:
|
||
"""成功下载音频并上传到 OSS，返回永久 URL 和 storage_key。"""
|
||
mock_download.return_value = b"fake audio data"
|
||
|
||
storage = MagicMock()
|
||
storage.upload_file.return_value = "https://oss.example.com/tts-outputs/user_001/job_123.mp3"
|
||
|
||
workflow = _make_workflow(storage=storage)
|
||
url, key = workflow._transfer_audio_to_oss(
|
||
"https://cosyvoice-temp.com/audio.mp3",
|
||
"user_001",
|
||
"job_123",
|
||
"mp3",
|
||
)
|
||
|
||
assert url == "https://oss.example.com/tts-outputs/user_001/job_123.mp3"
|
||
assert key == "tts-outputs/user_001/job_123.mp3"
|
||
|
||
mock_download.assert_called_once_with(
|
||
"https://cosyvoice-temp.com/audio.mp3",
|
||
purpose="tts_audio_download",
|
||
allowed_mime_types=unittest.mock.ANY,
|
||
timeout=60.0,
|
||
)
|
||
storage.upload_file.assert_called_once()
|
||
call_args = storage.upload_file.call_args
|
||
assert call_args[0][1] == "tts-outputs/user_001/job_123.mp3"
|
||
assert call_args[1]["content_type"] == "audio/mpeg"
|
||
|
||
@patch("packages.application.tts_job.workflow.safe_download_bytes")
|
||
def test_download_failure_fallback(self, mock_download: MagicMock) -> None:
|
||
"""下载失败时回退到原始临时 URL，storage_key 为空。"""
|
||
mock_download.side_effect = Exception("Network error")
|
||
|
||
workflow = _make_workflow()
|
||
url, key = workflow._transfer_audio_to_oss(
|
||
"https://cosyvoice-temp.com/audio.mp3",
|
||
"user_001",
|
||
"job_123",
|
||
)
|
||
|
||
assert url == "https://cosyvoice-temp.com/audio.mp3"
|
||
assert key == ""
|
||
|
||
@patch("packages.application.tts_job.workflow.safe_download_bytes")
|
||
def test_upload_failure_fallback(self, mock_download: MagicMock) -> None:
|
||
"""上传 OSS 失败时回退到原始临时 URL。"""
|
||
mock_download.return_value = b"fake audio data"
|
||
|
||
storage = MagicMock()
|
||
storage.upload_file.side_effect = Exception("OSS bucket error")
|
||
|
||
workflow = _make_workflow(storage=storage)
|
||
url, key = workflow._transfer_audio_to_oss(
|
||
"https://cosyvoice-temp.com/audio.mp3",
|
||
"user_001",
|
||
"job_123",
|
||
)
|
||
|
||
assert url == "https://cosyvoice-temp.com/audio.mp3"
|
||
assert key == ""
|
||
|
||
@patch("packages.application.tts_job.workflow.safe_download_bytes")
|
||
def test_wav_content_type(self, mock_download: MagicMock) -> None:
|
||
"""wav 格式使用正确的 content_type。"""
|
||
mock_download.return_value = b"fake wav data"
|
||
|
||
storage = MagicMock()
|
||
storage.upload_file.return_value = "https://oss.example.com/audio.wav"
|
||
|
||
workflow = _make_workflow(storage=storage)
|
||
workflow._transfer_audio_to_oss(
|
||
"https://cosyvoice-temp.com/audio.wav",
|
||
"user_001",
|
||
"job_456",
|
||
"wav",
|
||
)
|
||
|
||
call_args = storage.upload_file.call_args
|
||
assert call_args[1]["content_type"] == "audio/wav"
|
||
assert call_args[0][1] == "tts-outputs/user_001/job_456.wav"
|
||
|
||
|
||
class TestProcessSynthesisResultWithOSS:
|
||
"""测试 process_synthesis_result 集成 OSS 转存。"""
|
||
|
||
@patch("packages.application.tts_job.workflow.safe_download_bytes")
|
||
def test_stores_permanent_url_and_key(self, mock_download: MagicMock) -> None:
|
||
"""合成结果存 OSS 永久 URL 和 storage_key。"""
|
||
mock_download.return_value = b"audio bytes"
|
||
|
||
storage = MagicMock()
|
||
storage.upload_file.return_value = "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3"
|
||
|
||
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, storage=storage)
|
||
result = workflow.process_synthesis_result(
|
||
"test_job_001",
|
||
audio_url="https://cosyvoice-temp.com/expiring.mp3",
|
||
duration=5.0,
|
||
file_size=50000,
|
||
)
|
||
|
||
assert result.status == TTSJobStatus.COMPLETED
|
||
assert result.output_audio_url == "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3"
|
||
assert result.output_audio_key == "tts-outputs/user_001/test_job_001.mp3"
|
||
assert result.duration == 5.0
|
||
assert result.file_size == 50000
|
||
|
||
@patch("packages.application.tts_job.workflow.safe_download_bytes")
|
||
def test_fallback_to_temp_url_on_oss_failure(self, mock_download: MagicMock) -> None:
|
||
"""OSS 转存失败时，使用 CosyVoice 临时 URL（不阻塞合成流程）。"""
|
||
mock_download.side_effect = Exception("Download 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)
|
||
result = workflow.process_synthesis_result(
|
||
"test_job_001",
|
||
audio_url="https://cosyvoice-temp.com/expiring.mp3",
|
||
)
|
||
|
||
assert result.status == TTSJobStatus.COMPLETED
|
||
assert result.output_audio_url == "https://cosyvoice-temp.com/expiring.mp3"
|
||
assert result.output_audio_key == ""
|
||
|
||
|
||
class TestStartSynthesisSyncWithOSS:
|
||
"""测试 start_synthesis 同步路径的 OSS 转存。"""
|
||
|
||
@patch("packages.application.tts_job.workflow.safe_download_bytes")
|
||
def test_sync_path_transfers_to_oss(self, mock_download: MagicMock) -> None:
|
||
"""CosyVoice 同步返回 audio_url 时，也走 OSS 转存。"""
|
||
mock_download.return_value = b"sync audio bytes"
|
||
|
||
storage = MagicMock()
|
||
storage.upload_file.return_value = "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3"
|
||
|
||
service = MagicMock(spec=CosyVoiceService)
|
||
service.submit_synthesize_task.return_value = {
|
||
"task_id": "",
|
||
"audio_url": "https://cosyvoice-temp.com/sync.mp3",
|
||
"duration": 2.0,
|
||
"file_size": 20000,
|
||
"request_id": "req_sync",
|
||
}
|
||
|
||
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)
|
||
job = workflow.start_synthesis("test_job_001")
|
||
|
||
assert job.status == TTSJobStatus.COMPLETED
|
||
assert job.output_audio_url == "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3"
|
||
assert job.output_audio_key == "tts-outputs/user_001/test_job_001.mp3"
|
||
assert job.duration == 2.0
|
||
|
||
@patch("packages.application.tts_job.workflow.safe_download_bytes")
|
||
def test_sync_path_oss_failure_stores_temp_url(self, mock_download: MagicMock) -> None:
|
||
"""同步路径 OSS 失败时，降级存储临时 URL。"""
|
||
mock_download.side_effect = Exception("Network error")
|
||
|
||
service = MagicMock(spec=CosyVoiceService)
|
||
service.submit_synthesize_task.return_value = {
|
||
"task_id": "",
|
||
"audio_url": "https://cosyvoice-temp.com/sync.mp3",
|
||
"duration": 2.0,
|
||
"file_size": 20000,
|
||
"request_id": "req_sync",
|
||
}
|
||
|
||
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_001")
|
||
|
||
assert job.status == TTSJobStatus.COMPLETED
|
||
assert job.output_audio_url == "https://cosyvoice-temp.com/sync.mp3"
|
||
assert job.output_audio_key == ""
|
||
|
||
def test_async_path_no_oss_transfer(self) -> None:
|
||
"""异步路径（返回 task_id，无 audio_url）不触发 OSS 转存。"""
|
||
service = MagicMock(spec=CosyVoiceService)
|
||
service.submit_synthesize_task.return_value = {
|
||
"task_id": "cosy_task_async",
|
||
"audio_url": "",
|
||
"duration": 0.0,
|
||
"file_size": 0,
|
||
"request_id": "req_async",
|
||
}
|
||
|
||
repo = MagicMock()
|
||
repo.get.return_value = _make_job()
|
||
repo.update.side_effect = lambda j: j
|
||
|
||
storage = MagicMock()
|
||
workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)
|
||
job = workflow.start_synthesis("test_job_001")
|
||
|
||
assert job.status == TTSJobStatus.PROCESSING
|
||
# 异步路径不应调用 OSS 上传
|
||
storage.upload_file.assert_not_called()
|