xiaoxia-saas/tests/unit/test_tts_segment_synthesis.py

"""P1 长文本分段合成单元测试。

覆盖：
- text_splitter.split_text 分段逻辑
- audio_merger.AudioMerger 合并逻辑
- workflow 分段合成路径（同步 / 异步 / 失败）
"""

from __future__ import annotations

import os
import tempfile
from datetime import datetime, timezone
from unittest.mock import MagicMock, call, patch

import pytest

from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
from packages.application.tts_job.audio_merger import AudioMergeError, AudioMerger
from packages.application.tts_job.text_splitter import split_text
from packages.application.tts_job.workflow import TTSWorkflowService
from packages.domain.tts_job import TTSJob, TTSJobStatus

# ── text_splitter ─────────────────────────────────────────────


class TestSplitText:
    """测试文本分段工具。"""

    def test_short_text_no_split(self) -> None:
        """短文本不拆分。"""
        assert split_text("你好世界", max_chars=500) == ["你好世界"]

    def test_empty_text(self) -> None:
        """空文本返回空列表。"""
        assert split_text("") == []
        assert split_text("   ") == []

    def test_exact_threshold(self) -> None:
        """恰好等于阈值不拆分。"""
        text = "a" * 500
        assert split_text(text, max_chars=500) == [text]

    def test_split_at_sentence_boundary(self) -> None:
        """在句子边界处分段。"""
        text = "第一句话。" * 60  # 300 chars
        text += "第二句话。" * 60  # 300 chars → total 600
        segments = split_text(text, max_chars=500)
        assert len(segments) >= 2
        for seg in segments:
            assert len(seg) <= 500

    def test_split_at_newline(self) -> None:
        """在换行符处分段。"""
        text = "段落一\n" * 100  # 300 chars
        text += "段落二\n" * 100  # 300 chars
        segments = split_text(text, max_chars=500)
        assert len(segments) >= 2

    def test_long_sentence_hard_split(self) -> None:
        """超长句子硬切。"""
        text = "a" * 1200
        segments = split_text(text, max_chars=500)
        assert len(segments) >= 3
        for seg in segments:
            assert len(seg) <= 500

    def test_merge_short_segments(self) -> None:
        """短段合并减少 API 调用。"""
        # 多个短句子应该被合并
        text = "你好。" * 120  # 360 chars, each sentence 3 chars
        segments = split_text(text, max_chars=500)
        # 短段应该被合并，段数应该比较少
        assert len(segments) < 120

    def test_preserves_order(self) -> None:
        """分段保持原始顺序。"""
        text = "第一段。第二段。第三段。" + "x" * 490
        segments = split_text(text, max_chars=500)
        # 第一个段应该以 "第一段" 开头
        assert segments[0].startswith("第一段")


# ── audio_merger ─────────────────────────────────────────────


class TestAudioMerger:
    """测试 FFmpeg 音频合并器。"""

    def test_empty_list_raises(self) -> None:
        """空列表抛出 AudioMergeError。"""
        merger = AudioMerger()
        with pytest.raises(AudioMergeError, match="没有可合并"):
            merger.merge([])

    def test_single_file_returns_bytes(self) -> None:
        """单文件直接返回内容。"""
        with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
            f.write(b"fake audio content")
            f.flush()
            path = f.name

        try:
            merger = AudioMerger()
            data = merger.merge([path])
            assert data == b"fake audio content"
        finally:
            os.unlink(path)

    @patch("packages.application.tts_job.audio_merger.subprocess.run")
    def test_ffmpeg_called_correctly(self, mock_run: MagicMock) -> None:
        """多文件调用 FFmpeg concat。"""
        mock_run.return_value = MagicMock(returncode=0)

        # 创建临时文件
        paths = []
        for i in range(3):
            with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
                f.write(b"audio")
                paths.append(f.name)

        try:
            merger = AudioMerger()
            # Mock open for reading the merged output
            with patch("builtins.open", create=True) as mock_open:
                mock_open.return_value.__enter__ = lambda s: s
                mock_open.return_value.read = lambda: b"merged audio"
                try:
                    merger.merge(paths, output_format="mp3")
                except (FileNotFoundError, OSError):
                    pass  # Expected since we're mocking

            # 验证 FFmpeg 被调用
            mock_run.assert_called_once()
            cmd = mock_run.call_args[0][0]
            assert cmd[0] == "ffmpeg"
            assert "-f" in cmd
            assert "concat" in cmd
        finally:
            for p in paths:
                os.unlink(p)

    @patch("packages.application.tts_job.audio_merger.subprocess.run")
    def test_ffmpeg_failure_raises(self, mock_run: MagicMock) -> None:
        """FFmpeg 失败抛出 AudioMergeError。"""
        mock_run.return_value = MagicMock(returncode=1, stderr="error details")

        paths = []
        for i in range(2):
            with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
                f.write(b"audio")
                paths.append(f.name)

        try:
            merger = AudioMerger()
            with pytest.raises(AudioMergeError, match="FFmpeg 合并失败"):
                merger.merge(paths)
        finally:
            for p in paths:
                os.unlink(p)


