xiaoxia-saas/tests/unit/test_tts_streaming_service.py

"""TTSStreamingService 纯逻辑单测 — 分段策略 + 分块推送 + 错误处理.

mock 掉 WebSocket 和 CosyVoiceService，验证核心逻辑：
- 文本长度路由（短文本/长文本）
- 空文本/超长文本校验
- 音频分块推送算法
- 错误处理路径
"""

from __future__ import annotations

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from packages.application.cosyvoice_service import CosyVoiceError
from packages.application.tts_job.streaming_service import (
    _AUDIO_CHUNK_SIZE,
    TTSStreamingError,
    TTSStreamingService,
)

# ── Fixtures ──────────────────────────────────────────────────────────────


@pytest.fixture
def mock_cosyvoice():
    """mock CosyVoiceService."""
    svc = MagicMock()
    svc.submit_synthesize_task.return_value = {
        "audio_url": "https://example.com/audio.mp3",
        "duration": 3.5,
    }
    return svc


@pytest.fixture
def streaming_service(mock_cosyvoice):
    """TTSStreamingService 实例."""
    return TTSStreamingService(mock_cosyvoice)


@pytest.fixture
def mock_ws():
    """mock WebSocket."""
    ws = AsyncMock()
    ws.send_bytes = AsyncMock()
    ws.send_json = AsyncMock()
    return ws


class FakeAudioBytes:
    """生成指定大小的假音频数据."""

    @staticmethod
    def make(size: int) -> bytes:
        return b"\x00" * size


# ── 合成路由测试 ──────────────────────────────────────────────────────────


class TestSynthesizeRouting:
    """synthesize_and_stream 路由逻辑测试."""

    @pytest.mark.asyncio
    async def test_empty_text_returns_error(self, streaming_service, mock_ws):
        """空文本返回错误，不调用合成."""
        await streaming_service.synthesize_and_stream(mock_ws, {"text": ""})

        # 应发送 error 消息
        mock_ws.send_json.assert_called()
        last_call = mock_ws.send_json.call_args
        assert last_call[0][0]["type"] == "error"
        assert "不能为空" in last_call[0][0]["message"]
        # 不应调用合成
        streaming_service._cosyvoice.submit_synthesize_task.assert_not_called()

    @pytest.mark.asyncio
    async def test_missing_text_key_returns_error(self, streaming_service, mock_ws):
        """缺少 text 字段返回错误."""
        await streaming_service.synthesize_and_stream(mock_ws, {})

        mock_ws.send_json.assert_called()
        last_call = mock_ws.send_json.call_args
        assert last_call[0][0]["type"] == "error"
        streaming_service._cosyvoice.submit_synthesize_task.assert_not_called()

    @pytest.mark.asyncio
    async def test_too_long_text_returns_error(self, streaming_service, mock_ws):
        """超长文本返回错误."""
        long_text = "你" * 10001
        await streaming_service.synthesize_and_stream(mock_ws, {"text": long_text})

        mock_ws.send_json.assert_called()
        last_call = mock_ws.send_json.call_args
        assert last_call[0][0]["type"] == "error"
        assert "最大" in last_call[0][0]["message"]
        streaming_service._cosyvoice.submit_synthesize_task.assert_not_called()

    @pytest.mark.asyncio
    async def test_short_text_uses_short_path(self, streaming_service, mock_ws):
        """短文本走 _stream_short_text 路径."""
        with patch.object(streaming_service, "_stream_short_text", new_callable=AsyncMock) as mock_short:
            await streaming_service.synthesize_and_stream(mock_ws, {"text": "hello"})
            mock_short.assert_called_once()

    @pytest.mark.asyncio
    async def test_long_text_uses_long_path(self, streaming_service, mock_ws):
        """长文本走 _stream_long_text 路径."""
        long_text = "你" * 501
        with patch.object(streaming_service, "_stream_long_text", new_callable=AsyncMock) as mock_long:
            await streaming_service.synthesize_and_stream(mock_ws, {"text": long_text})
            mock_long.assert_called_once()

    @pytest.mark.asyncio
    async def test_exactly_threshold_uses_short_path(self, streaming_service, mock_ws):
        """恰好等于阈值走短文本路径."""
        text = "你" * 500
        with (
            patch.object(streaming_service, "_stream_short_text", new_callable=AsyncMock) as mock_short,
            patch.object(streaming_service, "_stream_long_text", new_callable=AsyncMock) as mock_long,
        ):
            await streaming_service.synthesize_and_stream(mock_ws, {"text": text})
            mock_short.assert_called_once()
            mock_long.assert_not_called()


