feat: support for faster-qwen3-tts backend server

1 files changed, 218 insertions, 2 deletions

diff --git a/tests/test_tts.py b/tests/test_tts.py
index b605daa..dfeda6f 100644
--- a/tests/test_tts.py
+++ b/tests/test_tts.py
@@ -1,12 +1,15 @@
-"""Tests for the Qwen TTS client wrapper (language handling and payloads)."""
+"""Tests for the TTS client wrappers (language handling and payloads)."""
 
+import json
 import tempfile
 import unittest
+import wave
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 from converter import config
-from converter.tts import QwenTTSClient, normalize_language
+from converter.converter import AudiobookConverter
+from converter.tts import FasterTTSClient, QwenTTSClient, normalize_language
 
 
 class NormalizeLanguageTests(unittest.TestCase):
@@ -151,5 +154,218 @@ class PayloadLanguageTests(unittest.TestCase):
         self.assertNotIn("max_chunk_chars", kwargs)
 
 
+class FasterTTSClientHealthTests(unittest.TestCase):
+    """Connection behavior of the faster-qwen3-tts client."""
+
+    def _health_response(self, model_loaded=True):
+        response = MagicMock()
+        response.__enter__.return_value = response
+        response.read.return_value = json.dumps(
+            {"status": "ok", "model_loaded": model_loaded}).encode("utf-8")
+        return response
+
+    def test_unreachable_server_raises_with_readme_pointer(self):
+        import urllib.error
+        with patch("converter.tts.urllib.request.urlopen",
+                   side_effect=urllib.error.URLError("Connection refused")):
+            with self.assertRaises(RuntimeError) as ctx:
+                FasterTTSClient()
+        message = str(ctx.exception)
+        self.assertIn("not reachable", message)
+        self.assertIn("README", message)
+
+    def test_model_not_loaded_raises(self):
+        with patch("converter.tts.urllib.request.urlopen",
+                   return_value=self._health_response(model_loaded=False)):
+            with self.assertRaises(RuntimeError) as ctx:
+                FasterTTSClient()
+        self.assertIn("not loaded", str(ctx.exception))
+
+    def test_healthy_server_defaults_from_config(self):
+        with patch("converter.tts.urllib.request.urlopen",
+                   return_value=self._health_response()):
+            client = FasterTTSClient()
+        self.assertEqual(client.voice, config.FASTER_TTS_VOICE)
+        self.assertEqual(client.api_url, config.FASTER_TTS_API_URL.rstrip("/"))
+
+    def test_explicit_voice_and_url_override_config(self):
+        with patch("converter.tts.urllib.request.urlopen",
+                   return_value=self._health_response()):
+            client = FasterTTSClient(voice="narrator", api_url="http://10.0.0.5:9000/")
+        self.assertEqual(client.voice, "narrator")
+        self.assertEqual(client.api_url, "http://10.0.0.5:9000")
+
+
+class FasterTTSClientGenerateTests(unittest.TestCase):
+    """Chunk generation: sub-chunking, WAV output, retries, bookkeeping."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self._chunks = patch.object(config, "CHUNKS_FOLDER", Path(self._tmp.name))
+        self._chunks.start()
+        self._sleep = patch("converter.tts.time.sleep")
+        self._sleep.start()
+
+    def tearDown(self):
+        self._sleep.stop()
+        self._chunks.stop()
+        self._tmp.cleanup()
+
+    def _make_client(self):
+        client = FasterTTSClient.__new__(FasterTTSClient)
+        client.voice = "default"
+        client.api_url = "http://127.0.0.1:8000"
+        return client
+
+    def _read_wav(self, path):
+        with wave.open(str(path), "rb") as wav_file:
+            return (wav_file.getnchannels(), wav_file.getsampwidth(),
+                    wav_file.getframerate(), wav_file.readframes(wav_file.getnframes()))
+
+    def test_generate_chunk_writes_valid_wav(self):
+        client = self._make_client()
+        pcm = b"\x01\x00" * 100
+        with patch.object(client, "_request_pcm", return_value=pcm):
+            result = client.generate_chunk("Hello world.", 1)
+        self.assertIsNotNone(result)
+        path = Path(result)
+        self.assertEqual(path.name, "chunk_0001.wav")
+        channels, sampwidth, framerate, frames = self._read_wav(path)
+        self.assertEqual(channels, 1)
+        self.assertEqual(sampwidth, 2)
+        self.assertEqual(framerate, config.FASTER_TTS_SAMPLE_RATE)
+        self.assertEqual(frames, pcm)
+
+    def test_long_text_is_subchunked_and_concatenated_in_order(self):
+        client = self._make_client()
+        sentences = [" ".join(f"word{i}" for i in range(6)) + "." for _ in range(3)]
+        text = " ".join(sentences)
+        pcm_parts = [b"\x01\x00" * 10, b"\x02\x00" * 20, b"\x03\x00" * 30]
+        with patch.object(config, "FASTER_SUBCHUNK_WORDS", 10), \
+                patch.object(client, "_request_pcm", side_effect=pcm_parts) as mock_pcm:
+            result = client.generate_chunk(text, 1)
+        self.assertEqual(mock_pcm.call_count, 3)
+        _, _, _, frames = self._read_wav(Path(result))
+        self.assertEqual(frames, b"".join(pcm_parts))
+
+    def test_stale_chunk_files_are_removed(self):
+        stale = Path(self._tmp.name) / "chunk_0001.mp3"
+        stale.write_bytes(b"old")
+        client = self._make_client()
+        with patch.object(client, "_request_pcm", return_value=b"\x01\x00"):
+            client.generate_chunk("Hello.", 1)
+        remaining = sorted(path.name for path in Path(self._tmp.name).glob("chunk_0001.*"))
+        self.assertEqual(remaining, ["chunk_0001.wav"])
+
+    def test_transient_failure_is_retried(self):
+        client = self._make_client()
+        pcm = b"\x01\x00" * 10
+        with patch.object(client, "_request_pcm",
+                          side_effect=[RuntimeError("boom"), pcm]) as mock_pcm:
+            result = client.generate_chunk("Hello.", 1)
+        self.assertIsNotNone(result)
+        self.assertEqual(mock_pcm.call_count, 2)
+
+    def test_empty_pcm_response_is_treated_as_failure(self):
+        client = self._make_client()
+        pcm = b"\x01\x00" * 10
+
+        def _response(body):
+            response = MagicMock()
+            response.__enter__.return_value = response
+            response.read.return_value = body
+            return response
+
+        with patch("converter.tts.urllib.request.urlopen",
+                   side_effect=[_response(b""), _response(pcm)]) as mock_urlopen:
+            result = client.generate_chunk("Hello.", 1)
+        self.assertIsNotNone(result)
+        self.assertEqual(mock_urlopen.call_count, 2)
+        _, _, _, frames = self._read_wav(Path(result))
+        self.assertEqual(frames, pcm)
+
+    def test_exhausted_subchunk_retries_fail_the_chunk(self):
+        client = self._make_client()
+        with patch.object(client, "_request_pcm",
+                          side_effect=RuntimeError("down")) as mock_pcm:
+            result = client.generate_chunk("Hello.", 1)
+        self.assertIsNone(result)
+        self.assertEqual(mock_pcm.call_count, config.FASTER_SUBCHUNK_RETRIES)
+
+    def test_empty_text_fails_the_chunk(self):
+        client = self._make_client()
+        with patch.object(client, "_request_pcm") as mock_pcm:
+            result = client.generate_chunk("   ", 1)
+        self.assertIsNone(result)
+        mock_pcm.assert_not_called()
+
+    def test_request_payload_includes_voice_text_and_format(self):
+        client = self._make_client()
+        response = MagicMock()
+        response.__enter__.return_value = response
+        response.read.return_value = b"\x01\x00" * 10
+        with patch("converter.tts.urllib.request.urlopen",
+                   return_value=response) as mock_urlopen:
+            pcm = client._request_pcm("Hello world.")
+        self.assertEqual(pcm, b"\x01\x00" * 10)
+        request = mock_urlopen.call_args[0][0]
+        self.assertEqual(request.full_url, "http://127.0.0.1:8000/v1/audio/speech")
+        payload = json.loads(request.data.decode("utf-8"))
+        self.assertEqual(payload["input"], "Hello world.")
+        self.assertEqual(payload["voice"], "default")
+        self.assertEqual(payload["response_format"], "pcm")
+
+
+class FasterModeWiringTests(unittest.TestCase):
+    """AudiobookConverter wiring for the --faster backend."""
+
+    def test_faster_mode_uses_faster_client_without_reference(self):
+        with patch("converter.converter.FasterTTSClient") as mock_faster, \
+                patch("converter.converter.QwenTTSClient") as mock_qwen:
+            AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE,
+                               faster=True, faster_voice="narrator")
+        mock_faster.assert_called_once_with(voice="narrator")
+        mock_qwen.assert_not_called()
+
+    def test_non_faster_clone_mode_still_requires_reference(self):
+        with patch("converter.converter.QwenTTSClient"):
+            with self.assertRaises(ValueError):
+                AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE)
+
+    def test_faster_mode_still_validates_other_settings(self):
+        with patch("converter.converter.FasterTTSClient"):
+            with self.assertRaises(ValueError):
+                AudiobookConverter(faster=True, speed=0)
+            with self.assertRaises(ValueError):
+                AudiobookConverter(faster=True, language="klingon")
+
+    def _faster_converter(self, faster_voice=None):
+        with patch("converter.converter.FasterTTSClient"):
+            return AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE,
+                                      faster=True, faster_voice=faster_voice)
+
+    def test_narrator_tag_uses_faster_voice_name(self):
+        converter = self._faster_converter(faster_voice="male_richard_poe")
+        self.assertEqual(converter._narrator_tag(), "male_richard_poe")
+
+    def test_narrator_tag_falls_back_to_config_voice(self):
+        converter = self._faster_converter()
+        self.assertEqual(converter._narrator_tag(), config.FASTER_TTS_VOICE)
+
+    def test_banner_and_narrator_work_without_reference_audio(self):
+        converter = self._faster_converter(faster_voice="male_richard_poe")
+        converter._print_banner()  # must not raise (regression: Path(None))
+        self.assertIsNone(converter.voice_clone_ref_audio)
+
+    def test_non_faster_narrator_tag_unchanged(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            ref = Path(tmp) / "ref.wav"
+            ref.write_bytes(b"x")
+            with patch("converter.converter.QwenTTSClient"):
+                converter = AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE,
+                                               voice_clone_ref_audio=str(ref))
+        self.assertEqual(converter._narrator_tag(), "ref")
+
+
 if __name__ == "__main__":
     unittest.main()