Retry transient TTS synthesis failures

This commit is contained in:
2026-05-24 16:44:17 +01:00
parent f1e72f27e2
commit 803f532ff3
2 changed files with 77 additions and 12 deletions

48
tests/test_tts_retry.py Normal file
View File

@@ -0,0 +1,48 @@
"""Tests for transient Edge TTS retry behavior."""
from __future__ import annotations
import asyncio
from src.engines import Engine
from src.translation import TranslationConfig
def test_synthesize_retries_transient_edge_tts_failure(tmp_path, monkeypatch):
calls = {"count": 0}
class FakeCommunicate:
def __init__(self, text, voice, rate):
self.text = text
self.voice = voice
self.rate = rate
async def save(self, out_path):
calls["count"] += 1
if calls["count"] == 1:
raise RuntimeError("transient 503")
with open(out_path, "wb") as audio_file:
audio_file.write(b"0" * 2048)
async def no_sleep(_seconds):
return None
monkeypatch.setattr("src.engines.edge_tts.Communicate", FakeCommunicate)
monkeypatch.setattr("src.engines.asyncio.sleep", no_sleep)
monkeypatch.setattr("src.engines.DEFAULT_TTS_MAX_RETRIES", 2)
engine = Engine(
"cpu",
translation_config=TranslationConfig(
base_url="http://127.0.0.1:1234/v1",
api_key="test-key",
model="test-model",
),
)
out_path = tmp_path / "tts.mp3"
asyncio.run(engine.synthesize("Bonjour", "fr", out_path))
assert calls["count"] == 2
assert out_path.exists()
assert out_path.stat().st_size == 2048