xiaoxia-saas/tests/unit/test_tts_oss_transfer.py

"""TTS 音频转存 OSS 单元测试。

验证 TTSWorkflowService 在合成完成后将 CosyVoice 临时音频转存到 OSS，
存储永久 URL 到 TTSJob.output_audio_url，OSS key 到 output_audio_key。
"""

from __future__ import annotations

import unittest
from datetime import UTC, datetime, timezone
from unittest.mock import MagicMock, patch

from packages.application.cosyvoice_service import CosyVoiceService
from packages.application.tts_job.workflow import TTSWorkflowService
from packages.domain.tts_job import TTSJob, TTSJobStatus


def _make_job(**kwargs) -> TTSJob:
    defaults = {
        "id": "test_job_001",
        "user_id": "user_001",
        "input_text": "测试文本",
        "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(UTC),
        "updated_at": datetime.now(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/tts-outputs/user_001/test_job_001.mp3"
    return TTSWorkflowService(
        repository=repo,
        cosyvoice_service=cosyvoice_service,
        storage_service=storage,
    )


class TestTransferAudioToOSS:
    """测试 _transfer_audio_to_oss 方法。"""

    @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_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"

        workflow = _make_workflow(storage=storage)
        url, key = workflow._transfer_audio_to_oss(
            "https://cosyvoice-temp.com/audio.mp3",
            "user_001",
            "job_123",
            "mp3",
        )

        assert url == "https://oss.example.com/tts-outputs/user_001/job_123.mp3"
        assert key == "tts-outputs/user_001/job_123.mp3"

        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,
        )
        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.safe_download_bytes")
    def test_download_failure_fallback(self, mock_download: MagicMock) -> None:
        """下载失败时回退到原始临时 URL，storage_key 为空。"""
        mock_download.side_effect = Exception("Network error")

        workflow = _make_workflow()
        url, key = workflow._transfer_audio_to_oss(
            "https://cosyvoice-temp.com/audio.mp3",
            "user_001",
            "job_123",
        )

        assert url == "https://cosyvoice-temp.com/audio.mp3"
        assert key == ""

    @patch("packages.application.tts_job.workflow.safe_download_bytes")
    def test_upload_failure_fallback(self, mock_download: MagicMock) -> None:
        """上传 OSS 失败时回退到原始临时 URL。"""
        mock_download.return_value = b"fake audio data"

        storage = MagicMock()
        storage.upload_file.side_effect = Exception("OSS bucket error")

        workflow = _make_workflow(storage=storage)
        url, key = workflow._transfer_audio_to_oss(
            "https://cosyvoice-temp.com/audio.mp3",
            "user_001",
            "job_123",
        )

        assert url == "https://cosyvoice-temp.com/audio.mp3"
        assert key == ""

    @patch("packages.application.tts_job.workflow.safe_download_bytes")
    def test_wav_content_type(self, mock_download: MagicMock) -> None:
        """wav 格式使用正确的 content_type。"""
        mock_download.return_value = b"fake wav data"

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

        workflow = _make_workflow(storage=storage)
        workflow._transfer_audio_to_oss(
            "https://cosyvoice-temp.com/audio.wav",
            "user_001",
            "job_456",
            "wav",
        )

        call_args = storage.upload_file.call_args
        assert call_args[1]["content_type"] == "audio/wav"
        assert call_args[0][1] == "tts-outputs/user_001/job_456.wav"


class TestProcessSynthesisResultWithOSS:
    """测试 process_synthesis_result 集成 OSS 转存。"""

    @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_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"

        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, storage=storage)
        result = workflow.process_synthesis_result(
            "test_job_001",
            audio_url="https://cosyvoice-temp.com/expiring.mp3",
            duration=5.0,
            file_size=50000,
        )

        assert result.status == TTSJobStatus.COMPLETED
        assert result.output_audio_url == "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3"
        assert result.output_audio_key == "tts-outputs/user_001/test_job_001.mp3"
        assert result.duration == 5.0
        assert result.file_size == 50000

    @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_download.side_effect = Exception("Download 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)
        result = workflow.process_synthesis_result(
            "test_job_001",
            audio_url="https://cosyvoice-temp.com/expiring.mp3",
        )

        assert result.status == TTSJobStatus.COMPLETED
        assert result.output_audio_url == "https://cosyvoice-temp.com/expiring.mp3"
        assert result.output_audio_key == ""


class TestStartSynthesisSyncWithOSS:
    """测试 start_synthesis 同步路径的 OSS 转存。"""

    @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_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"

        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.return_value = {
            "task_id": "",
            "audio_url": "https://cosyvoice-temp.com/sync.mp3",
            "duration": 2.0,
            "file_size": 20000,
            "request_id": "req_sync",
        }

        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)
        job = workflow.start_synthesis("test_job_001")

        assert job.status == TTSJobStatus.COMPLETED
        assert job.output_audio_url == "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3"
        assert job.output_audio_key == "tts-outputs/user_001/test_job_001.mp3"
        assert job.duration == 2.0

    @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_download.side_effect = Exception("Network error")

        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.return_value = {
            "task_id": "",
            "audio_url": "https://cosyvoice-temp.com/sync.mp3",
            "duration": 2.0,
            "file_size": 20000,
            "request_id": "req_sync",
        }

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

        assert job.status == TTSJobStatus.COMPLETED
        assert job.output_audio_url == "https://cosyvoice-temp.com/sync.mp3"
        assert job.output_audio_key == ""

    def test_async_path_no_oss_transfer(self) -> None:
        """异步路径（返回 task_id，无 audio_url）不触发 OSS 转存。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.return_value = {
            "task_id": "cosy_task_async",
            "audio_url": "",
            "duration": 0.0,
            "file_size": 0,
            "request_id": "req_async",
        }

        repo = MagicMock()
        repo.get.return_value = _make_job()
        repo.update.side_effect = lambda j: j

        storage = MagicMock()
        workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)
        job = workflow.start_synthesis("test_job_001")

        assert job.status == TTSJobStatus.PROCESSING
        # 异步路径不应调用 OSS 上传
        storage.upload_file.assert_not_called()