# ── 短文本流测试 ──────────────────────────────────────────────────────────


class TestStreamShortText:
    """短文本流式合成测试."""

    @pytest.mark.asyncio
    async def test_happy_path_sends_started_then_done(self, streaming_service, mock_ws):
        """短文本正常流程：started → 音频块 → done."""
        audio_data = FakeAudioBytes.make(5000)
        with patch.object(streaming_service, "_download_audio", return_value=audio_data):
            await streaming_service._stream_short_text(
                mock_ws,
                {"text": "hello", "voice_id": "v1", "format": "mp3", "speed": 1.0},
            )

        # 检查 started 消息
        calls = mock_ws.send_json.call_args_list
        assert calls[0][0][0]["type"] == "started"
        assert calls[0][0][0]["segment_count"] == 1

        # 检查 done 消息
        last_msg = calls[-1][0][0]
        assert last_msg["type"] == "done"
        assert last_msg["file_size"] == 5000
        assert last_msg["format"] == "mp3"
        assert last_msg["duration"] == 3.5

    @pytest.mark.asyncio
    async def test_calls_cosyvoice_with_correct_params(self, streaming_service, mock_ws):
        """正确传递参数给 CosyVoice."""
        audio_data = FakeAudioBytes.make(1000)
        with patch.object(streaming_service, "_download_audio", return_value=audio_data):
            await streaming_service._stream_short_text(
                mock_ws,
                {
                    "text": "test text",
                    "voice_id": "voice-123",
                    "sample_rate": 22050,
                    "format": "wav",
                    "speed": 1.5,
                },
            )

        streaming_service._cosyvoice.submit_synthesize_task.assert_called_once_with(
            text="test text",
            voice_id="voice-123",
            sample_rate=22050,
            format="wav",
            speed=1.5,
        )

    @pytest.mark.asyncio
    async def test_cosyvoice_error_returns_error(self, streaming_service, mock_ws):
        """CosyVoice 错误返回 error 消息."""
        streaming_service._cosyvoice.submit_synthesize_task.side_effect = CosyVoiceError("API quota exceeded")

        await streaming_service._stream_short_text(mock_ws, {"text": "hello"})

        # 最后一条应该是 error
        last_msg = mock_ws.send_json.call_args_list[-1][0][0]
        assert last_msg["type"] == "error"
        assert "API quota exceeded" in last_msg["message"]

    @pytest.mark.asyncio
    async def test_generic_exception_returns_error(self, streaming_service, mock_ws):
        """普通异常返回 error 消息."""
        streaming_service._cosyvoice.submit_synthesize_task.side_effect = RuntimeError("boom")

        await streaming_service._stream_short_text(mock_ws, {"text": "hello"})

        last_msg = mock_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_no_audio_url_returns_error(self, streaming_service, mock_ws):
        """合成结果无 audio_url 返回错误."""
        streaming_service._cosyvoice.submit_synthesize_task.return_value = {"duration": 3.0}

        await streaming_service._stream_short_text(mock_ws, {"text": "hello"})

        last_msg = mock_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_download_failure_returns_error(self, streaming_service, mock_ws):
        """音频下载失败返回 error."""
        with patch.object(
            streaming_service,
            "_download_audio",
            side_effect=Exception("download failed"),
        ):
            await streaming_service._stream_short_text(mock_ws, {"text": "hello"})

        last_msg = mock_ws.send_json.call_args_list[-1][0][0]
        assert last_msg["type"] == "error"
        assert "音频推送失败" in last_msg["message"]


# ── 音频分块测试 ──────────────────────────────────────────────────────────


