refactor(#5430): reuse shared TTS transport scaffolding

api/routes.py

@@ -16670,26 +16670,19 @@ def _handle_tts(handler, parsed):
             "voice_settings": {"stability": 0.5, "similarity_boost": 0.75},
         }).encode("utf-8")
 
-        from urllib.request import Request, build_opener
         req = Request(url, data=req_body, headers={
             "xi-api-key": api_key,
             "Content-Type": "application/json",
             "Accept": "audio/mpeg",
         })
 
-        class _NoRedirectElevenlabsHandler(HTTPRedirectHandler):
-            """Refuse to follow redirects on the ElevenLabs call."""
-
-            def redirect_request(self, req, fp, code, msg, headers, newurl):
-                raise ValueError("ElevenLabs TTS upstream attempted a redirect")
-
         # Buffer the full response before sending first byte.
         # The streaming endpoint is designed for chunked delivery, but urllib's
         # chunked-read path adds per-chunk overhead that dominates short TTS
         # payloads. A hard cap keeps the buffered path bounded even if the
         # upstream misbehaves.
         try:
-            with _tts_open(req, timeout=30, opener_factory=lambda: build_opener(ProxyHandler({}), _NoRedirectElevenlabsHandler())) as resp:
+            with _tts_open(req, timeout=30, opener_factory=lambda: build_opener(ProxyHandler({}), _NoRedirectTtsHandler())) as resp:
                 audio_data = _buffer_tts_audio_response(resp)
         except ValueError:
             logger.warning("ElevenLabs TTS rejected an invalid upstream response", exc_info=True)
@@ -16759,7 +16752,6 @@ def _handle_tts(handler, parsed):
             "voice": oai_voice,
         }).encode("utf-8")
 
-        from urllib.request import Request, build_opener, urlopen as _urlopen
         req = Request(url, data=req_body, headers={
             "Authorization": f"Bearer {api_key}",
             "Content-Type": "application/json",

tests/test_issue3510_elevenlabs_tts.py

@@ -176,7 +176,6 @@ def test_elevenlabs_rejects_oversized_upstream_audio(monkeypatch):
 
 def test_elevenlabs_tts_does_not_follow_upstream_redirect(monkeypatch):
     # Regression proof: a fake redirect response should fail at _tts_open, not be followed.
-    import urllib.request as _ur
     from urllib.request import HTTPRedirectHandler
 
     class _Resp:
@@ -194,9 +193,6 @@ def test_elevenlabs_tts_does_not_follow_upstream_redirect(monkeypatch):
         def __exit__(self, *a):
             return False
 
-    def _fake_urlopen(req, timeout=30):
-        return _Resp()
-
     def _fake_tts_open(req, timeout=30, opener_factory=None, **_kw):
         assert opener_factory is not None
         opener = opener_factory()
@@ -216,12 +212,9 @@ def test_elevenlabs_tts_does_not_follow_upstream_redirect(monkeypatch):
             )
         raise ValueError("redirect blocked 302")
 
-    # Keep the legacy path patched as fake audio to keep base and head checks
-    # behaviorally distinct.
     monkeypatch.setenv("ELEVENLABS_API_KEY", "sk-test")
     import api.config as _cfg
     monkeypatch.setattr(_cfg, "get_config", lambda: {"tts": {"elevenlabs": {"voice_id": "pNInz6obpgDQGcFmaJgB", "model": "eleven_multilingual_v2"}}})
-    monkeypatch.setattr(_ur, "urlopen", _fake_urlopen)
     monkeypatch.setattr(routes, "_tts_open", _fake_tts_open)
     h = _post({"text": "hello world", "engine": "elevenlabs"}, client="9.9.9.6")
     routes._handle_tts(h, None)
@@ -232,7 +225,6 @@ def test_elevenlabs_tts_does_not_follow_upstream_redirect(monkeypatch):
 
 def test_elevenlabs_tts_uses_no_proxy_opener(monkeypatch):
     captured = {}
-    import urllib.request as _ur
 
     class _Resp:
         def __init__(self):
@@ -248,9 +240,6 @@ def test_elevenlabs_tts_uses_no_proxy_opener(monkeypatch):
         def __exit__(self, *a):
             return False
 
-    def _fake_urlopen(req, timeout=30):
-        return _Resp()
-
     def _fake_tts_open(req, timeout=30, opener_factory=None, **_kw):
         captured["opener_factory"] = opener_factory
         opener = opener_factory()
@@ -270,7 +259,6 @@ def test_elevenlabs_tts_uses_no_proxy_opener(monkeypatch):
     monkeypatch.setenv("ELEVENLABS_API_KEY", "sk-test")
     import api.config as _cfg
     monkeypatch.setattr(_cfg, "get_config", lambda: {"tts": {"elevenlabs": {"voice_id": "pNInz6obpgDQGcFmaJgB", "model": "eleven_multilingual_v2"}}})
-    monkeypatch.setattr(_ur, "urlopen", _fake_urlopen)
     monkeypatch.setattr(routes, "_tts_open", _fake_tts_open)
 
     h = _post({"text": "hello world", "engine": "elevenlabs"}, client="9.9.9.7")