From 7051e21204191f26cf0f4e23c74c8e44ebc8fdb8 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Wed, 22 Jul 2026 17:58:55 +0800 Subject: [PATCH] =?UTF-8?q?test:=20P3-1=20=E7=AC=AC=E4=BA=8C=E5=8D=81?= =?UTF-8?q?=E4=BA=8C=E6=B3=A2=20TTS=E9=9F=B3=E9=A2=91=E5=90=88=E5=B9=B6+?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E6=9C=8D=E5=8A=A1=E5=8D=95=E5=85=83=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=2014=E4=B8=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - AudioMerger 7个（空列表/单文件/多文件/wav格式/异常/临时目录清理） - TTSStreamingService 7个（空文本/过长文本/短文本成功/合成失败/无audio_url/分块推送/send_json容错） - 合计 14 个测试全部通过 --- tests/unit/test_tts_audio_merger.py | 296 ++++++++++++++++++++++++++++ 1 file changed, 296 insertions(+) create mode 100755 tests/unit/test_tts_audio_merger.py diff --git a/tests/unit/test_tts_audio_merger.py b/tests/unit/test_tts_audio_merger.py new file mode 100755 index 000000000..0c847ca2e --- /dev/null +++ b/tests/unit/test_tts_audio_merger.py @@ -0,0 +1,296 @@ +""" +TTS 相关单元测试（第二十二波） + +覆盖： +- AudioMerger (空列表/单文件/多文件合并/格式/异常) +""" + +import os +import subprocess +import tempfile + +import pytest + +from packages.application.tts_job.audio_merger import ( + AudioMergeError, + AudioMerger, +) + + +def _make_silence(duration: float = 0.5, sample_rate: int = 22050, fmt: str = "mp3") -> str: + """生成一段静音音频文件，返回路径。""" + tmp = tempfile.NamedTemporaryFile(suffix=f".{fmt}", delete=False) + tmp.close() + cmd = [ + "ffmpeg", "-y", "-f", "lavfi", + "-i", f"anullsrc=r={sample_rate}:cl=mono", + "-t", str(duration), + "-q:a", "9", + tmp.name, + ] + subprocess.run(cmd, capture_output=True, check=True) + return tmp.name + + +class TestAudioMerger: + """AudioMerger 音频合并器""" + + def test_empty_list_raises(self): + """空列表抛错""" + merger = AudioMerger() + with pytest.raises(AudioMergeError, match="没有可合并"): + merger.merge([]) + + def test_single_file_returns_content(self): + """单个文件直接返回内容""" + path = _make_silence(duration=0.3) + try: + merger = AudioMerger() + result = merger.merge([path]) + assert isinstance(result, bytes) + assert len(result) > 100 # 应该有有效数据 + # 应该和文件本身一致 + with open(path, "rb") as f: + original = f.read() + assert result == original + finally: + os.unlink(path) + + def test_two_files_merged(self): + """两个文件合并""" + p1 = _make_silence(duration=0.3) + p2 = _make_silence(duration=0.4) + try: + merger = AudioMerger() + result = merger.merge([p1, p2]) + assert isinstance(result, bytes) + assert len(result) > 200 # 合并后应该有数据 + # 写出来用 ffprobe 验证时长 + tmp = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) + tmp.write(result) + tmp.close() + try: + probe = subprocess.run( + ["ffprobe", "-v", "error", "-show_entries", "format=duration", + "-of", "default=noprint_wrappers=1:nokey=1", tmp.name], + capture_output=True, text=True, check=True, + ) + duration = float(probe.stdout.strip()) + # 0.3 + 0.4 = 0.7 秒左右，允许一定误差 + assert 0.5 < duration < 1.0 + finally: + os.unlink(tmp.name) + finally: + os.unlink(p1) + os.unlink(p2) + + def test_three_files_merged(self): + """三个文件合并""" + paths = [_make_silence(duration=0.2) for _ in range(3)] + try: + merger = AudioMerger() + result = merger.merge(paths) + assert isinstance(result, bytes) + assert len(result) > 200 + finally: + for p in paths: + os.unlink(p) + + def test_wav_format(self): + """wav 格式合并""" + p1 = _make_silence(duration=0.2, fmt="wav") + p2 = _make_silence(duration=0.2, fmt="wav") + try: + merger = AudioMerger() + result = merger.merge([p1, p2], output_format="wav") + assert isinstance(result, bytes) + # WAV 头部以 RIFF 开头 + assert result[:4] == b"RIFF" + finally: + os.unlink(p1) + os.unlink(p2) + + def test_nonexistent_file_raises(self): + """不存在的文件会抛错""" + merger = AudioMerger() + with pytest.raises(AudioMergeError): + merger.merge(["/nonexistent/path/a.mp3", "/nonexistent/path/b.mp3"]) + + def test_cleanup_temp_dir(self): + """临时目录会被清理""" + p1 = _make_silence(duration=0.2) + p2 = _make_silence(duration=0.2) + try: + import tempfile as _tf + before = set(os.listdir(_tf.gettempdir())) + merger = AudioMerger() + merger.merge([p1, p2]) + after = set(os.listdir(_tf.gettempdir())) + # 不应该残留 tts_merge_ 前缀的目录 + new_items = after - before + tts_items = [i for i in new_items if i.startswith("tts_merge_")] + assert len(tts_items) == 0, f"残留临时目录: {tts_items}" + finally: + os.unlink(p1) + os.unlink(p2) + + +# ============================================================ +# TTSStreamingService - 入口路由与边界 +# ============================================================ + + +class TestTTSStreamingServiceRouting: + """TTSStreamingService 入口路由与边界条件""" + + @pytest.mark.asyncio + async def test_empty_text_returns_error(self): + """空文本返回错误""" + from unittest.mock import AsyncMock, MagicMock + + from packages.application.tts_job.streaming_service import TTSStreamingService + + mock_cosy = MagicMock() + svc = TTSStreamingService(cosyvoice_service=mock_cosy) + ws = AsyncMock() + + await svc.synthesize_and_stream(ws, {"text": ""}) + + ws.send_json.assert_called_once() + call_args = ws.send_json.call_args[0][0] + assert call_args["type"] == "error" + assert "不能为空" in call_args["message"] + # 不应该调用 cosyvoice + mock_cosy.submit_synthesize_task.assert_not_called() + + @pytest.mark.asyncio + async def test_text_too_long_returns_error(self): + """文本过长返回错误""" + from unittest.mock import AsyncMock, MagicMock + + from packages.application.tts_job.streaming_service import ( + _MAX_TEXT_LENGTH, + TTSStreamingService, + ) + + mock_cosy = MagicMock() + svc = TTSStreamingService(cosyvoice_service=mock_cosy) + ws = AsyncMock() + + long_text = "a" * (_MAX_TEXT_LENGTH + 1) + await svc.synthesize_and_stream(ws, {"text": long_text}) + + ws.send_json.assert_called_once() + call_args = ws.send_json.call_args[0][0] + assert call_args["type"] == "error" + assert "过长" in call_args["message"] + mock_cosy.submit_synthesize_task.assert_not_called() + + @pytest.mark.asyncio + async def test_short_text_routes_to_short_path(self): + """短文本走短文本路径（单段合成）""" + from unittest.mock import AsyncMock, MagicMock, patch + + from packages.application.tts_job.streaming_service import TTSStreamingService + + mock_cosy = MagicMock() + mock_cosy.submit_synthesize_task.return_value = { + "audio_url": "https://example.com/audio.mp3", + "duration": 3.5, + } + svc = TTSStreamingService(cosyvoice_service=mock_cosy) + ws = AsyncMock() + + # mock 掉音频下载 + fake_audio = b"fake_audio_data" * 100 + with patch.object(svc, "_download_audio", return_value=fake_audio): + await svc.synthesize_and_stream(ws, {"text": "你好世界", "voice_id": "v1"}) + + # 应该调用了 cosy + mock_cosy.submit_synthesize_task.assert_called_once() + # 应该有 started 和 done 消息 + msg_types = [c[0][0]["type"] for c in ws.send_json.call_args_list] + assert "started" in msg_types + assert "done" in msg_types + # 应该有音频分块发送 + assert ws.send_bytes.call_count > 0 + + @pytest.mark.asyncio + async def test_short_text_cosy_error(self): + """短文本合成失败返回错误""" + from unittest.mock import AsyncMock, MagicMock + + from packages.application.cosyvoice_service import CosyVoiceError + from packages.application.tts_job.streaming_service import TTSStreamingService + + mock_cosy = MagicMock() + mock_cosy.submit_synthesize_task.side_effect = CosyVoiceError("音色不存在") + svc = TTSStreamingService(cosyvoice_service=mock_cosy) + ws = AsyncMock() + + await svc.synthesize_and_stream(ws, {"text": "你好", "voice_id": "v-bad"}) + + # 最后一条消息应该是 error + last_msg = ws.send_json.call_args_list[-1][0][0] + assert last_msg["type"] == "error" + assert "音色不存在" in last_msg["message"] + + @pytest.mark.asyncio + async def test_short_text_no_audio_url(self): + """合成结果没有 audio_url 返回错误""" + from unittest.mock import AsyncMock, MagicMock + + from packages.application.tts_job.streaming_service import TTSStreamingService + + mock_cosy = MagicMock() + mock_cosy.submit_synthesize_task.return_value = {"duration": 1.0} # 没有 audio_url + svc = TTSStreamingService(cosyvoice_service=mock_cosy) + ws = AsyncMock() + + await svc.synthesize_and_stream(ws, {"text": "你好"}) + + last_msg = ws.send_json.call_args_list[-1][0][0] + assert last_msg["type"] == "error" + assert "音频 URL" in last_msg["message"] + + @pytest.mark.asyncio + async def test_stream_audio_chunks_returns_total(self): + """_stream_audio_chunks 返回正确字节数，分块正确""" + from unittest.mock import AsyncMock, MagicMock + + from packages.application.tts_job.streaming_service import ( + _AUDIO_CHUNK_SIZE, + TTSStreamingService, + ) + + mock_cosy = MagicMock() + svc = TTSStreamingService(cosyvoice_service=mock_cosy) + ws = AsyncMock() + + # 生成 10000 字节的假音频 + audio_data = b"x" * 10000 + total = await svc._stream_audio_chunks(ws, audio_data) + + assert total == 10000 + # 应该分 ceil(10000/4096) = 3 块 + expected_chunks = (10000 + _AUDIO_CHUNK_SIZE - 1) // _AUDIO_CHUNK_SIZE + assert ws.send_bytes.call_count == expected_chunks + # 验证所有块拼接起来等于原数据 + all_bytes = b"".join(c[0][0] for c in ws.send_bytes.call_args_list) + assert all_bytes == audio_data + + @pytest.mark.asyncio + async def test_send_json_handles_error(self): + """_send_json 发送失败不抛出异常""" + from unittest.mock import AsyncMock, MagicMock + + from packages.application.tts_job.streaming_service import TTSStreamingService + + mock_cosy = MagicMock() + svc = TTSStreamingService(cosyvoice_service=mock_cosy) + ws = AsyncMock() + ws.send_json.side_effect = Exception("连接已断开") + + # 不应该抛异常 + await svc._send_json(ws, {"type": "done"}) + ws.send_json.assert_called_once()