xiaoxia-saas/tests/unit/test_tts_audio_merger.py

"""
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()