xiaoxia-saas/tests/unit/test_tts_workflow_service.py

"""TTS Workflow service unit tests.

Covers TTSWorkflowService - start_synthesis, poll_and_process_synthesis,
process_synthesis_result, process_synthesis_failure, segment synthesis.
"""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest

from packages.application.cosyvoice_service import CosyVoiceAuthError, CosyVoiceError
from packages.application.tts_job.workflow import (
    TTSJobNotFoundError,
    TTSWorkflowError,
    TTSWorkflowService,
)
from packages.domain.tts_job import TTSJob, TTSJobStatus

# ── Helpers ─────────────────────────────────────────────


def make_job(**kwargs):
    defaults = dict(
        id="job-123",
        user_id="user-1",
        input_text="Hello world",
        voice_id="voice-1",
        sample_rate=22050,
        format="mp3",
    )
    defaults.update(kwargs)
    return TTSJob(**defaults)


class FakeTTSJobRepository:
    def __init__(self, job=None):
        self._job = job
        self.updated_jobs = []
        self.get_called = 0

    def get(self, job_id):
        self.get_called += 1
        if self._job and self._job.id == job_id:
            return self._job
        return None

    def update(self, job):
        self.updated_jobs.append(job)
        self._job = job
        return job


class FakeCosyVoiceService:
    def __init__(self, submit_result=None, submit_error=None, poll_result=None, poll_error=None):
        self._submit_result = submit_result or {
            "audio_url": "https://temp.example.com/audio.mp3",
            "task_id": "",
            "request_id": "req-1",
            "duration": 5.0,
            "file_size": 1024,
        }
        self._submit_error = submit_error
        self._poll_result = poll_result
        self._poll_error = poll_error
        self.submit_calls = []
        self.poll_calls = []

    def submit_synthesize_task(self, **kwargs):
        self.submit_calls.append(kwargs)
        if self._submit_error:
            raise self._submit_error
        return self._submit_result

    def poll_synthesize_task(self, task_id, timeout=120.0):
        self.poll_calls.append({"task_id": task_id, "timeout": timeout})
        if self._poll_error:
            raise self._poll_error
        return self._poll_result or {
            "audio_url": "https://temp.example.com/audio.mp3",
            "duration": 5.0,
            "file_size": 1024,
        }


class FakeStorageService:
    def __init__(self, upload_url="https://oss.example.com/tts-outputs/user-1/job-123.mp3", upload_error=None):
        self._upload_url = upload_url
        self._upload_error = upload_error
        self.uploads = []

    def upload_file(self, file_obj, storage_key, content_type=None):
        self.uploads.append({"storage_key": storage_key, "content_type": content_type})
        if self._upload_error:
            raise self._upload_error
        return self._upload_url


# ── start_synthesis tests ───────────────────────────────


