diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_audio.py | 52 | ||||
| -rw-r--r-- | tests/test_converter.py | 70 | ||||
| -rw-r--r-- | tests/test_make_audiocpp_server_json.py | 129 | ||||
| -rw-r--r-- | tests/test_tts.py | 65 |
4 files changed, 270 insertions, 46 deletions
diff --git a/tests/test_audio.py b/tests/test_audio.py index c127311..ef5e92a 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -1,11 +1,13 @@ """Tests for audio helpers: speed parameters, chunk cleanup, encoding, command construction, duration verification, and audio concatenation.""" +import io import tempfile import unittest import wave +from contextlib import redirect_stdout from pathlib import Path -from unittest.mock import patch +from unittest.mock import MagicMock, patch from converter import audio from converter import config @@ -470,5 +472,53 @@ class ConcatAudioFilesTests(unittest.TestCase): self.assertIn("ffmpeg", str(ctx.exception)) +class CombineChunksPrintTests(unittest.TestCase): + """Single-request runs (audiocpp whole-chapter) omit the chunks suffix.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._chunks = patch.object(audio, "CHUNKS_FOLDER", Path(self._tmp.name)) + self._chunks.start() + self.addCleanup(self._chunks.stop) + + def _combine(self, total_chunks, chunk_results, intermediate=False): + buf = io.StringIO() + with patch.object(audio.shutil, "which", return_value="/usr/bin/ffmpeg"), \ + patch.object(audio, "atempo_filters", return_value=False), \ + patch.object(audio, "build_concat_command", + return_value=["ffmpeg"]), \ + patch.object(audio.subprocess, "run", + return_value=MagicMock(returncode=0)), \ + patch.object(audio, "probe_duration_ms", return_value=1000), \ + patch.object(audio, "verify_output_duration", + return_value=True), \ + redirect_stdout(buf): + ok = audio.combine_chunks( + total_chunks, Path("out.m4b"), chunk_results, + output_format="m4b", intermediate=intermediate) + self.assertTrue(ok) + return buf.getvalue() + + def test_single_chunk_omits_chunks_suffix(self): + chunk = Path(self._tmp.name) / "chunk_0001.wav" + chunk.write_bytes(b"x") + out = self._combine(1, {1: chunk}) + self.assertEqual(out.strip(), "[INFO] Saved audiobook: out.m4b") + + def test_multi_chunk_keeps_chunks_suffix(self): + chunk = Path(self._tmp.name) / "chunk_0001.wav" + chunk.write_bytes(b"x") + out = self._combine(1, {1: chunk}, intermediate=True) + self.assertEqual(out.strip(), + "[INFO] Saved chapter audio (intermediate): out.m4b") + + def test_partial_chunk_run_keeps_chunks_suffix(self): + chunk = Path(self._tmp.name) / "chunk_0001.wav" + chunk.write_bytes(b"x") + out = self._combine(2, {1: chunk, 2: chunk}) + self.assertEqual(out.strip(), + "[INFO] Saved audiobook: out.m4b (2/2 chunks)") + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_converter.py b/tests/test_converter.py index 8402b30..d525c18 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -1,8 +1,11 @@ """Tests for the audiobook converter orchestration helpers.""" +import io import logging import tempfile +import time import unittest +from contextlib import redirect_stdout from pathlib import Path from unittest.mock import MagicMock, patch @@ -181,6 +184,7 @@ class DebugDumpTests(unittest.TestCase): self._debug_folder.start() self.debug_root = Path(self._tmp.name) self.converter = AudiobookConverter.__new__(AudiobookConverter) + self.converter.client_chunks = True self.converter.tts = MagicMock() def tearDown(self): @@ -323,6 +327,7 @@ class SynthesizeChunkLoggingTests(unittest.TestCase): def setUp(self): self.converter = AudiobookConverter.__new__(AudiobookConverter) + self.converter.client_chunks = True self.converter.tts = MagicMock() def test_failed_chunk_logs_single_error(self): @@ -342,6 +347,71 @@ class SynthesizeChunkLoggingTests(unittest.TestCase): self.assertIn("Chunk 1/1 error: boom", logs.output[0]) +class ServerSideChunkingOutputTests(unittest.TestCase): + """With client-side chunking off (audiocpp default), the console skips + the chunk vocabulary because the whole request is one server call.""" + + def _converter(self, client_chunks: bool): + converter = AudiobookConverter.__new__(AudiobookConverter) + converter.client_chunks = client_chunks + converter.backend = tts.BACKEND_AUDIOCPP + converter.speed = 1.0 + converter.output_format = "mp3" + converter.tts = MagicMock() + converter.tts.process_chunk_with_retry.return_value = "chunk.wav" + return converter + + def test_client_chunking_prints_chunk_progress(self): + buf = io.StringIO() + with redirect_stdout(buf): + self._converter(client_chunks=True)._synthesize_chunks(["Hello."]) + out = buf.getvalue() + self.assertIn("PROCESSING 1 CHUNKS", out) + self.assertIn("Chunk 1/1 completed", out) + self.assertIn("Successful: 1/1", out) + + def test_server_side_chunking_suppresses_chunk_output(self): + buf = io.StringIO() + with redirect_stdout(buf): + self._converter(client_chunks=False)._synthesize_chunks(["Hello."]) + self.assertEqual(buf.getvalue(), "") + + def test_server_side_chunking_suppresses_chapter_chunk_suffix(self): + buf = io.StringIO() + with patch.object(converter_mod.audio, "combine_chunks", return_value=True), \ + redirect_stdout(buf): + ok = self._converter(client_chunks=False)._convert_text( + "Hello world.", Path("out.mp3"), time.time(), chapter=(2, 5)) + self.assertTrue(ok) + out = buf.getvalue() + self.assertIn("Chapter 2/5 converted", out) + self.assertNotIn("chunk", out.lower()) + + def test_single_request_run_notes_long_wait(self): + buf = io.StringIO() + with patch.object(converter_mod.audio, "combine_chunks", return_value=True), \ + redirect_stdout(buf): + ok = self._converter(client_chunks=False)._convert_text( + "Hello world.", Path("out.mp3"), time.time(), chapter=(2, 5)) + self.assertTrue(ok) + out = buf.getvalue() + self.assertIn("Sending the chapter 2/5 to the audio.cpp server as a " + "single request", out) + self.assertIn("expected for this to take a very long time", out) + + def test_client_chunking_run_keeps_chunk_phrasing(self): + buf = io.StringIO() + with patch.object(converter_mod.audio, "combine_chunks", return_value=True), \ + redirect_stdout(buf): + ok = self._converter(client_chunks=True)._convert_text( + "Hello world.", Path("out.mp3"), time.time(), chapter=(2, 5)) + self.assertTrue(ok) + out = buf.getvalue() + self.assertIn("Processing 1 chunks via audio.cpp server", out) + self.assertNotIn("single request", out) + self.assertIn("Chapter 2/5 converted (1/1 chunks)", out) + + class PromptOverwriteTests(unittest.TestCase): def test_single_file_yes(self): with patch("builtins.input", return_value="y"): diff --git a/tests/test_make_audiocpp_server_json.py b/tests/test_make_audiocpp_server_json.py index f9fa794..2bd262a 100644 --- a/tests/test_make_audiocpp_server_json.py +++ b/tests/test_make_audiocpp_server_json.py @@ -304,20 +304,6 @@ class PromptHelperTests(unittest.TestCase): 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) @@ -341,6 +327,46 @@ class PromptHelperTests(unittest.TestCase): "one") +class ResolveWavDirArgTests(unittest.TestCase): + """Path normalization for the required WAV_DIR argument.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_resolves_to_absolute(self): + self.assertEqual(make_server.resolve_wav_dir_arg(str(self.folder)), + self.folder.resolve()) + + def test_strips_surrounding_quotes(self): + quoted = f'"{self.folder}"' + self.assertEqual(make_server.resolve_wav_dir_arg(quoted), + self.folder.resolve()) + + def test_strips_single_quotes(self): + quoted = f"'{self.folder}'" + self.assertEqual(make_server.resolve_wav_dir_arg(quoted), + self.folder.resolve()) + + def test_strips_whitespace(self): + self.assertEqual(make_server.resolve_wav_dir_arg(f" {self.folder} "), + self.folder.resolve()) + + def test_expands_tilde(self): + with patch.object(make_server.os.path, "expanduser", + return_value=str(self.folder)) as mock_expand: + result = make_server.resolve_wav_dir_arg("~/voices") + mock_expand.assert_called_once_with("~/voices") + self.assertEqual(result, self.folder.resolve()) + + def test_trailing_slash_preserved_as_dir(self): + self.assertEqual(make_server.resolve_wav_dir_arg(f"{self.folder}/"), + self.folder.resolve()) + + class MainTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() @@ -370,14 +396,21 @@ class MainTests(unittest.TestCase): return make_server.main() def _defaults(self, models="", host="", port="", backend="", - lazy="", custom_path="", clone_path="", wav_dir="", + lazy="", custom_path="", clone_path="", confirm="y", prefix=()): - # First input selects the model family (default: Qwen3-TTS). + # First input selects the model family (default: Qwen3-TTS). The + # wav directory is always a positional argument, never prompted. return list(prefix) + ["", models, host, port, backend, lazy, - custom_path, clone_path, wav_dir, confirm] + custom_path, clone_path, confirm] + + def test_required_wav_dir_missing_prints_usage(self): + with self.assertRaises(SystemExit) as ctx: + self._run(["--output", str(self.output)], inputs=[]) + self.assertEqual(ctx.exception.code, 2) + self.assertFalse(self.output.exists()) def test_default_run_hosts_both_models(self): - exit_code = self._run(["--output", str(self.output)], + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=self._defaults()) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -395,7 +428,7 @@ class MainTests(unittest.TestCase): self.assertNotIn("voice_presets", data["models"][1]) def test_eof_uses_all_defaults(self): - exit_code = self._run(["--output", str(self.output)]) + exit_code = self._run([str(self.folder), "--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") @@ -427,7 +460,7 @@ class MainTests(unittest.TestCase): def test_custom_only_single_model(self): inputs = ["", "", "", "", "", "", "y"] exit_code = self._run( - ["--output", str(self.output), "--models", "custom"], + [str(self.folder), "--output", str(self.output), "--models", "custom"], inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -437,8 +470,8 @@ class MainTests(unittest.TestCase): 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 = ["", "1", "qwen-clone-2", "", "", "", "", "", "", "y"] + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -449,7 +482,7 @@ class MainTests(unittest.TestCase): 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._run([str(self.folder), "--output", str(self.output)]) self.assertNotEqual(ctx.exception.code, 0) self.assertFalse(self.output.exists()) @@ -457,7 +490,7 @@ class MainTests(unittest.TestCase): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:9999"): inputs = ["", "", "", "y", "", "", "", "", "", "y"] - exit_code = self._run(["--output", str(self.output), + exit_code = self._run([str(self.folder), "--output", str(self.output), "--port", "8080"], inputs=inputs) self.assertEqual(exit_code, 0) @@ -470,7 +503,7 @@ class MainTests(unittest.TestCase): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:9999"): inputs = ["", "", "", "n", "", "", "", "", "", "y"] - exit_code = self._run(["--output", str(self.output), + exit_code = self._run([str(self.folder), "--output", str(self.output), "--port", "8080"], inputs=inputs) self.assertEqual(exit_code, 0) @@ -481,7 +514,7 @@ class MainTests(unittest.TestCase): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:8080"): inputs = self._defaults() - exit_code = self._run(["--output", str(self.output)], + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=inputs) self.assertEqual(exit_code, 0) self.assertEqual(self.fake_config.read_text(encoding="utf-8"), @@ -489,8 +522,8 @@ class MainTests(unittest.TestCase): def test_invalid_menu_choice_reprompts(self): # Family menu default, then an invalid models-menu choice retried. - inputs = ["", "9", "", "", "", "", "", "", "", "", "y"] - exit_code = self._run(["--output", str(self.output)], + inputs = ["", "9", "", "", "", "", "", "", "", "y"] + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -498,14 +531,14 @@ class MainTests(unittest.TestCase): def test_confirm_declined_writes_nothing(self): inputs = self._defaults(confirm="n") - exit_code = self._run(["--output", str(self.output)], + exit_code = self._run([str(self.folder), "--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)], + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=["n"]) self.assertEqual(exit_code, 1) self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), @@ -514,7 +547,7 @@ class MainTests(unittest.TestCase): 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)], + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -523,7 +556,8 @@ class MainTests(unittest.TestCase): 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"], + exit_code = self._run([str(self.folder), "--output", str(self.output), + "--force"], inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -531,13 +565,13 @@ class MainTests(unittest.TestCase): def test_flags_skip_prompts(self): # Family still asked (no --family flag); port 9000 differs from the - # config port so its sync prompt fires; custom/clone paths and the - # wav dir use their defaults. + # config port so its sync prompt fires; custom/clone paths use + # their defaults. exit_code = self._run( - ["--output", str(self.output), "--models", "both", + [str(self.folder), "--output", str(self.output), "--models", "both", "--host", "0.0.0.0", "--port", "9000", "--backend", "cpu", "--lazy-load"], - inputs=["", "y", "", "", "", "y"]) + inputs=["", "y", "", "", "y"]) self.assertEqual(exit_code, 0) self.assertIn('"http://127.0.0.1:9000"', self.fake_config.read_text(encoding="utf-8")) @@ -548,11 +582,15 @@ class MainTests(unittest.TestCase): 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)], + missing = self.folder / "nope" + with self.assertRaises(SystemExit) as ctx, \ + patch("sys.stderr") as mock_stderr: + self._run([str(missing), "--output", str(self.output)], inputs=self._defaults()) self.assertEqual(ctx.exception.code, 2) + shown = "".join(call[0][0] for call in mock_stderr.write.call_args_list) + self.assertIn(f"WAV directory not found: {missing.resolve()}", shown) + self.assertIn("directory containing the .wav", shown) class NonQwenFamilyMainTests(unittest.TestCase): @@ -613,10 +651,10 @@ class NonQwenFamilyMainTests(unittest.TestCase): self.fake_config.read_text(encoding="utf-8")) def test_model_id_sync_declined_keeps_config(self): - # sync declined, host, port, backend, lazy, wav dir skipped, confirm - inputs = ["n", "", "", "", "", "", "y"] + # sync declined, host, port, backend, lazy, confirm + inputs = ["n", "", "", "", "", "y"] exit_code = self._run( - ["--output", str(self.output), "--family", "voxcpm2", + [str(self.folder), "--output", str(self.output), "--family", "voxcpm2", "--model-id", "voxcpm2", "--model-path", "models/VoxCPM2-GGUF"], inputs=inputs) self.assertEqual(exit_code, 0) @@ -628,13 +666,14 @@ class NonQwenFamilyMainTests(unittest.TestCase): def test_no_voice_presets_warns(self): buf = io.StringIO() - # sync accepted, host, port, backend, lazy, wav dir skipped, confirm + # sync accepted, host, port, backend, lazy, confirm with patch.object(sys, "argv", ["make_audiocpp_server_json.py", + str(self.folder), "--output", str(self.output), "--family", "index_tts2", "--model-id", "indextts2", "--model-path", "models/IndexTTS2-GGUF"]), \ - patch("builtins.input", side_effect=["y", "", "", "", "", "", "y"]), \ + patch("builtins.input", side_effect=["y", "", "", "", "", "y"]), \ patch.object(make_server, "transcribe_reference_audio"), \ patch.object(make_server, "whisper_backend_available", return_value="faster_whisper"), \ @@ -649,7 +688,7 @@ class NonQwenFamilyMainTests(unittest.TestCase): def test_models_flag_rejected_for_non_qwen_family(self): with self.assertRaises(SystemExit) as ctx: - self._run(["--output", str(self.output), + self._run([str(self.folder), "--output", str(self.output), "--family", "higgs_audio_tts", "--models", "both"]) self.assertEqual(ctx.exception.code, 2) diff --git a/tests/test_tts.py b/tests/test_tts.py index e2fe921..0b6da02 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -3,8 +3,10 @@ import io import json import tempfile +import time import unittest import wave +from contextlib import redirect_stdout from pathlib import Path from unittest.mock import MagicMock, patch @@ -1016,6 +1018,69 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): self.assertEqual(remaining, ["chunk_0001.wav"]) +class AudioCppHeartbeatTests(unittest.TestCase): + """The heartbeat label drops 'Chunk' when the server does its own + long-form chunking (chunk_text=False, the default).""" + + 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() + + @staticmethod + def _client(chunk_text): + client = AudioCppTTSClient.__new__(AudioCppTTSClient) + client.api_url = "http://127.0.0.1:8080" + client.model_id = config.AUDIOCPP_MODEL_ID + client.preset_mode = False + client.voice = "Vivian" + client.language = "English" + client._seed = -1 + client.chunk_text = chunk_text + client.family = "qwen3_tts" + client.profile = tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE + return client + + @staticmethod + def _wav_bytes(): + buffer = io.BytesIO() + with wave.open(buffer, "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(tts.SAMPLE_RATE) + wav_file.writeframes(b"\x01\x00" * 10) + return buffer.getvalue() + + def _run(self, chunk_text): + client = self._client(chunk_text) + + def slow_request(*_args, **_kwargs): + time.sleep(0.12) + return self._wav_bytes() + + buf = io.StringIO() + with patch.object(config, "HEARTBEAT_INTERVAL_SECONDS", 0.03), \ + patch.object(client, "_request_wav_with_retry", + side_effect=slow_request), \ + redirect_stdout(buf): + result = client.generate_chunk("Hello.", 1) + self.assertTrue(result) + return buf.getvalue() + + def test_server_side_chunking_heartbeat_has_no_chunk_word(self): + out = self._run(chunk_text=False) + self.assertIn("Request still generating", out) + self.assertNotIn("Chunk", out) + + def test_client_side_chunking_heartbeat_keeps_chunk_word(self): + out = self._run(chunk_text=True) + self.assertIn("Chunk 1 still generating", out) + + class AudioCppTTSClientTruncationTests(unittest.TestCase): """Audio far shorter than its text implies fails the request.""" |