From f00249db9d1ea051d29aa1bcca869fc4b88e83eb Mon Sep 17 00:00:00 2001 From: historia Date: Mon, 24 Aug 2026 02:59:26 -0400 Subject: refactor: add app directory, dir structure change --- app/tests/__init__.py | 0 app/tests/cover_test.png | Bin 0 -> 6801 bytes app/tests/gen_test_cover.py | 8 + app/tests/test_audio.py | 524 ++++++++++++ app/tests/test_backends.py | 178 +++++ app/tests/test_backends_audiocpp.py | 1062 ++++++++++++++++++++++++ app/tests/test_backends_envs.py | 204 +++++ app/tests/test_backends_faster.py | 172 ++++ app/tests/test_backends_servers.py | 146 ++++ app/tests/test_chunking.py | 118 +++ app/tests/test_cleaning.py | 54 ++ app/tests/test_converter.py | 619 ++++++++++++++ app/tests/test_cover.py | 189 +++++ app/tests/test_extractors.py | 213 +++++ app/tests/test_hub.py | 502 ++++++++++++ app/tests/test_tts.py | 1513 +++++++++++++++++++++++++++++++++++ app/tests/test_tui.py | 661 +++++++++++++++ 17 files changed, 6163 insertions(+) create mode 100644 app/tests/__init__.py create mode 100644 app/tests/cover_test.png create mode 100644 app/tests/gen_test_cover.py create mode 100644 app/tests/test_audio.py create mode 100644 app/tests/test_backends.py create mode 100644 app/tests/test_backends_audiocpp.py create mode 100644 app/tests/test_backends_envs.py create mode 100644 app/tests/test_backends_faster.py create mode 100644 app/tests/test_backends_servers.py create mode 100644 app/tests/test_chunking.py create mode 100644 app/tests/test_cleaning.py create mode 100644 app/tests/test_converter.py create mode 100644 app/tests/test_cover.py create mode 100644 app/tests/test_extractors.py create mode 100644 app/tests/test_hub.py create mode 100644 app/tests/test_tts.py create mode 100644 app/tests/test_tui.py (limited to 'app/tests') diff --git a/app/tests/__init__.py b/app/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/tests/cover_test.png b/app/tests/cover_test.png new file mode 100644 index 0000000..0c252db Binary files /dev/null and b/app/tests/cover_test.png differ diff --git a/app/tests/gen_test_cover.py b/app/tests/gen_test_cover.py new file mode 100644 index 0000000..292469a --- /dev/null +++ b/app/tests/gen_test_cover.py @@ -0,0 +1,8 @@ +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from converter.cover import generate_cover + +p = generate_cover('The Count of Monte Cristo', + Path(__file__).resolve().parent / 'cover_test.png') +print('written:', p) diff --git a/app/tests/test_audio.py b/app/tests/test_audio.py new file mode 100644 index 0000000..ef5e92a --- /dev/null +++ b/app/tests/test_audio.py @@ -0,0 +1,524 @@ +"""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 MagicMock, patch + +from converter import audio +from converter import config +from converter.audio import ( + TrackMeta, + _collect_chunk_files, + _cover_args, + _encode_args, + _tag_args, + build_concat_command, + build_ffmetadata, + build_m4b_chapters_command, + cleanup_chunks, + concat_audio_files, + speed_export_params, + verify_output_duration, +) + + +class SpeedExportParamsTests(unittest.TestCase): + def test_normal_speed_no_filter(self): + self.assertEqual(speed_export_params(1.0), []) + + def test_simple_speedup(self): + self.assertEqual(speed_export_params(1.5), ["-filter:a", "atempo=1.5"]) + + def test_simple_slowdown(self): + self.assertEqual(speed_export_params(0.75), ["-filter:a", "atempo=0.75"]) + + def test_chained_speedup_beyond_2x(self): + self.assertEqual(speed_export_params(3.0), ["-filter:a", "atempo=2.0,atempo=1.5"]) + + def test_chained_slowdown_below_half(self): + self.assertEqual(speed_export_params(0.25), ["-filter:a", "atempo=0.5,atempo=0.5"]) + + def test_zero_speed_rejected(self): + with self.assertRaises(ValueError): + speed_export_params(0) + + def test_negative_speed_rejected(self): + with self.assertRaises(ValueError): + speed_export_params(-1.5) + + +class CleanupChunksTests(unittest.TestCase): + def test_removes_only_chunk_files(self): + with tempfile.TemporaryDirectory() as tmp: + chunks_dir = Path(tmp) + (chunks_dir / "chunk_0001.wav").write_bytes(b"stale") + (chunks_dir / "chunk_0002.wav").write_bytes(b"stale") + (chunks_dir / "keep.txt").write_bytes(b"keep") + + original = audio.CHUNKS_FOLDER + audio.CHUNKS_FOLDER = chunks_dir + try: + cleanup_chunks() + finally: + audio.CHUNKS_FOLDER = original + + self.assertFalse((chunks_dir / "chunk_0001.wav").exists()) + self.assertFalse((chunks_dir / "chunk_0002.wav").exists()) + self.assertTrue((chunks_dir / "keep.txt").exists()) + + def test_removes_chapter_files(self): + with tempfile.TemporaryDirectory() as tmp: + chunks_dir = Path(tmp) + (chunks_dir / "chapter_0001.m4b").write_bytes(b"stale") + (chunks_dir / "chunk_0001.wav").write_bytes(b"stale") + + original = audio.CHUNKS_FOLDER + audio.CHUNKS_FOLDER = chunks_dir + try: + cleanup_chunks() + finally: + audio.CHUNKS_FOLDER = original + + self.assertFalse((chunks_dir / "chapter_0001.m4b").exists()) + self.assertFalse((chunks_dir / "chunk_0001.wav").exists()) + + +class EncodeArgsTests(unittest.TestCase): + def test_mp3_uses_bitrate_only(self): + self.assertEqual(_encode_args("mp3"), ["-b:a", config.AUDIO_BITRATE]) + + def test_m4b_uses_aac(self): + self.assertEqual(_encode_args("m4b"), ["-c:a", "aac", "-b:a", config.AUDIO_BITRATE]) + + def test_wav_is_lossless_pcm(self): + self.assertEqual(_encode_args("wav"), ["-c:a", "pcm_s16le"]) + + def test_ogg_uses_libvorbis(self): + self.assertEqual(_encode_args("ogg"), ["-c:a", "libvorbis", "-b:a", config.AUDIO_BITRATE]) + + def test_flac_is_lossless(self): + self.assertEqual(_encode_args("flac"), ["-c:a", "flac"]) + + +class M4bContainerArgsTests(unittest.TestCase): + def setUp(self): + self._original = audio._brand_supported + audio._brand_supported = True + + def tearDown(self): + audio._brand_supported = self._original + + def test_includes_faststart_and_brand(self): + args = audio._m4b_container_args() + self.assertIn("+faststart", args) + self.assertIn("M4B ", args) + + def test_brand_omitted_when_unsupported(self): + audio._brand_supported = False + self.assertEqual(audio._m4b_container_args(), ["-movflags", "+faststart"]) + + +class BuildConcatCommandTests(unittest.TestCase): + def setUp(self): + self._original = audio._brand_supported + audio._brand_supported = True + + def tearDown(self): + audio._brand_supported = self._original + + def test_mp3_has_no_container_flags(self): + cmd = build_concat_command(Path("list.txt"), Path("out.mp3"), "mp3") + self.assertEqual(cmd[:6], ["ffmpeg", "-y", "-f", "concat", "-safe", "0"]) + self.assertNotIn("-movflags", cmd) + self.assertEqual(cmd[-1], "out.mp3") + + def test_m4b_gets_faststart_and_brand(self): + cmd = build_concat_command(Path("list.txt"), Path("out.m4b"), "m4b") + self.assertIn("+faststart", cmd) + self.assertIn("M4B ", cmd) + self.assertEqual(cmd[-1], "out.m4b") + + def test_speed_copy_writes_two_outputs(self): + cmd = build_concat_command(Path("list.txt"), Path("out.m4b"), "m4b", + speed=1.5, speed_path=Path("out_1.5.m4b")) + self.assertIn("out.m4b", cmd) + self.assertIn("out_1.5.m4b", cmd) + # faststart must apply to both outputs + self.assertEqual(cmd.count("+faststart"), 2) + self.assertTrue(any("atempo=1.5" in arg for arg in cmd)) + + def test_wav_intermediate(self): + cmd = build_concat_command(Path("list.txt"), Path("chapter.wav"), "wav") + self.assertIn("pcm_s16le", cmd) + self.assertNotIn("-movflags", cmd) + + def test_speed_without_speed_path_rejected(self): + with self.assertRaises(ValueError): + build_concat_command(Path("list.txt"), Path("out.mp3"), "mp3", speed=1.5) + + +class CollectChunkFilesTests(unittest.TestCase): + def test_uses_recorded_paths_exactly(self): + with tempfile.TemporaryDirectory() as tmp: + present = Path(tmp) / "chunk_0001.wav" + present.write_bytes(b"audio") + chunk_results = { + 1: present, + 2: None, # failed chunk + 3: Path(tmp) / "chunk_0003.wav", # recorded but deleted + } + files, missing = _collect_chunk_files(3, chunk_results) + + self.assertEqual(files, [present]) + self.assertEqual(missing, [2, 3]) + + +class BuildM4bChaptersCommandTests(unittest.TestCase): + def setUp(self): + self._original = audio._brand_supported + audio._brand_supported = True + + def tearDown(self): + audio._brand_supported = self._original + + def test_base_output_maps_metadata_and_chapters(self): + cmd = build_m4b_chapters_command(Path("list.txt"), Path("meta.txt"), Path("out.m4b")) + self.assertIn("-map_metadata", cmd) + self.assertIn("-map_chapters", cmd) + self.assertIn("out.m4b", cmd) + self.assertIn("+faststart", cmd) + self.assertNotIn("filter_complex", cmd) + + def test_speed_outputs_get_their_own_chapter_metadata(self): + cmd = build_m4b_chapters_command( + Path("list.txt"), Path("meta.txt"), Path("out.m4b"), + speed=2.0, speed_path=Path("out_2.m4b"), speed_metadata_file=Path("meta2.txt"), + ) + self.assertEqual(cmd.count("+faststart"), 2) + self.assertEqual(cmd.count("-map_chapters"), 2) + self.assertTrue(any("atempo=2" in arg for arg in cmd)) + # base output chapters come from metadata input 1, speed copy from 2 + chapter_flags = [i for i, v in enumerate(cmd) if v == "-map_chapters"] + self.assertEqual(cmd[chapter_flags[0] + 1], "1") + self.assertEqual(cmd[chapter_flags[1] + 1], "2") + base_idx, speed_idx = cmd.index("out.m4b"), cmd.index("out_2.m4b") + self.assertLess(chapter_flags[0], base_idx) + self.assertGreater(chapter_flags[1], base_idx) + self.assertLess(chapter_flags[1], speed_idx) + + +class VerifyOutputDurationTests(unittest.TestCase): + def _patch_probe(self, ms): + audio.probe_duration_ms = lambda path: ms + + def setUp(self): + self._original_probe = audio.probe_duration_ms + + def tearDown(self): + audio.probe_duration_ms = self._original_probe + + def test_close_duration_passes(self): + self._patch_probe(100_000) + self.assertTrue(verify_output_duration(Path("x.m4b"), 101_000)) + + def test_unverifiable_duration_passes(self): + self._patch_probe(0) + self.assertTrue(verify_output_duration(Path("x.m4b"), 100_000)) + + def test_zero_expected_passes(self): + self._patch_probe(50_000) + self.assertTrue(verify_output_duration(Path("x.m4b"), 0)) + + def test_large_drift_fails(self): + self._patch_probe(3_600_000) # bogus "1 hour" for a 1 minute book + with self.assertLogs(level="ERROR"): + self.assertFalse(verify_output_duration(Path("x.m4b"), 60_000)) + + +class BuildFFMetadataTests(unittest.TestCase): + def test_writes_chapters(self): + chapters = [(0, 1200, "One"), (1200, 2500, "Two")] + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "meta.txt" + build_ffmetadata(chapters, path) + content = path.read_text(encoding="utf-8") + + self.assertTrue(content.startswith(";FFMETADATA1\n")) + self.assertIn("[CHAPTER]", content) + self.assertIn("TIMEBASE=1/1000", content) + self.assertIn("START=0", content) + self.assertIn("END=1200", content) + self.assertIn("title=One", content) + self.assertIn("START=1200", content) + self.assertIn("title=Two", content) + + def test_escapes_special_characters(self): + # ffmpeg's FFMETADATA format treats = ; # and \ as structural. + chapters = [(0, 1000, "A = B; C# D\\E")] + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "meta.txt" + build_ffmetadata(chapters, path) + content = path.read_text(encoding="utf-8") + + self.assertIn(r"title=A \= B\; C\# D\\E", content) + + def test_collapses_newlines_in_titles(self): + chapters = [(0, 1000, "Two\nLines")] + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "meta.txt" + build_ffmetadata(chapters, path) + content = path.read_text(encoding="utf-8") + + self.assertIn("title=Two Lines\n", content) + self.assertNotIn("title=Two\n", content) + + +class TagArgsTests(unittest.TestCase): + META = TrackMeta(title="Dune", artist="Frank Herbert", album="Dune", + track=2, total_tracks=5) + + def test_full_meta_written(self): + args = _tag_args(self.META, "mp3") + for pair in ("title=Dune", "artist=Frank Herbert", + "album=Dune", "track=2/5"): + self.assertIn(pair, args) + + def test_mp3_gets_id3v23(self): + mp3_args = _tag_args(self.META, "mp3") + self.assertIn("-id3v2_version", mp3_args) + self.assertEqual(mp3_args[mp3_args.index("-id3v2_version") + 1], "3") + self.assertNotIn("-id3v2_version", _tag_args(self.META, "flac")) + + def test_empty_fields_omitted(self): + meta = TrackMeta(title="Only Title") + args = _tag_args(meta, "flac") + self.assertNotIn("artist", args) + self.assertNotIn("album", args) + self.assertNotIn("track", args) + + def test_track_requires_total(self): + meta = TrackMeta(title="T", track=3) + self.assertNotIn("track", _tag_args(meta, "mp3")) + + +class CoverArgsTests(unittest.TestCase): + def test_mp3_copies_png_stream(self): + args = _cover_args("mp3", 1) + self.assertIn("copy", args) + self.assertIn("attached_pic", args) + self.assertIn("1:v", args) + + def test_m4b_reencodes_to_jpeg(self): + args = _cover_args("m4b", 2) + self.assertIn("mjpeg", args) + self.assertIn("attached_pic", args) + self.assertIn("3", args) # jpeg quality + + def test_ogg_and_wav_have_no_cover(self): + self.assertEqual(_cover_args("ogg", 1), []) + self.assertEqual(_cover_args("wav", 1), []) + + +class BuildConcatCommandMetaTests(unittest.TestCase): + META = TrackMeta(title="Chapter 1", artist="Author", album="Book", + track=1, total_tracks=3) + + def test_cover_added_as_second_input(self): + cmd = build_concat_command(Path("list.txt"), Path("out.mp3"), "mp3", + meta=self.META, cover=Path("cover.png")) + # The cover is the second input, after the concat list + self.assertIn("cover.png", cmd) + self.assertLess(cmd.index("list.txt"), cmd.index("cover.png")) + self.assertIn("-map", cmd) + self.assertIn("1:v", cmd) + self.assertIn("attached_pic", cmd) + self.assertEqual(cmd[-1], "out.mp3") + + def test_audio_explicitly_mapped_when_cover_present(self): + cmd = build_concat_command(Path("list.txt"), Path("out.flac"), "flac", + cover=Path("cover.png")) + self.assertIn("0:a", cmd) + + def test_no_cover_keeps_single_input(self): + cmd = build_concat_command(Path("list.txt"), Path("out.mp3"), "mp3", + meta=self.META) + self.assertEqual(cmd.count("-i"), 1) + self.assertNotIn("attached_pic", cmd) + + def test_ogg_never_gets_cover_input(self): + cmd = build_concat_command(Path("list.txt"), Path("out.ogg"), "ogg", + meta=self.META, cover=Path("cover.png")) + self.assertEqual(cmd.count("-i"), 1) + self.assertNotIn("attached_pic", cmd) + + def test_speed_copy_gets_tags_and_cover(self): + cmd = build_concat_command(Path("list.txt"), Path("out.mp3"), "mp3", + speed=1.5, speed_path=Path("out_1.5.mp3"), + meta=self.META, cover=Path("cover.png")) + self.assertEqual(cmd.count("attached_pic"), 2) + self.assertEqual(cmd.count("title=Chapter 1"), 2) + self.assertEqual(cmd.count("1:v"), 2) + + def test_tags_without_cover_present(self): + cmd = build_concat_command(Path("list.txt"), Path("out.mp3"), "mp3", + meta=self.META) + self.assertIn("title=Chapter 1", cmd) + self.assertIn("artist=Author", cmd) + self.assertIn("album=Book", cmd) + self.assertIn("track=1/3", cmd) + + +class BuildM4bChaptersCommandMetaTests(unittest.TestCase): + def setUp(self): + self._original = audio._brand_supported + audio._brand_supported = True + + def tearDown(self): + audio._brand_supported = self._original + + def test_cover_indexed_after_metadata_inputs(self): + cmd = build_m4b_chapters_command(Path("list.txt"), Path("meta.txt"), + Path("out.m4b"), cover=Path("cover.png")) + # Inputs: 0=audio, 1=ffmetadata, 2=cover + self.assertIn("-i", cmd) + self.assertIn("2:v", cmd) + self.assertIn("attached_pic", cmd) + + def test_speed_variant_cover_is_input_three(self): + cmd = build_m4b_chapters_command( + Path("list.txt"), Path("meta.txt"), Path("out.m4b"), + speed=2.0, speed_path=Path("out_2.m4b"), speed_metadata_file=Path("meta2.txt"), + meta=TrackMeta(title="Book"), cover=Path("cover.png"), + ) + self.assertEqual(cmd.count("3:v"), 2) # both outputs attach the cover + self.assertNotIn("2:v", cmd) + self.assertEqual(cmd.count("title=Book"), 2) + # Chapter metadata inputs keep their 1/2 mapping + chapter_flags = [i for i, v in enumerate(cmd) if v == "-map_chapters"] + self.assertEqual(cmd[chapter_flags[0] + 1], "1") + self.assertEqual(cmd[chapter_flags[1] + 1], "2") + + def test_without_cover_regression(self): + cmd = build_m4b_chapters_command(Path("list.txt"), Path("meta.txt"), + Path("out.m4b")) + self.assertNotIn("attached_pic", cmd) + self.assertNotIn("-metadata", cmd) + + +class ConcatAudioFilesTests(unittest.TestCase): + """Concatenation of sub-request audio into one chunk file.""" + + @staticmethod + def _write_wav(path: Path, frames: bytes, framerate: int = 24000) -> Path: + with wave.open(str(path), "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(framerate) + wav_file.writeframes(frames) + return path + + def test_wav_files_are_merged_in_order(self): + with tempfile.TemporaryDirectory() as tmp: + first = self._write_wav(Path(tmp) / "a.wav", b"\x01\x00" * 10) + second = self._write_wav(Path(tmp) / "b.wav", b"\x02\x00" * 20) + destination = Path(tmp) / "out.wav" + concat_audio_files([first, second], destination) + with wave.open(str(destination), "rb") as wav_file: + self.assertEqual(wav_file.getframerate(), 24000) + self.assertEqual(wav_file.getnchannels(), 1) + self.assertEqual(wav_file.getsampwidth(), 2) + frames = wav_file.readframes(wav_file.getnframes()) + self.assertEqual(frames, b"\x01\x00" * 10 + b"\x02\x00" * 20) + + def test_single_wav_file_is_copied(self): + with tempfile.TemporaryDirectory() as tmp: + source = self._write_wav(Path(tmp) / "a.wav", b"\x03\x00" * 15) + destination = Path(tmp) / "out.wav" + concat_audio_files([source], destination) + with wave.open(str(destination), "rb") as wav_file: + self.assertEqual(wav_file.readframes(wav_file.getnframes()), + b"\x03\x00" * 15) + + def test_empty_source_list_raises(self): + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(ValueError): + concat_audio_files([], Path(tmp) / "out.wav") + + def test_mismatched_wav_parameters_fall_back_to_ffmpeg(self): + with tempfile.TemporaryDirectory() as tmp: + first = self._write_wav(Path(tmp) / "a.wav", b"\x01\x00" * 10, framerate=24000) + second = self._write_wav(Path(tmp) / "b.wav", b"\x02\x00" * 10, framerate=16000) + destination = Path(tmp) / "out.wav" + with patch("converter.audio.shutil.which", return_value=None), \ + self.assertRaises(RuntimeError) as ctx: + concat_audio_files([first, second], destination) + self.assertIn("ffmpeg", str(ctx.exception)) + # The wave-module path must not have written a partial output. + self.assertFalse(destination.exists()) + + def test_non_wav_input_falls_back_to_ffmpeg(self): + with tempfile.TemporaryDirectory() as tmp: + source = Path(tmp) / "part.mp3" + source.write_bytes(b"not a wav file") + destination = Path(tmp) / "out.wav" + with patch("converter.audio.shutil.which", return_value=None), \ + self.assertRaises(RuntimeError) as ctx: + concat_audio_files([source], destination) + 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/app/tests/test_backends.py b/app/tests/test_backends.py new file mode 100644 index 0000000..c0e8d4a --- /dev/null +++ b/app/tests/test_backends.py @@ -0,0 +1,178 @@ +"""Tests for the backends package registry and detection aggregation.""" + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from backends import REGISTRY, detect_all, get + + +class RegistryTests(unittest.TestCase): + def setUp(self): + # The registry is built lazily on first access (the backend modules + # pull in converter.tts and its deps, which are only available inside + # the managed venv). Trigger the build so these tests don't depend on + # another test class having called detect_all() first. + get("audiocpp") + + def test_registry_has_the_three_backends(self): + keys = [info.key for info in REGISTRY] + self.assertEqual(keys, ["audiocpp", "qwen", "faster"]) + + def test_every_entry_has_detect_and_setup_tui(self): + for info in REGISTRY: + self.assertTrue(callable(info.detect), info.key) + self.assertTrue(callable(info.setup_tui), info.key) + self.assertIsInstance(info.configure_actions, list) + for action in info.configure_actions: + self.assertTrue(callable(action.run)) + + def test_get_returns_entry_by_key(self): + self.assertIs(get("audiocpp").key, "audiocpp") + self.assertIsNone(get("nonexistent")) + + +class DetectAllTests(unittest.TestCase): + def test_detect_all_returns_one_status_per_backend(self): + with patch("backends.common.server_running", return_value=False): + statuses = detect_all() + self.assertEqual([s.key for s in statuses], + ["audiocpp", "qwen", "faster"]) + for s in statuses: + self.assertIn(s.key, ("audiocpp", "qwen", "faster")) + # ready requires both installed and configured; on a clean + # machine none are ready. + if s.ready: + self.assertTrue(s.installed and s.configured) + # running is always probed; patched False here so a dev machine + # running a real server can't flake the test. + self.assertFalse(s.running) + + def test_audiocpp_status_when_cloned_built_configured(self): + with tempfile.TemporaryDirectory() as td: + root = Path(td) + checkout = root / "audio.cpp" + checkout.mkdir() + (checkout / "model_specs").mkdir() + (checkout / "build" / "linux-cuda-release" / "bin").mkdir( + parents=True) + (checkout / "build" / "linux-cuda-release" / "bin" + / "audiocpp_server").write_bytes(b"x") + (checkout / "server.json").write_text('{"models":[]}', + encoding="utf-8") + from backends import audiocpp + with patch.object(audiocpp, "find_local_checkout", + return_value=checkout), \ + patch("backends.common.server_running", + return_value=False): + status = audiocpp.detect() + self.assertTrue(status.installed) + self.assertTrue(status.configured) + self.assertTrue(status.ready) + self.assertFalse(status.running) + self.assertIn("audiocpp_server", status.launch_hint) + + def test_audiocpp_running_when_server_probe_succeeds(self): + from backends import audiocpp + with patch.object(audiocpp, "find_local_checkout", + return_value=None), \ + patch("backends.common.server_running", return_value=True): + status = audiocpp.detect() + # Not installed (no checkout) but an external server is up. + self.assertFalse(status.installed) + self.assertTrue(status.running) + + def test_qwen_status_reflects_install(self): + from backends import qwen + with patch.object(qwen, "_is_installed", return_value=True), \ + patch("backends.common.server_running", return_value=False): + status = qwen.detect() + self.assertTrue(status.installed) + self.assertTrue(status.configured) + self.assertFalse(status.running) + self.assertIn("qwen-tts-demo", status.launch_hint) + with patch.object(qwen, "_is_installed", return_value=False), \ + patch("backends.common.server_running", return_value=False): + status = qwen.detect() + self.assertFalse(status.installed) + self.assertFalse(status.configured) + + def test_qwen_running_when_either_port_is_up(self): + # Either the CustomVoice port or the Base port counts as running. + from backends import qwen + with patch.object(qwen, "_is_installed", return_value=False), \ + patch("backends.common.server_running", + side_effect=[True, False]): + status = qwen.detect() + self.assertTrue(status.running) + with patch.object(qwen, "_is_installed", return_value=False), \ + patch("backends.common.server_running", + side_effect=[False, True]): + status = qwen.detect() + self.assertTrue(status.running) + + def test_faster_status_reflects_install_clone_voices(self): + from backends import faster + with tempfile.TemporaryDirectory() as td: + checkout = Path(td) / "faster-qwen3-tts" + (checkout / "examples").mkdir(parents=True) + (checkout / "examples" / "openai_server.py").write_text("x") + (checkout / "voices.json").write_text('{"default":{}}', + encoding="utf-8") + with patch.object(faster, "_is_installed", return_value=True), \ + patch.object(faster, "_checkout", + return_value=checkout), \ + patch("backends.common.server_running", + return_value=False): + status = faster.detect() + self.assertTrue(status.installed) + self.assertTrue(status.configured) + self.assertFalse(status.running) + self.assertIn("openai_server.py", status.launch_hint) + + def test_faster_running_when_server_probe_succeeds(self): + from backends import faster + with patch.object(faster, "_is_installed", return_value=False), \ + patch.object(faster, "_is_cloned", return_value=False), \ + patch("backends.common.server_running", return_value=True): + status = faster.detect() + self.assertTrue(status.running) + + +class ServerRunningTests(unittest.TestCase): + """backends.common.server_running: TCP probe against a real socket.""" + + def test_true_for_open_port(self): + import socket + + from backends import common + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.bind(("127.0.0.1", 0)) + server.listen(1) + host, port = server.getsockname() + url = f"http://127.0.0.1:{port}" + try: + self.assertTrue(common.server_running(url)) + finally: + server.close() + + def test_false_for_closed_port(self): + # Pick an unused port by opening + closing a socket, then probe it. + import socket + + from backends import common + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.bind(("127.0.0.1", 0)) + _, port = s.getsockname() + s.close() + self.assertFalse(common.server_running(f"http://127.0.0.1:{port}")) + + def test_false_for_invalid_url(self): + from backends import common + self.assertFalse(common.server_running("not a url")) + self.assertFalse(common.server_running("")) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py new file mode 100644 index 0000000..9882ce1 --- /dev/null +++ b/app/tests/test_backends_audiocpp.py @@ -0,0 +1,1062 @@ +"""Tests for the audio.cpp backend setup module (backends/audiocpp.py).""" + +import io +import json +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest.mock import MagicMock, patch + +from converter import config +from backends import audiocpp 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" +) + +FAKE_CONFIG_WITH_MODEL_IDS = ( + 'AUDIOCPP_API_URL = "http://127.0.0.1:9999" # audio.cpp audiocpp_server\n' + "\n" + 'AUDIOCPP_MODEL_ID = "qwen" # server entry for speaker mode\n' + 'AUDIOCPP_CLONE_MODEL_ID = "qwen-clone"\n' +) + + +def _write_spec(checkout: Path, family: str, *, display_name=None, + tasks=("tts", "clone"), languages=("en",), packages=None, + category="tts"): + """Write a minimal model_specs/.json into a fake checkout.""" + specs = checkout / "model_specs" + specs.mkdir(parents=True, exist_ok=True) + if packages is None: + packages = [{ + "id": f"{family}_q8_0", "default": True, "format": "gguf", + "target_directory": f"{family}-GGUF", + }] + spec = { + "family": family, + "display_name": display_name or family, + "category": category, + "tasks": list(tasks), + "languages": list(languages), + "packages": packages, + } + (specs / f"{family}.json").write_text(json.dumps(spec), encoding="utf-8") + return spec + + +def _make_checkout(tmp: Path) -> Path: + """Create a fake audio.cpp checkout with a realistic model_specs set.""" + checkout = tmp / "audio.cpp" + checkout.mkdir() + _write_spec(checkout, "qwen3_tts", display_name="Qwen3-TTS", + tasks=("tts", "clone", "design"), + languages=("zh", "en", "ja"), + packages=[ + {"id": "qwen3_tts_1_7b_base_q8_0", "default": True, + "format": "gguf", + "target_directory": "Qwen3-TTS-12Hz-1.7B-Base-GGUF"}, + {"id": "qwen3_tts_1_7b_customvoice_q8_0", + "format": "gguf", + "target_directory": "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"}, + {"id": "qwen3_tts_1_7b_voicedesign_q8_0", + "format": "gguf", + "target_directory": "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF"}, + ]) + _write_spec(checkout, "higgs_audio_tts", display_name="Higgs Audio v3 TTS 4B", + languages=("auto",), + packages=[{ + "id": "higgs_audio_tts_4b_q8_0", "default": True, + "format": "gguf", + "target_directory": "Higgs-Audio-v3-TTS-4B-GGUF", + }]) + _write_spec(checkout, "voxcpm2", display_name="VoxCPM2-2B", + languages=("en", "zh"), + packages=[{ + "id": "voxcpm2_q8_0", "default": True, "format": "gguf", + "target_directory": "VoxCPM2-GGUF", + }]) + _write_spec(checkout, "index_tts2", display_name="IndexTTS-2", + languages=("zh", "en"), + packages=[{ + "id": "index_tts2_q8_0", "default": True, "format": "gguf", + "target_directory": "IndexTTS2-GGUF", + }]) + _write_spec(checkout, "pocket_tts", display_name="PocketTTS-100M", + tasks=("tts", "clone"), languages=("en", "de"), + packages=[{ + "id": "pocket_tts_q8_0", "default": True, "format": "gguf", + "target_directory": "PocketTTS-GGUF", + }]) + _write_spec(checkout, "supertonic", display_name="Supertonic 3", + tasks=("tts",), languages=("en", "ko"), + packages=[{ + "id": "supertonic_q8_0", "default": True, "format": "gguf", + "target_directory": "Supertonic-GGUF", + }]) + # An ASR family that must be filtered out. + _write_spec(checkout, "qwen3_asr", display_name="Qwen3-ASR", + tasks=("asr",), category="asr") + # A TTS family with no installable packages (must be skipped). + _write_spec(checkout, "empty_tts", display_name="Empty TTS", + tasks=("tts",), packages=[]) + return checkout + + +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 DetectWavDirTests(unittest.TestCase): + """Shallow .wav-directory discovery across the two checkout roots.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.root = Path(self._td.name) + self.audiocpp = self.root / "audio.cpp" + self.tts_root = self.root / "tts-audiobook-generator" + self.audiocpp.mkdir() + self.tts_root.mkdir() + + def tearDown(self): + self._td.cleanup() + + def _wav_dir(self, where, name="voices"): + directory = where / name + directory.mkdir(parents=True, exist_ok=True) + (directory / "voice.wav").write_bytes(b"x") + return directory + + def test_unique_wav_dir_in_tts_root_returned(self): + found = self._wav_dir(self.tts_root, "voices") + self.assertEqual(make_server.detect_wav_dir(self.audiocpp, + self.tts_root), + found) + + def test_unique_wav_dir_in_audiocpp_root_returned(self): + found = self._wav_dir(self.audiocpp, "reference") + self.assertEqual(make_server.detect_wav_dir(self.audiocpp, + self.tts_root), + found) + + def test_root_itself_containing_wavs_returned(self): + (self.tts_root / "direct.wav").write_bytes(b"x") + self.assertEqual(make_server.detect_wav_dir(self.audiocpp, + self.tts_root), + self.tts_root) + + def test_multiple_wav_dirs_returns_none(self): + self._wav_dir(self.tts_root, "one") + self._wav_dir(self.audiocpp, "two") + self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.tts_root)) + + def test_output_dir_of_tts_root_excluded(self): + self._wav_dir(self.tts_root, "output") + self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.tts_root)) + + def test_no_wavs_returns_none(self): + self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.tts_root)) + + def test_nested_wav_dir_not_seen(self): + nested = self.tts_root / "outer" / "inner" + nested.mkdir(parents=True) + (nested / "voice.wav").write_bytes(b"x") + self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.tts_root)) + + +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 UpdateConfigModelIdsTests(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_WITH_MODEL_IDS, + encoding="utf-8") + + def tearDown(self): + self._tmp.cleanup() + + def test_rewrites_both_ids_preserving_lines(self): + changed = make_server.update_config_model_ids( + "higgs", "higgs", config_path=self.config_path) + self.assertTrue(changed) + text = self.config_path.read_text(encoding="utf-8") + self.assertIn('AUDIOCPP_MODEL_ID = "higgs" # server entry for speaker mode', + text) + self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "higgs"', text) + self.assertIn('AUDIOCPP_API_URL = "http://127.0.0.1:9999"', text) + + def test_clone_id_optional(self): + changed = make_server.update_config_model_ids( + "voxcpm2", config_path=self.config_path) + self.assertTrue(changed) + text = self.config_path.read_text(encoding="utf-8") + self.assertIn('AUDIOCPP_MODEL_ID = "voxcpm2"', text) + self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "qwen-clone"', text) + + def test_returns_false_when_ids_unchanged(self): + changed = make_server.update_config_model_ids( + "qwen", "qwen-clone", config_path=self.config_path) + self.assertFalse(changed) + self.assertEqual(self.config_path.read_text(encoding="utf-8"), + FAKE_CONFIG_WITH_MODEL_IDS) + + def test_returns_false_when_lines_missing(self): + path = Path(self._tmp.name) / "other.py" + path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") + self.assertFalse(make_server.update_config_model_ids( + "higgs", "higgs", config_path=path)) + + def test_returns_false_when_file_missing(self): + self.assertFalse(make_server.update_config_model_ids( + "higgs", "higgs", + config_path=Path(self._tmp.name) / "nope.py")) + + +class ResolveWavDirArgTests(unittest.TestCase): + 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 NormalizeDirArgTests(unittest.TestCase): + """Path normalization for the audio.cpp checkout argument.""" + + def test_expands_tilde_and_resolves(self): + with patch.object(make_server.os.path, "expanduser", + return_value="/home/u/audio.cpp") as mock_expand: + result = make_server.normalize_dir_arg("~/audio.cpp") + mock_expand.assert_called_once_with("~/audio.cpp") + self.assertEqual(result, Path("/home/u/audio.cpp").resolve()) + + def test_strips_quotes_and_whitespace(self): + with patch.object(make_server.os.path, "expanduser", + side_effect=lambda s: s): + result = make_server.normalize_dir_arg(' "/tmp/foo" ') + self.assertEqual(result, Path("/tmp/foo").resolve()) + + +class CheckoutAutoSelectTests(unittest.TestCase): + """TUI browser auto-accept callback for an audio.cpp checkout.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.root = Path(self._td.name) + + def tearDown(self): + self._td.cleanup() + + def test_accepts_audio_cpp_containing_model_specs(self): + checkout = self.root / "audio.cpp" + checkout.mkdir() + (checkout / "model_specs").mkdir() + self.assertEqual(make_server._checkout_auto_select(checkout), + checkout) + + def test_rejects_audio_cpp_without_model_specs(self): + checkout = self.root / "audio.cpp" + checkout.mkdir() + self.assertIsNone(make_server._checkout_auto_select(checkout)) + + def test_rejects_other_name_even_with_model_specs(self): + other = self.root / "not-audiocpp" + other.mkdir() + (other / "model_specs").mkdir() + self.assertIsNone(make_server._checkout_auto_select(other)) + + def test_rejects_plain_directory(self): + plain = self.root / "somewhere" + plain.mkdir() + self.assertIsNone(make_server._checkout_auto_select(plain)) + + +class DefaultModelIdTests(unittest.TestCase): + def test_preferred_ids_for_tested_families(self): + self.assertEqual(make_server.default_model_id("qwen3_tts"), "qwen") + self.assertEqual(make_server.default_model_id("higgs_audio_tts"), "higgs") + self.assertEqual(make_server.default_model_id("voxcpm2"), "voxcpm2") + self.assertEqual(make_server.default_model_id("index_tts2"), "indextts2") + + def test_derived_id_strips_trailing_tts_and_underscores(self): + self.assertEqual(make_server.default_model_id("pocket_tts"), "pocket") + self.assertEqual(make_server.default_model_id("dots_tts"), "dots") + self.assertEqual(make_server.default_model_id("moss_tts_local"), + "mossttslocal") + + +class LoadModelCatalogTests(unittest.TestCase): + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = _make_checkout(Path(self._td.name)) + + def tearDown(self): + self._td.cleanup() + + def test_includes_tts_families_excludes_asr(self): + catalog = make_server.load_model_catalog(self.checkout) + families = [entry["family"] for entry in catalog] + self.assertIn("qwen3_tts", families) + self.assertIn("higgs_audio_tts", families) + self.assertIn("pocket_tts", families) + self.assertIn("supertonic", families) + self.assertNotIn("qwen3_asr", families) + + def test_skips_families_with_no_packages(self): + catalog = make_server.load_model_catalog(self.checkout) + self.assertNotIn("empty_tts", + [entry["family"] for entry in catalog]) + + def test_families_sorted_alphabetically_by_display_name(self): + catalog = make_server.load_model_catalog(self.checkout) + names = [entry["display_name"].lower() for entry in catalog] + self.assertEqual(names, sorted(names)) + self.assertNotIn("tested", catalog[0]) + self.assertNotIn("TESTED_FAMILIES", dir(make_server)) + + def test_default_package_and_target_directory_resolved(self): + catalog = make_server.load_model_catalog(self.checkout) + by_family = {entry["family"]: entry for entry in catalog} + higgs = by_family["higgs_audio_tts"] + self.assertEqual(higgs["install_id"], "higgs_audio_tts_4b_q8_0") + self.assertEqual(higgs["default_path"], + "models/Higgs-Audio-v3-TTS-4B-GGUF") + + def test_picks_first_gguf_when_no_default_flag(self): + _write_spec(self.checkout, "voxcpm2", display_name="VoxCPM2-2B", + packages=[ + {"id": "voxcpm2_bf16", "format": "gguf", + "target_directory": "VoxCPM2-GGUF"}, + {"id": "voxcpm2_q8_0", "format": "gguf", + "target_directory": "VoxCPM2-GGUF"}, + ]) + catalog = make_server.load_model_catalog(self.checkout) + by_family = {entry["family"]: entry for entry in catalog} + self.assertEqual(by_family["voxcpm2"]["install_id"], "voxcpm2_bf16") + + def test_clone_capability_from_tasks(self): + catalog = make_server.load_model_catalog(self.checkout) + by_family = {entry["family"]: entry for entry in catalog} + self.assertTrue(by_family["higgs_audio_tts"]["clone_capable"]) + self.assertFalse(by_family["supertonic"]["clone_capable"]) + + def test_missing_model_specs_dir_raises(self): + empty = Path(self._td.name) / "empty" + empty.mkdir() + with self.assertRaises(NotADirectoryError): + make_server.load_model_catalog(empty) + + +class DetectBackendTests(unittest.TestCase): + """Backend detection from audio.cpp build directory names.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = Path(self._td.name) / "audio.cpp" + self.checkout.mkdir() + + def tearDown(self): + self._td.cleanup() + + def _build(self, name, binary="audiocpp_server"): + build_dir = self.checkout / "build" / name + bin_dir = build_dir / "bin" + bin_dir.mkdir(parents=True) + (bin_dir / binary).write_bytes(b"x") + return build_dir + + def test_no_build_dir_returns_none(self): + self.assertIsNone(make_server.detect_backend(self.checkout)) + + def test_unique_linux_backend_detected(self): + self._build("linux-cuda-release") + self.assertEqual(make_server.detect_backend(self.checkout), "cuda") + + def test_windows_exe_backend_detected(self): + self._build("windows-vulkan-debug", binary="audiocpp_server.exe") + self.assertEqual(make_server.detect_backend(self.checkout), "vulkan") + + def test_hip_backend_detected(self): + self._build("linux-hip-release") + self.assertEqual(make_server.detect_backend(self.checkout), "hip") + + def test_cpu_backend_detected(self): + self._build("linux-cpu-release") + self.assertEqual(make_server.detect_backend(self.checkout), "cpu") + + def test_metal_maps_to_cpu(self): + self._build("macos-metal-release") + self.assertEqual(make_server.detect_backend(self.checkout), "cpu") + + def test_multiple_backends_returns_none(self): + self._build("linux-cuda-release") + self._build("linux-cpu-release") + self.assertIsNone(make_server.detect_backend(self.checkout)) + + def test_multiple_builds_same_backend_detected(self): + self._build("linux-cuda-release") + self._build("windows-cuda-debug") + self.assertEqual(make_server.detect_backend(self.checkout), "cuda") + + def test_build_dir_without_binary_ignored(self): + (self.checkout / "build" / "linux-cuda-release").mkdir(parents=True) + self.assertIsNone(make_server.detect_backend(self.checkout)) + + def test_non_matching_build_dir_name_ignored(self): + self._build("linux-mybuild-release") + self.assertIsNone(make_server.detect_backend(self.checkout)) + + +class BackendOptionsTests(unittest.TestCase): + """Aligned backend menu labels and the [auto-detected] default.""" + + def test_options_have_aligned_dashes(self): + options, default_index = make_server._backend_options() + dash_columns = {label.index(" - ") for label, _ in options} + self.assertEqual(len(dash_columns), 1) + self.assertEqual(default_index, 0) + + def test_detected_backend_marked_and_defaulted(self): + options, default_index = make_server._backend_options("vulkan") + labels = [label for label, _ in options] + self.assertEqual(default_index, labels.index(next( + label for label, value in options + if value == "vulkan" and label.endswith("[auto-detected]")))) + self.assertTrue(labels[default_index].endswith("[auto-detected]")) + self.assertEqual(options[default_index][1], "vulkan") + + def test_unknown_detected_backend_is_ignored(self): + options, default_index = make_server._backend_options("opencl") + self.assertEqual(default_index, 0) + self.assertFalse(any("[auto-detected]" in label + for label, _ in options)) + + def test_labels_keep_backend_values(self): + options, _ = make_server._backend_options() + self.assertEqual([value for _, value in options], + list(make_server.BACKENDS)) + + +class BuildServerConfigTests(unittest.TestCase): + def test_single_entry_without_voice_dir(self): + entry = make_server.build_model_entry( + "higgs_audio_tts", "higgs", "models/Higgs-GGUF") + cfg = make_server.build_server_config( + "127.0.0.1", 8080, "cuda", False, [entry]) + self.assertEqual(cfg["host"], "127.0.0.1") + self.assertEqual(cfg["port"], 8080) + self.assertEqual(cfg["backend"], "cuda") + self.assertFalse(cfg["lazy_load"]) + self.assertEqual(cfg["models"], [entry]) + self.assertNotIn("voice_dir", cfg) + + def test_voice_dir_added_when_given(self): + entry = make_server.build_model_entry("voxcpm2", "voxcpm2", "models/V") + cfg = make_server.build_server_config( + "0.0.0.0", 9000, "cpu", True, [entry], + voice_dir="/abs/voices") + self.assertTrue(cfg["lazy_load"]) + self.assertEqual(cfg["voice_dir"], "/abs/voices") + + def test_model_entry_shape(self): + entry = make_server.build_model_entry("index_tts2", "indextts2", "p") + self.assertEqual(entry["id"], "indextts2") + self.assertEqual(entry["family"], "index_tts2") + self.assertEqual(entry["path"], "p") + self.assertEqual(entry["task"], "tts") + self.assertEqual(entry["mode"], "offline") + + def test_model_entry_design_task(self): + entry = make_server.build_model_entry( + "qwen3_tts", "qwen-design", "p", task="vdes") + self.assertEqual(entry["task"], "vdes") + self.assertEqual(entry["mode"], "offline") + + +class InstallModelsTests(unittest.TestCase): + """Printing or auto-running the model install commands.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = Path(self._td.name) / "audio.cpp" + self.checkout.mkdir() + self.manager = self.checkout / "tools" / "model_manager_v2.py" + self.manager.parent.mkdir() + self.manager.write_text("#!/usr/bin/env python3\n", encoding="utf-8") + self.guidance = [("Higgs Audio v3 TTS 4B", "higgs_audio_tts_4b_q8_0"), + ("Qwen3-TTS", "qwen3_tts_1_7b_base_q8_0"), + ("Qwen3-TTS", "qwen3_tts_1_7b_base_q8_0")] + + def tearDown(self): + self._td.cleanup() + + def test_declined_download_prints_commands_deduped(self): + buf = io.StringIO() + with redirect_stdout(buf), \ + patch.object(make_server.subprocess, "run") as run: + make_server._install_models(self.checkout, self.guidance, + download=False) + out = buf.getvalue() + self.assertEqual(out.count("install higgs_audio_tts_4b_q8_0"), 1) + self.assertEqual(out.count("install qwen3_tts_1_7b_base_q8_0"), 1) + run.assert_not_called() + + def test_accepted_download_runs_each_command(self): + with patch.object(make_server.subprocess, "run", + return_value=MagicMock(returncode=0)) as run: + make_server._install_models(self.checkout, self.guidance, + download=True) + self.assertEqual(run.call_count, 2) + commands = [call[0][0] for call in run.call_args_list] + self.assertEqual(commands[0], + [sys.executable, str(self.manager), "install", + "higgs_audio_tts_4b_q8_0"]) + self.assertEqual(commands[1], + [sys.executable, str(self.manager), "install", + "qwen3_tts_1_7b_base_q8_0"]) + for call in run.call_args_list: + self.assertEqual(call[1]["cwd"], str(self.checkout)) + + def test_missing_manager_falls_back_to_printing(self): + self.manager.unlink() + buf = io.StringIO() + with redirect_stdout(buf), \ + patch.object(make_server.subprocess, "run") as run: + make_server._install_models(self.checkout, self.guidance, + download=True) + self.assertIn("install higgs_audio_tts_4b_q8_0", buf.getvalue()) + run.assert_not_called() + + def test_failed_install_reports_warning_and_continues(self): + results = iter([MagicMock(returncode=1), MagicMock(returncode=0)]) + buf = io.StringIO() + with redirect_stdout(buf), \ + patch.object(make_server.subprocess, "run", + side_effect=lambda *a, **k: next(results)) as run: + make_server._install_models(self.checkout, self.guidance, + download=True) + self.assertEqual(run.call_count, 2) + self.assertIn("exited with code 1", buf.getvalue()) + + def test_decide_download_skips_prompt_without_manager(self): + self.manager.unlink() + confirm = MagicMock() + self.assertFalse(make_server._decide_download(self.checkout, confirm)) + confirm.assert_not_called() + + def test_decide_download_asks_when_manager_present(self): + confirm = MagicMock(return_value=True) + self.assertTrue(make_server._decide_download(self.checkout, confirm)) + confirm.assert_called_once() + + +class TranscribeWavDirTests(unittest.TestCase): + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.folder = Path(self._td.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._td.cleanup() + + def test_transcribes_to_stem_map_with_absolute_paths(self): + transcripts = {str(self.narrator): "First.", + str(self.other): "Second."} + with patch.object(make_server, "transcribe_reference_audio", + side_effect=lambda path, model_name="base": + transcripts[path]): + result = make_server.transcribe_wav_dir( + [self.narrator, self.other], "base") + self.assertEqual(list(result), ["narrator", "other"]) + self.assertEqual(result["narrator"], "First.") + + def test_failed_transcription_keeps_empty_string(self): + with patch.object(make_server, "transcribe_reference_audio", + return_value=None): + result = make_server.transcribe_wav_dir([self.narrator], "base") + self.assertEqual(result["narrator"], "") + + def test_whisper_model_name_passed_through(self): + with patch.object(make_server, "transcribe_reference_audio", + return_value="text") as mock_transcribe: + make_server.transcribe_wav_dir([self.narrator], "large-v3") + self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], + "large-v3") + + def test_write_prompt_text_format(self): + path = make_server.write_prompt_text( + self.folder, {"narrator": "Hello.", "other": "World."}) + self.assertEqual(path, self.folder / make_server.PROMPT_TEXT_FILENAME) + text = path.read_text(encoding="utf-8") + self.assertIn("narrator|Hello.", text) + self.assertIn("other|World.", text) + + +class DesignPackageTests(unittest.TestCase): + """Voice-design package detection.""" + + def test_detects_voicedesign_in_id(self): + self.assertTrue(make_server.is_design_package( + {"id": "qwen3_tts_1_7b_voicedesign_q8_0"})) + + def test_detects_voicedesign_in_directory(self): + self.assertTrue(make_server.is_design_package( + {"target_directory": "Foo-VoiceDesign-GGUF"})) + + def test_detects_separated_voice_design(self): + self.assertTrue(make_server.is_design_package( + {"display_name": "Voice Design Q8_0"})) + + def test_ignores_other_packages(self): + self.assertFalse(make_server.is_design_package( + {"id": "higgs_audio_tts_4b_q8_0"})) + self.assertFalse(make_server.is_design_package({})) + + +class PackageDirOptionsTests(unittest.TestCase): + """Grouping a family's packages into distinct target directories.""" + + def test_groups_precisions_and_marks_recommended(self): + entry = { + "family": "qwen3_tts", + "packages": [ + {"id": "base_q8", "default": True, "format": "gguf", + "target_directory": "Base-GGUF"}, + {"id": "base_bf16", "format": "gguf", + "target_directory": "Base-GGUF"}, + {"id": "voicedesign_q8", "format": "gguf", + "target_directory": "VoiceDesign-GGUF"}, + ], + } + options = make_server.package_dir_options(entry) + self.assertEqual([o["target_directory"] for o in options], + ["Base-GGUF", "VoiceDesign-GGUF"]) + self.assertTrue(options[0]["recommended"]) + self.assertFalse(options[0]["design"]) + self.assertFalse(options[1]["recommended"]) + self.assertTrue(options[1]["design"]) + self.assertEqual(options[0]["install_id"], "base_q8") + + def test_recommended_comes_first_even_if_listed_later(self): + entry = { + "family": "demo_tts", + "packages": [ + {"id": "demo_other", "format": "gguf", + "target_directory": "Other-GGUF"}, + {"id": "demo_default", "default": True, "format": "gguf", + "target_directory": "Default-GGUF"}, + ], + } + options = make_server.package_dir_options(entry) + self.assertEqual([o["target_directory"] for o in options], + ["Default-GGUF", "Other-GGUF"]) + + +class FindAudiocppServerBinTests(unittest.TestCase): + """Locating the built audiocpp_server binary.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = Path(self._td.name) / "audio.cpp" + self.checkout.mkdir() + + def tearDown(self): + self._td.cleanup() + + def _build(self, name, binary="audiocpp_server"): + bin_dir = self.checkout / "build" / name / "bin" + bin_dir.mkdir(parents=True) + (bin_dir / binary).write_bytes(b"x") + + def test_no_build_dir_returns_none(self): + self.assertIsNone(make_server.find_audiocpp_server_bin(self.checkout)) + + def test_finds_built_binary(self): + self._build("linux-cuda-release") + self.assertEqual( + make_server.find_audiocpp_server_bin(self.checkout), + self.checkout / "build" / "linux-cuda-release" / "bin" + / "audiocpp_server") + + def test_finds_windows_exe(self): + self._build("windows-vulkan-debug", binary="audiocpp_server.exe") + self.assertEqual( + make_server.find_audiocpp_server_bin(self.checkout).name, + "audiocpp_server.exe") + + def test_build_dir_without_binary_returns_none(self): + (self.checkout / "build" / "linux-cuda-release" / "bin").mkdir( + parents=True) + self.assertIsNone(make_server.find_audiocpp_server_bin(self.checkout)) + + +class BuildAudiocppTests(unittest.TestCase): + """Running the audio.cpp build helper script.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = Path(self._td.name) / "audio.cpp" + self.checkout.mkdir() + self.scripts = self.checkout / "scripts" + self.scripts.mkdir() + (self.scripts / "build_linux.sh").write_text("#!/bin/sh\n", + encoding="utf-8") + + def tearDown(self): + self._td.cleanup() + + def test_runs_build_script_with_backend_and_target(self): + with patch.object(make_server.common, "run_console_subprocess", + return_value=0) as run: + rc = make_server.build_audiocpp(self.checkout, "cuda") + self.assertEqual(rc, 0) + argv = run.call_args[0][0] + self.assertEqual(argv[:3], ["sh", str(self.scripts / "build_linux.sh"), + "--backend"]) + self.assertIn("cuda", argv) + self.assertIn("--target", argv) + self.assertIn("audiocpp_server", argv) + self.assertEqual(run.call_args[1]["cwd"], self.checkout) + + def test_missing_script_returns_nonzero(self): + for f in self.scripts.iterdir(): + f.unlink() + rc = make_server.build_audiocpp(self.checkout, "cuda") + self.assertNotEqual(rc, 0) + + +class AudiocppDetectTests(unittest.TestCase): + """backends.audiocpp.detect() status reporting.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.root = Path(self._td.name) + self.checkout = _make_checkout(self.root) + + def tearDown(self): + self._td.cleanup() + + def test_not_cloned(self): + with patch.object(make_server, "find_local_checkout", return_value=None): + status = make_server.detect() + self.assertFalse(status.installed) + self.assertFalse(status.configured) + self.assertIn("not cloned", status.details[0]) + + def test_cloned_not_built_not_configured(self): + with patch.object(make_server, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server, "find_audiocpp_server_bin", + return_value=None): + status = make_server.detect() + self.assertFalse(status.installed) + self.assertFalse(status.configured) + self.assertEqual(status.launch_hint, "") + + def test_built_and_configured_ready(self): + binary = self.checkout / "build" / "linux-cuda-release" / "bin" \ + / "audiocpp_server" + binary.parent.mkdir(parents=True) + binary.write_bytes(b"x") + server_json = self.checkout / "server.json" + server_json.write_text('{"models":[]}', encoding="utf-8") + with patch.object(make_server, "find_local_checkout", + return_value=self.checkout): + status = make_server.detect() + self.assertTrue(status.installed) + self.assertTrue(status.configured) + self.assertIn(str(binary), status.launch_hint) + self.assertIn(str(server_json), status.launch_hint) + + +class NonInteractiveMainTests(unittest.TestCase): + """The flag-only (non-TUI) path through main(), end to end.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.root = Path(self._td.name) + self.folder = self.root / "wavs" + self.folder.mkdir() + self.output = self.root / "server.json" + self.checkout = _make_checkout(self.root) + # Isolate config.py rewrites so no test touches the real one. + self.fake_config = self.root / "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) + # Tests run without a tty -> main() takes the non-interactive path. + patcher = patch.object(make_server, "_interactive", return_value=False) + patcher.start() + self.addCleanup(patcher.stop) + + def tearDown(self): + self._td.cleanup() + + def _run(self, argv, transcribe=None, whisper="faster_whisper"): + argv = ["backends/audiocpp.py"] + argv + transcribe_effect = transcribe if transcribe is not None \ + else MagicMock() + with patch.object(sys, "argv", argv), \ + patch.object(make_server, "transcribe_reference_audio", + side_effect=transcribe_effect), \ + patch.object(make_server, "whisper_backend_available", + return_value=whisper): + return make_server.main() + + def _args(self, *extra): + return ["--wavs", str(self.folder), "--output", str(self.output), + "--audiocpp-dir", str(self.checkout)] + list(extra) + + def test_default_run_hosts_recommended_entry(self): + exit_code = self._run( + self._args("--families", "higgs_audio_tts", "--no-sync-model-ids")) + 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([m["id"] for m in data["models"]], ["higgs"]) + self.assertNotIn("voice_dir", data) + + def test_port_sync_accepted_updates_config(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:9999"): + exit_code = self._run( + self._args("--families", "higgs_audio_tts", "--port", "8080", + "--no-sync-model-ids")) + 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"): + exit_code = self._run( + self._args("--families", "higgs_audio_tts", "--port", "8080", + "--no-sync-port", "--no-sync-model-ids")) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:9999"', + self.fake_config.read_text(encoding="utf-8")) + + def test_model_id_sync_accepted_updates_config(self): + self.fake_config.write_text(FAKE_CONFIG_WITH_MODEL_IDS, + encoding="utf-8") + exit_code = self._run(self._args("--families", "higgs_audio_tts")) + self.assertEqual(exit_code, 0) + text = self.fake_config.read_text(encoding="utf-8") + self.assertIn('AUDIOCPP_MODEL_ID = "higgs"', text) + self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "higgs"', text) + + def test_multi_family_lazy_with_voice_dir(self): + (self.folder / "narrator.wav").write_bytes(b"x") + exit_code = self._run( + self._args("--families", "qwen3_tts,higgs_audio_tts", + "--no-sync-model-ids"), + transcribe=lambda path, model_name="base": "a transcript") + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual([m["id"] for m in data["models"]], ["qwen", "higgs"]) + self.assertTrue(data["lazy_load"]) + self.assertEqual(data["voice_dir"], str(self.folder.resolve())) + prompt = (self.folder / make_server.PROMPT_TEXT_FILENAME).read_text( + encoding="utf-8") + self.assertIn("narrator|a transcript", prompt) + + def test_force_overwrites_existing_output(self): + self.output.write_text('{"old": true}', encoding="utf-8") + exit_code = self._run( + self._args("--families", "higgs_audio_tts", "--force", + "--no-sync-model-ids")) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 1) + + def test_existing_output_declined_keeps_file(self): + self.output.write_text('{"old": true}', encoding="utf-8") + exit_code = self._run( + self._args("--families", "higgs_audio_tts", "--no-sync-model-ids")) + self.assertEqual(exit_code, 1) + self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), + {"old": True}) + + def test_all_packages_hosts_design_as_vdes(self): + exit_code = self._run( + self._args("--families", "qwen3_tts", "--all-packages", + "--no-sync-model-ids")) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + by_id = {m["id"]: m for m in data["models"]} + self.assertIn("qwen-design", by_id) + self.assertEqual(by_id["qwen-design"]["task"], "vdes") + # The non-design packages are hosted with task "tts". + self.assertTrue(any(m["id"] in ("qwen", "qwen-2") and m["task"] == "tts" + for m in data["models"])) + + def test_unknown_family_rejected(self): + with self.assertRaises(SystemExit) as ctx: + self._run(self._args("--families", "not_a_family", + "--no-sync-model-ids")) + self.assertEqual(ctx.exception.code, 2) + + def test_missing_checkout_rejected(self): + with patch.object(make_server, "find_local_checkout", + return_value=None), \ + self.assertRaises(SystemExit) as ctx: + self._run(["--families", "higgs_audio_tts", "--output", + str(self.output), "--no-sync-model-ids"]) + self.assertEqual(ctx.exception.code, 2) + + def test_missing_wav_dir_rejected(self): + missing = self.root / "nope" + with self.assertRaises(SystemExit) as ctx: + self._run(["--wavs", str(missing), "--output", str(self.output), + "--audiocpp-dir", str(self.checkout), + "--families", "higgs_audio_tts", "--no-sync-model-ids"]) + self.assertEqual(ctx.exception.code, 2) + + def test_families_required_in_noninteractive_run(self): + with self.assertRaises(SystemExit) as ctx: + self._run(self._args("--no-sync-model-ids")) + self.assertEqual(ctx.exception.code, 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_backends_envs.py b/app/tests/test_backends_envs.py new file mode 100644 index 0000000..cf4ecc6 --- /dev/null +++ b/app/tests/test_backends_envs.py @@ -0,0 +1,204 @@ +"""Tests for the managed Python environment (backends/envs.py).""" + +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +from backends import envs + + +class EnvPathTests(unittest.TestCase): + """Platform-aware path helpers (no venv actually created).""" + + def test_env_dir_under_envs_tts(self): + self.assertEqual(envs.ENV_DIR.name, "tts") + self.assertEqual(envs.ENV_DIR.parent.name, "envs") + + def test_env_python_posix(self): + with patch.object(envs, "_is_windows", return_value=False): + self.assertEqual(envs.env_python(), + envs.ENV_DIR / "bin" / "python") + + def test_env_python_windows(self): + with patch.object(envs, "_is_windows", return_value=True): + self.assertEqual(envs.env_python(), + envs.ENV_DIR / "Scripts" / "python.exe") + + def test_env_script_posix(self): + with patch.object(envs, "_is_windows", return_value=False): + self.assertEqual(envs.env_script("qwen-tts-demo"), + envs.ENV_DIR / "bin" / "qwen-tts-demo") + + def test_env_script_windows(self): + with patch.object(envs, "_is_windows", return_value=True): + self.assertEqual(envs.env_script("qwen-tts-demo"), + envs.ENV_DIR / "Scripts" / "qwen-tts-demo.exe") + + def test_env_exists_false_when_python_missing(self): + with patch.object(envs, "env_python", + return_value=Path("/no/such/path/python")): + self.assertFalse(envs.env_exists()) + + def test_is_managed_env_compares_resolved_executable(self): + fake_env_python = Path("/tmp/opencode/managed-env/bin/python") + with patch.object(envs, "env_python", return_value=fake_env_python), \ + patch.object(sys, "executable", str(fake_env_python)): + self.assertTrue(envs.is_managed_env()) + with patch.object(envs, "env_python", return_value=fake_env_python), \ + patch.object(sys, "executable", "/usr/bin/python3"): + self.assertFalse(envs.is_managed_env()) + + +class CreateEnvTests(unittest.TestCase): + def test_create_env_invokes_venv_module(self): + with patch.object(envs.common, "run_console_subprocess", + return_value=0) as run: + rc = envs.create_env() + self.assertEqual(rc, 0) + argv = run.call_args[0][0] + self.assertEqual(argv[0], sys.executable) + self.assertEqual(argv[1], "-m") + self.assertEqual(argv[2], "venv") + self.assertEqual(argv[3], str(envs.ENV_DIR)) + + def test_create_env_reports_remediation_on_failure(self): + with patch.object(envs.common, "run_console_subprocess", + return_value=1): + rc = envs.create_env() + self.assertEqual(rc, 1) + + +class PipInstallTests(unittest.TestCase): + def test_creates_env_first_when_missing(self): + calls = [] + + def fake_run(argv): + calls.append(list(argv)) + return 0 + + with patch.object(envs, "env_exists", return_value=False), \ + patch.object(envs, "create_env", return_value=0) as mk, \ + patch.object(envs.common, "run_console_subprocess", + side_effect=fake_run): + rc = envs.pip_install(["qwen-tts"]) + self.assertEqual(rc, 0) + mk.assert_called_once_with() + # The actual pip call targets the venv's python. + self.assertEqual(calls[0][0], str(envs.env_python())) + self.assertIn("pip", calls[0]) + self.assertIn("qwen-tts", calls[0]) + + def test_skips_create_when_env_exists(self): + with patch.object(envs, "env_exists", return_value=True), \ + patch.object(envs, "create_env") as mk, \ + patch.object(envs.common, "run_console_subprocess", + return_value=0): + envs.pip_install(["qwen-tts"]) + mk.assert_not_called() + + def test_returns_nonzero_when_create_fails(self): + with patch.object(envs, "env_exists", return_value=False), \ + patch.object(envs, "create_env", return_value=1), \ + patch.object(envs.common, "run_console_subprocess") as run: + rc = envs.pip_install(["qwen-tts"]) + self.assertEqual(rc, 1) + run.assert_not_called() + + +class ModuleAvailableTests(unittest.TestCase): + def test_false_when_env_missing(self): + with patch.object(envs, "env_exists", return_value=False): + self.assertFalse(envs.module_available("qwen_tts")) + + def test_true_when_subprocess_exits_zero(self): + import subprocess + fake = subprocess.CompletedProcess(args=["x"], returncode=0) + with patch.object(envs, "env_exists", return_value=True), \ + patch("subprocess.run", return_value=fake) as run: + self.assertTrue(envs.module_available("qwen_tts")) + argv = run.call_args[0][0] + self.assertEqual(argv[0], str(envs.env_python())) + self.assertIn("import qwen_tts", argv[2]) + + def test_false_when_subprocess_exits_nonzero(self): + import subprocess + fake = subprocess.CompletedProcess(args=["x"], returncode=1) + with patch.object(envs, "env_exists", return_value=True), \ + patch("subprocess.run", return_value=fake): + self.assertFalse(envs.module_available("qwen_tts")) + + def test_false_on_timeout(self): + import subprocess + with patch.object(envs, "env_exists", return_value=True), \ + patch("subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="x", timeout=1)): + self.assertFalse(envs.module_available("qwen_tts")) + + +class EnsureAppEnvTests(unittest.TestCase): + def test_creates_env_then_installs_when_marker_invalid(self): + with patch.object(envs, "env_exists", return_value=False), \ + patch.object(envs, "create_env", return_value=0), \ + patch.object(envs, "_marker_valid", return_value=False), \ + patch.object(envs, "install_requirements", return_value=0), \ + patch.object(envs, "_write_marker") as mk: + envs.ensure_app_env() + mk.assert_called_once_with() + + def test_raises_when_create_fails(self): + with patch.object(envs, "env_exists", return_value=False), \ + patch.object(envs, "create_env", return_value=1): + with self.assertRaises(RuntimeError): + envs.ensure_app_env() + + def test_raises_when_install_fails(self): + with patch.object(envs, "env_exists", return_value=True), \ + patch.object(envs, "_marker_valid", return_value=False), \ + patch.object(envs, "install_requirements", return_value=1): + with self.assertRaises(RuntimeError): + envs.ensure_app_env() + + def test_skips_install_when_marker_valid(self): + with patch.object(envs, "env_exists", return_value=True), \ + patch.object(envs, "_marker_valid", return_value=True), \ + patch.object(envs, "install_requirements") as mk: + envs.ensure_app_env() + mk.assert_not_called() + + +class BootstrapTests(unittest.TestCase): + def test_noop_when_already_managed(self): + with patch.object(envs, "is_managed_env", return_value=True), \ + patch.object(envs, "ensure_app_env") as mk, \ + patch("os.execv") as ex: + envs.bootstrap("/path/to/audiobook.py") + mk.assert_not_called() + ex.assert_not_called() + + def test_ensures_env_then_execvs(self): + with patch.object(envs, "is_managed_env", return_value=False), \ + patch.object(envs, "ensure_app_env") as mk_env, \ + patch("os.execv") as ex, \ + patch.object(sys, "argv", ["audiobook.py", "--backend", "qwen"]): + envs.bootstrap("/path/to/audiobook.py") + mk_env.assert_called_once_with() + py = str(envs.env_python()) + args = ex.call_args[0] + self.assertEqual(args[0], py) + self.assertEqual(args[1][0], py) + self.assertTrue(args[1][1].endswith("audiobook.py")) + self.assertEqual(args[1][2:], ["--backend", "qwen"]) + + def test_exits_when_ensure_raises(self): + with patch.object(envs, "is_managed_env", return_value=False), \ + patch.object(envs, "ensure_app_env", + side_effect=RuntimeError("boom")), \ + patch("os.execv") as ex, \ + self.assertRaises(SystemExit): + envs.bootstrap("/path/to/audiobook.py") + ex.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_backends_faster.py b/app/tests/test_backends_faster.py new file mode 100644 index 0000000..641f6ee --- /dev/null +++ b/app/tests/test_backends_faster.py @@ -0,0 +1,172 @@ +"""Tests for the faster-qwen3-tts backend setup module (backends/faster.py).""" + +import json +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from backends import faster as 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): + """The flag-only (non-TUI) path through main(), end to end.""" + + 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" + # Avoid touching the real converter/config.py and pip/git. + patcher = patch.object(make_voices.common, "update_config_value", + return_value=False) + patcher.start() + self.addCleanup(patcher.stop) + patcher = patch.object(make_voices, "_interactive", return_value=False) + patcher.start() + self.addCleanup(patcher.stop) + + def tearDown(self): + self._tmp.cleanup() + + def _run(self, argv): + with patch.object(sys, "argv", ["backends/faster.py"] + argv), \ + patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"): + return make_voices.main() + + def test_writes_json_with_alphabetical_voice_order(self): + exit_code = self._run([str(self.folder), "--output", str(self.output), + "--skip-install", "--skip-clone"]) + 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" + exit_code = self._run([str(self.folder), "--output", str(custom), + "--skip-install", "--skip-clone"]) + self.assertEqual(exit_code, 0) + 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), "--output", str(self.output), + "--language", "klingon", "--skip-install", + "--skip-clone"]) + 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"), "--output", str(self.output), + "--skip-install", "--skip-clone"]) + self.assertEqual(ctx.exception.code, 2) + + def test_no_wav_files_returns_error(self): + empty = Path(tempfile.mkdtemp()) + try: + exit_code = self._run([str(empty), "--output", + str(empty / "voices.json"), + "--skip-install", "--skip-clone"]) + self.assertEqual(exit_code, 1) + 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: + exit_code = self._run([str(self.folder), "--output", str(self.output), + "--skip-install", "--skip-clone"]) + 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_force_overwrites_without_prompt(self): + self.output.write_text('{"old": true}', encoding="utf-8") + exit_code = self._run([str(self.folder), "--output", str(self.output), + "--force", "--skip-install", "--skip-clone"]) + 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/app/tests/test_backends_servers.py b/app/tests/test_backends_servers.py new file mode 100644 index 0000000..02b65e6 --- /dev/null +++ b/app/tests/test_backends_servers.py @@ -0,0 +1,146 @@ +"""Tests for the server lifecycle module (backends/servers.py).""" + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from backends import ServerSpec, servers + + +class StartTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + # A fake executable so Path(argv[0]).exists() passes. + self.exe = self.dir / "fake_server" + self.exe.write_bytes(b"#!/bin/sh\n") + self.spec = ServerSpec("test", "http://127.0.0.1:9999", + [str(self.exe), "--port", "9999"]) + + def tearDown(self): + self._tmp.cleanup() + + def test_returns_false_when_executable_missing(self): + spec = ServerSpec("nope", "http://127.0.0.1:1", ["/no/such/binary"]) + with patch.object(servers, "LOG_DIR", self.dir): + self.assertFalse(servers.start(spec)) + + def test_noop_when_already_running(self): + with patch.object(servers, "LOG_DIR", self.dir), \ + patch("backends.common.server_running", return_value=True), \ + patch("subprocess.Popen") as mk: + self.assertTrue(servers.start(self.spec)) + mk.assert_not_called() + + def test_happy_path_spawns_and_polls_until_ready(self): + proc = MagicMock() + proc.pid = 4242 + proc.poll.return_value = None # process still running + # server_running: False on the pre-check, True once inside the loop. + with patch.object(servers, "LOG_DIR", self.dir), \ + patch("subprocess.Popen", return_value=proc) as mk, \ + patch("backends.common.server_running", + side_effect=[False, True]), \ + patch("time.sleep"): + ok = servers.start(self.spec) + self.assertTrue(ok) + mk.assert_called_once() + # Pid file written. + self.assertEqual( + (self.dir / "test-server.pid").read_text(encoding="utf-8"), + "4242") + + def test_returns_false_when_process_exits_early(self): + proc = MagicMock() + proc.pid = 99 + proc.poll.return_value = 1 # exited with code 1 + with patch.object(servers, "LOG_DIR", self.dir), \ + patch("subprocess.Popen", return_value=proc), \ + patch("backends.common.server_running", return_value=False), \ + patch("time.sleep"): + ok = servers.start(self.spec) + self.assertFalse(ok) + # Pid file cleaned up after early exit. + self.assertFalse((self.dir / "test-server.pid").exists()) + + def test_returns_false_on_timeout(self): + proc = MagicMock() + proc.pid = 7 + proc.poll.return_value = None + # time.time: first call < deadline loop entry, then past deadline. + times = iter([0.0, float(servers.SERVER_START_TIMEOUT + 1)]) + with patch.object(servers, "LOG_DIR", self.dir), \ + patch("subprocess.Popen", return_value=proc), \ + patch("backends.common.server_running", return_value=False), \ + patch("time.sleep"), \ + patch("time.time", side_effect=lambda: next(times)): + ok = servers.start(self.spec) + self.assertFalse(ok) + + +class StopTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def _write_pid(self, name, pid): + (self.dir / f"{name}-server.pid").write_text(str(pid), + encoding="utf-8") + + def test_returns_false_when_no_pid_file(self): + with patch.object(servers, "LOG_DIR", self.dir): + self.assertFalse(servers.stop("test")) + + def test_stops_alive_process_and_removes_pid_file(self): + self._write_pid("test", 1234) + with patch.object(servers, "LOG_DIR", self.dir), \ + patch.object(servers, "_pid_alive", return_value=True), \ + patch.object(servers, "_kill_pid", return_value=True) as mk: + ok = servers.stop("test") + self.assertTrue(ok) + mk.assert_called_once_with(1234) + self.assertFalse((self.dir / "test-server.pid").exists()) + + def test_already_dead_returns_true_and_cleans_pid_file(self): + self._write_pid("test", 1234) + with patch.object(servers, "LOG_DIR", self.dir), \ + patch.object(servers, "_pid_alive", return_value=False), \ + patch.object(servers, "_kill_pid") as mk: + ok = servers.stop("test") + self.assertTrue(ok) + mk.assert_not_called() + self.assertFalse((self.dir / "test-server.pid").exists()) + + def test_corrupt_pid_file_returns_false_and_cleans(self): + (self.dir / "test-server.pid").write_text("not-a-number", + encoding="utf-8") + with patch.object(servers, "LOG_DIR", self.dir): + self.assertFalse(servers.stop("test")) + self.assertFalse((self.dir / "test-server.pid").exists()) + + +class PidForTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_none_when_no_pid_file(self): + with patch.object(servers, "LOG_DIR", self.dir): + self.assertIsNone(servers.pid_for("test")) + + def test_returns_pid_from_file(self): + (self.dir / "test-server.pid").write_text("555\n", + encoding="utf-8") + with patch.object(servers, "LOG_DIR", self.dir): + self.assertEqual(servers.pid_for("test"), 555) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_chunking.py b/app/tests/test_chunking.py new file mode 100644 index 0000000..2904e40 --- /dev/null +++ b/app/tests/test_chunking.py @@ -0,0 +1,118 @@ +"""Tests for text chunking.""" + +import unittest +from unittest.mock import patch + +from converter import config +from converter.chunking import split_into_chunks + + +class ChunkSizeDefaultTests(unittest.TestCase): + """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 Qwen + 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_single_generation_budget(self): + self.assertLessEqual(config.CHUNK_SIZE, 300) + + def test_default_chunk_size_is_positive(self): + self.assertGreaterEqual(config.CHUNK_SIZE, 1) + + +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)) + "." + chunks = split_into_chunks(text, max_words=5000) + self.assertEqual(len(chunks), 1) + self.assertEqual(len(chunks[0].split()), 30) + + 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)) + with patch.object(config, "CHUNK_SIZE", 120): + chunks = split_into_chunks(sentences) + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(chunk.split()) <= 120 for chunk in chunks)) + + +class SplitIntoChunksTests(unittest.TestCase): + def test_empty_input(self): + self.assertEqual(split_into_chunks(""), []) + self.assertEqual(split_into_chunks(" \n "), []) + + def test_short_text_single_chunk(self): + self.assertEqual(split_into_chunks("One short sentence."), ["One short sentence."]) + + def test_respects_word_limit_across_sentences(self): + # 10 sentences of 9 words each = 90 words total + sentences = [f"S{i} " + " ".join(["word"] * 8) + "." for i in range(10)] + chunks = split_into_chunks(" ".join(sentences), max_words=25) + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(c.split()) <= 25 for c in chunks)) + self.assertEqual(sum(len(c.split()) for c in chunks), 90) + + def test_long_sentence_split_keeps_punctuation(self): + # 10 clauses of 5 words each, joined by comma+space + sentence = ", ".join([" ".join(["w"] * 5) for _ in range(10)]) + "." + chunks = split_into_chunks(sentence, max_words=12) + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(c.split()) <= 12 for c in chunks)) + self.assertIn(",", chunks[0]) # commas retained for TTS prosody + + def test_clause_split_never_breaks_numbers(self): + # Regression: the clause split used to fire at every comma even + # without whitespace, mutating "1,000,000" into "1, 000, 000". + sentence = ("There were exactly 1,000,000 soldiers marching at 12:30, " + + "and they kept marching onward " * 30) + "endlessly." + chunks = split_into_chunks(sentence, max_words=25) + self.assertGreater(len(chunks), 1) + joined = " ".join(chunks) + self.assertIn("1,000,000", joined) + self.assertIn("12:30", joined) + self.assertNotIn("1, 000", joined) + self.assertNotIn("000, 000", joined) + self.assertNotIn("12: 30", joined) + + def test_clause_split_requires_whitespace_after_punctuation(self): + # Run-on clauses without spaces after commas have no clause split + # point, so the last-resort word-boundary split fires instead. + # Tokens themselves (and numbers like "1,000,000") stay intact. + sentence = ",".join([" ".join(["w"] * 5) for _ in range(10)]) + "." + chunks = split_into_chunks(sentence, max_words=12) + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(chunk.split()) <= 12 for chunk in chunks)) + tokens = sentence.replace(",", " , ").split() + rejoined = " ".join(chunks).replace(",", " , ").split() + self.assertEqual(rejoined, tokens) + + def test_single_oversized_sentence_is_word_split(self): + # A punctuation-free sentence longer than the limit is split at word + # 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) + self.assertTrue(all(len(chunk.split()) <= 10 for chunk in chunks)) + self.assertEqual(sum(len(chunk.split()) for chunk in chunks), 30) + + def test_word_split_never_breaks_number_tokens(self): + # Numbers and other punctuation-bearing tokens are single words and + # must never be broken apart by the last-resort word split. + sentence = ("There were exactly 1,000,000 soldiers marching at 12:30 " + "and " + "they kept marching onward " * 20) + "endlessly." + chunks = split_into_chunks(sentence, max_words=10) + self.assertGreater(len(chunks), 1) + joined = " ".join(chunks) + self.assertIn("1,000,000", joined) + self.assertIn("12:30", joined) + self.assertNotIn("1, 000", joined) + self.assertNotIn("12: 30", joined) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_cleaning.py b/app/tests/test_cleaning.py new file mode 100644 index 0000000..41f4ed7 --- /dev/null +++ b/app/tests/test_cleaning.py @@ -0,0 +1,54 @@ +"""Tests for text and HTML cleaning.""" + +import unittest + +from converter.extractors import clean_html, clean_text + + +class CleanTextTests(unittest.TestCase): + def test_empty_input(self): + self.assertEqual(clean_text(""), "") + self.assertEqual(clean_text(None), "") + + def test_collapses_whitespace(self): + self.assertEqual(clean_text("a\n\n b \t c"), "a b c") + + def test_preserves_inline_numbers(self): + self.assertEqual(clean_text("He was 42 years old."), "He was 42 years old.") + + def test_preserves_grouped_and_decimal_numbers(self): + self.assertEqual( + clean_text("Over 1,000 pages and 3.5 stars."), + "Over 1,000 pages and 3.5 stars.", + ) + + def test_removes_standalone_page_numbers(self): + self.assertEqual( + clean_text("End of page.\n7\nNext page text."), + "End of page. Next page text.", + ) + + def test_page_number_removal_leaves_single_spacing(self): + result = clean_text("Chapter one\n\n12\n\nChapter two") + self.assertEqual(result, "Chapter one Chapter two") + self.assertNotIn(" ", result) + + +class CleanHtmlTests(unittest.TestCase): + def test_strips_tags(self): + self.assertEqual(clean_html("