class TestStreamAudioChunks:
    """_stream_audio_chunks 分块推送测试."""

    @pytest.mark.asyncio
    async def test_exact_one_chunk(self, streaming_service, mock_ws):
        """恰好一个 chunk 大小的数据."""
        data = FakeAudioBytes.make(_AUDIO_CHUNK_SIZE)
        total = await streaming_service._stream_audio_chunks(mock_ws, data)

        assert total == _AUDIO_CHUNK_SIZE
        assert mock_ws.send_bytes.call_count == 1
        assert len(mock_ws.send_bytes.call_args[0][0]) == _AUDIO_CHUNK_SIZE

    @pytest.mark.asyncio
    async def test_smaller_than_one_chunk(self, streaming_service, mock_ws):
        """小于一个 chunk 的数据."""
        data = FakeAudioBytes.make(1000)
        total = await streaming_service._stream_audio_chunks(mock_ws, data)

        assert total == 1000
        assert mock_ws.send_bytes.call_count == 1

    @pytest.mark.asyncio
    async def test_multiple_full_chunks(self, streaming_service, mock_ws):
        """多个完整 chunk."""
        num_chunks = 5
        data = FakeAudioBytes.make(_AUDIO_CHUNK_SIZE * num_chunks)
        total = await streaming_service._stream_audio_chunks(mock_ws, data)

        assert total == _AUDIO_CHUNK_SIZE * num_chunks
        assert mock_ws.send_bytes.call_count == num_chunks
        for c in mock_ws.send_bytes.call_args_list:
            assert len(c[0][0]) == _AUDIO_CHUNK_SIZE

    @pytest.mark.asyncio
    async def test_partial_last_chunk(self, streaming_service, mock_ws):
        """最后一个 chunk 不完整."""
        data = FakeAudioBytes.make(_AUDIO_CHUNK_SIZE * 2 + 1234)
        total = await streaming_service._stream_audio_chunks(mock_ws, data)

        assert total == _AUDIO_CHUNK_SIZE * 2 + 1234
        assert mock_ws.send_bytes.call_count == 3
        # 最后一块是 1234 字节
        last_chunk = mock_ws.send_bytes.call_args_list[-1][0][0]
        assert len(last_chunk) == 1234

    @pytest.mark.asyncio
    async def test_empty_audio_sends_zero_chunks(self, streaming_service, mock_ws):
        """空音频不发送任何 chunk."""
        total = await streaming_service._stream_audio_chunks(mock_ws, b"")
        assert total == 0
        mock_ws.send_bytes.assert_not_called()

    @pytest.mark.asyncio
    async def test_chunks_are_consecutive(self, streaming_service, mock_ws):
        """所有 chunk 拼接起来等于原始数据."""
        data = bytes(range(256)) * 50  # 12800 bytes
        total = await streaming_service._stream_audio_chunks(mock_ws, data)

        assert total == len(data)
        # 收集所有 chunk
        all_bytes = b"".join(c[0][0] for c in mock_ws.send_bytes.call_args_list)
        assert all_bytes == data


# ── 长文本分段流测试 ──────────────────────────────────────────────────────


