"""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 == ""