class TestStartSynthesis:
    def test_successful_sync_completion(self):
        """start_synthesis with sync audio_url → job marked completed."""
        job = make_job()
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/audio.mp3",
                "request_id": "req-abc",
                "task_id": "",
                "duration": 3.5,
                "file_size": 5000,
            }
        )
        storage = FakeStorageService(upload_url="https://oss.example.com/final.mp3")
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"fake audio data"):
            result = svc.start_synthesis("job-123")

        assert result.id == "job-123"
        assert result.status == TTSJobStatus.COMPLETED.value
        assert result.output_audio_url == "https://oss.example.com/final.mp3"
        assert result.output_audio_key == "tts-outputs/user-1/job-123.mp3"
        assert result.duration == 3.5
        assert result.file_size == 5000
        assert result.metadata["cosyvoice_request_id"] == "req-abc"

        # verify cosyvoice was called
        assert len(cosy.submit_calls) == 1
        assert cosy.submit_calls[0]["text"] == "Hello world"
        assert cosy.submit_calls[0]["voice_id"] == "voice-1"

        # verify storage upload
        assert len(storage.uploads) == 1
        assert storage.uploads[0]["storage_key"] == "tts-outputs/user-1/job-123.mp3"

    def test_job_not_found_raises(self):
        repo = FakeTTSJobRepository()  # no job
        cosy = FakeCosyVoiceService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)

        with pytest.raises(TTSJobNotFoundError):
            svc.start_synthesis("nonexistent")

    def test_job_marked_processing_before_submit(self):
        job = make_job()
        assert job.status == TTSJobStatus.PENDING.value

        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "",
                "task_id": "task-abc",
                "request_id": "req-1",
            }
        )
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=FakeStorageService())

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            svc.start_synthesis("job-123")

        # first update should mark processing
        first_update = repo.updated_jobs[0]
        assert first_update.status == TTSJobStatus.PROCESSING.value

    def test_cosyvoice_error_marks_failed(self):
        job = make_job()
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(submit_error=CosyVoiceError("API rate limit"))
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)

        result = svc.start_synthesis("job-123")

        assert result.status == TTSJobStatus.FAILED.value
        assert "API rate limit" in result.error_message

    def test_cosyvoice_auth_error_marks_failed(self):
        job = make_job()
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(submit_error=CosyVoiceAuthError("Invalid key"))
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)

        result = svc.start_synthesis("job-123")

        assert result.status == TTSJobStatus.FAILED.value
        assert "Invalid key" in result.error_message

    def test_value_error_marks_failed(self):
        job = make_job()
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(submit_error=ValueError("text is empty"))
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)

        result = svc.start_synthesis("job-123")

        assert result.status == TTSJobStatus.FAILED.value
        assert "text is empty" in result.error_message

    def test_oss_transfer_failure_falls_back_to_temp_url(self):
        job = make_job()
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/original.mp3",
                "request_id": "req-1",
                "task_id": "",
                "duration": 2.0,
                "file_size": 1000,
            }
        )
        storage = FakeStorageService(upload_error=RuntimeError("OSS down"))
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"fake"):
            result = svc.start_synthesis("job-123")

        assert result.status == TTSJobStatus.COMPLETED.value
        # falls back to temp URL
        assert result.output_audio_url == "https://temp.example.com/original.mp3"
        assert result.output_audio_key == ""

    def test_no_audio_url_task_id_saved(self):
        """Async path: no audio_url, save task_id to metadata."""
        job = make_job()
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "",
                "task_id": "async-task-123",
                "request_id": "req-async",
            }
        )
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=FakeStorageService())

        result = svc.start_synthesis("job-123")

        assert result.status == TTSJobStatus.PROCESSING.value
        assert result.metadata["cosyvoice_task_id"] == "async-task-123"
        assert result.metadata["cosyvoice_request_id"] == "req-async"


# ── poll_and_process_synthesis tests ────────────────────


class TestPollAndProcessSynthesis:
    def test_already_completed_returns_immediately(self):
        job = make_job()
        job.mark_processing()
        job.mark_completed(output_audio_url="https://oss.example.com/done.mp3", output_audio_key="key")
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)

        result = svc.poll_and_process_synthesis("job-123")

        assert result.status == TTSJobStatus.COMPLETED.value
        # no new submit calls
        assert len(cosy.submit_calls) == 0
        assert len(cosy.poll_calls) == 0

    def test_job_not_found_raises(self):
        repo = FakeTTSJobRepository()
        cosy = FakeCosyVoiceService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)

        with pytest.raises(TTSJobNotFoundError):
            svc.poll_and_process_synthesis("nonexistent")

    def test_no_task_id_resynthesizes(self):
        """No task_id in metadata → re-sync synthesize."""
        job = make_job()
        job.mark_processing()
        job.metadata = {}  # no task_id
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/retry.mp3",
                "request_id": "req-retry",
                "task_id": "",
                "duration": 3.0,
                "file_size": 2048,
            }
        )
        storage = FakeStorageService(upload_url="https://oss.example.com/retry-final.mp3")
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            result = svc.poll_and_process_synthesis("job-123")

        assert result.status == TTSJobStatus.COMPLETED.value
        assert result.output_audio_url == "https://oss.example.com/retry-final.mp3"
        assert len(cosy.submit_calls) == 1

    def test_old_task_id_poll_fails_resynthesizes(self):
        """Old task_id poll fails → fallback to re-synthesize."""
        job = make_job()
        job.mark_processing()
        job.metadata = {"cosyvoice_task_id": "old-task-id"}
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            poll_error=CosyVoiceError("task not found"),
            submit_result={
                "audio_url": "https://temp.example.com/retry2.mp3",
                "request_id": "req-retry2",
                "task_id": "",
                "duration": 3.0,
                "file_size": 1000,
            },
        )
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            result = svc.poll_and_process_synthesis("job-123")

        assert len(cosy.poll_calls) == 1
        assert cosy.poll_calls[0]["task_id"] == "old-task-id"
        assert len(cosy.submit_calls) == 1  # resynthesize
        assert result.status == TTSJobStatus.COMPLETED.value


