Compare commits

...

2 Commits

Author SHA1 Message Date
CI Auto Fix Bot ddbe444cb6 fix(ci): auto-fix lint/format issues
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 18s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 33s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 37s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 1m28s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 33s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 49s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 1m8s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Failing after 1m18s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 50s
AI Code Review / AI Code Review (pull_request) Successful in 4m18s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 6m49s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Unit Tests (pull_request) Successful in 26s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Failing after 2m36s
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 29m17s
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 57s
2026-07-22 19:18:35 +08:00
CI Bot 0182274b28 test(p3-1): add 25 unit tests for TTS workflow service
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 49s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 37s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 31s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m0s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 2m24s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 1m27s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 34s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Failing after 54s
AI Code Review / AI Code Review (pull_request) Successful in 4m52s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 7m12s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 11m13s
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been cancelled
Covers:
- start_synthesis: sync completion, job not found, processing state,
  CosyVoice error, auth error, ValueError, OSS transfer fallback,
  async task_id saved
- poll_and_process_synthesis: already completed, job not found,
  no task_id resynthesize, old task_id poll fail fallback
- process_synthesis_result: success, not found, OSS upload failure
- process_synthesis_failure: marks failed, not found
- segment synthesis: long text triggers segments, segment failure,
  async segment task_ids saved
- resynthesize: success and failure paths
- error classes + lazy storage property
2026-07-22 18:38:25 +08:00
+549
View File
@@ -0,0 +1,549 @@
"""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