diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_chunking.py | 48 | ||||
| -rw-r--r-- | tests/test_make_audiocpp_server_json.py | 457 | ||||
| -rw-r--r-- | tests/test_make_faster_voices_json.py (renamed from tests/test_make_voices.py) | 4 | ||||
| -rw-r--r-- | tests/test_tts.py | 27 |
4 files changed, 497 insertions, 39 deletions
diff --git a/tests/test_chunking.py b/tests/test_chunking.py index 7f37f80..3cd926b 100644 --- a/tests/test_chunking.py +++ b/tests/test_chunking.py @@ -4,42 +4,41 @@ import unittest from unittest.mock import patch from converter import config -from converter.chunking import MAX_REQUEST_WORDS, split_into_chunks +from converter.chunking import split_into_chunks class ChunkSizeDefaultTests(unittest.TestCase): - """Guard the request-size settings: each API call is one model - generation, and the servers silently truncate audio past their caps - (~2.5 min faster backend, ~11 min Gradio demo), so both the default - chunk size and the hard ceiling must stay well inside that budget.""" + """Guard the request-size setting: each API call is one model + generation, and the servers silently truncate audio when a single + generation runs too long (~2.5 min faster backend, ~11 min Gradio + demo), so the default chunk size must stay well inside that budget. + There is no hard ceiling beyond CHUNK_SIZE; users raising it accept + the truncation risk themselves.""" - def test_default_chunk_size_within_request_ceiling(self): - self.assertLessEqual(config.CHUNK_SIZE, MAX_REQUEST_WORDS) + def test_default_chunk_size_within_single_generation_budget(self): + self.assertLessEqual(config.CHUNK_SIZE, 300) - def test_request_ceiling_within_single_generation_budget(self): - self.assertLessEqual(MAX_REQUEST_WORDS, 300) - - def test_sizes_are_positive(self): + def test_default_chunk_size_is_positive(self): self.assertGreaterEqual(config.CHUNK_SIZE, 1) - self.assertGreaterEqual(MAX_REQUEST_WORDS, 1) -class RequestCeilingClampTests(unittest.TestCase): - def test_oversized_chunk_size_is_clamped_with_warning(self): +class RequestSizeTests(unittest.TestCase): + def test_oversized_chunk_size_is_honored(self): + # No clamping: whatever size is configured (or requested) is used. text = " ".join(f"word{i}" for i in range(30)) + "." - with patch("converter.chunking.MAX_REQUEST_WORDS", 10), \ - self.assertLogs("converter.chunking", level="WARNING") as logs: - chunks = split_into_chunks(text, max_words=5000) - self.assertTrue(all(len(chunk.split()) <= 10 for chunk in chunks)) - self.assertIn("clamped", " ".join(logs.output)) + chunks = split_into_chunks(text, max_words=5000) + self.assertEqual(len(chunks), 1) + self.assertEqual(len(chunks[0].split()), 30) - def test_default_ceiling_clamps_realistic_configuration(self): + def test_default_uses_runtime_config_chunk_size(self): + # The default resolves config.CHUNK_SIZE at call time, so + # patching the config changes the default split size. sentences = " ".join( f"S{i} " + " ".join(["word"] * 8) + "." for i in range(60)) - chunks = split_into_chunks(sentences, max_words=5000) + with patch.object(config, "CHUNK_SIZE", 120): + chunks = split_into_chunks(sentences) self.assertGreater(len(chunks), 1) - self.assertTrue(all(len(chunk.split()) <= MAX_REQUEST_WORDS - for chunk in chunks)) + self.assertTrue(all(len(chunk.split()) <= 120 for chunk in chunks)) class SplitIntoChunksTests(unittest.TestCase): @@ -94,8 +93,7 @@ class SplitIntoChunksTests(unittest.TestCase): def test_single_oversized_sentence_is_word_split(self): # A punctuation-free sentence longer than the limit is split at word - # boundaries: the request-size ceiling is a hard limit because the - # TTS servers silently truncate oversized generations. + # boundaries so no single request exceeds the configured size. sentence = " ".join(["word"] * 30) + "." chunks = split_into_chunks(sentence, max_words=10) self.assertGreater(len(chunks), 1) diff --git a/tests/test_make_audiocpp_server_json.py b/tests/test_make_audiocpp_server_json.py new file mode 100644 index 0000000..63260ca --- /dev/null +++ b/tests/test_make_audiocpp_server_json.py @@ -0,0 +1,457 @@ +"""Tests for the audio.cpp server.json generator tool.""" + +import json +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from converter import config +from tools import make_audiocpp_server_json as make_server + +FAKE_CONFIG = ( + 'LANGUAGE = "English"\n' + "\n" + 'AUDIOCPP_API_URL = "http://127.0.0.1:9999" # audio.cpp audiocpp_server\n' + "\n" + "CHUNK_SIZE = 250\n" +) + + +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_server.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_server.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_server.find_wav_files(self.folder), []) + + +class ConfigPortTests(unittest.TestCase): + def test_port_parsed_from_config_url(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:8080"): + self.assertEqual(make_server.config_port(), 8080) + + def test_missing_port_falls_back(self): + with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1"): + self.assertEqual(make_server.config_port(), + make_server.FALLBACK_PORT) + + def test_invalid_url_falls_back(self): + with patch.object(config, "AUDIOCPP_API_URL", "not a url"): + self.assertEqual(make_server.config_port(), + make_server.FALLBACK_PORT) + + def test_url_with_port_replaces_port(self): + self.assertEqual( + make_server._url_with_port("http://127.0.0.1:8080", 9000), + "http://127.0.0.1:9000") + + def test_url_without_port_adds_port(self): + self.assertEqual( + make_server._url_with_port("http://localhost", 8080), + "http://localhost:8080") + + +class UpdateConfigPortTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.config_path = Path(self._tmp.name) / "config.py" + self.config_path.write_text(FAKE_CONFIG, encoding="utf-8") + + def tearDown(self): + self._tmp.cleanup() + + def test_rewrites_port_preserving_comment(self): + changed = make_server.update_config_api_url_port( + 8080, config_path=self.config_path) + self.assertTrue(changed) + text = self.config_path.read_text(encoding="utf-8") + self.assertIn( + 'AUDIOCPP_API_URL = "http://127.0.0.1:8080" # audio.cpp audiocpp_server', + text) + self.assertIn('LANGUAGE = "English"', text) + self.assertIn("CHUNK_SIZE = 250", text) + + def test_returns_false_when_no_url_line(self): + path = Path(self._tmp.name) / "other.py" + path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") + self.assertFalse(make_server.update_config_api_url_port( + 8080, config_path=path)) + + def test_returns_false_when_port_unchanged(self): + self.assertFalse(make_server.update_config_api_url_port( + 9999, config_path=self.config_path)) + self.assertEqual(self.config_path.read_text(encoding="utf-8"), + FAKE_CONFIG) + + def test_returns_false_when_file_missing(self): + self.assertFalse(make_server.update_config_api_url_port( + 8080, config_path=Path(self._tmp.name) / "nope.py")) + + +class BuildVoicePresetsTests(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_presets_named_after_basenames_with_absolute_paths(self): + transcripts = {str(self.narrator): "First transcript.", + str(self.other): "Second transcript."} + with patch.object(make_server, "transcribe_reference_audio", + side_effect=lambda path, model_name="base": + transcripts[path]): + presets = make_server.build_voice_presets( + [self.narrator, self.other], "base") + self.assertEqual(list(presets), ["narrator", "other"]) + self.assertEqual(presets["narrator"]["reference_text"], + "First transcript.") + self.assertEqual(Path(presets["narrator"]["voice_ref"]), + self.narrator.resolve()) + + def test_failed_transcription_keeps_entry_with_empty_text(self): + with patch.object(make_server, "transcribe_reference_audio", + return_value=None): + presets = make_server.build_voice_presets([self.narrator], "base") + self.assertEqual(presets["narrator"]["reference_text"], "") + + def test_whisper_model_name_is_passed_through(self): + with patch.object(make_server, "transcribe_reference_audio", + return_value="text") as mock_transcribe: + make_server.build_voice_presets([self.narrator], "large-v3") + self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], + "large-v3") + + +class BuildServerConfigTests(unittest.TestCase): + def test_both_models_with_presets(self): + presets = {"narrator": {"voice_ref": "/x.wav", + "reference_text": "hi"}} + server_config = make_server.build_server_config( + host="127.0.0.1", port=8080, backend="cuda", lazy_load=False, + include_custom=True, include_clone=True, + custom_voice_id="qwen", clone_model_id="qwen-clone", + custom_voice_path="models/custom", base_path="models/base", + voice_presets=presets) + self.assertEqual(server_config["host"], "127.0.0.1") + self.assertEqual(server_config["port"], 8080) + self.assertEqual(server_config["backend"], "cuda") + self.assertFalse(server_config["lazy_load"]) + self.assertEqual([model["id"] for model in server_config["models"]], + ["qwen", "qwen-clone"]) + custom_entry, clone_entry = server_config["models"] + self.assertNotIn("voice_presets", custom_entry) + self.assertEqual(custom_entry["family"], "qwen3_tts") + self.assertEqual(custom_entry["path"], "models/custom") + self.assertEqual(clone_entry["path"], "models/base") + self.assertEqual(clone_entry["voice_presets"], presets) + + def test_custom_only_has_single_entry(self): + server_config = make_server.build_server_config( + host="0.0.0.0", port=9000, backend="cpu", lazy_load=True, + include_custom=True, include_clone=False, + custom_voice_id="qwen", clone_model_id="qwen-clone", + custom_voice_path="models/custom", base_path=None, + voice_presets={}) + self.assertEqual(len(server_config["models"]), 1) + self.assertEqual(server_config["models"][0]["id"], "qwen") + self.assertNotIn("voice_presets", server_config["models"][0]) + + def test_clone_only_without_presets_omits_key(self): + server_config = make_server.build_server_config( + host="127.0.0.1", port=8080, backend="vulkan", lazy_load=False, + include_custom=False, include_clone=True, + custom_voice_id="qwen", clone_model_id="qwen-clone", + custom_voice_path=None, base_path="models/base", + voice_presets={}) + self.assertEqual(len(server_config["models"]), 1) + self.assertEqual(server_config["models"][0]["id"], "qwen-clone") + self.assertNotIn("voice_presets", server_config["models"][0]) + + +class PromptHelperTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_ask_wav_dir_reprompts_until_valid(self): + with patch("builtins.input", + side_effect=[str(self.folder / "nope"), + str(self.folder)]): + self.assertEqual(make_server.ask_wav_dir(), self.folder) + + def test_ask_wav_dir_empty_skips(self): + with patch("builtins.input", return_value=""): + self.assertIsNone(make_server.ask_wav_dir()) + + def test_ask_wav_dir_eof_returns_none(self): + with patch("builtins.input", side_effect=EOFError): + self.assertIsNone(make_server.ask_wav_dir()) + + def test_ask_port_reprompts_until_valid(self): + with patch("builtins.input", side_effect=["abc", "8081"]): + self.assertEqual(make_server.ask_port(8080), 8081) + + def test_ask_port_eof_returns_default(self): + with patch("builtins.input", side_effect=EOFError): + self.assertEqual(make_server.ask_port(8080), 8080) + + def test_ask_menu_reprompts_until_valid(self): + options = [("One", "one"), ("Two", "two")] + with patch("builtins.input", side_effect=["9", "2"]): + self.assertEqual( + make_server.ask_menu("Pick:", options, default_index=1), + "two") + + def test_ask_menu_eof_returns_default(self): + options = [("One", "one"), ("Two", "two")] + with patch("builtins.input", side_effect=EOFError): + self.assertEqual( + make_server.ask_menu("Pick:", options, default_index=1), + "one") + + +class MainTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self.output = self.folder / "server.json" + # Isolate the config.py rewrite target so no test can ever + # modify the repository's real converter/config.py. + self.fake_config = self.folder / "config.py" + self.fake_config.write_text(FAKE_CONFIG, encoding="utf-8") + patcher = patch.object(make_server, "CONFIG_PATH", self.fake_config) + patcher.start() + self.addCleanup(patcher.stop) + + def tearDown(self): + self._tmp.cleanup() + + def _run(self, argv, inputs=None, transcribe=None): + argv = ["make_audiocpp_server_json.py"] + argv + input_effect = inputs if inputs is not None else EOFError + transcribe_effect = transcribe if transcribe is not None else MagicMock() + with patch.object(sys, "argv", argv), \ + patch("builtins.input", side_effect=input_effect), \ + patch.object(make_server, "transcribe_reference_audio", + side_effect=transcribe_effect): + return make_server.main() + + def _defaults(self, models="", host="", port="", backend="", + lazy="", custom_path="", clone_path="", wav_dir="", + confirm="y", prefix=()): + return list(prefix) + [models, host, port, backend, lazy, + custom_path, clone_path, wav_dir, confirm] + + def test_default_run_hosts_both_models(self): + exit_code = self._run(["--output", str(self.output)], + inputs=self._defaults()) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["host"], "127.0.0.1") + self.assertEqual(data["port"], make_server.config_port()) + self.assertEqual(data["backend"], "cuda") + self.assertFalse(data["lazy_load"]) + self.assertEqual( + [model["id"] for model in data["models"]], + [config.AUDIOCPP_MODEL_ID, config.AUDIOCPP_CLONE_MODEL_ID]) + self.assertEqual( + [model["path"] for model in data["models"]], + [make_server.DEFAULT_CUSTOM_VOICE_PATH, + make_server.DEFAULT_BASE_PATH]) + self.assertNotIn("voice_presets", data["models"][1]) + + def test_eof_uses_all_defaults(self): + exit_code = self._run(["--output", str(self.output)]) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["host"], "127.0.0.1") + self.assertEqual(data["port"], make_server.config_port()) + self.assertEqual(data["backend"], "cuda") + self.assertFalse(data["lazy_load"]) + self.assertEqual(len(data["models"]), 2) + + def test_clone_only_with_positional_wav_dir(self): + (self.folder / "narrator.wav").write_bytes(b"x") + (self.folder / "alpha.wav").write_bytes(b"x") + inputs = ["3", "", "", "", "", "", "y"] + exit_code = self._run( + [str(self.folder), "--output", str(self.output)], + inputs=inputs, + transcribe=lambda path, model_name="base": + f"transcript of {Path(path).name}") + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 1) + clone_entry = data["models"][0] + self.assertEqual(clone_entry["id"], config.AUDIOCPP_CLONE_MODEL_ID) + self.assertEqual(sorted(clone_entry["voice_presets"]), + ["alpha", "narrator"]) + self.assertEqual(clone_entry["voice_presets"]["narrator"], + {"voice_ref": str((self.folder / "narrator.wav").resolve()), + "reference_text": "transcript of narrator.wav"}) + + def test_custom_only_single_model(self): + inputs = ["", "", "", "", "", "y"] + exit_code = self._run( + ["--output", str(self.output), "--models", "custom"], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual([model["id"] for model in data["models"]], + [config.AUDIOCPP_MODEL_ID]) + + def test_duplicate_ids_prompt_for_distinct_clone_id(self): + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen"): + inputs = ["1", "qwen-clone-2", "", "", "", "", "", "", "", "y"] + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual([model["id"] for model in data["models"]], + ["qwen", "qwen-clone-2"]) + + def test_duplicate_ids_eof_exits(self): + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen"): + with self.assertRaises(SystemExit) as ctx: + self._run(["--output", str(self.output)]) + self.assertNotEqual(ctx.exception.code, 0) + self.assertFalse(self.output.exists()) + + def test_port_sync_accepted_updates_config(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:9999"): + inputs = ["", "", "y", "", "", "", "", "", "y"] + exit_code = self._run(["--output", str(self.output), + "--port", "8080"], + inputs=inputs) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:8080"', + self.fake_config.read_text(encoding="utf-8")) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["port"], 8080) + + def test_port_sync_declined_keeps_config(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:9999"): + inputs = ["", "", "n", "", "", "", "", "", "y"] + exit_code = self._run(["--output", str(self.output), + "--port", "8080"], + inputs=inputs) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:9999"', + self.fake_config.read_text(encoding="utf-8")) + + def test_matching_port_does_not_prompt_for_sync(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:8080"): + inputs = self._defaults() + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 0) + self.assertEqual(self.fake_config.read_text(encoding="utf-8"), + FAKE_CONFIG) + + def test_invalid_menu_choice_reprompts(self): + inputs = ["9", "", "", "", "", "", "", "", "", "y"] + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 2) + + def test_confirm_declined_writes_nothing(self): + inputs = self._defaults(confirm="n") + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 1) + self.assertFalse(self.output.exists()) + + def test_existing_output_declined_keeps_file(self): + self.output.write_text('{"old": true}', encoding="utf-8") + exit_code = self._run(["--output", str(self.output)], + inputs=["n"]) + self.assertEqual(exit_code, 1) + 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") + inputs = ["y"] + self._defaults() + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 2) + + def test_force_overwrites_without_prompt(self): + self.output.write_text('{"old": true}', encoding="utf-8") + inputs = self._defaults() + exit_code = self._run(["--output", str(self.output), "--force"], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 2) + + def test_flags_skip_prompts(self): + exit_code = self._run( + ["--output", str(self.output), "--models", "both", + "--host", "0.0.0.0", "--port", "9000", "--backend", "cpu", + "--lazy-load"], + inputs=["y", "", "", "", "y"]) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:9000"', + self.fake_config.read_text(encoding="utf-8")) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["host"], "0.0.0.0") + self.assertEqual(data["port"], 9000) + self.assertEqual(data["backend"], "cpu") + self.assertTrue(data["lazy_load"]) + + def test_missing_positional_wav_dir_errors(self): + with self.assertRaises(SystemExit) as ctx: + self._run([str(self.folder / "nope"), + "--output", str(self.output)], + inputs=self._defaults()) + self.assertEqual(ctx.exception.code, 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_make_voices.py b/tests/test_make_faster_voices_json.py index cf76a57..f26e645 100644 --- a/tests/test_make_voices.py +++ b/tests/test_make_faster_voices_json.py @@ -1,4 +1,4 @@ -"""Tests for the voices.json generator tool.""" +"""Tests for the faster-qwen3-tts voices.json generator tool.""" import json import sys @@ -7,7 +7,7 @@ import unittest from pathlib import Path from unittest.mock import patch -from tools import make_voices +from tools import make_faster_voices_json as make_voices class FindWavFilesTests(unittest.TestCase): diff --git a/tests/test_tts.py b/tests/test_tts.py index 813332d..f2dda0e 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -296,23 +296,23 @@ class FasterTTSClientGenerateTests(unittest.TestCase): 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(tts, "MAX_REQUEST_WORDS", 10), \ + with patch.object(config, "CHUNK_SIZE", 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_subchunk_size_is_clamped_to_request_ceiling(self): + def test_subchunk_size_follows_config_chunk_size(self): client = self._make_client() text = " ".join(f"word{i}" for i in range(8)) pcm = b"\x01\x00" * 10 with patch.object(config, "CHUNK_SIZE", 4), \ patch.object(client, "_request_pcm", return_value=pcm) as mock_pcm: result = client.generate_chunk(text, 1) - # CHUNK_SIZE no longer drives request size: the hard ceiling - # does, so the whole (8-word) text is one request here. - self.assertEqual(mock_pcm.call_count, 1) + # The sub-chunk split follows config.CHUNK_SIZE, so the whole + # (8-word) text needs two 4-word requests here. + self.assertEqual(mock_pcm.call_count, 2) self.assertIsNotNone(result) def test_stale_chunk_files_are_removed(self): @@ -489,7 +489,7 @@ class QwenTTSClientGenerateTests(unittest.TestCase): first = self._write_wav(Path(self._tmp.name) / "one.wav", b"\x01\x00" * 10) second = self._write_wav(Path(self._tmp.name) / "two.wav", b"\x02\x00" * 20) text = " ".join(f"word{i}" for i in range(12)) - with patch.object(tts, "MAX_REQUEST_WORDS", 5), \ + with patch.object(config, "CHUNK_SIZE", 5), \ patch.object(client, "_generate_custom_voice", side_effect=[(str(first),), (str(second),), (str(first),)]) as mock_generate: @@ -627,26 +627,29 @@ class AudioCppTTSClientHealthTests(unittest.TestCase): self.assertGreaterEqual(client._seed, 0) def test_preset_mode_routes_to_clone_model_when_configured(self): - with patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): client = self._client( voice="narrator", models={"data": [{"id": "qwen3-tts"}, {"id": "qwen3-tts-clone"}]}) self.assertEqual(client.model_id, "qwen3-tts-clone") def test_preset_mode_falls_back_when_clone_model_not_on_server(self): - with patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"), \ + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"), \ self.assertLogs("converter.tts", level="WARNING") as logs: client = self._client( voice="narrator", models={"data": [{"id": "qwen3-tts"}, {"id": "pocket-tts"}]}) - self.assertEqual(client.model_id, config.AUDIOCPP_MODEL_ID) + self.assertEqual(client.model_id, "qwen3-tts") self.assertTrue(any("qwen3-tts-clone" in line for line in logs.output)) def test_clone_model_id_ignored_for_speaker_mode(self): - with patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): client = self._client( models={"data": [{"id": "qwen3-tts"}, {"id": "qwen3-tts-clone"}]}) - self.assertEqual(client.model_id, config.AUDIOCPP_MODEL_ID) + self.assertEqual(client.model_id, "qwen3-tts") def test_clone_model_id_equal_to_primary_is_noop(self): with patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", @@ -787,7 +790,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): parts = [self._wav_bytes(b"\x01\x00" * 10), self._wav_bytes(b"\x02\x00" * 20), self._wav_bytes(b"\x03\x00" * 30)] - with patch.object(tts, "MAX_REQUEST_WORDS", 10), \ + with patch.object(config, "CHUNK_SIZE", 10), \ patch.object(client, "_request_wav", side_effect=parts) as mock_request: result = client.generate_chunk(text, 1) self.assertEqual(mock_request.call_count, 3) |