From d72d274bd9895bdf97d31d56690b7df07fa74ad4 Mon Sep 17 00:00:00 2001 From: historia Date: Tue, 18 Aug 2026 01:56:25 -0400 Subject: fix: timing/punctuation edge cases, process chunks as uncompressed wavs --- tests/test_audio.py | 132 ++++++++++++++++++++++++++++++++++++++++++++++++- tests/test_chunking.py | 25 +++++++++- 2 files changed, 153 insertions(+), 4 deletions(-) (limited to 'tests') diff --git a/tests/test_audio.py b/tests/test_audio.py index 29bf4ea..f36ed54 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -1,11 +1,21 @@ -"""Tests for audio helpers: speed parameters, chunk cleanup, and encoding.""" +"""Tests for audio helpers: speed parameters, chunk cleanup, encoding, +command construction, and duration verification.""" import tempfile import unittest from pathlib import Path +from converter import audio from converter import config -from converter.audio import _encode_args, build_ffmetadata, cleanup_chunks, speed_export_params +from converter.audio import ( + _encode_args, + build_concat_command, + build_ffmetadata, + build_m4b_chapters_command, + cleanup_chunks, + speed_export_params, + verify_output_duration, +) class SpeedExportParamsTests(unittest.TestCase): @@ -76,6 +86,124 @@ class EncodeArgsTests(unittest.TestCase): 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"]) + + +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) + + +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): diff --git a/tests/test_chunking.py b/tests/test_chunking.py index da45f65..659a771 100644 --- a/tests/test_chunking.py +++ b/tests/test_chunking.py @@ -22,13 +22,34 @@ class SplitIntoChunksTests(unittest.TestCase): 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 commas - sentence = ",".join([" ".join(["w"] * 5) for _ in range(10)]) + "." + # 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 split point and + # must stay byte-identical rather than being re-joined with spaces. + sentence = ",".join([" ".join(["w"] * 5) for _ in range(10)]) + "." + chunks = split_into_chunks(sentence, max_words=12) + self.assertEqual(chunks, [sentence]) + def test_single_oversized_sentence_stays_intact(self): sentence = " ".join(["word"] * 30) + "." chunks = split_into_chunks(sentence, max_words=10) -- cgit v1.2.3