From 9d4d7ef806c17387af9778725cd65a5e7ed10e39 Mon Sep 17 00:00:00 2001 From: historia Date: Wed, 19 Aug 2026 04:45:28 -0400 Subject: fix: limit chunk size to 250 --- tests/test_tts.py | 155 +++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 154 insertions(+), 1 deletion(-) (limited to 'tests/test_tts.py') diff --git a/tests/test_tts.py b/tests/test_tts.py index 47a8309..26f663e 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -241,13 +241,25 @@ 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, "CHUNK_SIZE_WORDS", 10), \ + with patch.object(config, "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) _, _, _, frames = self._read_wav(Path(result)) self.assertEqual(frames, b"".join(pcm_parts)) + def test_subchunk_size_is_clamped_to_request_ceiling(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_WORDS", 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 + # does, so the whole (8-word) text is one request here. + self.assertEqual(mock_pcm.call_count, 1) + self.assertIsNotNone(result) + def test_stale_chunk_files_are_removed(self): stale = Path(self._tmp.name) / "chunk_0001.mp3" stale.write_bytes(b"old") @@ -316,6 +328,147 @@ class FasterTTSClientGenerateTests(unittest.TestCase): self.assertEqual(payload["response_format"], "pcm") +class TruncationDetectionTests(unittest.TestCase): + """check_for_truncation: servers cut audio silently past their caps, so + grossly short audio must fail the request (then retry / fail visibly).""" + + def test_duration_below_ratio_raises(self): + with self.assertRaises(RuntimeError) as ctx: + tts.check_for_truncation(" ".join(["w"] * 150), 25.0, "Chunk 1") + self.assertIn("truncated", str(ctx.exception)) + + def test_duration_at_ratio_passes(self): + tts.check_for_truncation(" ".join(["w"] * 150), 30.0, "Chunk 1") + + def test_unknown_duration_skips_check(self): + tts.check_for_truncation(" ".join(["w"] * 150), None, "Chunk 1") + + def test_short_requests_skip_check(self): + tts.check_for_truncation(" ".join(["w"] * 9), 0.0, "Chunk 1") + + def test_zero_duration_fails_checked_requests(self): + with self.assertRaises(RuntimeError): + tts.check_for_truncation(" ".join(["w"] * 150), 0.0, "Chunk 1") + + +class FasterTTSClientTruncationTests(unittest.TestCase): + """Audio far shorter than its text implies fails the faster request.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) + self._chunks.start() + + def tearDown(self): + 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 test_truncated_pcm_fails_the_chunk(self): + client = self._make_client() + text = " ".join(f"word{i}" for i in range(12)) + with patch.object(client, "_request_pcm", return_value=b"\x01\x00" * 24), \ + self.assertLogs("converter.tts", level="ERROR") as logs: + result = client.generate_chunk(text, 1) + self.assertIsNone(result) + self.assertTrue(any("truncated" in line for line in logs.output)) + + def test_full_length_pcm_passes(self): + client = self._make_client() + text = " ".join(f"word{i}" for i in range(12)) + # 12 words -> expected 4.8s, half is 2.4s -> 2.5s of audio passes. + pcm = b"\x01\x00" * int(2.5 * tts.SAMPLE_RATE) + with patch.object(client, "_request_pcm", return_value=pcm): + result = client.generate_chunk(text, 1) + self.assertIsNotNone(result) + + +class QwenTTSClientGenerateTests(unittest.TestCase): + """Qwen chunk generation: sub-request splitting and concatenation.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) + self._chunks.start() + + def tearDown(self): + self._chunks.stop() + self._tmp.cleanup() + + def _make_client(self): + client = QwenTTSClient.__new__(QwenTTSClient) + client.voice_mode = tts.VOICE_MODE_CUSTOM + return client + + @staticmethod + def _write_wav(path: Path, frames: bytes) -> Path: + with wave.open(str(path), "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(tts.SAMPLE_RATE) + wav_file.writeframes(frames) + return path + + def _read_wav_frames(self, path: Path) -> bytes: + with wave.open(str(path), "rb") as wav_file: + return wav_file.readframes(wav_file.getnframes()) + + def test_single_request_copies_audio(self): + client = self._make_client() + source = self._write_wav(Path(self._tmp.name) / "server.wav", b"\x01\x00" * 50) + with patch.object(client, "_generate_custom_voice", + return_value=(str(source),)) as mock_generate: + result = client.generate_chunk("Hello world.", 1) + mock_generate.assert_called_once_with("Hello world.") + path = Path(result) + self.assertEqual(path.name, "chunk_0001.wav") + self.assertEqual(self._read_wav_frames(path), b"\x01\x00" * 50) + + def test_oversized_input_is_split_and_concatenated_in_order(self): + client = self._make_client() + 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), \ + patch.object(client, "_generate_custom_voice", + side_effect=[(str(first),), (str(second),), + (str(first),)]) as mock_generate: + result = client.generate_chunk(text, 1) + self.assertEqual(mock_generate.call_count, 3) + path = Path(result) + self.assertEqual(path.name, "chunk_0001.wav") + self.assertEqual(self._read_wav_frames(path), + b"\x01\x00" * 10 + b"\x02\x00" * 20 + b"\x01\x00" * 10) + for call in mock_generate.call_args_list: + self.assertLessEqual(len(call[0][0].split()), 5) + + def test_truncated_sub_request_fails_the_chunk(self): + client = self._make_client() + # 0.1s of audio for 12 words (expected >= 2.4s). + source = self._write_wav(Path(self._tmp.name) / "short.wav", + b"\x01\x00" * int(0.1 * tts.SAMPLE_RATE)) + text = " ".join(f"word{i}" for i in range(12)) + with patch.object(client, "_generate_custom_voice", + return_value=(str(source),)) as mock_generate, \ + self.assertLogs("converter.tts", level="ERROR") as logs: + result = client.generate_chunk(text, 1) + self.assertIsNone(result) + self.assertEqual(mock_generate.call_count, 1) + self.assertTrue(any("truncated" in line for line in logs.output)) + + def test_empty_text_fails_the_chunk(self): + client = self._make_client() + with patch.object(client, "_generate_custom_voice") as mock_generate: + result = client.generate_chunk(" ", 1) + self.assertIsNone(result) + mock_generate.assert_not_called() + + class FasterModeWiringTests(unittest.TestCase): """AudiobookConverter wiring for the --faster backend.""" -- cgit v1.2.3