From f3e21980320c1708ff17cc6f699a9aa4758accdf Mon Sep 17 00:00:00 2001 From: historia Date: Tue, 18 Aug 2026 23:27:42 -0400 Subject: feat: support for faster-qwen3-tts backend server --- tests/test_tts.py | 220 +++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 218 insertions(+), 2 deletions(-) (limited to 'tests/test_tts.py') 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() -- cgit v1.2.3