Hello world

"), "Hello world") + + def test_removes_script_and_style(self): + html = "

Text

" + self.assertEqual(clean_html(html), "Text") + + def test_unescapes_entities(self): + self.assertEqual(clean_html("Tom & Jerry"), "Tom & Jerry") + + def test_empty(self): + self.assertEqual(clean_html(""), "") + self.assertEqual(clean_html(None), "") + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_converter.py b/app/tests/test_converter.py new file mode 100644 index 0000000..2fe0f5d --- /dev/null +++ b/app/tests/test_converter.py @@ -0,0 +1,619 @@ +"""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 + +from converter import config, tts +from converter import converter as converter_mod +from converter.converter import ( + AudiobookConverter, + find_existing_outputs, + prompt_overwrite, + setup_logging, +) + + +class SanitizeFilenameTests(unittest.TestCase): + def test_removes_invalid_characters(self): + self.assertEqual(AudiobookConverter._sanitize_filename('A "bad" name: here'), + "A bad name here") + + def test_collapses_whitespace(self): + self.assertEqual(AudiobookConverter._sanitize_filename(" spaced\tout "), "spaced out") + + def test_empty_falls_back(self): + self.assertEqual(AudiobookConverter._sanitize_filename("///"), "chapter") + + +class ConfigurationValidationTests(unittest.TestCase): + def test_invalid_voice_mode_rejected(self): + with self.assertRaises(ValueError): + AudiobookConverter(voice_mode="custon_voice") + + def test_nonpositive_speed_rejected(self): + with self.assertRaises(ValueError): + AudiobookConverter(speed=0) + + def test_unknown_format_rejected(self): + with self.assertRaises(ValueError): + AudiobookConverter(output_format="wma") + + def test_unknown_language_rejected(self): + with self.assertRaises(ValueError): + AudiobookConverter(language="klingon") + + def test_unknown_backend_rejected(self): + with self.assertRaises(ValueError) as ctx: + AudiobookConverter(backend="piper") + self.assertIn("piper", str(ctx.exception)) + self.assertIn("audiocpp", str(ctx.exception)) + + def test_language_defaults_to_config(self): + with patch("converter.converter.QwenTTSClient") as mock_tts: + AudiobookConverter(backend=tts.BACKEND_QWEN) + self.assertEqual(mock_tts.call_args.kwargs["language"], config.LANGUAGE) + + def test_output_format_defaults_to_config(self): + with patch("converter.converter.QwenTTSClient"): + converter = AudiobookConverter(backend=tts.BACKEND_QWEN) + self.assertEqual(converter.output_format, config.AUDIO_FORMAT) + + def test_language_normalized_before_tts_client(self): + with patch("converter.converter.QwenTTSClient") as mock_tts: + converter = AudiobookConverter(language="ja", backend=tts.BACKEND_QWEN) + self.assertEqual(converter.language, "Japanese") + self.assertEqual(mock_tts.call_args.kwargs["language"], "Japanese") + + +class FindExistingOutputsTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self._original = converter_mod.AUDIOBOOKS_FOLDER + converter_mod.AUDIOBOOKS_FOLDER = self.folder + + def tearDown(self): + converter_mod.AUDIOBOOKS_FOLDER = self._original + self._tmp.cleanup() + + def _touch(self, name): + path = self.folder / name + path.write_bytes(b"x") + return path + + def test_no_existing_output(self): + self.assertEqual(find_existing_outputs("dune", "mp3"), []) + + def test_primary_output_detected(self): + self._touch("dune.mp3") + self.assertEqual([p.name for p in find_existing_outputs("dune", "mp3")], + ["dune.mp3"]) + + def test_chapter_and_speed_copies_detected(self): + for name in ("dune_01_Dune.mp3", "dune_02_Barony.mp3", "dune_1.5x.mp3"): + self._touch(name) + self._touch("dune2_01.mp3") # different book stem; must not match + found = [p.name for p in find_existing_outputs("dune", "mp3")] + self.assertEqual(len(found), 3) + + def test_other_extensions_ignored(self): + self._touch("dune.mp3") + self.assertEqual(find_existing_outputs("dune", "m4b"), []) + + def test_glob_metacharacters_in_stem(self): + self._touch("book [1].mp3") + self._touch("book [1]_1.5x.mp3") + found = [p.name for p in find_existing_outputs("book [1]", "mp3")] + self.assertEqual(sorted(found), ["book [1].mp3", "book [1]_1.5x.mp3"]) + + def test_narrator_named_outputs_detected(self): + for name in ("dune_Vivian.mp3", "dune_Vivian_1.5.mp3", "dune_Vivian_01_Dune.mp3"): + self._touch(name) + found = [p.name for p in find_existing_outputs("dune_Vivian", "mp3")] + self.assertEqual(len(found), 3) + + def test_legacy_outputs_without_narrator_ignored(self): + self._touch("dune.mp3") + self._touch("dune_1.5.mp3") + self.assertEqual(find_existing_outputs("dune_Vivian", "mp3"), []) + + +class NarratorTagTests(unittest.TestCase): + def _converter(self, voice_mode, ref_audio=None, instructions=None): + converter = AudiobookConverter.__new__(AudiobookConverter) + converter.voice_mode = voice_mode + converter.voice_clone_ref_audio = ref_audio + converter.backend = tts.BACKEND_QWEN + converter.voice = None + converter.instructions = instructions + return converter + + def test_custom_voice_uses_speaker_display_name(self): + self.assertEqual(self._converter(tts.VOICE_MODE_CUSTOM)._narrator_tag(), + "Vivian") + + def test_multi_word_display_name_gets_underscores(self): + with patch.object(config, "SPEAKER", "uncle_fu"): + self.assertEqual(self._converter(tts.VOICE_MODE_CUSTOM)._narrator_tag(), + "Uncle_Fu") + + def test_clone_uses_reference_audio_stem(self): + self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/ref.wav")._narrator_tag(), + "ref") + + def test_clone_stem_spaces_become_underscores(self): + self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/my voice.wav")._narrator_tag(), + "my_voice") + + def test_invalid_characters_sanitized(self): + self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/bad:name?.wav")._narrator_tag(), + "bad_name") + + def test_empty_after_sanitize_falls_back(self): + self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/???.wav")._narrator_tag(), + "narrator") + + def _audiocpp_converter(self, voice=None, instructions=None): + converter = self._converter(tts.VOICE_MODE_CUSTOM, + instructions=instructions) + converter.backend = tts.BACKEND_AUDIOCPP + converter.voice = voice + return converter + + def test_audiocpp_design_run_uses_designed_tag(self): + # An instruction without a voice (voice design, or instruction- + # defined voices) must not be named after the built-in speaker. + converter = self._audiocpp_converter(instructions="A warm narrator") + self.assertEqual(converter._narrator_tag(), "designed") + + def test_audiocpp_instruction_with_voice_keeps_voice_tag(self): + converter = self._audiocpp_converter( + voice="narrator", instructions="Calm delivery") + self.assertEqual(converter._narrator_tag(), "narrator") + + def test_audiocpp_speaker_mode_keeps_speaker_tag(self): + converter = self._audiocpp_converter() + self.assertEqual(converter._narrator_tag(), "Vivian") + + def test_preflight_design_run_uses_designed_tag(self): + with tempfile.TemporaryDirectory() as books_tmp, \ + tempfile.TemporaryDirectory() as output_tmp: + original = (converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER) + converter_mod.BOOKS_FOLDER = Path(books_tmp) + converter_mod.AUDIOBOOKS_FOLDER = Path(output_tmp) + try: + (converter_mod.BOOKS_FOLDER / "book.txt").write_text( + "hello world", encoding="utf-8") + with patch("builtins.input", + side_effect=AssertionError("should not prompt")): + _, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_AUDIOCPP, None, tts.VOICE_MODE_CUSTOM, + None, "mp3", instructions="A warm narrator") + self.assertEqual(planned, [(converter_mod.BOOKS_FOLDER / "book.txt", + "book_designed")]) + finally: + converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = original + + +class ChapterDebugDirTests(unittest.TestCase): + """Per-chapter debug subfolder naming (chunk numbering restarts per chapter).""" + + def test_none_when_not_debugging(self): + self.assertIsNone(AudiobookConverter._chapter_debug_dir(None, 3, "The Trial")) + + def test_chapter_subfolder_named_by_index_and_title(self): + book_dir = Path("debug") / "dune_Vivian" + chapter_dir = AudiobookConverter._chapter_debug_dir(book_dir, 3, "The Trial") + self.assertEqual(chapter_dir, book_dir / "03_The Trial") + + def test_untitled_chapter_uses_fallback(self): + chapter_dir = AudiobookConverter._chapter_debug_dir(Path("d"), 1, "") + self.assertEqual(chapter_dir, Path("d") / "01_chapter") + + +class DebugDumpTests(unittest.TestCase): + """--debug: per-chunk text/audio dumps and request/response logging.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._debug_folder = patch.object(converter_mod, "DEBUG_FOLDER", Path(self._tmp.name)) + 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): + self._debug_folder.stop() + self._tmp.cleanup() + + def _chunk_source(self, name, body=b"audio"): + path = self.debug_root / "sources" / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(body) + return path + + def test_successful_chunk_dumps_text_and_audio(self): + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.return_value = audio + results = self.converter._synthesize_chunks(["Hello world."], + debug_dir=self.debug_root / "book") + self.assertEqual(results, {1: audio}) + debug_dir = self.debug_root / "book" + self.assertEqual((debug_dir / "chunk_0001.txt").read_text(encoding="utf-8"), + "Hello world.") + self.assertEqual((debug_dir / "chunk_0001.wav").read_bytes(), b"audio") + + def test_failed_chunk_dumps_text_but_no_audio(self): + self.converter.tts.process_chunk_with_retry.return_value = None + results = self.converter._synthesize_chunks(["Hello again."], + debug_dir=self.debug_root / "book") + self.assertEqual(results, {1: None}) + debug_dir = self.debug_root / "book" + self.assertEqual([path.name for path in sorted(debug_dir.iterdir())], + ["chunk_0001.txt"]) + + def test_text_dumped_even_when_request_raises(self): + self.converter.tts.process_chunk_with_retry.side_effect = RuntimeError("boom") + results = self.converter._synthesize_chunks(["Crash text."], + debug_dir=self.debug_root / "book") + self.assertEqual(results, {1: None}) + self.assertEqual((self.debug_root / "book" / "chunk_0001.txt").read_text( + encoding="utf-8"), "Crash text.") + + def test_audio_suffix_preserved_and_nested_dirs_created(self): + audio = self._chunk_source("generated.mp3") + self.converter.tts.process_chunk_with_retry.return_value = audio + self.converter._synthesize_chunks(["Hello."], + debug_dir=self.debug_root / "nested" / "book") + self.assertTrue((self.debug_root / "nested" / "book" / "chunk_0001.mp3").exists()) + + def test_no_debug_dir_writes_nothing(self): + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.return_value = audio + results = self.converter._synthesize_chunks(["Hello world."]) + self.assertEqual(results, {1: audio}) + self.assertEqual([path.name for path in self.debug_root.iterdir()], ["sources"]) + + def test_request_and_response_are_logged(self): + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.return_value = audio + with self.assertLogs("converter.converter", level="DEBUG") as logs: + self.converter._synthesize_chunks(["Hello world."], + debug_dir=self.debug_root / "book") + joined = "\n".join(logs.output) + self.assertIn("Chunk 1/1 request text: Hello world.", joined) + self.assertIn("Chunk 1/1 response in", joined) + self.assertIn("chunk_0001.wav", joined) + + def test_debug_write_failure_does_not_abort_conversion(self): + blocker = self.debug_root / "blocker" + blocker.write_bytes(b"") + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.return_value = audio + results = self.converter._synthesize_chunks(["Hello."], debug_dir=blocker / "book") + self.assertEqual(results, {1: audio}) + + def test_failed_chunk_stops_remaining_chunks(self): + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.side_effect = [audio, None, audio] + results = self.converter._synthesize_chunks(["One.", "Two.", "Three."]) + self.assertEqual(results, {1: audio, 2: None}) + self.assertEqual(self.converter.tts.process_chunk_with_retry.call_count, 2) + + def test_raising_chunk_stops_remaining_chunks(self): + audio = self._chunk_source("chunk_0001.wav") + self.converter.tts.process_chunk_with_retry.side_effect = [audio, RuntimeError("boom")] + results = self.converter._synthesize_chunks(["One.", "Two.", "Three."]) + self.assertEqual(results, {1: audio, 2: None}) + self.assertEqual(self.converter.tts.process_chunk_with_retry.call_count, 2) + + def test_debug_flag_wiring(self): + with patch("converter.converter.QwenTTSClient"): + self.assertFalse(AudiobookConverter(backend=tts.BACKEND_QWEN).debug) + self.assertTrue(AudiobookConverter(debug=True, backend=tts.BACKEND_QWEN).debug) + + +class SetupLoggingTests(unittest.TestCase): + """Console handler stays quiet; the log file keeps the full record.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._logs_folder = patch.object(converter_mod, "LOGS_FOLDER", Path(self._tmp.name)) + self._logs_folder.start() + self._root = logging.getLogger() + self._saved_handlers = self._root.handlers[:] + self._saved_level = self._root.level + self._saved_converter_level = logging.getLogger("converter").level + self._root.handlers.clear() + + def tearDown(self): + for handler in self._root.handlers: + if handler not in self._saved_handlers: + handler.close() + self._root.handlers[:] = self._saved_handlers + self._root.setLevel(self._saved_level) + logging.getLogger("converter").setLevel(self._saved_converter_level) + self._logs_folder.stop() + self._tmp.cleanup() + + def _console_handler(self): + matches = [h for h in logging.getLogger().handlers + if isinstance(h, logging.StreamHandler) + and not isinstance(h, logging.FileHandler)] + self.assertEqual(len(matches), 1) + return matches[0] + + def _file_handler(self): + matches = [h for h in logging.getLogger().handlers + if isinstance(h, logging.FileHandler)] + self.assertEqual(len(matches), 1) + return matches[0] + + def test_console_quiet_and_file_verbose_by_default(self): + setup_logging() + self.assertEqual(self._console_handler().level, logging.WARNING) + self.assertEqual(self._file_handler().level, logging.INFO) + + def test_debug_flag_lowers_both_handlers(self): + setup_logging(debug=True) + self.assertEqual(self._console_handler().level, logging.DEBUG) + self.assertEqual(self._file_handler().level, logging.DEBUG) + + def test_http_logs_filtered_from_console_only(self): + setup_logging(debug=True) + console = self._console_handler() + http_record = logging.LogRecord("httpx", logging.INFO, "httpx", 1, + "HTTP Request: GET ...", None, None) + self.assertFalse(console.filter(http_record)) + chunk_record = logging.LogRecord("converter.converter", logging.DEBUG, + "converter", 1, + "Chunk 1/1 request text", None, None) + self.assertTrue(console.filter(chunk_record)) + + +class SynthesizeChunkLoggingTests(unittest.TestCase): + """Chunk failures surface as a single ERROR record (no print echo).""" + + 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): + self.converter.tts.process_chunk_with_retry.return_value = None + with self.assertLogs("converter.converter", level="ERROR") as logs: + results = self.converter._synthesize_chunks(["Hello."]) + self.assertEqual(results, {1: None}) + self.assertEqual(len(logs.output), 1) + self.assertIn("Chunk 1/1 failed", logs.output[0]) + + def test_raising_chunk_logs_single_error(self): + self.converter.tts.process_chunk_with_retry.side_effect = RuntimeError("boom") + with self.assertLogs("converter.converter", level="ERROR") as logs: + results = self.converter._synthesize_chunks(["Hello."]) + self.assertEqual(results, {1: None}) + self.assertEqual(len(logs.output), 1) + 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) + + def test_partial_chunks_abort_without_assembling(self): + converter = self._converter(client_chunks=True) + converter.tts.process_chunk_with_retry.side_effect = ["chunk_0001.wav", None] + text = " ".join(f"word{i}" for i in range(8)) + with patch.object(config, "CHUNK_SIZE", 5), \ + patch.object(converter_mod.audio, "combine_chunks") as mock_combine: + ok = converter._convert_text(text, Path("out.mp3"), time.time()) + self.assertFalse(ok) + mock_combine.assert_not_called() + + +class PromptOverwriteTests(unittest.TestCase): + def test_single_file_yes(self): + with patch("builtins.input", return_value="y"): + self.assertTrue(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_single_file_no(self): + with patch("builtins.input", return_value="n"): + self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_accepts_full_words(self): + with patch("builtins.input", return_value="yes"): + self.assertTrue(prompt_overwrite([Path("dune.mp3")], "dune")) + with patch("builtins.input", return_value="No"): + self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_invalid_answer_reasked(self): + with patch("builtins.input", side_effect=["maybe", "n"]) as mock_input: + self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) + self.assertEqual(mock_input.call_count, 2) + + def test_empty_answer_defaults_yes(self): + # Pressing Enter (empty input) accepts the default of yes, matching + # the make_audiocpp_server_json tool's ask_bool(default=True) prompt. + with patch("builtins.input", return_value=""): + self.assertTrue(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_eof_keeps_existing_output(self): + with patch("builtins.input", side_effect=EOFError): + self.assertFalse(prompt_overwrite([Path("dune.mp3")], "dune")) + + def test_multiple_files_prompt_names_them(self): + files = [Path("dune_01_Dune.mp3"), Path("dune_02_Barony.mp3")] + with patch("builtins.input", return_value="y") as mock_input: + self.assertTrue(prompt_overwrite(files, "dune")) + prompt_text = mock_input.call_args[0][0] + self.assertIn("2 output files for 'dune'", prompt_text) + self.assertIn("dune_01_Dune.mp3", prompt_text) + self.assertIn("overwrite them", prompt_text) + + +class PreflightOverwritesTests(unittest.TestCase): + """The pre-flight overwrite check runs without a TTS server connection.""" + + def setUp(self): + self._books_tmp = tempfile.TemporaryDirectory() + self._output_tmp = tempfile.TemporaryDirectory() + self._original_folders = (converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER) + converter_mod.BOOKS_FOLDER = Path(self._books_tmp.name) + converter_mod.AUDIOBOOKS_FOLDER = Path(self._output_tmp.name) + (converter_mod.BOOKS_FOLDER / "book.txt").write_text("hello world", encoding="utf-8") + + def tearDown(self): + converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = self._original_folders + self._books_tmp.cleanup() + self._output_tmp.cleanup() + + def test_no_books_returns_empty(self): + (converter_mod.BOOKS_FOLDER / "book.txt").unlink() + with patch("builtins.input", side_effect=AssertionError("should not prompt")): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(book_files, []) + self.assertEqual(planned, []) + + def test_new_book_planned_without_prompt(self): + with patch("builtins.input", side_effect=AssertionError("should not prompt")): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(len(book_files), 1) + self.assertEqual(planned, [(book_files[0], "book_Vivian")]) + + def test_existing_output_enter_defaults_yes(self): + (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value=""): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(planned, [(book_files[0], "book_Vivian")]) + + def test_existing_output_declined_is_skipped(self): + (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value="n"): + book_files, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + self.assertEqual(len(book_files), 1) + self.assertEqual(planned, []) + + +class RunOverwritePromptTests(unittest.TestCase): + """The full run() flow: prompts collected before any conversion starts.""" + + def setUp(self): + self._books_tmp = tempfile.TemporaryDirectory() + self._output_tmp = tempfile.TemporaryDirectory() + self._original_folders = (converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER) + converter_mod.BOOKS_FOLDER = Path(self._books_tmp.name) + converter_mod.AUDIOBOOKS_FOLDER = Path(self._output_tmp.name) + (converter_mod.BOOKS_FOLDER / "book.txt").write_text("hello world", encoding="utf-8") + self.converter = AudiobookConverter.__new__(AudiobookConverter) + self.converter.voice_mode = tts.VOICE_MODE_CUSTOM + self.converter.voice_clone_ref_audio = None + self.converter.backend = tts.BACKEND_QWEN + self.converter.voice = None + self.converter.instructions = None + self.converter.speed = 1.0 + self.converter.single_file = False + self.converter.output_format = "mp3" + self.converter.language = "English" + self.converter.debug = False + self.converted = [] + self.converter.convert_book = ( + lambda file_path, output_name=None: + not self.converted.append((file_path.name, output_name)) or True) + + def tearDown(self): + converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = self._original_folders + self._books_tmp.cleanup() + self._output_tmp.cleanup() + + def test_declined_book_is_skipped(self): + (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value="n"): + self.assertTrue(self.converter.run()) + self.assertEqual(self.converted, []) + self.assertTrue((converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").exists()) + + def test_accepted_book_is_converted(self): + (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") + with patch("builtins.input", return_value="y"): + self.assertTrue(self.converter.run()) + self.assertEqual(self.converted, [("book.txt", "book_Vivian")]) + + def test_new_book_converted_without_prompt(self): + with patch("builtins.input", side_effect=AssertionError("should not prompt")): + self.assertTrue(self.converter.run()) + self.assertEqual(self.converted, [("book.txt", "book_Vivian")]) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_cover.py b/app/tests/test_cover.py new file mode 100644 index 0000000..f19db5a --- /dev/null +++ b/app/tests/test_cover.py @@ -0,0 +1,189 @@ +"""Tests for stdlib-only cover generation: PNG structure, gradient, text.""" + +import random +import struct +import tempfile +import unittest +import zlib +from pathlib import Path + +from converter.cover import ( + _random_light_color, + _text_width, + _wrap_title, + generate_cover, +) + + +def _decode_png(data: bytes): + """Parse a PNG into (width, height, rows of RGB tuples).""" + assert data[:8] == b"\x89PNG\r\n\x1a\n", "bad PNG signature" + pos = 8 + idat = b"" + width = height = None + while pos < len(data): + length, chunk_type = struct.unpack(">I4s", data[pos:pos + 8]) + chunk_data = data[pos + 8:pos + 8 + length] + crc = struct.unpack(">I", data[pos + 8 + length:pos + 12 + length])[0] + assert crc == zlib.crc32(chunk_type + chunk_data) & 0xFFFFFFFF, "bad CRC" + if chunk_type == b"IHDR": + width, height, depth, color_type = struct.unpack(">IIBB", chunk_data[:10]) + assert depth == 8 and color_type == 2 # 8-bit RGB + elif chunk_type == b"IDAT": + idat += chunk_data + pos += 12 + length + raw = zlib.decompress(idat) + stride = 1 + width * 3 + assert len(raw) == height * stride, "unexpected decompressed size" + rows = [] + for y in range(height): + row = raw[y * stride + 1:(y + 1) * stride] + rows.append([tuple(row[x * 3:x * 3 + 3]) for x in range(width)]) + return width, height, rows + + +def _black_pixels(rows): + return sum(1 for row in rows for pixel in row if pixel == (0, 0, 0)) + + +class GenerateCoverTests(unittest.TestCase): + def _write(self, title, width=120, height=180, seed=7): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "cover.png" + result = generate_cover(title, path, width=width, height=height, seed=seed) + data = path.read_bytes() + return result, data + + def test_valid_png_with_requested_dimensions(self): + _, data = self._write("Hello") + width, height, rows = _decode_png(data) + self.assertEqual((width, height), (120, 180)) + self.assertEqual(len(rows), 180) + + def test_gradient_matches_seeded_colors(self): + _, data = self._write("Hello", seed=42) + width, height, rows = _decode_png(data) + rng = random.Random(42) + top = _random_light_color(rng) + bottom = _random_light_color(rng) + # Corners of the text-free top/bottom rows match the endpoints + self.assertEqual(rows[0][0], top) + self.assertEqual(rows[0][-1], top) + self.assertEqual(rows[height - 1][0], bottom) + self.assertEqual(rows[height - 1][-1], bottom) + + def test_gradient_colors_are_light(self): + # Text-free bottom row: every channel must stay in pastel territory + _, data = self._write("Hello", seed=1) + _, height, rows = _decode_png(data) + for channel in rows[height - 1][0]: + self.assertGreaterEqual(channel, 90) + + def test_title_renders_black_pixels(self): + _, data = self._write("Hello") + _, _, rows = _decode_png(data) + self.assertGreater(_black_pixels(rows), 50) + + def test_title_renders_white_pixels(self): + _, data = self._write("Hello") + _, _, rows = _decode_png(data) + white = sum(1 for row in rows for pixel in row if pixel == (255, 255, 255)) + self.assertGreater(white, 50) + + def test_white_text_sits_on_black_stroke(self): + # Directly above a white pixel row there must be a black stroke row: + # sample white pixels and confirm black neighbors within stroke width. + _, data = self._write("Hi", width=200, height=100, seed=3) + _, _, rows = _decode_png(data) + whites = [(x, y) for y, row in enumerate(rows) + for x, pixel in enumerate(row) if pixel == (255, 255, 255)] + self.assertTrue(whites) + checked = near_stroke = 0 + for x, y in whites[::5]: + neighborhood = [] + for dy in range(-3, 4): + for dx in range(-3, 4): + if 0 <= y + dy < len(rows) and 0 <= x + dx < len(rows[0]): + neighborhood.append(rows[y + dy][x + dx]) + checked += 1 + if (0, 0, 0) in neighborhood: + near_stroke += 1 + # Interior white pixels are surrounded by white; every sampled pixel + # should still see stroke black within 3px (font strokes are 5-6 px thick) + self.assertEqual(near_stroke, checked) + + def test_empty_title_renders_gradient_only(self): + _, data = self._write("") + _, _, rows = _decode_png(data) + self.assertEqual(_black_pixels(rows), 0) + + def test_unrenderable_title_degrades_to_gradient(self): + # CJK glyphs are not in the bitmap font; no crash, no text pixels + _, data = self._write("书名") + _, _, rows = _decode_png(data) + self.assertEqual(_black_pixels(rows), 0) + + def test_write_failure_returns_none(self): + result = generate_cover("Hello", Path("/nonexistent_dir/cover.png")) + self.assertIsNone(result) + + +class WrapTitleTests(unittest.TestCase): + def test_short_title_one_line(self): + self.assertEqual(len(_wrap_title("Dune", 500)), 1) + + def test_long_title_wraps(self): + lines = _wrap_title("The Extremely Long Windy Title of a Very Long Book", 600) + self.assertGreater(len(lines), 1) + for line in lines: + self.assertLessEqual(_text_width(line), 600) + + def test_single_long_word_kept_intact(self): + lines = _wrap_title("Antidisestablishmentarianism", 10) + self.assertEqual(lines, ["Antidisestablishmentarianism"]) + + def test_empty_title_no_lines(self): + self.assertEqual(_wrap_title("", 500), []) + + +class TextWidthTests(unittest.TestCase): + def test_empty(self): + self.assertEqual(_text_width(""), 0) + + def test_single_char_is_scaled_glyph(self): + self.assertEqual(_text_width("A"), 30) # 5 px * scale 6 + + def test_chars_include_spacing(self): + self.assertEqual(_text_width("AB"), 66) # (2 glyphs * 6 - 1) * 6 + + +class DropShadowTests(unittest.TestCase): + def _cover_rows(self, title, seed=7): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "cover.png" + generate_cover(title, path, width=200, height=200, seed=seed) + return _decode_png(path.read_bytes())[2] + + def test_shadow_pixels_survive_next_to_text(self): + rows = self._cover_rows("Hi") + # The shadow lives down-right of the glyphs: there must be darkened + # (but not pure black, not full-brightness) pixels beyond the text + # block's bottom edge. + blacks = {(x, y) for y, row in enumerate(rows) + for x, pixel in enumerate(row) if pixel == (0, 0, 0)} + self.assertTrue(blacks, "no text rendered") + text_bottom = max(y for _, y in blacks) + darkened = [pixel for y, row in enumerate(rows) + if y > text_bottom for pixel in row + if pixel != (0, 0, 0) and max(pixel) < 130] + self.assertTrue(darkened, "no shadow pixels below the text") + + def test_empty_title_has_no_shadow(self): + rows = self._cover_rows("") + for row in rows: + for pixel in row: + self.assertNotEqual(pixel, (0, 0, 0)) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_extractors.py b/app/tests/test_extractors.py new file mode 100644 index 0000000..ae1794c --- /dev/null +++ b/app/tests/test_extractors.py @@ -0,0 +1,213 @@ +"""Tests for file text extraction.""" + +import tempfile +import unittest +from pathlib import Path + +from converter.extractors import extract_text + + +class TxtExtractionTests(unittest.TestCase): + def _extract(self, data: bytes) -> str: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "book.txt" + path.write_bytes(data) + return extract_text(path) + + def test_utf8(self): + self.assertEqual(self._extract("héllo wörld".encode("utf-8")), "héllo wörld") + + def test_utf16_with_bom(self): + self.assertEqual(self._extract("héllo".encode("utf-16")), "héllo") + + def test_cp1252(self): + self.assertEqual(self._extract("“quotes”".encode("cp1252")), "“quotes”") + + def test_latin1_fallback(self): + # 0x81 is undefined in cp1252, forcing the latin-1 catch-all + self.assertEqual(self._extract(b"caf\x81"), "caf\x81") + + def test_unsupported_format(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "book.xyz" + path.write_bytes(b"data") + with self.assertRaises(ValueError): + extract_text(path) + + +def _build_test_epub(path: Path, chapters=(("One", "First chapter text."), + ("Two", "Second chapter text."))) -> None: + from ebooklib import epub + + book = epub.EpubBook() + book.set_identifier("test-id") + book.set_title("Test Book") + book.set_language("en") + book.add_author("Test Author") + + items = [] + for index, (title, text) in enumerate(chapters, 1): + chapter = epub.EpubHtml(title=title, file_name=f"chap{index}.xhtml", lang="en") + chapter.content = f"