# ── workflow segment methods ─────────────────────────────────


def _make_job(**kwargs) -> TTSJob:
    defaults = {
        "id": "test_job_seg",
        "user_id": "user_001",
        "input_text": "x" * 600,  # > 500 threshold
        "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/merged.mp3"
    return TTSWorkflowService(
        repository=repo,
        cosyvoice_service=cosyvoice_service,
        storage_service=storage,
    )


class TestStartSegmentSynthesis:
    """测试 _start_segment_synthesis 分段合成入口。"""

    def test_short_text_no_segment(self) -> None:
        """短文本不触发分段。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.return_value = {
            "task_id": "task_1",
            "audio_url": "",
        }
        repo = MagicMock()
        repo.get.return_value = _make_job(input_text="短文本")
        repo.update.side_effect = lambda j: j

        workflow = _make_workflow(cosyvoice_service=service, repo=repo)
        job = workflow.start_synthesis("test_job_seg")

        # 短文本走普通路径，不调用分段
        assert job.status == TTSJobStatus.PROCESSING

    @patch("packages.application.tts_job.workflow.httpx")
    def test_long_text_sync_segments(self, mock_httpx: 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

        service = MagicMock(spec=CosyVoiceService)
        # 每个分段都同步返回 audio_url
        service.submit_synthesize_task.side_effect = [
            {"task_id": "", "audio_url": "https://temp.com/seg1.mp3", "duration": 2.0, "file_size": 1000},
            {"task_id": "", "audio_url": "https://temp.com/seg2.mp3", "duration": 3.0, "file_size": 1500},
        ]

        storage = MagicMock()
        storage.upload_file.return_value = "https://oss.example.com/merged.mp3"

        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)

        # Mock AudioMerger 避免真实 FFmpeg 调用
        with patch("packages.application.tts_job.workflow.AudioMerger") as MockMerger:
            mock_merger = MagicMock()
            mock_merger.merge.return_value = b"merged audio data"
            MockMerger.return_value = mock_merger

            job = workflow.start_synthesis("test_job_seg")

        assert job.status == TTSJobStatus.COMPLETED
        assert job.output_audio_url == "https://oss.example.com/merged.mp3"
        assert job.duration == 5.0  # 2.0 + 3.0

    def test_long_text_async_segments(self) -> None:
        """长文本异步分段：返回 task_id，存入 metadata。"""
        service = MagicMock(spec=CosyVoiceService)
        # 每个分段返回 task_id（异步）
        service.submit_synthesize_task.side_effect = [
            {"task_id": "seg_task_1", "audio_url": "", "duration": 0.0, "file_size": 0},
            {"task_id": "seg_task_2", "audio_url": "", "duration": 0.0, "file_size": 0},
        ]

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

        assert job.status == TTSJobStatus.PROCESSING
        assert "segment_task_ids" in job.metadata
        assert job.metadata["segment_task_ids"] == ["seg_task_1", "seg_task_2"]

    def test_segment_submit_failure_marks_failed(self) -> None:
        """分段提交失败时标记 job 为 failed。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.side_effect = CosyVoiceError("API error")

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

        assert job.status == TTSJobStatus.FAILED


class TestUploadMergedToOSS:
    """测试 _upload_merged_to_oss 辅助方法。"""

    def test_success(self) -> None:
        """成功上传返回 URL 和 key。"""
        storage = MagicMock()
        storage.upload_file.return_value = "https://oss.example.com/merged.mp3"

        workflow = _make_workflow(storage=storage)
        url, key = workflow._upload_merged_to_oss(b"audio data", "user_001", "job_001", "mp3")

        assert url == "https://oss.example.com/merged.mp3"
        assert key == "tts-outputs/user_001/job_001.mp3"
        storage.upload_file.assert_called_once()
        call_args = storage.upload_file.call_args
        assert call_args[1]["content_type"] == "audio/mpeg"

    def test_failure_returns_empty(self) -> None:
        """上传失败返回空字符串。"""
        storage = MagicMock()
        storage.upload_file.side_effect = Exception("OSS error")

        workflow = _make_workflow(storage=storage)
        url, key = workflow._upload_merged_to_oss(b"audio data", "user_001", "job_001", "mp3")

        assert url == ""
        assert key == ""


class TestHandleSegmentFailure:
    """测试 _handle_segment_failure。"""

    def test_marks_job_failed(self) -> None:
        """标记 job 为 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)
        workflow._handle_segment_failure(job, "分段 1 合成失败")

        assert job.status == TTSJobStatus.FAILED
        assert "分段 1 合成失败" in job.error_message
        repo.update.assert_called_once()


class TestPollSegmentTasks:
    """测试 _poll_segment_tasks 异步轮询。"""

    @patch("packages.application.tts_job.workflow.time")
    @patch("packages.application.tts_job.workflow.httpx")
    def test_all_segments_done(self, mock_httpx: MagicMock, mock_time: MagicMock) -> None:
        """所有分段完成后合并并标记完成。"""
        # Mock time.monotonic 让循环只执行一次
        mock_time.monotonic.side_effect = [0.0, 1.0, 2.0]
        mock_time.sleep = MagicMock()

        # 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

        service = MagicMock(spec=CosyVoiceService)
        service.poll_synthesize_task.side_effect = [
            {"audio_url": "https://temp.com/seg1.mp3", "duration": 2.0, "file_size": 100},
            {"audio_url": "https://temp.com/seg2.mp3", "duration": 3.0, "file_size": 200},
        ]

        storage = MagicMock()
        storage.upload_file.return_value = "https://oss.example.com/merged.mp3"

        repo = MagicMock()
        job = _make_job(
            status=TTSJobStatus.PROCESSING,
            metadata={
                "segment_task_ids": ["task_1", "task_2"],
                "segment_audio_urls": ["", ""],
                "segment_format": "mp3",
            },
        )
        repo.get.return_value = job
        repo.update.side_effect = lambda j: j

        workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)

        with patch("packages.application.tts_job.workflow.AudioMerger") as MockMerger:
            mock_merger = MagicMock()
            mock_merger.merge.return_value = b"merged data"
            MockMerger.return_value = mock_merger

            result = workflow._poll_segment_tasks(job)

        assert result.status == TTSJobStatus.COMPLETED

    @patch("packages.application.tts_job.workflow.time")
    def test_segment_poll_failure(self, mock_time: MagicMock) -> None:
        """分段轮询失败时标记 job failed。"""
        mock_time.monotonic.side_effect = [0.0, 1.0]
        mock_time.sleep = MagicMock()

        service = MagicMock(spec=CosyVoiceService)
        service.poll_synthesize_task.side_effect = CosyVoiceError("Poll failed")

        repo = MagicMock()
        job = _make_job(
            status=TTSJobStatus.PROCESSING,
            metadata={
                "segment_task_ids": ["task_1"],
                "segment_audio_urls": [""],
            },
        )
        repo.get.return_value = job
        repo.update.side_effect = lambda j: j

        workflow = _make_workflow(cosyvoice_service=service, repo=repo)
        result = workflow._poll_segment_tasks(job)

        assert result.status == TTSJobStatus.FAILED


class TestPollAndProcessSynthesisSegmentDetection:
    """测试 poll_and_process_synthesis 正确识别分段任务。"""

    def test_detects_segment_task(self) -> None:
        """metadata 中有 segment_task_ids 时走分段轮询路径。"""
        service = MagicMock(spec=CosyVoiceService)

        repo = MagicMock()
        job = _make_job(
            status=TTSJobStatus.PROCESSING,
            metadata={
                "segment_task_ids": ["task_1", "task_2"],
                "segment_audio_urls": ["", ""],
            },
        )
        repo.get.return_value = job
        repo.update.side_effect = lambda j: j

        workflow = _make_workflow(cosyvoice_service=service, repo=repo)

        with patch.object(workflow, "_poll_segment_tasks") as mock_poll:
            mock_poll.return_value = job
            workflow.poll_and_process_synthesis("test_job_seg")
            mock_poll.assert_called_once()

    def test_normal_task_no_segment(self) -> None:
        """普通任务不走分段路径。"""
        service = MagicMock(spec=CosyVoiceService)
        service.poll_synthesize_task.return_value = {
            "audio_url": "https://temp.com/audio.mp3",
            "duration": 5.0,
            "file_size": 5000,
        }

        repo = MagicMock()
        job = _make_job(
            status=TTSJobStatus.PROCESSING,
            metadata={"cosyvoice_task_id": "task_normal"},
        )
        repo.get.return_value = job
        repo.update.side_effect = lambda j: j

        storage = MagicMock()
        storage.upload_file.return_value = "https://oss.example.com/audio.mp3"

        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

            result = workflow.poll_and_process_synthesis("test_job_seg")

        assert result.status == TTSJobStatus.COMPLETED