# ── process_synthesis_result tests ──────────────────────


class TestProcessSynthesisResult:
    def test_success_marks_completed(self):
        job = make_job()
        job.mark_processing()
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService()
        storage = FakeStorageService(upload_url="https://oss.example.com/final.mp3")
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"fake audio"):
            result = svc.process_synthesis_result(
                "job-123",
                audio_url="https://temp.example.com/audio.mp3",
                duration=10.5,
                file_size=50000,
            )

        assert result.status == TTSJobStatus.COMPLETED.value
        assert result.output_audio_url == "https://oss.example.com/final.mp3"
        assert result.output_audio_key == "tts-outputs/user-1/job-123.mp3"
        assert result.duration == 10.5
        assert result.file_size == 50000

    def test_job_not_found_raises(self):
        repo = FakeTTSJobRepository()
        cosy = FakeCosyVoiceService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)

        with pytest.raises(TTSJobNotFoundError):
            svc.process_synthesis_result("nonexistent", audio_url="https://x.mp3")

    def test_oss_upload_failure_uses_temp_url(self):
        job = make_job()
        job.mark_processing()
        repo = FakeTTSJobRepository(job=job)
        storage = FakeStorageService(upload_error=RuntimeError("network error"))
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=FakeCosyVoiceService(), storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"data"):
            result = svc.process_synthesis_result("job-123", audio_url="https://temp.example.com/x.mp3")

        assert result.status == TTSJobStatus.COMPLETED.value
        assert result.output_audio_url == "https://temp.example.com/x.mp3"
        assert result.output_audio_key == ""


# ── process_synthesis_failure tests ─────────────────────


class TestProcessSynthesisFailure:
    def test_marks_job_failed(self):
        job = make_job()
        job.mark_processing()
        repo = FakeTTSJobRepository(job=job)
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=FakeCosyVoiceService())

        result = svc.process_synthesis_failure("job-123", "CosyVoice 429 rate limited")

        assert result.status == TTSJobStatus.FAILED.value
        assert "429 rate limited" in result.error_message

    def test_job_not_found_raises(self):
        repo = FakeTTSJobRepository()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=FakeCosyVoiceService())

        with pytest.raises(TTSJobNotFoundError):
            svc.process_synthesis_failure("nonexistent", "error")


# ── Segment synthesis tests ─────────────────────────────