{text}

" + book.add_item(chapter) + items.append(chapter) + + book.toc = tuple(items) + book.spine = ["nav", *items] + book.add_item(epub.EpubNcx()) + book.add_item(epub.EpubNav()) + + epub.write_epub(str(path), book) + + +class EpubExtractionTests(unittest.TestCase): + def setUp(self): + try: + import ebooklib # noqa: F401 + except ImportError: + self.skipTest("ebooklib not installed") + + def test_ebooklib_extraction(self): + # Regression test: the ebooklib path used to silently return "" due to + # isinstance(item, ebooklib.ITEM_DOCUMENT) (an int, not a class). + from converter.extractors import _read_epub_ebooklib + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "book.epub" + _build_test_epub(path) + items = _read_epub_ebooklib(path) + + html = "\n".join(content for _, content in items) + self.assertIn("First chapter text.", html) + self.assertIn("Second chapter text.", html) + + def test_epub_extraction_follows_spine_order(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "book.epub" + _build_test_epub(path) + text = extract_text(path) + + self.assertIn("First chapter text.", text) + self.assertIn("Second chapter text.", text) + self.assertLess(text.index("First chapter text."), + text.index("Second chapter text.")) + + +class ExtractSectionsTests(unittest.TestCase): + def setUp(self): + try: + import ebooklib # noqa: F401 + except ImportError: + self.skipTest("ebooklib not installed") + + def test_epub_sections_split_on_chapters(self): + from converter.extractors import extract_sections + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "book.epub" + _build_test_epub(path) + sections = extract_sections(path) + + self.assertEqual(len(sections), 2) + self.assertEqual(sections[0].title, "One") + self.assertEqual(sections[1].title, "Two") + self.assertIn("First chapter text.", sections[0].text) + self.assertIn("Second chapter text.", sections[1].text) + + def test_txt_is_single_section(self): + from converter.extractors import extract_sections + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "book.txt" + path.write_text("Hello world.", encoding="utf-8") + sections = extract_sections(path) + + self.assertEqual(len(sections), 1) + self.assertEqual(sections[0].title, "book") + self.assertEqual(sections[0].text, "Hello world.") + + def test_single_chapter_epub_keeps_chapter_title(self): + from converter.extractors import extract_sections + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "book.epub" + _build_test_epub(path, chapters=(("Only", "Just one chapter."),)) + sections = extract_sections(path) + + self.assertEqual(len(sections), 1) + self.assertEqual(sections[0].title, "Only") + self.assertIn("Just one chapter.", sections[0].text) + + +class ExtractBookTests(unittest.TestCase): + def test_txt_falls_back_to_stem_and_blank_author(self): + from converter.extractors import extract_book + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "mybook.txt" + path.write_text("Hello world.", encoding="utf-8") + book = extract_book(path) + + self.assertEqual(book.title, "mybook") + self.assertEqual(book.author, "") + self.assertEqual(len(book.sections), 1) + + def test_epub_metadata_harvested(self): + try: + import ebooklib # noqa: F401 + except ImportError: + self.skipTest("ebooklib not installed") + from converter.extractors import extract_book + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "book.epub" + _build_test_epub(path) + book = extract_book(path) + + self.assertEqual(book.title, "Test Book") + self.assertEqual(book.author, "Test Author") + self.assertEqual([s.title for s in book.sections], ["One", "Two"]) + + def test_pdf_metadata_harvested(self): + from converter.extractors import extract_book + + try: + from pypdf import PdfWriter + except ImportError: + self.skipTest("pypdf not installed") + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "book.pdf" + writer = PdfWriter() + writer.add_metadata({"/Title": "PDF Title", "/Author": "PDF Author"}) + writer.add_blank_page(width=612, height=792) + with open(path, "wb") as handle: + writer.write(handle) + book = extract_book(path) + + self.assertEqual(book.title, "PDF Title") + self.assertEqual(book.author, "PDF Author") + self.assertEqual(len(book.sections), 1) + + def test_pdf_without_metadata_falls_back(self): + from converter.extractors import extract_book + + try: + from pypdf import PdfWriter + except ImportError: + self.skipTest("pypdf not installed") + + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "plain.pdf" + writer = PdfWriter() + writer.add_blank_page(width=612, height=792) + with open(path, "wb") as handle: + writer.write(handle) + book = extract_book(path) + + self.assertEqual(book.title, "plain") + self.assertEqual(book.author, "") + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py new file mode 100644 index 0000000..b3a17e8 --- /dev/null +++ b/app/tests/test_hub.py @@ -0,0 +1,502 @@ +"""Tests for the TUI hub (ui/hub.py) menu and helpers. + +The hub drives the same curses widgets as ui/tui.py, so these tests reuse +the fake curses/screen from test_tui to run the menu without a terminal. +""" + +import unittest +from pathlib import Path +from unittest.mock import patch + +from backends import BackendStatus, ServerSpec +from tests.test_tui import FakeCurses, FakeScreen +from ui import hub, tui + + +class HubHelperTests(unittest.TestCase): + """Pure helpers in hub.py (no curses).""" + + def test_is_float(self): + self.assertTrue(hub._is_float("1.0")) + self.assertTrue(hub._is_float("2")) + self.assertFalse(hub._is_float("abc")) + self.assertFalse(hub._is_float("")) + + def test_list_voices_from_dir(self): + with __import__("tempfile").TemporaryDirectory() as td: + d = Path(td) + (d / "Narrator.wav").write_bytes(b"x") + (d / "Alpha.WAV").write_bytes(b"x") + (d / "notes.txt").write_bytes(b"x") + voices = hub._list_voices(str(d)) + # Stems preserve case; sorting is case-insensitive. + self.assertEqual(voices, ["Alpha", "Narrator"]) + + def test_list_voices_missing_dir(self): + self.assertEqual(hub._list_voices("/no/such/dir"), []) + + def test_status_mark(self): + from backends import BackendStatus + running = BackendStatus("k", "l", installed=True, configured=True, + running=True) + installed = BackendStatus("k", "l", installed=True, + configured=False) + none = BackendStatus("k", "l", installed=False, configured=False) + # running beats installed (a server is up even if not configured); + # only a backend that is neither installed nor running is dimmed. + self.assertEqual(hub._status_mark(running), + ("running", "ok", "body")) + self.assertEqual(hub._status_mark(installed), + ("installed", "warn", "body")) + self.assertEqual(hub._status_mark(none), + ("unavailable", "err", "dim")) + self.assertEqual(hub._status_mark(None), + ("unavailable", "err", "dim")) + + +class HubMenuTests(unittest.TestCase): + """Drive _hub_menu with a fake screen (no terminal).""" + + def setUp(self): + tui._THEME.clear() + self.curses = FakeCurses() + from unittest.mock import patch as _patch + self._patcher = _patch.dict("sys.modules", {"curses": self.curses}) + self._patcher.start() + self.addCleanup(self._patcher.stop) + self.addCleanup(tui._THEME.clear) + + def _none_status(self, key="k", label="l"): + from backends import BackendStatus + return BackendStatus(key, label, installed=False, configured=False) + + def test_quit_returns_none_when_no_backend(self): + # No backends installed/running: menu is [Set up, Settings, Quit]. + # Quit is the 3rd option (Down twice) then Enter. + screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10]) + with patch.object(hub, "detect_all", return_value=[]): + result = hub._hub_menu(screen) + self.assertIsNone(result) + + def test_menu_has_only_setup_settings_and_quit_without_backends(self): + # Capture the options handed to tui.menu: with nothing installed or + # running, Convert/Configure must be absent. + captured = {} + + def fake_menu(stdscr, title, options, **kwargs): + captured["options"] = options + return "quit" + + screen = FakeScreen() + with patch.object(hub.tui, "menu", fake_menu), \ + patch.object(hub, "detect_all", return_value=[]): + hub._hub_menu(screen) + labels = [label for label, _ in captured["options"]] + self.assertEqual(labels, ["Set up a backend...", "Settings...", + "Quit"]) + + def test_menu_has_all_six_when_one_installed(self): + captured = {} + + def fake_menu(stdscr, title, options, **kwargs): + captured["options"] = options + captured["rows"] = kwargs.get("table_rows") + return "quit" + + screen = FakeScreen() + st = self._none_status("qwen", "qwen-tts") + st.installed = True + with patch.object(hub.tui, "menu", fake_menu), \ + patch.object(hub, "detect_all", return_value=[st]): + hub._hub_menu(screen) + labels = [label for label, _ in captured["options"]] + self.assertEqual( + labels, + ["Convert books...", "Set up a backend...", + "Configure a backend...", "Server...", "Settings...", "Quit"]) + # The status table is passed through, one row per backend. + self.assertEqual(captured["rows"], + [("qwen-tts", "installed", "warn", "body")]) + + def test_table_dims_name_when_not_installed_and_not_running(self): + captured = {} + + def fake_menu(stdscr, title, options, **kwargs): + captured["rows"] = kwargs.get("table_rows") + return "quit" + + screen = FakeScreen() + dead = self._none_status("audiocpp", "audio.cpp") + external = self._none_status("qwen", "qwen-tts") + external.running = True + with patch.object(hub.tui, "menu", fake_menu), \ + patch.object(hub, "detect_all", + return_value=[dead, external]): + hub._hub_menu(screen) + # Unusable backend: dim name. Running-but-not-installed stays bright. + self.assertEqual( + captured["rows"], + [("audio.cpp", "unavailable", "err", "dim"), + ("qwen-tts", "running", "ok", "body")]) + + def test_menu_has_all_six_when_one_running_only(self): + # Running but not installed (an external server) still unlocks the + # Convert/Configure/Server entries. + captured = {} + + def fake_menu(stdscr, title, options, **kwargs): + captured["options"] = options + return "quit" + + screen = FakeScreen() + st = self._none_status("qwen", "qwen-tts") + st.running = True + with patch.object(hub.tui, "menu", fake_menu), \ + patch.object(hub, "detect_all", return_value=[st]): + hub._hub_menu(screen) + labels = [label for label, _ in captured["options"]] + self.assertEqual( + labels, + ["Convert books...", "Set up a backend...", + "Configure a backend...", "Server...", "Settings...", "Quit"]) + + def test_convert_with_no_available_backend_offers_setup(self): + # One installed-but-not-ready backend → Convert is offered. The + # convert menu lists no available backend, so only "Set up a + # backend..." is shown; Enter selects it → setup menu lists 3 + # backends; Esc goes back → convert returns None → main menu loops. + # Then quit: main menu now has 5 options, Quit is the 5th (Down x4). + from backends import BackendInfo, BackendStatus + none = BackendStatus("k", "l", installed=True, configured=False) + infos = [BackendInfo("audiocpp", "audio.cpp", lambda: none, + lambda: 0), + BackendInfo("qwen", "qwen-tts", lambda: none, lambda: 0), + BackendInfo("faster", "faster", lambda: none, lambda: 0)] + # installed=True so the main menu shows Convert; but ready/running + # is False so the convert menu's available list is empty. + statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True, + configured=False), + BackendStatus("qwen", "qwen-tts", installed=True, + configured=False), + BackendStatus("faster", "faster", installed=True, + configured=False)] + with patch.object(hub, "detect_all", return_value=statuses), \ + patch.object(hub, "REGISTRY", infos): + # Convert(Enter), setup-entry(Enter), Esc on setup menu, + # back at main menu -> Down x5 -> Enter (Quit; Settings sits + # just before it). + screen = FakeScreen(keys=[10, 10, 27, + FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + FakeCurses.KEY_DOWN, 10]) + result = hub._hub_menu(screen) + self.assertIsNone(result) + + +class SelectSpecTests(unittest.TestCase): + """_select_spec: mode-aware server selection (qwen has two servers).""" + + def _qwen_status(self): + return BackendStatus( + "qwen", "qwen-tts", installed=True, configured=True, + servers=[ServerSpec("qwen-custom", "http://127.0.0.1:7860", []), + ServerSpec("qwen-clone", "http://127.0.0.1:7861", [])]) + + def test_qwen_custom_mode(self): + spec = hub._select_spec(self._qwen_status(), {"clone": None}) + self.assertEqual(spec.name, "qwen-custom") + + def test_qwen_clone_mode(self): + spec = hub._select_spec(self._qwen_status(), {"clone": "ref.wav"}) + self.assertEqual(spec.name, "qwen-clone") + + def test_audiocpp_returns_single_spec(self): + st = BackendStatus("audiocpp", "audio.cpp", installed=True, + configured=True, + servers=[ServerSpec("audiocpp", "http://x", [])]) + spec = hub._select_spec(st, {}) + self.assertEqual(spec.name, "audiocpp") + + def test_none_when_no_servers(self): + st = BackendStatus("qwen", "qwen-tts", installed=False, + configured=False) + self.assertIsNone(hub._select_spec(st, {})) + + +class RunConversionTests(unittest.TestCase): + """_run_conversion: autostart, hint-when-manual, and stop-after.""" + + def test_autostart_starts_server_then_converts(self): + spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) + status = BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, running=False, + servers=[spec]) + kwargs = {"autostart": "qwen-custom"} + with patch.object(hub, "detect_all", return_value=[status]), \ + patch.object(hub, "_find_spec", return_value=spec), \ + patch.object(hub.servers, "start", return_value=True) as mk_start, \ + patch.object(hub.audiobook, "convert", return_value=0) as mk_conv, \ + patch("builtins.input", return_value="n") as mk_input, \ + patch.object(hub.servers, "stop") as mk_stop: + hub._run_conversion("qwen", kwargs) + mk_start.assert_called_once_with(spec) + mk_conv.assert_called_once() + # User declined stopping → stop not called. + mk_stop.assert_not_called() + + def test_autostart_stop_when_user_says_yes(self): + spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) + status = BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, running=False, + servers=[spec]) + kwargs = {"autostart": "qwen-custom"} + with patch.object(hub, "detect_all", return_value=[status]), \ + patch.object(hub, "_find_spec", return_value=spec), \ + patch.object(hub.servers, "start", return_value=True), \ + patch.object(hub.audiobook, "convert", return_value=0), \ + patch("builtins.input", return_value="y"), \ + patch.object(hub.servers, "stop") as mk_stop: + hub._run_conversion("qwen", kwargs) + mk_stop.assert_called_once_with("qwen-custom") + + def test_autostart_aborts_when_server_fails(self): + spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) + status = BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, running=False, + launch_hint="hint cmd", servers=[spec]) + kwargs = {"autostart": "qwen-custom"} + with patch.object(hub, "detect_all", return_value=[status]), \ + patch.object(hub, "_find_spec", return_value=spec), \ + patch.object(hub.servers, "start", return_value=False), \ + patch.object(hub.audiobook, "convert") as mk_conv, \ + patch.object(hub.servers, "stop") as mk_stop: + hub._run_conversion("qwen", kwargs) + mk_conv.assert_not_called() + mk_stop.assert_not_called() + + def test_no_autostart_prints_hint_when_not_running(self): + status = BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, running=False, + launch_hint="the-hint") + with patch.object(hub, "detect_all", return_value=[status]), \ + patch.object(hub.audiobook, "convert", return_value=0) as mk_conv: + hub._run_conversion("qwen", {}) + mk_conv.assert_called_once() + + +class AddAutostartTests(unittest.TestCase): + """_add_autostart: offers to start the server when it isn't running.""" + + def setUp(self): + tui._THEME.clear() + self.curses = FakeCurses() + self._patcher = patch.dict("sys.modules", {"curses": self.curses}) + self._patcher.start() + self.addCleanup(self._patcher.stop) + self.addCleanup(tui._THEME.clear) + + def _status(self): + spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) + return BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, running=False, + servers=[spec]) + + def test_sets_autostart_when_user_confirms(self): + screen = FakeScreen(keys=[10]) # Enter = Yes + cmd = ("convert", "qwen", {"clone": None}) + with patch.object(hub, "detect_all", return_value=[self._status()]), \ + patch("backends.common.server_running", return_value=False): + hub._add_autostart(screen, cmd, [self._status()]) + self.assertEqual(cmd[2]["autostart"], "qwen-custom") + + def test_no_autostart_when_server_already_running(self): + screen = FakeScreen(keys=[10]) + cmd = ("convert", "qwen", {"clone": None}) + with patch.object(hub, "detect_all", return_value=[self._status()]), \ + patch("backends.common.server_running", return_value=True): + hub._add_autostart(screen, cmd, [self._status()]) + self.assertNotIn("autostart", cmd[2]) + + +class SettingsTests(unittest.TestCase): + """Settings menu: field collection, validation, config.py writing.""" + + def test_write_config_preserves_comments_and_other_lines(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + path = Path(td) / "config.py" + path.write_text( + "# Default output options\n" + 'AUDIO_FORMAT = "m4b"\n' + 'AUDIO_BITRATE = "128k"\n' + 'LANGUAGE = "English"\n' + "\n" + "CHUNK_SIZE = 250 # words per request\n", + encoding="utf-8") + with patch.object(hub.config, "__file__", str(path)): + hub._write_config({"AUDIO_FORMAT": "mp3", + "AUDIO_BITRATE": "192k", + "LANGUAGE": "Japanese", + "CHUNK_SIZE": 300}) + text = path.read_text(encoding="utf-8") + self.assertEqual( + text, + "# Default output options\n" + 'AUDIO_FORMAT = "mp3"\n' + 'AUDIO_BITRATE = "192k"\n' + 'LANGUAGE = "Japanese"\n' + "\n" + "CHUNK_SIZE = 300 # words per request\n") + + def test_write_config_missing_key_raises(self): + import tempfile + with tempfile.TemporaryDirectory() as td: + path = Path(td) / "config.py" + path.write_text("X = 1\n", encoding="utf-8") + with patch.object(hub.config, "__file__", str(path)): + with self.assertRaises(ValueError): + hub._write_config({"AUDIO_FORMAT": "mp3"}) + + def test_apply_settings_writes_and_reloads_in_memory(self): + written = {} + + def fake_write(updates): + written.update(updates) + + original = {name: getattr(hub.config, name) for name in + ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", + "CHUNK_SIZE")} + self.addCleanup(lambda: [setattr(hub.config, name, value) + for name, value in original.items()]) + values = {"audio_format": "ogg", "audio_bitrate": " 192k ", + "language": "en", "chunk_size": "300"} + with patch.object(hub, "_write_config", fake_write): + hub._apply_settings(values) + # Values are trimmed and language normalized to a display name. + self.assertEqual(written, {"AUDIO_FORMAT": "ogg", + "AUDIO_BITRATE": "192k", + "LANGUAGE": "English", + "CHUNK_SIZE": 300}) + # In-memory config is reloaded so this session sees the change. + self.assertEqual(hub.config.AUDIO_FORMAT, "ogg") + self.assertEqual(hub.config.AUDIO_BITRATE, "192k") + self.assertEqual(hub.config.LANGUAGE, "English") + self.assertEqual(hub.config.CHUNK_SIZE, 300) + + def test_apply_settings_rejects_bad_values(self): + original = {name: getattr(hub.config, name) for name in + ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", + "CHUNK_SIZE")} + self.addCleanup(lambda: [setattr(hub.config, name, value) + for name, value in original.items()]) + with patch.object(hub, "_write_config") as mk_write: + with self.assertRaises(ValueError): + hub._apply_settings({"audio_format": "m4b", + "audio_bitrate": "128k", + "language": "Klingon", + "chunk_size": "250"}) + with self.assertRaises(ValueError): + hub._apply_settings({"audio_format": "m4b", + "audio_bitrate": "128k", + "language": "English", + "chunk_size": "0"}) + mk_write.assert_not_called() + + def test_field_validators(self): + self.assertIsNone(hub._validate_bitrate("128k")) + self.assertIsNotNone(hub._validate_bitrate(" ")) + self.assertIsNone(hub._validate_language("English")) + self.assertIsNone(hub._validate_language("en")) + self.assertIsNotNone(hub._validate_language("Klingon")) + self.assertIsNone(hub._validate_chunk_size("250")) + self.assertIsNotNone(hub._validate_chunk_size("abc")) + self.assertIsNotNone(hub._validate_chunk_size("0")) + + def test_settings_menu_builds_form_and_saves(self): + captured = {} + + def fake_form(stdscr, title, fields, back_value=None): + captured["fields"] = fields + return {"audio_format": "ogg", "audio_bitrate": "192k", + "language": "English", "chunk_size": "300"} + + applied = [] + + def fake_apply(values): + applied.append(values) + + def fake_flash(stdscr, text, kind="warn"): + captured["flash"] = (text, kind) + + with patch.object(hub.tui, "form", fake_form), \ + patch.object(hub, "_apply_settings", fake_apply), \ + patch.object(hub.tui, "flash", fake_flash): + hub._settings_menu(None) + self.assertEqual([f["key"] for f in captured["fields"]], + ["audio_format", "audio_bitrate", "language", + "chunk_size"]) + kinds = {f["key"]: f["kind"] for f in captured["fields"]} + self.assertEqual(kinds["audio_format"], "choice") + self.assertEqual(kinds["audio_bitrate"], "text") + self.assertEqual(applied, [{"audio_format": "ogg", + "audio_bitrate": "192k", + "language": "English", + "chunk_size": "300"}]) + self.assertEqual(captured["flash"], ("Settings saved.", "ok")) + + def test_settings_menu_cancel_does_not_apply(self): + def fake_form(stdscr, title, fields, back_value=None): + return back_value # user pressed Cancel + + applied = [] + + def fake_apply(values): + applied.append(values) + + with patch.object(hub.tui, "form", fake_form), \ + patch.object(hub, "_apply_settings", fake_apply): + hub._settings_menu(None) + self.assertEqual(applied, []) + + def test_settings_menu_writes_config_end_to_end(self): + import tempfile + tui._THEME.clear() + self.addCleanup(tui._THEME.clear) + curses = FakeCurses() + patcher = patch.dict("sys.modules", {"curses": curses}) + patcher.start() + self.addCleanup(patcher.stop) + + original = {name: getattr(hub.config, name) for name in + ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", + "CHUNK_SIZE")} + self.addCleanup(lambda: [setattr(hub.config, name, value) + for name, value in original.items()]) + + with tempfile.TemporaryDirectory() as td: + path = Path(td) / "config.py" + path.write_text( + "# Default output options\n" + 'AUDIO_FORMAT = "m4b"\n' + 'AUDIO_BITRATE = "128k"\n' + 'LANGUAGE = "English"\n' + "\n" + "CHUNK_SIZE = 250\n", + encoding="utf-8") + with patch.object(hub.config, "__file__", str(path)): + # Down to Chunk size, Enter -> editor, Ctrl-U + '300', + # Enter; Tab -> Save, Enter; a key dismisses the flash. + screen = FakeScreen(keys=[ + FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + FakeCurses.KEY_DOWN, 10, 21, ord("3"), ord("0"), + ord("0"), 10, 9, 10, 10]) + hub._settings_menu(screen) + text = path.read_text(encoding="utf-8") + self.assertIn('AUDIO_FORMAT = "m4b"', text) + self.assertIn("CHUNK_SIZE = 300", text) + # The running session also picked up the change in-memory. + self.assertEqual(hub.config.CHUNK_SIZE, 300) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_tts.py b/app/tests/test_tts.py new file mode 100644 index 0000000..a2df07f --- /dev/null +++ b/app/tests/test_tts.py @@ -0,0 +1,1513 @@ +"""Tests for the TTS client wrappers (language handling and payloads).""" + +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 + +from converter import config, tts +from converter.converter import AudiobookConverter +from converter.tts import ( + AudioCppTTSClient, + FasterTTSClient, + QwenTTSClient, + normalize_language, +) + + +class NormalizeLanguageTests(unittest.TestCase): + def test_display_names_case_insensitive(self): + self.assertEqual(normalize_language("english"), "English") + self.assertEqual(normalize_language("ENGLISH"), "English") + self.assertEqual(normalize_language(" Japanese "), "Japanese") + + def test_auto_accepted(self): + self.assertEqual(normalize_language("auto"), "Auto") + self.assertEqual(normalize_language("Auto"), "Auto") + + def test_iso_aliases(self): + self.assertEqual(normalize_language("en"), "English") + self.assertEqual(normalize_language("ja"), "Japanese") + self.assertEqual(normalize_language("zh"), "Chinese") + self.assertEqual(normalize_language("ko"), "Korean") + self.assertEqual(normalize_language("de"), "German") + self.assertEqual(normalize_language("fr"), "French") + self.assertEqual(normalize_language("ru"), "Russian") + self.assertEqual(normalize_language("pt"), "Portuguese") + self.assertEqual(normalize_language("es"), "Spanish") + self.assertEqual(normalize_language("it"), "Italian") + + def test_all_supported_languages_round_trip(self): + for name in tts.TTS_LANGUAGES: + self.assertEqual(normalize_language(name.lower()), name) + + def test_unknown_language_rejected_with_guidance(self): + with self.assertRaises(ValueError) as ctx: + normalize_language("klingon") + message = str(ctx.exception) + self.assertIn("klingon", message) + self.assertIn("English", message) + + def test_none_and_empty_rejected(self): + with self.assertRaises(ValueError): + normalize_language(None) + with self.assertRaises(ValueError): + normalize_language(" ") + + +class QwenTTSClientLanguageTests(unittest.TestCase): + """Language validation and defaults, without touching the network.""" + + def _make_client(self, **kwargs): + with patch.object(QwenTTSClient, "_connect"): + return QwenTTSClient(**kwargs) + + def test_default_follows_config_for_each_mode(self): + custom = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM) + self.assertEqual(custom.language, config.LANGUAGE) + clone = self._make_client(voice_mode=tts.VOICE_MODE_CLONE, + voice_clone_ref_audio="ref.wav") + self.assertEqual(clone.language, config.LANGUAGE) + + def test_explicit_language_normalized(self): + client = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM, language="ja") + self.assertEqual(client.language, "Japanese") + + def test_invalid_language_fails_before_connect(self): + with patch.object(QwenTTSClient, "_connect") as mock_connect: + with self.assertRaises(ValueError): + QwenTTSClient(language="klingon") + 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.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.ref_audio = Path(self._tmp.name) / "reference.wav" + self.ref_audio.write_bytes(b"x") + + def tearDown(self): + self._tmp.cleanup() + + def _custom_client(self, language, endpoint, api_info=None): + 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: {}} + } + client.client = MagicMock() + return client + + def _clone_client(self, language, endpoint, api_info=None, ref_text="hello"): + 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 { + "named_endpoints": {endpoint: {}} + } + client.clone_client = MagicMock() + client._ref_audio_filedata = {"dummy": "payload"} + return client + + def test_custom_voice_run_instruct_uses_language(self): + client = self._custom_client("Japanese", "/run_instruct") + client._generate_custom_voice("text") + kwargs = client.client.predict.call_args.kwargs + self.assertEqual(kwargs["lang_disp"], "Japanese") + + def test_custom_voice_alt_endpoint_uses_language(self): + client = self._custom_client("French", "/run_custom_voice") + client._generate_custom_voice("text") + kwargs = client.client.predict.call_args.kwargs + self.assertEqual(kwargs["language"], "French") + + def test_voice_clone_run_voice_clone_uses_language(self): + client = self._clone_client("Japanese", "/run_voice_clone") + client._generate_voice_clone("text") + kwargs = client.clone_client.predict.call_args.kwargs + self.assertEqual(kwargs["lang_disp"], "Japanese") + + def test_voice_clone_alt_endpoint_uses_language(self): + client = self._clone_client("Korean", "/generate_voice_clone") + client._generate_voice_clone("text") + kwargs = client.clone_client.predict.call_args.kwargs + self.assertEqual(kwargs["language"], "Korean") + + def test_voice_clone_alt_endpoint_includes_optional_params(self): + api_info = { + "named_endpoints": { + "/generate_voice_clone": { + "parameters": [ + {"parameter_name": "model_size"}, + {"parameter_name": "seed"}, + ] + } + } + } + client = self._clone_client("English", "/generate_voice_clone", api_info=api_info) + client._generate_voice_clone("text") + kwargs = client.clone_client.predict.call_args.kwargs + self.assertEqual(kwargs["model_size"], tts.MODEL_SIZE) + self.assertEqual(kwargs["seed"], config.SEED) + + +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_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", + 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(tts, "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, 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, "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_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) + # 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): + 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.MAX_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") + + 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, "CHUNK_SIZE", 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_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 AudioCppTTSClientHealthTests(unittest.TestCase): + """Connection behavior of the audio.cpp client.""" + + @staticmethod + def _json_response(payload): + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = json.dumps(payload).encode("utf-8") + return response + + def _get_responses(self, health=None, models=None, voices=None): + """Side effect dispatching GET responses by URL.""" + def _dispatch(request, **_kwargs): + url = request if isinstance(request, str) else request.full_url + if url.endswith("/health"): + return self._json_response(health if health is not None + else {"status": "ok"}) + if url.endswith("/v1/models"): + return self._json_response(models if models is not None else + {"data": [{"id": config.AUDIOCPP_MODEL_ID}]}) + if "/v1/audio/voices" in url: + if voices is Exception: + raise Exception("voices endpoint down") + return self._json_response(voices if voices is not None + else {"voices": ["narrator"]}) + raise AssertionError(f"unexpected URL: {url}") + return _dispatch + + def _client(self, voice=None, language=None, model_id=None, **kwargs): + with patch("converter.tts.urllib.request.urlopen", + side_effect=self._get_responses(**kwargs)): + return AudioCppTTSClient(voice=voice, language=language, + model_id=model_id) + + 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: + AudioCppTTSClient() + message = str(ctx.exception) + self.assertIn("not reachable", message) + self.assertIn("README", message) + + def test_unhealthy_status_raises(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(health={"status": "starting"}) + self.assertIn("starting", str(ctx.exception)) + + def test_unknown_model_id_raises_with_configured_ids(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(models={"data": [{"id": "pocket-tts"}, {"id": "other"}]}) + message = str(ctx.exception) + self.assertIn(config.AUDIOCPP_MODEL_ID, message) + self.assertIn("pocket-tts", message) + self.assertIn("other", message) + + def test_healthy_server_speaker_mode_defaults(self): + client = self._client() + self.assertEqual(client.api_url, config.AUDIOCPP_API_URL.rstrip("/")) + self.assertEqual(client.model_id, config.AUDIOCPP_MODEL_ID) + self.assertEqual(client.language, config.LANGUAGE) + self.assertEqual(client.voice, "Vivian") + self.assertFalse(client.preset_mode) + + def test_speaker_mode_uses_configured_speaker(self): + with patch.object(config, "SPEAKER", "uncle_fu"): + client = self._client() + self.assertEqual(client.voice, "Uncle Fu") + + def test_preset_mode_uses_requested_voice(self): + client = self._client(voice="narrator") + self.assertEqual(client.voice, "narrator") + self.assertTrue(client.preset_mode) + + def test_preset_mode_validates_voice_against_server_list(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(voice="ghost", voices={"voices": ["narrator", "obama"]}) + message = str(ctx.exception) + self.assertIn("ghost", message) + self.assertIn("narrator", message) + self.assertIn("obama", message) + + def test_preset_mode_skips_validation_when_voices_endpoint_fails(self): + client = self._client(voice="narrator", voices=Exception) + self.assertEqual(client.voice, "narrator") + + def test_invalid_language_fails_before_connect(self): + with patch("converter.tts.urllib.request.urlopen") as mock_urlopen: + with self.assertRaises(ValueError): + AudioCppTTSClient(language="klingon") + mock_urlopen.assert_not_called() + + def test_explicit_language_normalized(self): + client = self._client(language="ja") + self.assertEqual(client.language, "Japanese") + + def test_seed_resolved_once_per_run(self): + with patch.object(config, "CONSTANT_SEED", True), \ + patch.object(config, "SEED", -1): + client = self._client() + self.assertGreaterEqual(client._seed, 0) + + def test_preset_mode_routes_to_clone_model_when_configured(self): + 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_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, "qwen3-tts") + self.assertTrue(any("qwen3-tts-clone" in line for line in logs.output)) + + def test_empty_model_id_auto_picks_single_server_entry(self): + # A multi-model server used without editing config.py: an empty + # --model resolves to the only hosted entry automatically. + client = self._client( + voice="narrator", model_id="", + models={"data": [{"id": "higgs", "family": "higgs_audio_tts"}]}, + voices={"voices": ["narrator"]}) + self.assertEqual(client.model_id, "higgs") + + def test_empty_model_id_with_multiple_entries_requires_explicit_choice(self): + with self.assertRaises(RuntimeError) as ctx: + self._client( + voice="narrator", model_id="", + models={"data": [{"id": "higgs"}, {"id": "voxcpm2"}]}, + voices={"voices": ["narrator"]}) + message = str(ctx.exception) + self.assertIn("--model", message) + self.assertIn("higgs", message) + self.assertIn("voxcpm2", message) + + def test_model_id_override_reaches_request(self): + # --model overrides AUDIOCPP_MODEL_ID for the run. + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"): + client = self._client( + voice="narrator", model_id="higgs", + models={"data": [{"id": "higgs", "family": "higgs_audio_tts"}]}, + voices={"voices": ["narrator"]}) + self.assertEqual(client.model_id, "higgs") + + def test_clone_model_id_ignored_for_speaker_mode(self): + 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, "qwen3-tts") + + def test_clone_model_id_equal_to_primary_is_noop(self): + with patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", + config.AUDIOCPP_MODEL_ID): + client = self._client(voice="narrator") + self.assertEqual(client.model_id, config.AUDIOCPP_MODEL_ID) + + def test_preset_mode_with_clone_only_server_uses_clone_model(self): + 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-clone"}]}) + self.assertEqual(client.model_id, "qwen3-tts-clone") + + def test_speaker_mode_with_clone_only_server_suggests_voice(self): + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): + with self.assertRaises(RuntimeError) as ctx: + self._client(models={"data": [{"id": "qwen3-tts-clone"}]}) + message = str(ctx.exception) + self.assertIn("qwen3-tts", message) + self.assertIn("--voice", message) + + def test_preset_mode_with_no_matching_model_lists_both_ids(self): + 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"): + with self.assertRaises(RuntimeError) as ctx: + self._client(voice="narrator", + models={"data": [{"id": "pocket-tts"}]}) + message = str(ctx.exception) + self.assertIn("qwen3-tts", message) + self.assertIn("qwen3-tts-clone", message) + self.assertIn("pocket-tts", message) + + +class AudioCppTaskDetectionTests(unittest.TestCase): + """Task auto-detection (tts/clon/vdes) and voice design validation.""" + + @staticmethod + def _json_response(payload): + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = json.dumps(payload).encode("utf-8") + return response + + def _client(self, voice=None, instructions=None, request_options=None, + models=None): + if models is None: + models = {"data": [{"id": config.AUDIOCPP_MODEL_ID, + "family": "qwen3_tts"}]} + + def _dispatch(request, **_kwargs): + url = request if isinstance(request, str) else request.full_url + if url.endswith("/health"): + return self._json_response({"status": "ok"}) + if url.endswith("/v1/models"): + return self._json_response(models) + if "/v1/audio/voices" in url: + return self._json_response({"voices": ["narrator"]}) + raise AssertionError(f"unexpected URL: {url}") + + with patch("converter.tts.urllib.request.urlopen", + side_effect=_dispatch): + return AudioCppTTSClient(voice=voice, instructions=instructions, + request_options=request_options) + + def test_missing_task_falls_back_to_tts(self): + # Servers that predate the task field hosted plain TTS models. + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts"}]}) + self.assertEqual(client.task, tts.AUDIOCPP_TASK_TTS) + self.assertFalse(client.design_mode) + + def test_task_detected_from_models_endpoint(self): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="A warm adult narrator") + self.assertEqual(client.task, tts.AUDIOCPP_TASK_VDES) + self.assertTrue(client.design_mode) + + def test_clon_task_entry_connects_in_preset_mode(self): + client = self._client(voice="narrator", models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "chatterbox", + "task": "clon"}]}) + self.assertEqual(client.task, "clon") + self.assertFalse(client.design_mode) + self.assertTrue(client.preset_mode) + + def test_unsupported_task_rejected_with_available_entries(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_asr", + "task": "asr"}, + {"id": "tts-1", "family": "qwen3_tts", "task": "tts"}]}, + instructions="unused") + message = str(ctx.exception) + self.assertIn("'asr'", message) + self.assertIn("--model", message) + self.assertIn("tts-1", message) + + def test_vdes_without_instructions_requires_description(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}) + message = str(ctx.exception) + self.assertIn("voice design", message) + self.assertIn("--instructions", message) + + def test_vdes_with_voice_rejected(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(voice="narrator", models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="A warm adult narrator") + self.assertIn("--voice", str(ctx.exception)) + self.assertIn("--instructions", str(ctx.exception)) + + def test_vdes_with_instructions_connects_in_design_mode(self): + buf = io.StringIO() + with redirect_stdout(buf): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="A warm adult narrator") + self.assertTrue(client.design_mode) + self.assertEqual(client.instructions, "A warm adult narrator") + out = buf.getvalue() + self.assertIn("voice design", out) + self.assertIn("A warm adult narrator", out) + + def test_instructions_without_voice_on_generic_family_connects(self): + # Families without built-in speakers can get their voice from the + # instruction alone (e.g. OmniVoice voice design). + buf = io.StringIO() + with redirect_stdout(buf): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "omnivoice", + "task": "tts"}]}, + instructions="female, young adult, moderate pitch") + self.assertFalse(client.design_mode) + self.assertTrue(client.instruction_voice) + self.assertIn("instruction voice", buf.getvalue()) + + def test_instructions_with_builtin_speaker_family_stays_speaker_mode(self): + buf = io.StringIO() + with redirect_stdout(buf): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "tts"}]}, + instructions="Very happy.") + self.assertFalse(client.design_mode) + self.assertFalse(client.instruction_voice) + self.assertIn("speaker 'Vivian'", buf.getvalue()) + + def test_config_instructions_used_when_flag_omitted(self): + with patch.object(config, "AUDIOCPP_INSTRUCTIONS", + "A calm elderly storyteller"): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}) + self.assertEqual(client.instructions, "A calm elderly storyteller") + + def test_explicit_instructions_override_config_default(self): + with patch.object(config, "AUDIOCPP_INSTRUCTIONS", "from config"): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="from flag") + self.assertEqual(client.instructions, "from flag") + + +class AudioCppFamilyDetectionTests(unittest.TestCase): + """Family auto-detection and per-family adaptations.""" + + @staticmethod + def _json_response(payload): + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = json.dumps(payload).encode("utf-8") + return response + + def _client(self, voice="narrator", models=None): + def _dispatch(request, **_kwargs): + url = request if isinstance(request, str) else request.full_url + if url.endswith("/health"): + return self._json_response({"status": "ok"}) + if url.endswith("/v1/models"): + return self._json_response(models) + if "/v1/audio/voices" in url: + return self._json_response({"voices": [voice] if voice else []}) + raise AssertionError(f"unexpected URL: {url}") + + with patch("converter.tts.urllib.request.urlopen", + side_effect=_dispatch): + return AudioCppTTSClient(voice=voice) + + def test_family_detected_from_models_endpoint(self): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "higgs_audio_tts"}]}) + self.assertEqual(client.family, "higgs_audio_tts") + self.assertIs(client.profile, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) + + def test_missing_family_falls_back_to_qwen3_tts(self): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID}]}) + self.assertEqual(client.family, "qwen3_tts") + self.assertTrue(client.profile.builtin_speakers) + + def test_unknown_family_uses_generic_profile(self): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "future_tts"}]}) + self.assertEqual(client.family, "future_tts") + self.assertIs(client.profile, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) + self.assertFalse(client.profile.builtin_speakers) + self.assertEqual(client.profile.language_style, tts.AUDIOCPP_LANG_OMIT) + + def test_speaker_mode_rejected_for_clone_only_family(self): + client = None + try: + client = self._client(voice=None, models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "voxcpm2"}]}) + except RuntimeError as exc: + message = str(exc) + self.assertIn("voxcpm2", message) + self.assertIn("--voice", message) + self.assertIn("no built-in speakers", message) + self.assertIsNone(client) + + def test_speaker_mode_allowed_for_qwen_family(self): + client = self._client(voice=None, models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts"}]}) + self.assertEqual(client.family, "qwen3_tts") + + def test_clone_model_id_of_different_family_is_ignored(self): + with patch.object(config, "AUDIOCPP_MODEL_ID", "higgs"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen-clone"), \ + self.assertLogs("converter.tts", level="WARNING") as logs: + client = self._client(models={"data": [ + {"id": "higgs", "family": "higgs_audio_tts"}, + {"id": "qwen-clone", "family": "qwen3_tts"}]}) + self.assertEqual(client.model_id, "higgs") + self.assertTrue(any("different family" in line.lower() or + "hosts family" in line.lower() + for line in logs.output)) + + def test_clone_model_id_missing_on_non_qwen_server_is_debug_only(self): + with patch.object(config, "AUDIOCPP_MODEL_ID", "higgs"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen-clone"), \ + self.assertNoLogs("converter.tts", level="WARNING"): + client = self._client(models={"data": [ + {"id": "higgs", "family": "higgs_audio_tts"}]}) + self.assertEqual(client.model_id, "higgs") + + def test_clone_model_id_missing_on_qwen_server_still_warns(self): + 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(models={"data": [ + {"id": "qwen3-tts", "family": "qwen3_tts"}, + {"id": "pocket-tts", "family": "pocket_tts"}]}) + self.assertEqual(client.model_id, "qwen3-tts") + self.assertTrue(any("qwen3-tts-clone" in line for line in logs.output)) + + def test_iso_language_code_helper(self): + self.assertEqual(tts.LANGUAGE_ISO_CODES["English"], "en") + self.assertIsNone(tts.LANGUAGE_ISO_CODES.get("Auto")) + + +class AudioCppTTSClientRequestTests(unittest.TestCase): + """The /v1/audio/speech payload and response validation.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._chunks = patch.object(tts, "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() + + @staticmethod + def _make_client(preset_mode=False, voice="Vivian", language="English", seed=-1, + chunk_text=True, family="qwen3_tts", task="tts", + instructions=None, request_options=None): + client = AudioCppTTSClient.__new__(AudioCppTTSClient) + client.api_url = "http://127.0.0.1:8080" + client.model_id = config.AUDIOCPP_MODEL_ID + client.preset_mode = preset_mode + client.voice = voice + client.language = language + client._seed = seed + client.chunk_text = chunk_text + client.family = family + client.task = task + client.profile = tts.AUDIOCPP_FAMILY_PROFILES.get( + family, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) + client.instructions = instructions or "" + client.request_options = dict(request_options or {}) + client.design_mode = task == tts.AUDIOCPP_TASK_VDES + # Mirrors the connect-time rule: an instruction-defined voice on a + # family without built-in speakers (design mode takes precedence). + client.instruction_voice = ( + not preset_mode and not client.design_mode + and not client.profile.builtin_speakers + and bool(client.instructions)) + return client + + @staticmethod + def _wav_bytes(frames=b"\x01\x00" * 10, rate=tts.SAMPLE_RATE): + buffer = io.BytesIO() + with wave.open(buffer, "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(rate) + wav_file.writeframes(frames) + return buffer.getvalue() + + def _post_response(self, body): + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = body + return response + + def test_payload_includes_model_input_voice_language_and_seed(self): + client = self._make_client(preset_mode=True, voice="narrator", + language="Japanese", seed=1234) + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello world.") + request = mock_urlopen.call_args[0][0] + self.assertEqual(request.full_url, + "http://127.0.0.1:8080/v1/audio/speech") + payload = json.loads(request.data.decode("utf-8")) + self.assertEqual(payload["model"], config.AUDIOCPP_MODEL_ID) + self.assertEqual(payload["input"], "Hello world.") + self.assertEqual(payload["voice"], "narrator") + self.assertEqual(payload["language"], "Japanese") + self.assertEqual(payload["seed"], 1234) + self.assertNotIn("instructions", payload) + + def test_negative_seed_omitted_from_payload(self): + client = self._make_client(preset_mode=True, voice="narrator", seed=-1) + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello world.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertNotIn("seed", payload) + + def test_whole_text_sent_as_one_request_without_client_chunking(self): + client = self._make_client(chunk_text=False) + # 9 words with CHUNK_SIZE=5 would split in two if client chunking + # were on. + text = " ".join(f"word{i}" for i in range(9)) + with patch.object(config, "CHUNK_SIZE", 5), \ + patch.object(client, "_request_wav", + return_value=self._wav_bytes()) as mock_request: + result = client.generate_chunk(text, 1) + self.assertIsNotNone(result) + self.assertEqual(mock_request.call_count, 1) + self.assertEqual(mock_request.call_args[0][0], text) + + def test_single_request_timeout_scales_with_text_length(self): + client = self._make_client(chunk_text=False) + long_text = " ".join(f"word{i}" for i in range(1500)) # ~10 min of audio + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav(long_text) + timeout = mock_urlopen.call_args[1]["timeout"] + self.assertGreater(timeout, config.API_TIMEOUT) + + def test_client_chunking_keeps_configured_timeout(self): + client = self._make_client(chunk_text=True) + long_text = " ".join(f"word{i}" for i in range(1500)) + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav(long_text) + timeout = mock_urlopen.call_args[1]["timeout"] + self.assertEqual(timeout, config.API_TIMEOUT) + + def test_speaker_mode_sends_instruct(self): + client = self._make_client(preset_mode=False) + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["instructions"], config.INSTRUCT) + + def test_explicit_instructions_replace_config_instruct(self): + # --instructions overrides the INSTRUCT default in speaker mode. + client = self._make_client(preset_mode=False, + instructions="Read whisper quiet.") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["instructions"], "Read whisper quiet.") + + def test_preset_mode_sends_instructions_alongside_voice(self): + # Clone + style control: both the server-side voice and the + # instruction reach the model. + client = self._make_client(preset_mode=True, voice="narrator", + instructions="Calm and steady.") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["voice"], "narrator") + self.assertEqual(payload["instructions"], "Calm and steady.") + + def test_design_mode_payload_omits_voice_and_sends_instructions(self): + client = self._make_client(task="vdes", + instructions="A warm adult narrator") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertNotIn("voice", payload) + self.assertEqual(payload["instructions"], "A warm adult narrator") + + def test_design_mode_language_follows_family_profile(self): + # The VoiceDesign package is family qwen3_tts, whose language field + # takes Qwen display names like the other variants. + client = self._make_client(task="vdes", language="Japanese", + instructions="A warm adult narrator") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["language"], "Japanese") + + def test_instruction_voice_payload_omits_voice(self): + # Instruction-defined voice on a family without built-in speakers: + # no speaker name is invented, the instruction carries the voice. + client = self._make_client(family="omnivoice", + instructions="female, young adult") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertNotIn("voice", payload) + self.assertNotIn("language", payload) # generic profile: omitted + self.assertEqual(payload["instructions"], "female, young adult") + + def test_request_options_forwarded_in_payload(self): + client = self._make_client(preset_mode=True, voice="narrator", + request_options={"emotion": "neutral", + "speed": "1.1"}) + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["options"], {"emotion": "neutral", + "speed": "1.1"}) + + def test_empty_request_options_omit_options_field(self): + client = self._make_client(preset_mode=True, voice="narrator") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertNotIn("options", payload) + + def test_generic_family_omits_language_and_instructions(self): + # Clone-only families (higgs_audio_tts, voxcpm2, ...) detect the + # language themselves and take no style instruction. + client = self._make_client(preset_mode=False, family="higgs_audio_tts") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertNotIn("language", payload) + self.assertNotIn("instructions", payload) + + def test_iso_family_sends_language_code(self): + client = self._make_client(preset_mode=True, voice="narrator", + language="Japanese", family="index_tts2") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["language"], "ja") + + def test_iso_family_auto_omits_language(self): + client = self._make_client(preset_mode=True, voice="narrator", + language="Auto", family="index_tts2") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertNotIn("language", payload) + + def test_qwen_language_display_name_still_sent(self): + client = self._make_client(preset_mode=True, voice="narrator", + language="Japanese", family="qwen3_tts") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["language"], "Japanese") + + def test_non_wav_response_rejected(self): + client = self._make_client() + for body in (b"", b"RIFFxxxx", b"MP3DATA-MP3DATA", b"RIFF\x00\x00\x00\x00mpeg"): + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(body)): + with self.assertRaises(RuntimeError): + client._request_wav("Hello.") + + def test_http_error_body_surfaced(self): + import urllib.error + client = self._make_client() + error = urllib.error.HTTPError( + "http://127.0.0.1:8080/v1/audio/speech", 500, + "Server Error", {}, io.BytesIO(b'{"error":"bad voice"}')) + with patch("converter.tts.urllib.request.urlopen", side_effect=error): + with self.assertRaises(RuntimeError) as ctx: + client._request_wav("Hello.") + self.assertIn("500", str(ctx.exception)) + self.assertIn("bad voice", str(ctx.exception)) + + def test_transient_failure_is_retried(self): + client = self._make_client() + wav = self._wav_bytes() + with patch.object(client, "_request_wav", + side_effect=[RuntimeError("boom"), wav]) as mock_request: + result = client.generate_chunk("Hello.", 1) + self.assertIsNotNone(result) + self.assertEqual(mock_request.call_count, 2) + + def test_exhausted_retries_fail_the_chunk(self): + client = self._make_client() + with patch.object(client, "_request_wav", + side_effect=RuntimeError("down")) as mock_request: + result = client.generate_chunk("Hello.", 1) + self.assertIsNone(result) + self.assertEqual(mock_request.call_count, config.MAX_RETRIES) + + def test_empty_text_fails_the_chunk(self): + client = self._make_client() + with patch.object(client, "_request_wav") as mock_request: + result = client.generate_chunk(" ", 1) + self.assertIsNone(result) + mock_request.assert_not_called() + + def test_generate_chunk_writes_valid_wav(self): + client = self._make_client() + frames = b"\x01\x00" * 100 + with patch.object(client, "_request_wav", return_value=self._wav_bytes(frames)): + result = client.generate_chunk("Hello world.", 1) + self.assertIsNotNone(result) + path = Path(result) + self.assertEqual(path.name, "chunk_0001.wav") + with wave.open(str(path), "rb") as wav_file: + self.assertEqual(wav_file.getnchannels(), 1) + self.assertEqual(wav_file.getsampwidth(), 2) + self.assertEqual(wav_file.getframerate(), tts.SAMPLE_RATE) + self.assertEqual(wav_file.readframes(wav_file.getnframes()), frames) + + 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) + 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(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) + with wave.open(str(Path(result)), "rb") as wav_file: + self.assertEqual(wav_file.readframes(wav_file.getnframes()), + b"\x01\x00" * 10 + b"\x02\x00" * 20 + b"\x03\x00" * 30) + + 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_wav", return_value=self._wav_bytes()): + 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"]) + + +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.""" + + 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 = AudioCppTTSClient.__new__(AudioCppTTSClient) + client.api_url = "http://127.0.0.1:8080" + client.model_id = config.AUDIOCPP_MODEL_ID + client.preset_mode = True + client.voice = "narrator" + client.language = "English" + client._seed = -1 + client.chunk_text = True + client.family = "qwen3_tts" + client.profile = tts.AUDIOCPP_FAMILY_PROFILES["qwen3_tts"] + return client + + @staticmethod + def _wav_bytes(frames): + 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(frames) + return buffer.getvalue() + + def test_full_length_wav_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. + wav = self._wav_bytes(b"\x01\x00" * int(2.5 * tts.SAMPLE_RATE)) + with patch.object(client, "_request_wav", return_value=wav): + result = client.generate_chunk(text, 1) + self.assertIsNotNone(result) + + +class BackendWiringTests(unittest.TestCase): + """AudiobookConverter wiring for the --backend selector.""" + + def test_faster_backend_uses_faster_client_without_reference(self): + with patch("converter.converter.FasterTTSClient") as mock_faster, \ + patch("converter.converter.QwenTTSClient") as mock_qwen, \ + patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: + AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, + backend=tts.BACKEND_FASTER, voice="narrator") + mock_faster.assert_called_once_with(voice="narrator") + mock_qwen.assert_not_called() + mock_audiocpp.assert_not_called() + + def test_audiocpp_backend_with_voice_uses_audiocpp_client(self): + with patch("converter.converter.FasterTTSClient") as mock_faster, \ + patch("converter.converter.QwenTTSClient") as mock_qwen, \ + patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: + AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, + backend=tts.BACKEND_AUDIOCPP, voice="narrator", + language="ja") + mock_audiocpp.assert_called_once_with(voice="narrator", language="Japanese", + chunk_text=False, model_id=None, + instructions=None, + request_options={}) + mock_faster.assert_not_called() + mock_qwen.assert_not_called() + + def test_audiocpp_backend_without_voice_uses_audiocpp_client(self): + with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: + AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, + backend=tts.BACKEND_AUDIOCPP) + mock_audiocpp.assert_called_once_with(voice=None, language=config.LANGUAGE, + chunk_text=False, model_id=None, + instructions=None, + request_options={}) + + def test_audiocpp_backend_chunk_flag_forces_client_chunking(self): + with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: + converter = AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, + backend=tts.BACKEND_AUDIOCPP, + voice="narrator", chunk=True) + mock_audiocpp.assert_called_once_with(voice="narrator", + language=config.LANGUAGE, + chunk_text=True, model_id=None, + instructions=None, + request_options={}) + self.assertTrue(converter.client_chunks) + + def test_audiocpp_backend_model_id_is_wired_through(self): + with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: + AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, + backend=tts.BACKEND_AUDIOCPP, voice="narrator", + model_id="higgs") + mock_audiocpp.assert_called_once_with( + voice="narrator", language=config.LANGUAGE, + chunk_text=False, model_id="higgs", instructions=None, + request_options={}) + + def test_audiocpp_backend_instructions_and_options_are_wired_through(self): + with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: + AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, + backend=tts.BACKEND_AUDIOCPP, + instructions="A warm adult narrator", + request_options={"emotion": "neutral", + "speed": "1.1"}) + mock_audiocpp.assert_called_once_with( + voice=None, language=config.LANGUAGE, + chunk_text=False, model_id=None, + instructions="A warm adult narrator", + request_options={"emotion": "neutral", "speed": "1.1"}) + + def test_qwen_backend_uses_qwen_client(self): + with patch("converter.converter.FasterTTSClient") as mock_faster, \ + patch("converter.converter.QwenTTSClient") as mock_qwen, \ + patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: + AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, + backend=tts.BACKEND_QWEN) + mock_qwen.assert_called_once() + mock_faster.assert_not_called() + mock_audiocpp.assert_not_called() + + def test_qwen_clone_mode_still_requires_reference(self): + with patch("converter.converter.QwenTTSClient"): + with self.assertRaises(ValueError): + AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, + backend=tts.BACKEND_QWEN) + + def test_audiocpp_clone_mode_does_not_require_reference(self): + # Cloning is server-side for the audiocpp backend, so the + # clone-mode voice can be selected without local reference audio. + with patch("converter.converter.AudioCppTTSClient"): + converter = AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, + backend=tts.BACKEND_AUDIOCPP, + voice="narrator") + self.assertIsNone(converter.voice_clone_ref_audio) + + def test_chapter_chunks_audiocpp_default_is_one_request(self): + converter = self._audiocpp_converter(voice="narrator") + text = " ".join(f"word{i}" for i in range(50)) + with patch.object(config, "CHUNK_SIZE", 10): + self.assertEqual(converter._chapter_chunks(text), [text]) + + def test_chapter_chunks_audiocpp_chunk_flag_splits(self): + with patch("converter.converter.AudioCppTTSClient"): + converter = AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, + backend=tts.BACKEND_AUDIOCPP, + voice="narrator", chunk=True) + text = " ".join(f"word{i}" for i in range(50)) + with patch.object(config, "CHUNK_SIZE", 10): + chunks = converter._chapter_chunks(text) + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(chunk.split()) <= 10 for chunk in chunks)) + + def test_chapter_chunks_qwen_always_splits(self): + with patch("converter.converter.QwenTTSClient"): + converter = AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, + backend=tts.BACKEND_QWEN) + text = " ".join(f"word{i}" for i in range(50)) + with patch.object(config, "CHUNK_SIZE", 10): + chunks = converter._chapter_chunks(text) + self.assertGreater(len(chunks), 1) + + def test_faster_backend_still_validates_other_settings(self): + with patch("converter.converter.FasterTTSClient"): + with self.assertRaises(ValueError): + AudiobookConverter(backend=tts.BACKEND_FASTER, speed=0) + with self.assertRaises(ValueError): + AudiobookConverter(backend=tts.BACKEND_FASTER, language="klingon") + + def test_audiocpp_backend_still_validates_other_settings(self): + with patch("converter.converter.AudioCppTTSClient"): + with self.assertRaises(ValueError): + AudiobookConverter(backend=tts.BACKEND_AUDIOCPP, speed=0) + with self.assertRaises(ValueError): + AudiobookConverter(backend=tts.BACKEND_AUDIOCPP, language="klingon") + + def _faster_converter(self, voice=None): + with patch("converter.converter.FasterTTSClient"): + return AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, + backend=tts.BACKEND_FASTER, voice=voice) + + def _audiocpp_converter(self, voice=None): + with patch("converter.converter.AudioCppTTSClient"): + return AudiobookConverter( + voice_mode=tts.VOICE_MODE_CLONE if voice else tts.VOICE_MODE_CUSTOM, + backend=tts.BACKEND_AUDIOCPP, voice=voice) + + def test_narrator_tag_uses_faster_voice_name(self): + converter = self._faster_converter(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_VOICE) + + def test_narrator_tag_audiocpp_uses_voice_name(self): + converter = self._audiocpp_converter(voice="female_narrator") + self.assertEqual(converter._narrator_tag(), "female_narrator") + + def test_narrator_tag_audiocpp_falls_back_to_speaker(self): + converter = self._audiocpp_converter() + self.assertEqual(converter._narrator_tag(), "Vivian") + + def test_banner_and_narrator_work_without_reference_audio(self): + converter = self._faster_converter(voice="male_richard_poe") + converter._print_banner() # must not raise (regression: Path(None)) + self.assertIsNone(converter.voice_clone_ref_audio) + + def test_audiocpp_banner_prints_without_reference_audio(self): + converter = self._audiocpp_converter(voice="narrator") + converter._print_banner() # must not raise + converter = self._audiocpp_converter() + converter._print_banner() + + def test_audiocpp_banner_prints_model_family(self): + from contextlib import redirect_stdout + converter = self._audiocpp_converter(voice="narrator") + converter.tts.family = "higgs_audio_tts" + buffer = io.StringIO() + with redirect_stdout(buffer): + converter._print_banner() + self.assertIn("higgs_audio_tts", buffer.getvalue()) + + 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=tts.VOICE_MODE_CLONE, + voice_clone_ref_audio=str(ref), + backend=tts.BACKEND_QWEN) + self.assertEqual(converter._narrator_tag(), "ref") + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_tui.py b/app/tests/test_tui.py new file mode 100644 index 0000000..c121d55 --- /dev/null +++ b/app/tests/test_tui.py @@ -0,0 +1,661 @@ +"""Tests for the DOS-style curses TUI widgets in tui.py. + +The widget module imports curses lazily, so these tests swap the +curses module for a small fake (patched into sys.modules) and drive +the widgets with scripted keys against a recording screen. That works +without a terminal and lets the tests assert exact drawing +coordinates: theme colors, the left margin of list rows, and that no +row ever paints over the dialog border. +""" + +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from ui import tui + + +class FakeCurses: + """Minimal curses stand-in: attributes, key codes, color pairs.""" + + A_BOLD = 1 + A_DIM = 2 + A_REVERSE = 4 + + COLOR_BLACK = 0 + COLOR_RED = 1 + COLOR_GREEN = 2 + COLOR_YELLOW = 3 + COLOR_BLUE = 4 + COLOR_MAGENTA = 5 + COLOR_CYAN = 6 + COLOR_WHITE = 7 + + KEY_DOWN = 0x101 + KEY_UP = 0x102 + KEY_LEFT = 0x103 + KEY_RIGHT = 0x104 + KEY_HOME = 0x105 + KEY_END = 0x106 + KEY_PPAGE = 0x107 + KEY_NPAGE = 0x108 + KEY_BACKSPACE = 0x109 + KEY_BTAB = 0x10A + + ACS_ULCORNER = "ul" + ACS_URCORNER = "ur" + ACS_LLCORNER = "ll" + ACS_LRCORNER = "lr" + ACS_VLINE = "v" + ACS_HLINE = "h" + + class error(Exception): + pass + + def __init__(self): + self.pairs = {} # pair number -> (fg, bg) + self.colors = True + + def has_colors(self): + return self.colors + + def start_color(self): + pass + + def init_pair(self, number, fg, bg): + self.pairs[number] = (fg, bg) + + def color_pair(self, number): + return number << 8 + + def curs_set(self, visibility): + pass + + def endwin(self): + pass + + +class FakeScreen: + """Recording curses window; getch() replays scripted keys.""" + + def __init__(self, keys=(), width=80, height=24): + self.keys = list(keys) + self.width = width + self.height = height + self.strings = [] # (y, x, text, attr) from addstr + self.chars = [] # (y, x, ch, attr) from addch + + def getmaxyx(self): + return self.height, self.width + + def erase(self): + pass + + def refresh(self): + pass + + def bkgd(self, ch, attr): + pass + + def addstr(self, y, x, text, attr=0): + self.strings.append((y, x, text, attr)) + + def addch(self, y, x, ch, attr=0): + self.chars.append((y, x, ch, attr)) + + def hline(self, y, x, ch, n, attr=0): + pass + + def redrawwin(self): + pass + + def getch(self): + if not self.keys: + raise AssertionError("the script ran out of keys") + return self.keys.pop(0) + + +class TuiTestCase(unittest.TestCase): + """Base class: fresh theme + fake curses module for every test.""" + + def setUp(self): + tui._THEME.clear() + self.curses = FakeCurses() + self.screen = FakeScreen() + patcher = patch.dict(sys.modules, {"curses": self.curses}) + patcher.start() + self.addCleanup(patcher.stop) + self.addCleanup(tui._THEME.clear) + + def dialog_box(self, screen=None): + """(x0, x_right) border columns of the drawn dialog.""" + screen = screen or self.screen + corners = screen.chars[:2] + self.assertEqual(corners[0][2], FakeCurses.ACS_ULCORNER) + self.assertEqual(corners[1][2], FakeCurses.ACS_URCORNER) + return corners[0][1], corners[1][1] + + def assert_inside_border(self, screen=None): + """No drawn string may reach the right border column.""" + screen = screen or self.screen + _, x_right = self.dialog_box(screen) + for _, x, text, _ in screen.strings: + self.assertLessEqual( + x + len(text), x_right, + f"{text!r} painted over the border at x={x}") + + +class ThemeTests(TuiTestCase): + def test_desktop_and_message_backgrounds_are_black(self): + frame = tui.Frame(self.screen, "Title", "footer") + theme, pairs = frame.theme, self.curses.pairs + for name in ("desktop", "border", "title", "ok", "warn", "err", + "info", "input", "check", "accent"): + fg, bg = pairs[theme[name] >> 8] + self.assertEqual(bg, FakeCurses.COLOR_BLACK, name) + + def test_cursor_bar_and_active_button_stand_out(self): + frame = tui.Frame(self.screen, "Title", "footer") + theme, pairs = frame.theme, self.curses.pairs + fg, bg = pairs[theme["bar"] >> 8] + self.assertEqual((fg, bg), + (FakeCurses.COLOR_BLACK, FakeCurses.COLOR_CYAN)) + fg, bg = pairs[theme["btn_on"] >> 8] + self.assertEqual((fg, bg), + (FakeCurses.COLOR_BLACK, FakeCurses.COLOR_GREEN)) + + def test_without_colors_theme_uses_plain_attributes(self): + self.curses.colors = False + frame = tui.Frame(self.screen, "Title", "footer") + self.assertEqual(self.curses.pairs, {}) + self.assertEqual(frame.theme["desktop"], 0) + self.assertEqual(frame.theme["title"], FakeCurses.A_BOLD) + + +class MenuTests(TuiTestCase): + OPTIONS = [("first option", "one"), ("second option", "two")] + + def test_option_rows_left_justified_help_centered(self): + screen = FakeScreen(keys=[10]) + value = tui.menu(screen, "Pick one", self.OPTIONS, + help_lines=["Help text"]) + self.assertEqual(value, "one") + x0, x_right = self.dialog_box(screen) + margin = x0 + 1 + tui.Frame.LIST_MARGIN + for label in ("first option", "second option"): + x = next(x for _, x, text, _ in screen.strings if text == label) + self.assertEqual(x, margin, label) + inner_w = x_right - x0 - 1 + help_x = next(x for _, x, text, _ in screen.strings + if text == "Help text") + self.assertEqual(help_x, x0 + 1 + (inner_w - len("Help text")) // 2) + self.assertGreater(help_x, margin) + self.assert_inside_border(screen) + + def test_up_wraps_around_to_last_option(self): + screen = FakeScreen(keys=[FakeCurses.KEY_UP, 10]) + value = tui.menu(screen, "Pick one", self.OPTIONS) + self.assertEqual(value, "two") + + def test_no_option_painted_over_the_border(self): + screen = FakeScreen(keys=[FakeCurses.KEY_END, 10]) + tui.menu(screen, "Pick", [("a" * 60, "a"), ("b", "b")]) + self.assert_inside_border(screen) + + def test_empty_options_rejected(self): + with self.assertRaises(ValueError): + tui.menu(self.screen, "Pick", []) + + def test_esc_returns_back_value(self): + marker = object() + screen = FakeScreen(keys=[27]) + self.assertIs( + tui.menu(screen, "Pick", self.OPTIONS, back_value=marker), + marker) + + def test_q_still_aborts_with_back_value(self): + marker = object() + screen = FakeScreen(keys=[ord("q")]) + with self.assertRaises(tui.WizardCancelled): + tui.menu(screen, "Pick", self.OPTIONS, back_value=marker) + + +class MenuTableTests(TuiTestCase): + """The optional status table: aligned columns and colored statuses.""" + + ROWS = [("audio.cpp", "not installed", "err"), + ("qwen-tts", "installed", "warn"), + ("faster-qwen3-tts", "running", "ok")] + + def test_name_column_left_aligned_at_margin(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + x0, _ = self.dialog_box(screen) + margin = x0 + 1 + tui.Frame.LIST_MARGIN + for name, _, _ in self.ROWS: + x = next(x for _, x, text, _ in screen.strings + if text.rstrip() == name) + self.assertEqual(x, margin, name) + + def test_status_column_aligned_at_one_fixed_offset(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + x0, _ = self.dialog_box(screen) + margin = x0 + 1 + tui.Frame.LIST_MARGIN + name_w = max(len(name) for name, _, _ in self.ROWS) + expected_x = margin + name_w # the " status" segment starts here + for _, status, _ in self.ROWS: + x = next(x for _, x, text, _ in screen.strings + if text.strip() == status) + self.assertEqual(x, expected_x, status) + + def test_status_text_uses_the_theme_kind_color(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + want = {"err": tui._THEME["err"], "warn": tui._THEME["warn"], + "ok": tui._THEME["ok"]} + for _, status, kind in self.ROWS: + attr = next(a for _, _, text, a in screen.strings + if text.strip() == status) + self.assertEqual(attr, want[kind], status) + + def test_optional_name_kind_colors_the_name_column(self): + # 4-element rows: the 4th value is a theme kind for the name. + rows = [("gone", "unavailable", "err", "dim"), + ("here", "running", "ok", "body")] + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], table_rows=rows) + drawn = {text.rstrip(): attr for _, _, text, attr in screen.strings} + self.assertEqual(drawn["gone"], tui._THEME["dim"]) + self.assertEqual(drawn["here"], tui._THEME["body"]) + + def test_three_element_rows_default_to_body_names(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + for name, _, _ in self.ROWS: + attr = next(a for _, _, text, a in screen.strings + if text.rstrip() == name) + self.assertEqual(attr, tui._THEME["body"], name) + + def test_table_title_is_dim_and_left_aligned(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + x0, _ = self.dialog_box(screen) + margin = x0 + 1 + tui.Frame.LIST_MARGIN + x, attr = next((x, a) for _, x, text, a in screen.strings + if text == "Backend status") + self.assertEqual(x, margin) + self.assertEqual(attr, tui._THEME["dim"]) + + def test_table_does_not_paint_over_the_border(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + self.assert_inside_border(screen) + + +class ConfirmTests(TuiTestCase): + def test_tab_switches_and_enter_activates(self): + screen = FakeScreen(keys=[9, 10]) + self.assertFalse(tui.confirm(screen, "Overwrite?", default=True)) + + def test_y_answers_directly(self): + screen = FakeScreen(keys=[ord("y")]) + self.assertTrue(tui.confirm(screen, "Overwrite?", default=False)) + + def test_enter_takes_the_default(self): + screen = FakeScreen(keys=[10]) + self.assertTrue(tui.confirm(screen, "Overwrite?", default=True)) + + def test_esc_aborts_without_cancel_value(self): + screen = FakeScreen(keys=[27]) + with self.assertRaises(tui.WizardCancelled): + tui.confirm(screen, "Overwrite?", default=True) + + def test_esc_returns_cancel_value(self): + marker = object() + screen = FakeScreen(keys=[27]) + self.assertIs( + tui.confirm(screen, "Overwrite?", default=True, + cancel_value=marker), + marker) + + def test_q_returns_cancel_value(self): + marker = object() + screen = FakeScreen(keys=[ord("q")]) + self.assertIs( + tui.confirm(screen, "Overwrite?", default=True, + cancel_value=marker), + marker) + + +class LineEditTests(TuiTestCase): + def test_typing_backspace_and_enter(self): + keys = [ord("c"), ord("d"), FakeCurses.KEY_BACKSPACE, 10] + screen = FakeScreen(keys=keys) + value = tui.line_edit(screen, "Edit", "ab") + self.assertEqual(value, "abc") + + def test_validation_error_keeps_editing(self): + keys = [ord("x"), 10, FakeCurses.KEY_BACKSPACE, 10] + screen = FakeScreen(keys=keys) + value = tui.line_edit( + screen, "Edit", "5", + validate=lambda s: None if s.isdigit() else "digits only") + self.assertEqual(value, "5") + + def test_esc_returns_back_value(self): + marker = object() + screen = FakeScreen(keys=[27]) + self.assertIs( + tui.line_edit(screen, "Edit", "text", back_value=marker), + marker) + + def test_q_stays_typeable_with_back_value(self): + marker = object() + screen = FakeScreen(keys=[ord("q"), 10]) + value = tui.line_edit(screen, "Edit", "te", back_value=marker) + self.assertEqual(value, "teq") + + +class FormTests(TuiTestCase): + def _fields(self): + # Fresh dicts each call: form() edits field values in place, and a + # shared class attribute would leak edits between tests. + return [ + {"key": "fmt", "label": "Format", "kind": "choice", + "value": "m4b", "choices": ["mp3", "m4b", "ogg"]}, + {"key": "chunk", "label": "Chunk", "kind": "text", + "value": "250"}, + ] + + def test_save_returns_current_values(self): + screen = FakeScreen(keys=[9, 10]) # Tab -> Save, Enter + result = tui.form(screen, "Settings", self._fields()) + self.assertEqual(result, {"fmt": "m4b", "chunk": "250"}) + + def test_cancel_returns_back_value(self): + marker = object() + # Tab -> buttons, Left -> Cancel, Enter. + screen = FakeScreen(keys=[9, FakeCurses.KEY_LEFT, 10]) + result = tui.form(screen, "Settings", self._fields(), + back_value=marker) + self.assertIs(result, marker) + + def test_esc_returns_back_value(self): + marker = object() + screen = FakeScreen(keys=[27]) + self.assertIs( + tui.form(screen, "Settings", self._fields(), back_value=marker), + marker) + + def test_q_still_aborts_with_back_value(self): + marker = object() + screen = FakeScreen(keys=[ord("q")]) + with self.assertRaises(tui.WizardCancelled): + tui.form(screen, "Settings", self._fields(), back_value=marker) + + def test_choice_field_picks_another_value(self): + # Enter opens the choice menu on 'm4b' (index 1); Down moves to + # 'ogg', Enter accepts; Tab -> Save, Enter. + screen = FakeScreen(keys=[10, FakeCurses.KEY_DOWN, 10, 9, 10]) + result = tui.form(screen, "Settings", self._fields()) + self.assertEqual(result, {"fmt": "ogg", "chunk": "250"}) + + def test_text_field_edits_then_saves(self): + # Down to the text row, Enter opens the editor, type 'x', Enter, + # then Tab -> Save, Enter. + screen = FakeScreen( + keys=[FakeCurses.KEY_DOWN, 10, ord("x"), 10, 9, 10]) + result = tui.form(screen, "Settings", self._fields()) + self.assertEqual(result, {"fmt": "m4b", "chunk": "250x"}) + + def test_validation_error_refuses_save_then_recovers(self): + fields = [ + {"key": "chunk", "label": "Chunk", "kind": "text", "value": "bad", + "validate": lambda s: None if s.isdigit() else "digits only"}, + {"key": "fmt", "label": "Format", "kind": "text", "value": "x"}, + ] + # Tab->Save(Enter) fails, the flash consumes the next key; Enter + # reopens the editor, Ctrl-U clears 'bad', type '120', Enter; then + # Tab->Save(Enter). + keys = [9, 10, 10, 10, 21, ord("1"), ord("2"), ord("0"), 10, 9, 10] + screen = FakeScreen(keys=keys) + result = tui.form(screen, "Settings", fields) + self.assertEqual(result, {"chunk": "120", "fmt": "x"}) + + def test_rows_left_justified_inside_the_border(self): + screen = FakeScreen(keys=[9, 10]) + tui.form(screen, "Settings", self._fields()) + self.assert_inside_border(screen) + + def test_empty_fields_rejected(self): + with self.assertRaises(ValueError): + tui.form(self.screen, "Settings", []) + + +def _accept_audio_cpp(entry: Path): + """auto_select callback that accepts an 'audio.cpp' checkout root.""" + if entry.name == "audio.cpp" and (entry / "model_specs").is_dir(): + return entry + return None + + +class BrowseDirectoryTests(TuiTestCase): + def setUp(self): + super().setUp() + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.root = Path(tmp.name) + for name in ("alpha", "beta", "zulu"): + (self.root / name).mkdir() + (self.root / "noise.txt").write_text("x", encoding="utf-8") + + def _checkout_tree(self): + """A temp dir containing an 'audio.cpp' checkout + a sibling dir.""" + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + root = Path(tmp.name) + (root / "audio.cpp").mkdir() + (root / "audio.cpp" / "model_specs").mkdir() + (root / "other").mkdir() + return root + + def test_listing_rows_left_justified(self): + screen = FakeScreen(keys=[10]) + chosen = tui.browse_directory(screen, "Pick", start=self.root) + self.assertEqual(chosen, self.root.resolve()) + x0, x_right = self.dialog_box(screen) + margin = x0 + 1 + tui.Frame.LIST_MARGIN + for label in ("[ Use this directory ]", "..", + "alpha/", "beta/", "zulu/"): + x = next(x for _, x, text, _ in screen.strings if text == label) + self.assertEqual(x, margin, label) + drawn = " ".join(text for _, _, text, _ in screen.strings) + self.assertNotIn("noise.txt", drawn) + self.assert_inside_border(screen) + + def test_enter_opens_highlighted_subdirectory(self): + keys = [FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10, 10] + screen = FakeScreen(keys=keys) + chosen = tui.browse_directory(screen, "Pick", start=self.root) + self.assertEqual(chosen, (self.root / "alpha").resolve()) + self.assert_inside_border(screen) + + def test_enter_auto_accepts_matching_subdir(self): + root = self._checkout_tree() + # sel 0 = [ Use this directory ], 1 = .., 2 = audio.cpp/ + keys = [FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10] + screen = FakeScreen(keys=keys) + chosen = tui.browse_directory(screen, "Pick", start=root, + auto_select=_accept_audio_cpp) + self.assertEqual(chosen, (root / "audio.cpp").resolve()) + + def test_right_auto_accepts_matching_subdir(self): + root = self._checkout_tree() + keys = [FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + FakeCurses.KEY_RIGHT] + screen = FakeScreen(keys=keys) + chosen = tui.browse_directory(screen, "Pick", start=root, + auto_select=_accept_audio_cpp) + self.assertEqual(chosen, (root / "audio.cpp").resolve()) + + def test_use_this_directory_ignores_auto_select(self): + # Enter on '[ Use this directory ]' must accept the listed dir + # without ever consulting auto_select. + root = self._checkout_tree() + calls = [] + + def callback(entry): + calls.append(entry) + return entry # would auto-accept any subdir if consulted + + screen = FakeScreen(keys=[10]) + chosen = tui.browse_directory(screen, "Pick", start=root, + auto_select=callback) + self.assertEqual(chosen, root.resolve()) + self.assertEqual(calls, []) + + def test_auto_select_returning_none_descends_normally(self): + # A non-matching subdir (or a None reply) keeps browsing: Enter + # descends into it, then '[ Use this directory ]' accepts it. + root = self._checkout_tree() + calls = [] + + def callback(entry): + calls.append(entry) + return None + + # sel 0 = use, 1 = .., 2 = audio.cpp/, 3 = other/ + keys = [FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + FakeCurses.KEY_DOWN, 10, 10] + screen = FakeScreen(keys=keys) + chosen = tui.browse_directory(screen, "Pick", start=root, + auto_select=callback) + self.assertEqual(chosen, (root / "other").resolve()) + # auto_select was consulted only for the highlighted 'other/' row. + self.assertEqual([p.name for p in calls], ["other"]) + + def test_esc_returns_back_value(self): + marker = object() + screen = FakeScreen(keys=[27]) + self.assertIs( + tui.browse_directory(screen, "Pick", start=self.root, + back_value=marker), + marker) + + def test_q_still_aborts_with_back_value(self): + marker = object() + screen = FakeScreen(keys=[ord("q")]) + with self.assertRaises(tui.WizardCancelled): + tui.browse_directory(screen, "Pick", start=self.root, + back_value=marker) + + +class CheckboxTreeTests(TuiTestCase): + FAMILIES = [ + {"label": "Family one", "detail": "tts", + "options": [{"key": "pkg-a", "label": "pkg-a", "recommended": True}, + {"key": "pkg-b", "label": "pkg-b", "recommended": False}]}, + {"label": "Family two", "detail": "tts, cloning", + "options": [{"key": "pkg-c", "label": "pkg-c", "recommended": True}]}, + ] + + def test_nothing_selected_by_default(self): + # Nothing is pre-checked: Enter alone flashes and waits, and a + # selection only happens after Space checks an option. The first + # Enter and the flash each consume a key. + screen = FakeScreen(keys=[10, 10, ord(" "), 10]) + picked = tui.checkbox_tree(screen, "Pick models", self.FAMILIES) + self.assertEqual(picked, [(0, "pkg-a")]) + + def test_rows_left_justified_inside_the_border(self): + screen = FakeScreen(keys=[ord(" "), 10]) + tui.checkbox_tree(screen, "Pick models", self.FAMILIES) + x0, x_right = self.dialog_box(screen) + margin = x0 + 1 + tui.Frame.LIST_MARGIN + strings = sorted((y, x, text) for y, x, text, _ in screen.strings) + # Family row: its checkbox starts at the margin. + y_family = next(y for y, _, text in strings if text == "- Family one") + family_box_x = next(x for y, x, text in strings + if y == y_family and text == "[x] ") + self.assertEqual(family_box_x, margin) + # Option row: its checkbox sits one indent (2 columns) deeper. + option_box_x = next(x for y, x, text in strings + if text == "[x] " and y != y_family) + self.assertEqual(option_box_x, margin + 4) + self.assert_inside_border(screen) + + def test_long_indented_options_do_not_paint_over_the_border(self): + # A wide screen keeps the dialog width driven by the option row + # itself (not the footer), the geometry where the old centered + # segments drawing could paint over the right border. + families = [{"label": "F", "detail": "tts", + "options": [{"key": "long", "label": "x" * 60, + "recommended": True}]}] + screen = FakeScreen(keys=[ord(" "), 10], width=120) + picked = tui.checkbox_tree(screen, "Pick", families) + self.assertEqual(picked, [(0, "long")]) + self.assert_inside_border(screen) + + def test_space_checks_then_enter_accepts(self): + screen = FakeScreen(keys=[ord(" "), 10]) + picked = tui.checkbox_tree(screen, "Pick models", self.FAMILIES) + self.assertEqual(picked, [(0, "pkg-a")]) + + def test_empty_families_rejected(self): + with self.assertRaises(ValueError): + tui.checkbox_tree(self.screen, "Pick", []) + + def test_esc_returns_back_value(self): + marker = object() + screen = FakeScreen(keys=[27]) + self.assertIs( + tui.checkbox_tree(screen, "Pick models", self.FAMILIES, + back_value=marker), + marker) + + def test_q_still_aborts_with_back_value(self): + marker = object() + screen = FakeScreen(keys=[ord("q")]) + with self.assertRaises(tui.WizardCancelled): + tui.checkbox_tree(screen, "Pick models", self.FAMILIES, + back_value=marker) + + +class SuspendTests(TuiTestCase): + """tui.suspend leaves curses, runs code, then repaints.""" + + def test_suspend_runs_block_and_restores(self): + ran = [] + with tui.suspend(self.screen): + ran.append("inside") + self.assertEqual(ran, ["inside"]) + + def test_suspend_always_restores_on_exception(self): + class Boom(Exception): + pass + with self.assertRaises(Boom): + with tui.suspend(self.screen): + raise Boom() + + +class FlashTests(TuiTestCase): + """tui.flash shows a notice until any key is pressed.""" + + def test_notice_dismissed_by_any_key(self): + screen = FakeScreen(keys=[10]) + # Should return (None) after consuming one key; not raise. + tui.flash(screen, "a notice", kind="warn") + self.assertEqual(screen.keys, []) + + +if __name__ == "__main__": + unittest.main() -- cgit v1.2.3