From 94ddbb0634022a6e5209b5221c057228ec3d1418 Mon Sep 17 00:00:00 2001 From: historia Date: Wed, 19 Aug 2026 05:13:36 -0400 Subject: feat: reuse one seed per run for consistent voice across chunks --- tests/test_chunking.py | 14 +++++------ tests/test_converter.py | 2 +- tests/test_tts.py | 64 +++++++++++++++++++++++++++++++++++++++++++------ 3 files changed, 65 insertions(+), 15 deletions(-) (limited to 'tests') diff --git a/tests/test_chunking.py b/tests/test_chunking.py index 2062b4e..7f37f80 100644 --- a/tests/test_chunking.py +++ b/tests/test_chunking.py @@ -4,7 +4,7 @@ import unittest from unittest.mock import patch from converter import config -from converter.chunking import split_into_chunks +from converter.chunking import MAX_REQUEST_WORDS, split_into_chunks class ChunkSizeDefaultTests(unittest.TestCase): @@ -14,20 +14,20 @@ class ChunkSizeDefaultTests(unittest.TestCase): chunk size and the hard ceiling must stay well inside that budget.""" def test_default_chunk_size_within_request_ceiling(self): - self.assertLessEqual(config.CHUNK_SIZE_WORDS, config.MAX_REQUEST_WORDS) + self.assertLessEqual(config.CHUNK_SIZE, MAX_REQUEST_WORDS) def test_request_ceiling_within_single_generation_budget(self): - self.assertLessEqual(config.MAX_REQUEST_WORDS, 300) + self.assertLessEqual(MAX_REQUEST_WORDS, 300) def test_sizes_are_positive(self): - self.assertGreaterEqual(config.CHUNK_SIZE_WORDS, 1) - self.assertGreaterEqual(config.MAX_REQUEST_WORDS, 1) + 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): text = " ".join(f"word{i}" for i in range(30)) + "." - with patch.object(config, "MAX_REQUEST_WORDS", 10), \ + 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)) @@ -38,7 +38,7 @@ class RequestCeilingClampTests(unittest.TestCase): f"S{i} " + " ".join(["word"] * 8) + "." for i in range(60)) chunks = split_into_chunks(sentences, max_words=5000) self.assertGreater(len(chunks), 1) - self.assertTrue(all(len(chunk.split()) <= config.MAX_REQUEST_WORDS + self.assertTrue(all(len(chunk.split()) <= MAX_REQUEST_WORDS for chunk in chunks)) diff --git a/tests/test_converter.py b/tests/test_converter.py index 91ab0b5..737fc06 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -130,7 +130,7 @@ class NarratorTagTests(unittest.TestCase): "Vivian") def test_multi_word_display_name_gets_underscores(self): - with patch.object(config, "CUSTOM_VOICE_SPEAKER", "uncle_fu"): + with patch.object(config, "SPEAKER", "uncle_fu"): self.assertEqual(self._converter(tts.VOICE_MODE_CUSTOM)._narrator_tag(), "Uncle_Fu") diff --git a/tests/test_tts.py b/tests/test_tts.py index 26f663e..afaa2e3 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -77,6 +77,54 @@ class QwenTTSClientLanguageTests(unittest.TestCase): mock_connect.assert_not_called() +class SeedResolutionTests(unittest.TestCase): + """CONSTANT_SEED: one seed per run, reused for every request, so the + voice stays consistent across chunk boundaries (the servers + re-sample the voice when the seed changes between generations).""" + + def _make_client(self, **kwargs): + with patch.object(QwenTTSClient, "_connect"): + return QwenTTSClient(**kwargs) + + def test_constant_seed_draws_one_nonnegative_seed(self): + with patch.object(config, "CONSTANT_SEED", True), \ + patch.object(config, "SEED", -1): + client = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM) + self.assertGreaterEqual(client._seed, 0) + + def test_explicit_seed_wins_over_constant_seed(self): + with patch.object(config, "CONSTANT_SEED", True), \ + patch.object(config, "SEED", 42): + client = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM) + self.assertEqual(client._seed, 42) + + def test_without_constant_seed_minus_one_is_forwarded(self): + with patch.object(config, "CONSTANT_SEED", False), \ + patch.object(config, "SEED", -1): + client = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM) + self.assertEqual(client._seed, -1) + + def test_resolved_seed_is_reused_across_requests(self): + api_info = { + "named_endpoints": { + "/run_custom_voice": { + "parameters": [{"parameter_name": "seed"}] + } + } + } + client = QwenTTSClient.__new__(QwenTTSClient) + client.voice_mode = tts.VOICE_MODE_CUSTOM + client.language = "English" + client._seed = 1234 + client.api_info = api_info + client.client = MagicMock() + client._generate_custom_voice("first text") + client._generate_custom_voice("second text") + seeds = [call.kwargs["seed"] + for call in client.client.predict.call_args_list] + self.assertEqual(seeds, [1234, 1234]) + + class PayloadLanguageTests(unittest.TestCase): """The language must reach the API payload in every endpoint variant.""" @@ -92,6 +140,7 @@ class PayloadLanguageTests(unittest.TestCase): client = QwenTTSClient.__new__(QwenTTSClient) client.voice_mode = tts.VOICE_MODE_CUSTOM client.language = language + client._seed = config.SEED client.api_info = api_info if api_info is not None else { "named_endpoints": {endpoint: {}} } @@ -102,6 +151,7 @@ class PayloadLanguageTests(unittest.TestCase): client = QwenTTSClient.__new__(QwenTTSClient) client.voice_mode = tts.VOICE_MODE_CLONE client.language = language + client._seed = config.SEED client.voice_clone_ref_audio = str(self.ref_audio) client.voice_clone_ref_text = ref_text client.clone_api_info = api_info if api_info is not None else { @@ -185,8 +235,8 @@ class FasterTTSClientHealthTests(unittest.TestCase): 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("/")) + self.assertEqual(client.voice, config.FASTER_VOICE) + self.assertEqual(client.api_url, config.FASTER_API_URL.rstrip("/")) def test_explicit_voice_and_url_override_config(self): with patch("converter.tts.urllib.request.urlopen", @@ -241,7 +291,7 @@ 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(config, "MAX_REQUEST_WORDS", 10), \ + with patch.object(tts, "MAX_REQUEST_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) @@ -252,10 +302,10 @@ class FasterTTSClientGenerateTests(unittest.TestCase): 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_WORDS", 4), \ + 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_WORDS no longer drives request size: the hard ceiling + # 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) self.assertIsNotNone(result) @@ -434,7 +484,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(config, "MAX_REQUEST_WORDS", 5), \ + with patch.object(tts, "MAX_REQUEST_WORDS", 5), \ patch.object(client, "_generate_custom_voice", side_effect=[(str(first),), (str(second),), (str(first),)]) as mock_generate: @@ -503,7 +553,7 @@ class FasterModeWiringTests(unittest.TestCase): def test_narrator_tag_falls_back_to_config_voice(self): converter = self._faster_converter() - self.assertEqual(converter._narrator_tag(), config.FASTER_TTS_VOICE) + self.assertEqual(converter._narrator_tag(), config.FASTER_VOICE) def test_banner_and_narrator_work_without_reference_audio(self): converter = self._faster_converter(faster_voice="male_richard_poe") -- cgit v1.2.3