class TestSegmentSynthesis:
    def test_long_text_triggers_segment_synthesis(self):
        """Text over 500 chars → segment synthesis path."""
        long_text = "你好" * 300  # 600 chars, over 500 threshold
        job = make_job(input_text=long_text)
        repo = FakeTTSJobRepository(job=job)

        # Each segment returns audio_url (sync path)
        def mock_submit(**kwargs):
            return {
                "audio_url": f"https://seg.example.com/{kwargs.get('text', '')[:10]}.mp3",
                "task_id": "",
                "request_id": "r",
                "duration": 1.0,
                "file_size": 100,
            }

        cosy = FakeCosyVoiceService()
        cosy.submit_synthesize_task = MagicMock(side_effect=mock_submit)
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with (
            patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"),
            patch("packages.application.tts_job.workflow.safe_download_file") as mock_download,
            patch("packages.application.tts_job.workflow.AudioMerger") as mock_merger_class,
        ):
            mock_merger = MagicMock()
            mock_merger.merge.return_value = b"merged audio data"
            mock_merger_class.return_value = mock_merger

            result = svc.start_synthesis("job-123")

        assert result.status == TTSJobStatus.COMPLETED.value
        assert result.file_size == len(b"merged audio data")
        # Multiple segments submitted
        assert cosy.submit_synthesize_task.call_count >= 2

    def test_segment_failure_marks_job_failed(self):
        """One segment fails → whole job fails."""
        long_text = "A" * 600
        job = make_job(input_text=long_text)
        repo = FakeTTSJobRepository(job=job)

        call_count = {"n": 0}

        def mock_submit(**kwargs):
            call_count["n"] += 1
            if call_count["n"] == 2:
                raise CosyVoiceError("segment 2 failed")
            return {
                "audio_url": "https://seg.example.com/s.mp3",
                "task_id": "",
                "request_id": "r",
                "duration": 1.0,
                "file_size": 100,
            }

        cosy = FakeCosyVoiceService()
        cosy.submit_synthesize_task = MagicMock(side_effect=mock_submit)
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=FakeStorageService())

        result = svc.start_synthesis("job-123")

        assert result.status == TTSJobStatus.FAILED.value
        assert "分段" in result.error_message

    def test_segment_task_ids_saved_for_async(self):
        """Async segment results → task_ids saved to metadata."""
        long_text = "B" * 600
        job = make_job(input_text=long_text)
        repo = FakeTTSJobRepository(job=job)

        def mock_submit(**kwargs):
            return {"audio_url": "", "task_id": f"task-{kwargs.get('text', '')[:5]}", "request_id": "r"}

        cosy = FakeCosyVoiceService()
        cosy.submit_synthesize_task = MagicMock(side_effect=mock_submit)
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=FakeStorageService())

        result = svc.start_synthesis("job-123")

        assert result.status == TTSJobStatus.PROCESSING.value
        assert "segment_task_ids" in result.metadata
        assert len(result.metadata["segment_task_ids"]) >= 2
        assert result.metadata["segment_count"] >= 2


# ── _resynthesize_and_complete tests ────────────────────


class TestResynthesizeAndComplete:
    def test_resynthesize_success(self):
        job = make_job(input_text="Retry me")
        job.mark_processing()
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/retry.mp3",
                "request_id": "req-r",
                "task_id": "",
                "duration": 2.5,
                "file_size": 1500,
            }
        )
        storage = FakeStorageService(upload_url="https://oss.example.com/retry.mp3")
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            # call via poll_and_process_synthesis which uses _resynthesize_and_complete
            result = svc.poll_and_process_synthesis("job-123")

        assert result.status == TTSJobStatus.COMPLETED.value
        assert len(cosy.submit_calls) == 1

    def test_resynthesize_failure_marks_failed(self):
        job = make_job()
        job.mark_processing()
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(submit_error=CosyVoiceError("permanent failure"))
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)

        result = svc.poll_and_process_synthesis("job-123")

        assert result.status == TTSJobStatus.FAILED.value
        assert "permanent failure" in result.error_message


# ── Error classes tests ─────────────────────────────────


class TestErrorClasses:
    def test_workflow_error_inherits_from_exception(self):
        err = TTSWorkflowError("test error")
        assert isinstance(err, Exception)
        assert str(err) == "test error"

    def test_not_found_error_inherits_from_exception(self):
        err = TTSJobNotFoundError("not found")
        assert isinstance(err, Exception)
        assert "not found" in str(err)

    def test_storage_property_lazy_init(self):
        """_storage property lazily initializes storage service."""
        from packages.shared.storage import SharedStorageService

        # With provided storage_service
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=MagicMock(), cosyvoice_service=MagicMock(), storage_service=storage)
        assert svc._storage is storage


