diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_converter.py | 4 | ||||
| -rw-r--r-- | tests/test_make_voices.py | 169 | ||||
| -rw-r--r-- | tests/test_tts.py | 220 |
3 files changed, 391 insertions, 2 deletions
diff --git a/tests/test_converter.py b/tests/test_converter.py index ae307a2..48dc162 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -114,6 +114,8 @@ class NarratorTagTests(unittest.TestCase): converter = AudiobookConverter.__new__(AudiobookConverter) converter.voice_mode = voice_mode converter.voice_clone_ref_audio = ref_audio + converter.faster = False + converter.faster_voice = None return converter def test_custom_voice_uses_speaker_display_name(self): @@ -189,6 +191,8 @@ class RunOverwritePromptTests(unittest.TestCase): self.converter = AudiobookConverter.__new__(AudiobookConverter) self.converter.voice_mode = config.VOICE_MODE_CUSTOM self.converter.voice_clone_ref_audio = None + self.converter.faster = False + self.converter.faster_voice = None self.converter.speed = 1.0 self.converter.single_file = False self.converter.output_format = "mp3" diff --git a/tests/test_make_voices.py b/tests/test_make_voices.py new file mode 100644 index 0000000..cf76a57 --- /dev/null +++ b/tests/test_make_voices.py @@ -0,0 +1,169 @@ +"""Tests for the voices.json generator tool.""" + +import json +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from tools import make_voices + + +class FindWavFilesTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def _touch(self, name): + path = self.folder / name + path.write_bytes(b"x") + return path + + def test_finds_only_wavs_case_insensitive(self): + self._touch("b.wav") + self._touch("a.WAV") + self._touch("notes.txt") + (self.folder / "sub").mkdir() + (self.folder / "sub" / "c.wav").write_bytes(b"x") + names = [path.name for path in make_voices.find_wav_files(self.folder)] + self.assertEqual(names, ["a.WAV", "b.wav"]) + + def test_sorted_alphabetically_case_insensitive(self): + for name in ("Zed.wav", "alpha.wav", "Beta.wav"): + self._touch(name) + names = [path.name for path in make_voices.find_wav_files(self.folder)] + self.assertEqual(names, ["alpha.wav", "Beta.wav", "Zed.wav"]) + + def test_empty_directory_returns_empty_list(self): + self.assertEqual(make_voices.find_wav_files(self.folder), []) + + +class BuildVoicesTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self.narrator = self.folder / "narrator.wav" + self.narrator.write_bytes(b"x") + self.other = self.folder / "other.wav" + self.other.write_bytes(b"x") + + def tearDown(self): + self._tmp.cleanup() + + def test_voices_named_after_basenames_with_absolute_paths(self): + transcripts = {str(self.narrator): "First transcript.", + str(self.other): "Second transcript."} + with patch.object(make_voices, "transcribe_reference_audio", + side_effect=lambda path, model_name="base": transcripts[path]): + voices = make_voices.build_voices([self.narrator, self.other], + "English", "base") + self.assertEqual(list(voices), ["narrator", "other"]) + self.assertEqual(voices["narrator"]["ref_text"], "First transcript.") + self.assertEqual(voices["narrator"]["language"], "English") + self.assertTrue(Path(voices["narrator"]["ref_audio"]).is_absolute()) + self.assertEqual(Path(voices["narrator"]["ref_audio"]), self.narrator.resolve()) + + def test_failed_transcription_keeps_entry_with_empty_text(self): + with patch.object(make_voices, "transcribe_reference_audio", + return_value=None): + voices = make_voices.build_voices([self.narrator], "English", "base") + self.assertEqual(voices["narrator"]["ref_text"], "") + + def test_whisper_model_name_is_passed_through(self): + with patch.object(make_voices, "transcribe_reference_audio", + return_value="text") as mock_transcribe: + make_voices.build_voices([self.narrator], "English", "large-v3") + self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], "large-v3") + + +class MainTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + (self.folder / "narrator.wav").write_bytes(b"x") + (self.folder / "alpha.wav").write_bytes(b"x") + self.output = self.folder / "voices.json" + + def tearDown(self): + self._tmp.cleanup() + + def _run(self, argv): + with patch.object(sys, "argv", ["make_voices.py"] + argv): + return make_voices.main() + + def test_writes_json_with_alphabetical_voice_order(self): + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"): + exit_code = self._run([str(self.folder)]) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(list(data), ["alpha", "narrator"]) + self.assertEqual(data["alpha"]["ref_text"], "hello") + self.assertEqual(data["alpha"]["language"], "English") + + def test_custom_output_path(self): + custom = Path(self._tmp.name) / "custom.json" + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"): + self._run([str(self.folder), "--output", str(custom)]) + self.assertTrue(custom.exists()) + self.assertFalse(self.output.exists()) + + def test_invalid_language_errors_before_work(self): + with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe: + with self.assertRaises(SystemExit) as ctx: + self._run([str(self.folder), "--language", "klingon"]) + self.assertEqual(ctx.exception.code, 2) + mock_transcribe.assert_not_called() + + def test_missing_input_dir_errors(self): + with self.assertRaises(SystemExit) as ctx: + self._run([str(self.folder / "nope")]) + self.assertEqual(ctx.exception.code, 2) + + def test_no_wav_files_errors(self): + empty = Path(tempfile.mkdtemp()) + try: + with self.assertRaises(SystemExit) as ctx: + self._run([str(empty)]) + self.assertEqual(ctx.exception.code, 2) + finally: + empty.rmdir() + + def test_existing_output_declined_keeps_file(self): + self.output.write_text('{"old": true}', encoding="utf-8") + with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe, \ + patch("builtins.input", return_value="n"): + exit_code = self._run([str(self.folder)]) + self.assertEqual(exit_code, 1) + mock_transcribe.assert_not_called() + self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), + {"old": True}) + + def test_existing_output_accepted_overwrites(self): + self.output.write_text('{"old": true}', encoding="utf-8") + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"), \ + patch("builtins.input", return_value="y"): + exit_code = self._run([str(self.folder)]) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(list(data), ["alpha", "narrator"]) + + def test_force_overwrites_without_prompt(self): + self.output.write_text('{"old": true}', encoding="utf-8") + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"), \ + patch("builtins.input", side_effect=AssertionError("prompted")): + exit_code = self._run([str(self.folder), "--force"]) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(list(data), ["alpha", "narrator"]) + + +if __name__ == "__main__": + unittest.main() 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() |