tests/unit/test_tts_workflow_service.py

@@ -547,3 +547,428 @@ class TestErrorClasses:
         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 == ""