# ── Additional edge case tests ──────────────────────────


class TestTransferAudioToOSS:
    """_transfer_audio_to_oss 细节测试."""

    def test_mp3_content_type(self):
        """MP3格式使用audio/mpeg content-type."""
        job = make_job(format="mp3")
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/a.mp3",
                "request_id": "r",
                "task_id": "",
                "duration": 1.0,
                "file_size": 100,
            }
        )
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            svc.start_synthesis("job-123")

        assert storage.uploads[0]["content_type"] == "audio/mpeg"

    def test_wav_content_type(self):
        """WAV格式使用audio/wav content-type."""
        job = make_job(format="wav")
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/a.wav",
                "request_id": "r",
                "task_id": "",
                "duration": 1.0,
                "file_size": 100,
            }
        )
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            svc.start_synthesis("job-123")

        assert storage.uploads[0]["content_type"] == "audio/wav"

    def test_unknown_format_default_content_type(self):
        """未知格式使用application/octet-stream."""
        job = make_job(format="flac")
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/a.flac",
                "request_id": "r",
                "task_id": "",
                "duration": 1.0,
                "file_size": 100,
            }
        )
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            svc.start_synthesis("job-123")

        assert storage.uploads[0]["content_type"] == "application/octet-stream"

    def test_storage_key_format(self):
        """storage_key格式正确：tts-outputs/{user_id}/{job_id}.{format}."""
        job = make_job(id="custom-job", user_id="user-999", format="wav")
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/a.wav",
                "duration": 1.0,
                "file_size": 100,
            }
        )
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            svc.start_synthesis("custom-job")

        assert storage.uploads[0]["storage_key"] == "tts-outputs/user-999/custom-job.wav"


class TestResynthesizeParams:
    """重新合成时参数从metadata读取测试."""

    def test_speed_from_metadata(self):
        """重新合成时speed从metadata读取."""
        job = make_job(input_text="test")
        job.mark_processing()
        job.metadata = {"speed": 1.5}
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/r.mp3",
                "duration": 2.0,
                "file_size": 500,
            }
        )
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            svc.poll_and_process_synthesis("job-123")

        assert cosy.submit_calls[0]["speed"] == 1.5

    def test_volume_from_metadata(self):
        """重新合成时volume从metadata读取."""
        job = make_job(input_text="test")
        job.mark_processing()
        job.metadata = {"volume": 80}
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/r.mp3",
                "duration": 2.0,
                "file_size": 500,
            }
        )
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            svc.poll_and_process_synthesis("job-123")

        assert cosy.submit_calls[0]["volume"] == 80

    def test_default_speed_when_no_metadata(self):
        """无metadata时speed默认1.0."""
        job = make_job(input_text="test")
        job.mark_processing()
        job.metadata = {}
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/r.mp3",
                "duration": 1.0,
                "file_size": 100,
            }
        )
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            svc.poll_and_process_synthesis("job-123")

        assert cosy.submit_calls[0]["speed"] == 1.0

    def test_default_volume_when_no_metadata(self):
        """无metadata时volume默认50."""
        job = make_job(input_text="test")
        job.mark_processing()
        job.metadata = {}
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(
            submit_result={
                "audio_url": "https://temp.example.com/r.mp3",
                "duration": 1.0,
                "file_size": 100,
            }
        )
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with patch("packages.application.tts_job.workflow.safe_download_bytes", return_value=b"audio"):
            svc.poll_and_process_synthesis("job-123")

        assert cosy.submit_calls[0]["volume"] == 50

    def test_resynthesize_no_audio_url_marks_failed(self):
        """重新合成未返回audio_url时标记失败."""
        job = make_job(input_text="test")
        job.mark_processing()
        job.metadata = {}
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService(submit_result={"audio_url": "", "task_id": "", "request_id": "r"})
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy)

        result = svc.poll_and_process_synthesis("job-123")

        assert result.status == TTSJobStatus.FAILED.value
        assert "重新合成" in result.error_message


