diff --git a/apps/worker/worker_app/tasks/asset_analyzer.py b/apps/worker/worker_app/tasks/asset_analyzer.py index 424799841..1611e260a 100755 --- a/apps/worker/worker_app/tasks/asset_analyzer.py +++ b/apps/worker/worker_app/tasks/asset_analyzer.py @@ -144,29 +144,29 @@ class AssetAnalyzer: ] stdout, _ = run_ffprobe(cmd, timeout=30) data = json.loads(stdout) - streams = data.get("streams", []) - format_info = data.get("format", {}) + streams = data.get("streams", []) + format_info = data.get("format", {}) - for stream in streams: - if stream.get("codec_type") == "video": - info.width = int(stream.get("width", 0)) - info.height = int(stream.get("height", 0)) - info.codec = stream.get("codec_name", "") + for stream in streams: + if stream.get("codec_type") == "video": + info.width = int(stream.get("width", 0)) + info.height = int(stream.get("height", 0)) + info.codec = stream.get("codec_name", "") - # 解析帧率 - fps_str = stream.get("r_frame_rate", "0/1") - if "/" in fps_str: - num, denom = fps_str.split("/") - info.fps = float(num) / float(denom) if float(denom) != 0 else 0.0 - else: - info.fps = float(fps_str) + # 解析帧率 + fps_str = stream.get("r_frame_rate", "0/1") + if "/" in fps_str: + num, denom = fps_str.split("/") + info.fps = float(num) / float(denom) if float(denom) != 0 else 0.0 + else: + info.fps = float(fps_str) - elif stream.get("codec_type") == "audio": - info.has_audio = True + elif stream.get("codec_type") == "audio": + info.has_audio = True - info.duration = float(format_info.get("duration", 0)) - info.bitrate = int(format_info.get("bit_rate", 0)) - info.file_size = int(format_info.get("size", 0)) + info.duration = float(format_info.get("duration", 0)) + info.bitrate = int(format_info.get("bit_rate", 0)) + info.file_size = int(format_info.get("size", 0)) except Exception as e: logger.warning(f"Failed to get video info: {e}") diff --git a/tests/unit/test_tts_oss_transfer.py b/tests/unit/test_tts_oss_transfer.py old mode 100644 new mode 100755 index 99d5fc292..785a2260e --- a/tests/unit/test_tts_oss_transfer.py +++ b/tests/unit/test_tts_oss_transfer.py @@ -7,6 +7,7 @@ from __future__ import annotations from datetime import datetime, timezone +import unittest from unittest.mock import MagicMock, patch from packages.application.cosyvoice_service import CosyVoiceService @@ -67,13 +68,10 @@ def _make_workflow( class TestTransferAudioToOSS: """测试 _transfer_audio_to_oss 方法。""" - @patch("packages.application.tts_job.workflow.httpx") - def test_success_download_and_upload(self, mock_httpx: MagicMock) -> None: + @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_resp = MagicMock() - mock_resp.content = b"fake audio data" - mock_resp.raise_for_status.return_value = None - mock_httpx.get.return_value = mock_resp + 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" @@ -89,20 +87,21 @@ class TestTransferAudioToOSS: assert url == "https://oss.example.com/tts-outputs/user_001/job_123.mp3" assert key == "tts-outputs/user_001/job_123.mp3" - mock_httpx.get.assert_called_once_with( + 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, - follow_redirects=True, ) 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.httpx") - def test_download_failure_fallback(self, mock_httpx: MagicMock) -> None: + @patch("packages.application.tts_job.workflow.safe_download_bytes") + def test_download_failure_fallback(self, mock_download: MagicMock) -> None: """下载失败时回退到原始临时 URL，storage_key 为空。""" - mock_httpx.get.side_effect = Exception("Network error") + mock_download.side_effect = Exception("Network error") workflow = _make_workflow() url, key = workflow._transfer_audio_to_oss( @@ -114,13 +113,10 @@ class TestTransferAudioToOSS: assert url == "https://cosyvoice-temp.com/audio.mp3" assert key == "" - @patch("packages.application.tts_job.workflow.httpx") - def test_upload_failure_fallback(self, mock_httpx: MagicMock) -> None: + @patch("packages.application.tts_job.workflow.safe_download_bytes") + def test_upload_failure_fallback(self, mock_download: MagicMock) -> None: """上传 OSS 失败时回退到原始临时 URL。""" - mock_resp = MagicMock() - mock_resp.content = b"fake audio data" - mock_resp.raise_for_status.return_value = None - mock_httpx.get.return_value = mock_resp + mock_download.return_value = b"fake audio data" storage = MagicMock() storage.upload_file.side_effect = Exception("OSS bucket error") @@ -135,13 +131,10 @@ class TestTransferAudioToOSS: assert url == "https://cosyvoice-temp.com/audio.mp3" assert key == "" - @patch("packages.application.tts_job.workflow.httpx") - def test_wav_content_type(self, mock_httpx: MagicMock) -> None: + @patch("packages.application.tts_job.workflow.safe_download_bytes") + def test_wav_content_type(self, mock_download: MagicMock) -> None: """wav 格式使用正确的 content_type。""" - mock_resp = MagicMock() - mock_resp.content = b"fake wav data" - mock_resp.raise_for_status.return_value = None - mock_httpx.get.return_value = mock_resp + mock_download.return_value = b"fake wav data" storage = MagicMock() storage.upload_file.return_value = "https://oss.example.com/audio.wav" @@ -162,13 +155,10 @@ class TestTransferAudioToOSS: class TestProcessSynthesisResultWithOSS: """测试 process_synthesis_result 集成 OSS 转存。""" - @patch("packages.application.tts_job.workflow.httpx") - def test_stores_permanent_url_and_key(self, mock_httpx: MagicMock) -> None: + @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_resp = MagicMock() - mock_resp.content = b"audio bytes" - mock_resp.raise_for_status.return_value = None - mock_httpx.get.return_value = mock_resp + 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" @@ -192,10 +182,10 @@ class TestProcessSynthesisResultWithOSS: assert result.duration == 5.0 assert result.file_size == 50000 - @patch("packages.application.tts_job.workflow.httpx") - def test_fallback_to_temp_url_on_oss_failure(self, mock_httpx: MagicMock) -> None: + @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_httpx.get.side_effect = Exception("Download failed") + mock_download.side_effect = Exception("Download failed") repo = MagicMock() job = _make_job(status=TTSJobStatus.PROCESSING) @@ -216,13 +206,10 @@ class TestProcessSynthesisResultWithOSS: class TestStartSynthesisSyncWithOSS: """测试 start_synthesis 同步路径的 OSS 转存。""" - @patch("packages.application.tts_job.workflow.httpx") - def test_sync_path_transfers_to_oss(self, mock_httpx: MagicMock) -> None: + @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_resp = MagicMock() - mock_resp.content = b"sync audio bytes" - mock_resp.raise_for_status.return_value = None - mock_httpx.get.return_value = mock_resp + 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" @@ -248,10 +235,10 @@ class TestStartSynthesisSyncWithOSS: assert job.output_audio_key == "tts-outputs/user_001/test_job_001.mp3" assert job.duration == 2.0 - @patch("packages.application.tts_job.workflow.httpx") - def test_sync_path_oss_failure_stores_temp_url(self, mock_httpx: MagicMock) -> None: + @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_httpx.get.side_effect = Exception("Network error") + mock_download.side_effect = Exception("Network error") service = MagicMock(spec=CosyVoiceService) service.submit_synthesize_task.return_value = { diff --git a/tests/unit/test_tts_segment_synthesis.py b/tests/unit/test_tts_segment_synthesis.py old mode 100644 new mode 100755 index 583b3a2ba..6e9a4c2c3 --- a/tests/unit/test_tts_segment_synthesis.py +++ b/tests/unit/test_tts_segment_synthesis.py @@ -233,14 +233,11 @@ class TestStartSegmentSynthesis: # 短文本走普通路径，不调用分段 assert job.status == TTSJobStatus.PROCESSING - @patch("packages.application.tts_job.workflow.httpx") - def test_long_text_sync_segments(self, mock_httpx: MagicMock) -> None: + @patch("packages.application.tts_job.workflow.safe_download_file") + def test_long_text_sync_segments(self, mock_download: MagicMock) -> None: """长文本同步分段：所有段立即返回 audio_url，直接合并。""" # Mock 分段音频下载 - mock_resp = MagicMock() - mock_resp.content = b"segment audio" - mock_resp.raise_for_status.return_value = None - mock_httpx.get.return_value = mock_resp + mock_download.return_value = 1024 # 模拟文件大小 service = MagicMock(spec=CosyVoiceService) # 每个分段都同步返回 audio_url @@ -355,14 +352,11 @@ class TestHandleSegmentFailure: class TestPollSegmentTasks: """测试 _poll_segment_tasks 分段缺失重新合成（适配同步接口）。""" - @patch("packages.application.tts_job.workflow.httpx") - def test_all_segments_done(self, mock_httpx: MagicMock) -> None: + @patch("packages.application.tts_job.workflow.safe_download_file") + def test_all_segments_done(self, mock_download: MagicMock) -> None: """所有分段缺少 audio_url 时重新同步合成，合并后标记完成。""" # Mock 下载分段音频 - mock_resp = MagicMock() - mock_resp.content = b"seg audio" - mock_resp.raise_for_status.return_value = None - mock_httpx.get.return_value = mock_resp + mock_download.return_value = 1024 # 模拟文件大小 service = MagicMock(spec=CosyVoiceService) service.submit_synthesize_task.side_effect = [ @@ -422,13 +416,10 @@ class TestPollSegmentTasks: assert result.status == TTSJobStatus.FAILED - @patch("packages.application.tts_job.workflow.httpx") - def test_partial_audio_urls_reuse_existing(self, mock_httpx: MagicMock) -> None: + @patch("packages.application.tts_job.workflow.safe_download_file") + def test_partial_audio_urls_reuse_existing(self, mock_download: MagicMock) -> None: """部分分段已有 audio_url 时直接复用，缺失的重新合成。""" - mock_resp = MagicMock() - mock_resp.content = b"seg audio" - mock_resp.raise_for_status.return_value = None - mock_httpx.get.return_value = mock_resp + mock_download.return_value = 1024 # 模拟文件大小 service = MagicMock(spec=CosyVoiceService) # 只有 1 个分段需要重新合成 @@ -515,11 +506,8 @@ class TestPollAndProcessSynthesisSegmentDetection: workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage) - with patch("packages.application.tts_job.workflow.httpx") as mock_httpx: - mock_resp = MagicMock() - mock_resp.content = b"audio" - mock_resp.raise_for_status.return_value = None - mock_httpx.get.return_value = mock_resp + 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") diff --git a/tests/unit/test_tts_streaming.py b/tests/unit/test_tts_streaming.py old mode 100644 new mode 100755 index 8b2bd29ca..1f587aedd --- a/tests/unit/test_tts_streaming.py +++ b/tests/unit/test_tts_streaming.py @@ -221,17 +221,14 @@ class TestTTSStreamingService: @pytest.mark.asyncio async def test_download_audio(self): - """下载音频数据。""" + """下载音频数据（SSRF防护走safe_download_bytes，mock掉安全层）。""" cosyvoice = MagicMock(spec=CosyVoiceService) service = TTSStreamingService(cosyvoice) - with patch("packages.application.tts_job.streaming_service.httpx") as mock_httpx: - mock_resp = MagicMock() - mock_resp.content = b"audio data" - mock_resp.raise_for_status.return_value = None - mock_httpx.get.return_value = mock_resp + with patch("packages.application.tts_job.streaming_service.safe_download_bytes") as mock_download: + mock_download.return_value = b"audio data" result = service._download_audio("https://example.com/audio.mp3") assert result == b"audio data" - mock_httpx.get.assert_called_once() + mock_download.assert_called_once()