class TestStreamLongText:
    """长文本分段流式合成测试."""

    @pytest.mark.asyncio
    async def test_happy_path_all_segments_ok(self, streaming_service, mock_ws):
        """长文本正常流程：多个分段全部成功."""
        audio_data = FakeAudioBytes.make(2000)
        with patch.object(streaming_service, "_download_audio", return_value=audio_data):
            text = "你" * 1200  # 应该分成3段
            await streaming_service._stream_long_text(
                mock_ws,
                {"text": text, "voice_id": "v1", "format": "mp3", "speed": 1.0},
            )

        # 检查 started 消息
        calls = mock_ws.send_json.call_args_list
        assert calls[0][0][0]["type"] == "started"
        segment_count = calls[0][0][0]["segment_count"]
        assert segment_count >= 2  # 1200 字至少分 2 段

        # 检查有 segment_done 消息
        segment_dones = [c for c in calls if c[0][0].get("type") == "segment_done"]
        assert len(segment_dones) == segment_count

        # 检查最后是 done 消息
        last_msg = calls[-1][0][0]
        assert last_msg["type"] == "done"
        assert last_msg["file_size"] == 2000 * segment_count

    @pytest.mark.asyncio
    async def test_first_segment_fails_returns_error(self, streaming_service, mock_ws):
        """第一个分段失败，立即返回错误."""
        streaming_service._cosyvoice.submit_synthesize_task.side_effect = CosyVoiceError("segment 0 failed")

        text = "你" * 1200
        await streaming_service._stream_long_text(mock_ws, {"text": text, "voice_id": "v1"})

        calls = mock_ws.send_json.call_args_list
        last_msg = calls[-1][0][0]
        assert last_msg["type"] == "error"
        assert "分段" in last_msg["message"]
        assert "1" in last_msg["message"]  # 第1段失败

    @pytest.mark.asyncio
    async def test_segment_without_audio_url_fails(self, streaming_service, mock_ws):
        """分段结果无 audio_url 视为失败."""
        # 第一段正常，第二段返回空 audio_url
        call_results = [
            {"audio_url": "https://a.com/1.mp3", "duration": 2.0},
            {"audio_url": "", "duration": 0},
            {"audio_url": "https://a.com/3.mp3", "duration": 3.0},
        ]
        streaming_service._cosyvoice.submit_synthesize_task.side_effect = call_results

        with patch.object(
            streaming_service,
            "_download_audio",
            return_value=FakeAudioBytes.make(1000),
        ):
            text = "你" * 1500
            await streaming_service._stream_long_text(mock_ws, {"text": text, "voice_id": "v1"})

        calls = mock_ws.send_json.call_args_list
        # 应该有错误
        error_msgs = [c for c in calls if c[0][0].get("type") == "error"]
        assert len(error_msgs) >= 1

    @pytest.mark.asyncio
    async def test_each_segment_gets_correct_text(self, streaming_service, mock_ws):
        """每个分段都调用了合成，且 text 参数不同."""
        with patch.object(
            streaming_service,
            "_download_audio",
            return_value=FakeAudioBytes.make(500),
        ):
            text = "你" * 1200
            await streaming_service._stream_long_text(mock_ws, {"text": text, "voice_id": "v1"})

        # 分段数应大于1
        assert streaming_service._cosyvoice.submit_synthesize_task.call_count >= 2

        # 收集所有传进去的 text
        texts_called = [
            c.kwargs.get("text") or c.args[0]
            for c in streaming_service._cosyvoice.submit_synthesize_task.call_args_list
        ]
        # 每段文本都应该是原文的一部分（不全部相同）
        assert len(set(texts_called)) >= 2
        # 所有文本拼接起来应该约等于原文长度
        total_len = sum(len(t) for t in texts_called)
        assert total_len >= len(text) * 0.95  # 允许标点切分的小误差

    @pytest.mark.asyncio
    async def test_each_segment_has_unique_index(self, streaming_service, mock_ws):
        """segment_done 消息的序号不重复且正确."""
        with patch.object(
            streaming_service,
            "_download_audio",
            return_value=FakeAudioBytes.make(500),
        ):
            text = "你" * 1200
            await streaming_service._stream_long_text(mock_ws, {"text": text, "voice_id": "v1"})

        calls = mock_ws.send_json.call_args_list
        segment_dones = [c[0][0] for c in calls if c[0][0].get("type") == "segment_done"]
        indices = [s["segment"] for s in segment_dones]
        total = segment_dones[0]["total"]
        # 序号从 1 到 total，不重复
        assert sorted(indices) == list(range(1, total + 1))


# ── WebSocket 发送失败容错 ────────────────────────────────────────────────


class TestSendJsonErrorHandling:
    """_send_json 容错测试."""

    @pytest.mark.asyncio
    async def test_send_json_failure_logs_warning(self, streaming_service, mock_ws):
        """WebSocket send_json 失败不抛异常."""
        mock_ws.send_json.side_effect = Exception("connection closed")

        # 不应抛出异常
        await streaming_service._send_json(mock_ws, {"type": "done"})
        mock_ws.send_json.assert_called_once()


# ── TTSStreamingError 异常类 ──────────────────────────────────────────────


class TestTTSStreamingError:
    """TTSStreamingError 异常类测试."""

    def test_is_exception(self):
        """是 Exception 子类."""
        assert issubclass(TTSStreamingError, Exception)

    def test_carry_message(self):
        """携带错误消息."""
        err = TTSStreamingError("stream failed")
        assert str(err) == "stream failed"