class TestPollSegmentTasks:
    """分段任务轮询测试."""

    def test_poll_segment_with_no_audio_urls_triggers_resynth(self):
        """所有分段都缺audio_url时全部重新合成."""
        long_text = "x" * 600
        job = make_job(input_text=long_text, format="mp3")
        job.mark_processing()
        job.metadata = {
            "segment_task_ids": ["task1", "task2"],
            "segment_audio_urls": ["", ""],
            "segment_count": 2,
        }
        repo = FakeTTSJobRepository(job=job)

        def mock_submit(**kwargs):
            return {
                "audio_url": "https://resynth.example.com/r.mp3",
                "duration": 1.0,
                "file_size": 100,
            }

        cosy = FakeCosyVoiceService()
        cosy.submit_synthesize_task = MagicMock(side_effect=mock_submit)
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with (
            patch("packages.application.tts_job.workflow.safe_download_file"),
            patch("packages.application.tts_job.workflow.AudioMerger") as mock_merger_class,
        ):
            mock_merger = MagicMock()
            mock_merger.merge.return_value = b"merged"
            mock_merger_class.return_value = mock_merger
            result = svc.poll_and_process_synthesis("job-123")

        # 2个分段都需要重新合成
        assert cosy.submit_synthesize_task.call_count == 2
        assert result.status == TTSJobStatus.COMPLETED.value

    def test_existing_audio_urls_used_directly(self):
        """已有segment_audio_urls的分段直接使用，不重新合成."""
        long_text = "x" * 600
        job = make_job(input_text=long_text, format="mp3")
        job.mark_processing()
        job.metadata = {
            "segment_task_ids": ["task1", "task2"],
            "segment_audio_urls": ["https://seg1.mp3", "https://seg2.mp3"],
            "segment_count": 2,
        }
        repo = FakeTTSJobRepository(job=job)
        cosy = FakeCosyVoiceService()
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with (
            patch("packages.application.tts_job.workflow.safe_download_file"),
            patch("packages.application.tts_job.workflow.AudioMerger") as mock_merger_class,
        ):
            mock_merger = MagicMock()
            mock_merger.merge.return_value = b"merged"
            mock_merger_class.return_value = mock_merger
            result = svc.poll_and_process_synthesis("job-123")

        # 所有分段都有audio_url，不需要重新合成
        assert len(cosy.submit_calls) == 0
        assert result.status == TTSJobStatus.COMPLETED.value

    def test_missing_audio_url_resynthesized(self):
        """缺少audio_url的分段会重新合成."""
        long_text = "x" * 600
        job = make_job(input_text=long_text, format="mp3")
        job.mark_processing()
        job.metadata = {
            "segment_task_ids": ["task1", "task2"],
            "segment_audio_urls": ["https://seg1.mp3", ""],
            "segment_count": 2,
        }
        repo = FakeTTSJobRepository(job=job)

        def mock_submit(**kwargs):
            return {
                "audio_url": "https://resynth.mp3",
                "duration": 1.0,
                "file_size": 100,
            }

        cosy = FakeCosyVoiceService()
        cosy.submit_synthesize_task = MagicMock(side_effect=mock_submit)
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with (
            patch("packages.application.tts_job.workflow.safe_download_file"),
            patch("packages.application.tts_job.workflow.AudioMerger") as mock_merger_class,
        ):
            mock_merger = MagicMock()
            mock_merger.merge.return_value = b"merged"
            mock_merger_class.return_value = mock_merger
            result = svc.poll_and_process_synthesis("job-123")

        # 只有1个分段需要重新合成
        assert cosy.submit_synthesize_task.call_count == 1
        assert result.status == TTSJobStatus.COMPLETED.value


class TestSegmentSyncDetails:
    """分段同步路径细节测试."""

    def test_segment_count_in_metadata(self):
        """分段合成时segment_count写入metadata."""
        long_text = "x" * 1200
        job = make_job(input_text=long_text)
        repo = FakeTTSJobRepository(job=job)

        def mock_submit(**kwargs):
            return {
                "audio_url": "https://seg.mp3",
                "duration": 1.0,
                "file_size": 100,
            }

        cosy = FakeCosyVoiceService()
        cosy.submit_synthesize_task = MagicMock(side_effect=mock_submit)
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with (
            patch("packages.application.tts_job.workflow.safe_download_file"),
            patch("packages.application.tts_job.workflow.AudioMerger") as mock_merger_class,
        ):
            mock_merger = MagicMock()
            mock_merger.merge.return_value = b"merged audio"
            mock_merger_class.return_value = mock_merger
            result = svc.start_synthesis("job-123")

        # 检查完成状态和文件大小
        assert result.status == TTSJobStatus.COMPLETED.value
        assert result.file_size == len(b"merged audio")

    def test_segment_sync_duration_accumulated(self):
        """分段同步路径时长累加."""
        long_text = "x" * 600
        job = make_job(input_text=long_text)
        repo = FakeTTSJobRepository(job=job)

        call_idx = {"n": 0}

        def mock_submit(**kwargs):
            call_idx["n"] += 1
            return {
                "audio_url": f"https://seg{call_idx['n']}.mp3",
                "duration": 2.5 * call_idx["n"],  # 2.5 + 5.0 = 7.5
                "file_size": 100,
            }

        cosy = FakeCosyVoiceService()
        cosy.submit_synthesize_task = MagicMock(side_effect=mock_submit)
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with (
            patch("packages.application.tts_job.workflow.safe_download_file"),
            patch("packages.application.tts_job.workflow.AudioMerger") as mock_merger_class,
        ):
            mock_merger = MagicMock()
            mock_merger.merge.return_value = b"merged"
            mock_merger_class.return_value = mock_merger
            result = svc.start_synthesis("job-123")

        assert result.duration > 0
        assert result.status == TTSJobStatus.COMPLETED.value

    def test_segment_missing_audio_url_raises(self):
        """分段同步路径中某段无audio_url抛出TTSWorkflowError."""
        long_text = "x" * 600
        job = make_job(input_text=long_text)
        repo = FakeTTSJobRepository(job=job)

        call_idx = {"n": 0}

        def mock_submit(**kwargs):
            call_idx["n"] += 1
            if call_idx["n"] == 2:
                return {"audio_url": "", "duration": 0, "file_size": 0}
            return {
                "audio_url": "https://example.com/seg1.mp3",
                "duration": 1.0,
                "file_size": 100,
            }

        cosy = FakeCosyVoiceService()
        cosy.submit_synthesize_task = MagicMock(side_effect=mock_submit)
        storage = FakeStorageService()
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=cosy, storage_service=storage)

        with (
            patch("packages.application.tts_job.workflow.safe_download_file"),
            pytest.raises(TTSWorkflowError, match="没有返回 audio_url"),
        ):
            svc.start_synthesis("job-123")


class TestUploadMergedToOSS:
    """_upload_merged_to_oss 测试."""

    def test_upload_success_returns_url_and_key(self):
        """上传成功返回永久URL和storage_key."""
        job = make_job(id="job-merge", user_id="u1", format="mp3")
        repo = FakeTTSJobRepository(job=job)
        storage = FakeStorageService(upload_url="https://oss.example.com/merged.mp3")
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=MagicMock(), storage_service=storage)

        url, key = svc._upload_merged_to_oss(b"merged data", "u1", "job-merge", "mp3")

        assert url == "https://oss.example.com/merged.mp3"
        assert key == "tts-outputs/u1/job-merge.mp3"
        assert len(storage.uploads) == 1

    def test_upload_failure_returns_empty(self):
        """上传失败返回空字符串."""
        job = make_job()
        repo = FakeTTSJobRepository(job=job)
        storage = FakeStorageService(upload_error=RuntimeError("upload failed"))
        svc = TTSWorkflowService(repository=repo, cosyvoice_service=MagicMock(), storage_service=storage)

        url, key = svc._upload_merged_to_oss(b"data", "user", "job", "wav")

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