aboutsummaryrefslogtreecommitdiff
path: root/app/tests
diff options
context:
space:
mode:
Diffstat (limited to 'app/tests')
-rw-r--r--app/tests/test_audio.py30
-rw-r--r--app/tests/test_backends_audiocpp.py2
-rw-r--r--app/tests/test_backends_servers.py37
-rw-r--r--app/tests/test_chunking.py41
-rw-r--r--app/tests/test_cleaning.py61
-rw-r--r--app/tests/test_converter.py19
-rw-r--r--app/tests/test_extractors.py60
-rw-r--r--app/tests/test_hub.py46
8 files changed, 269 insertions, 27 deletions
diff --git a/app/tests/test_audio.py b/app/tests/test_audio.py
index 75de97c..04a6afc 100644
--- a/app/tests/test_audio.py
+++ b/app/tests/test_audio.py
@@ -51,6 +51,36 @@ class AtempoFiltersTests(unittest.TestCase):
with self.assertRaises(ValueError):
atempo_filters(-1.5)
+ def test_infinite_speed_rejected(self):
+ # inf passes a bare positivity check; without the finite guard
+ # atempo chaining would loop forever (inf / 2.0 stays inf).
+ with self.assertRaises(ValueError):
+ atempo_filters(float("inf"))
+
+ def test_overflowing_speed_rejected(self):
+ with self.assertRaises(ValueError):
+ atempo_filters(1e309)
+
+ def test_nan_speed_rejected(self):
+ with self.assertRaises(ValueError):
+ atempo_filters(float("nan"))
+
+ def test_asplit_used_for_audio_streams(self):
+ # split is a video filter and rejects an audio stream, so a
+ # speed-adjusted copy needs asplit — in both command builders.
+ concat_cmd = build_concat_command(
+ Path("/tmp/_concat.txt"), Path("/tmp/out.mp3"), "mp3",
+ speed=1.5, speed_path=Path("/tmp/out_1.5.mp3"))
+ m4b_cmd = build_m4b_chapters_command(
+ Path("/tmp/_concat.txt"), Path("/tmp/_meta.txt"),
+ Path("/tmp/out.m4b"), speed=1.5,
+ speed_path=Path("/tmp/out_1.5.m4b"),
+ speed_metadata_file=Path("/tmp/_meta2.txt"))
+ for cmd in (concat_cmd, m4b_cmd):
+ filter_complex = cmd[cmd.index("-filter_complex") + 1]
+ self.assertIn("[0:a]asplit=2", filter_complex)
+ self.assertNotIn("[0:a]split=2", filter_complex)
+
class CleanupChunksTests(unittest.TestCase):
def test_removes_only_chunk_files(self):
diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py
index 9392ba1..1800267 100644
--- a/app/tests/test_backends_audiocpp.py
+++ b/app/tests/test_backends_audiocpp.py
@@ -252,7 +252,7 @@ class UpdateConfigPortTests(unittest.TestCase):
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',
+ "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)
diff --git a/app/tests/test_backends_servers.py b/app/tests/test_backends_servers.py
index 9fd71f2..283df95 100644
--- a/app/tests/test_backends_servers.py
+++ b/app/tests/test_backends_servers.py
@@ -370,19 +370,44 @@ class ReapTests(unittest.TestCase):
class KillPidTests(unittest.TestCase):
- """_kill_pid: the reap check ends the grace wait before SIGKILL."""
+ """_kill_pid: the group probe decides the wait's end, not the reap.
- def test_reaped_child_ends_wait_without_sigkill(self):
+ Reaping the launcher proves nothing about the rest of its process
+ group: workers can outlive it and must still get the SIGKILL
+ escalation.
+ """
+
+ def test_empty_group_ends_wait_without_sigkill(self):
+ # Single-process server: the launcher is reaped and the group is
+ # empty (killpg(0) raises) — success immediately, no SIGKILL.
with patch("os.getpgid", return_value=4242), \
- patch("os.killpg") as mk_killpg, \
- patch("os.waitpid", return_value=(4242, 0)) as mk_waitpid, \
+ patch("os.killpg",
+ side_effect=[None, ProcessLookupError]) as mk_killpg, \
+ patch("os.waitpid", return_value=(4242, 0)), \
patch("time.sleep") as mk_sleep:
ok = servers._kill_pid(4242)
self.assertTrue(ok)
- mk_killpg.assert_called_once_with(4242, signal.SIGTERM)
- mk_waitpid.assert_called_once_with(4242, os.WNOHANG)
+ self.assertEqual(mk_killpg.call_args_list[0].args,
+ (4242, signal.SIGTERM))
+ self.assertEqual(mk_killpg.call_args_list[-1].args, (4242, 0))
mk_sleep.assert_not_called()
+ def test_reaped_leader_with_live_workers_escalates_to_sigkill(self):
+ # A launcher that dies on SIGTERM while its workers ignore it:
+ # reaping the leader must not end the stop, the surviving group
+ # keeps burning the grace period and then gets SIGKILLed.
+ with patch("os.getpgid", return_value=4242), \
+ patch("os.killpg", return_value=None) as mk_killpg, \
+ patch("os.waitpid", return_value=(4242, 0)), \
+ patch("time.sleep"):
+ ok = servers._kill_pid(4242)
+ self.assertTrue(ok)
+ calls = mk_killpg.call_args_list
+ self.assertEqual(calls[0].args, (4242, signal.SIGTERM))
+ self.assertEqual(calls[-1].args, (4242, signal.SIGKILL))
+ # The group was probed repeatedly while waiting for the workers.
+ self.assertGreater(len(calls), 2)
+
def test_escalates_to_sigkill_when_child_stays_alive(self):
with patch("os.getpgid", return_value=4242), \
patch("os.killpg") as mk_killpg, \
diff --git a/app/tests/test_chunking.py b/app/tests/test_chunking.py
index bcae6b2..5c6f60b 100644
--- a/app/tests/test_chunking.py
+++ b/app/tests/test_chunking.py
@@ -203,6 +203,47 @@ class SplitIntoChunksTests(unittest.TestCase):
self.assertTrue(all(len(c) <= 250 * 8 for c in chunks))
self.assertEqual("".join(chunks), text)
+ def test_non_spaced_script_respects_the_word_limit(self):
+ # A punctuation-free CJK token longer than the limit counts one
+ # word per character: it must be sliced by the word limit too,
+ # not only by the character cap (which alone would allow chunks
+ # up to CHUNK_SIZE * 8 characters).
+ from converter.chunking import _count_words
+
+ text = "字" * 500
+ chunks = split_into_chunks(text, max_words=250, smart=True)
+ self.assertGreater(len(chunks), 1)
+ self.assertTrue(all(_count_words(c) <= 250 for c in chunks),
+ [(_count_words(c), c[:20]) for c in chunks])
+ self.assertEqual("".join(chunks), text)
+
+ text = "字" * 5000
+ chunks = split_into_chunks(text, max_words=250, smart=True)
+ self.assertTrue(all(_count_words(c) <= 250 for c in chunks),
+ [(_count_words(c), c[:20]) for c in chunks])
+ self.assertEqual("".join(chunks), text)
+
+ def test_oversized_sentence_keeps_quote_state_through_fallback(self):
+ # An oversized opening sentence of a dialogue must not present
+ # its mid-quote remainder as a clean boundary: the fallback
+ # carries the sentence's quote state, so the close of the quote
+ # stays attached to the words before it in the next chunk.
+ text = "“" + "go " * 17 + "go. End now.”"
+ chunks = split_into_chunks(text, max_words=10, smart=True)
+ self._assert_bounded(chunks, 10)
+ self._assert_content(text, chunks)
+ self.assertEqual([len(c.split()) for c in chunks], [10, 10])
+
+ def test_oversized_sentence_breaks_cleanly_when_quote_closes_mid_sentence(self):
+ # A quote opened and closed inside one long sentence: the
+ # clause-split pieces after the closing quote are clean
+ # boundaries and can be packed together.
+ text = ("He answered " + "very " * 8 + "“calmly” and then kept "
+ "talking on and on for a while.")
+ chunks = split_into_chunks(text, max_words=6, smart=True)
+ self._assert_bounded(chunks, 6)
+ self._assert_content(text, chunks)
+
def test_non_spaced_script_stays_bounded(self):
# CJK text: sentence terminators break units without
# whitespace, and each ideograph counts as a word.
diff --git a/app/tests/test_cleaning.py b/app/tests/test_cleaning.py
index 41f4ed7..40f2d53 100644
--- a/app/tests/test_cleaning.py
+++ b/app/tests/test_cleaning.py
@@ -10,8 +10,24 @@ class CleanTextTests(unittest.TestCase):
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_collapses_whitespace_inside_paragraphs(self):
+ self.assertEqual(clean_text("a b \t c"), "a b c")
+
+ def test_paragraph_breaks_survive(self):
+ # Blank lines are the smart chunker's paragraph boundaries (and
+ # reset its quotation state), so they survive cleaning while a
+ # run of blank lines collapses to a single break.
+ self.assertEqual(clean_text("a\n\n b \t c"), "a\n\nb c")
+ self.assertEqual(clean_text("a\n\n\n\nb"), "a\n\nb")
+ self.assertEqual(clean_text("a\n \nb"), "a\n\nb")
+ self.assertEqual(clean_text("line one\nline two"), "line one line two")
+ self.assertEqual(clean_text("a \n\n b"), "a\n\nb")
+
+ def test_paragraph_breaks_not_glued_to_text(self):
+ self.assertEqual(
+ clean_text("End of chapter.\n\n\n New chapter. \n\nStarts here."),
+ "End of chapter.\n\nNew chapter.\n\nStarts here.",
+ )
def test_preserves_inline_numbers(self):
self.assertEqual(clean_text("He was 42 years old."), "He was 42 years old.")
@@ -23,14 +39,16 @@ class CleanTextTests(unittest.TestCase):
)
def test_removes_standalone_page_numbers(self):
+ # The page number's own line becomes a paragraph break (a safe
+ # chunk boundary), not a glued sentence.
self.assertEqual(
clean_text("End of page.\n7\nNext page text."),
- "End of page. Next page text.",
+ "End of page.\n\nNext page text.",
)
- def test_page_number_removal_leaves_single_spacing(self):
+ def test_page_number_removal_leaves_paragraph_break(self):
result = clean_text("Chapter one\n\n12\n\nChapter two")
- self.assertEqual(result, "Chapter one Chapter two")
+ self.assertEqual(result, "Chapter one\n\nChapter two")
self.assertNotIn(" ", result)
@@ -49,6 +67,39 @@ class CleanHtmlTests(unittest.TestCase):
self.assertEqual(clean_html(""), "")
self.assertEqual(clean_html(None), "")
+ def test_inline_markup_never_splits_words(self):
+ # Inline tags must not inject spaces mid-word (they become chunk
+ # boundaries and corrupt pronunciation).
+ self.assertEqual(
+ clean_html("<p>He was un<em>believ</em>able and didn<i>'</i>t stop.</p>"),
+ "He was unbelievable and didn't stop.",
+ )
+
+ def test_block_tags_become_paragraph_breaks(self):
+ self.assertEqual(
+ clean_html("<p>First para.</p><p>Second para.</p><h2>Head</h2>"),
+ "First para.\n\nSecond para.\n\nHead",
+ )
+
+ def test_div_sections_keep_paragraph_boundaries(self):
+ html = ('<div>“An unfinished quotation.</div>'
+ '<div>A new paragraph outside the quotation.</div>')
+ self.assertEqual(
+ clean_html(html),
+ "“An unfinished quotation.\n\nA new paragraph outside the quotation.",
+ )
+
+ def test_table_cells_do_not_glue(self):
+ self.assertEqual(
+ clean_html("<table><tr><td>A</td><td>B</td></tr>"
+ "<tr><td>C</td><td>D</td></tr></table>"),
+ "A\n\nB\n\nC\n\nD",
+ )
+
+ def test_regex_fallback_matches_bs4_behavior(self):
+ html = "<p>un<em>believ</em>able</p><div>After a block.</div>"
+ self.assertEqual(clean_html(html), "unbelievable\n\nAfter a block.")
+
if __name__ == "__main__":
unittest.main()
diff --git a/app/tests/test_converter.py b/app/tests/test_converter.py
index 9084898..3795fbc 100644
--- a/app/tests/test_converter.py
+++ b/app/tests/test_converter.py
@@ -709,6 +709,25 @@ class PreflightOverwritesTests(unittest.TestCase):
names = {name for _book, name in planned}
self.assertEqual(names, {"book_txt_m1_Vivian", "book_epub_m1_Vivian"})
+ def test_planned_names_are_unique_across_the_batch(self):
+ # "book.txt" and "book_txt.txt" both resolve to the stem
+ # "book_txt" (the suffix disambiguation's target), which without
+ # a uniqueness pass would make the later book silently overwrite
+ # the earlier one's audiobook.
+ (converter_mod.BOOKS_FOLDER / "book.epub").write_text("x",
+ encoding="utf-8")
+ (converter_mod.BOOKS_FOLDER / "book.txt").write_text("x",
+ encoding="utf-8")
+ (converter_mod.BOOKS_FOLDER / "book_txt.txt").write_text(
+ "x", encoding="utf-8")
+ with patch("builtins.input", side_effect=AssertionError("should not prompt")):
+ _book_files, planned = AudiobookConverter.preflight_overwrites(
+ BACKEND_QWEN, "Vivian", VOICE_MODE_CUSTOM, None, "mp3")
+ names = [name for _book, name in planned]
+ self.assertEqual(len(names), len(set(names)))
+ self.assertEqual(names, ["book_epub_Vivian", "book_txt_Vivian",
+ "book_txt_Vivian_2"])
+
class ComputeModelTagTests(unittest.TestCase):
"""compute_model_tag: the sanitized model id used in output names."""
diff --git a/app/tests/test_extractors.py b/app/tests/test_extractors.py
index 619fe5a..b7f14d9 100644
--- a/app/tests/test_extractors.py
+++ b/app/tests/test_extractors.py
@@ -133,6 +133,51 @@ class EpubZipfileFallbackTests(unittest.TestCase):
self.assertEqual([title for title, _ in items], ["a", "b"])
+ def test_spine_hrefs_are_normalized_before_matching(self):
+ # Hrefs are URL-encoded, entity-escaped and relative (possibly
+ # with ./ or ../ segments): every equivalent spelling must
+ # resolve to its archived chapter, or the chapter silently
+ # disappears from the book.
+ import zipfile
+
+ from converter.extractors import _read_epub_zipfile
+
+ container = ("<?xml version=\"1.0\"?>"
+ "<container><rootfiles>"
+ "<rootfile full-path=\"OEBPS/content.opf\"/>"
+ "</rootfiles></container>")
+ opf = ("<?xml version=\"1.0\"?>"
+ "<package xmlns=\"http://www.idpf.org/2007/opf\">"
+ "<manifest>"
+ "<item id=\"nav\" href=\"nav.xhtml\" properties=\"nav\"/>"
+ "<item id=\"c1\" href=\"./text/chapterA.xhtml\"/>"
+ "<item id=\"c2\" href=\"../OEBPS/text/chapterB.xhtml\"/>"
+ "<item id=\"c3\" href=\"text/chapter&#37;20C.xhtml\"/>"
+ "</manifest>"
+ "<spine><itemref idref=\"nav\"/>"
+ "<itemref idref=\"c1\"/><itemref idref=\"c2\"/>"
+ "<itemref idref=\"c3\"/></spine>"
+ "</package>")
+ with tempfile.TemporaryDirectory() as tmp:
+ path = Path(tmp) / "book.epub"
+ with zipfile.ZipFile(path, "w") as zf:
+ zf.writestr("mimetype", "application/epub+zip")
+ zf.writestr("META-INF/container.xml", container)
+ zf.writestr("OEBPS/content.opf", opf)
+ zf.writestr("OEBPS/nav.xhtml",
+ "<html><body><p>Contents</p></body></html>")
+ zf.writestr("OEBPS/text/chapterA.xhtml",
+ "<html><body><p>Alpha text.</p></body></html>")
+ zf.writestr("OEBPS/text/chapterB.xhtml",
+ "<html><body><p>Beta text.</p></body></html>")
+ zf.writestr("OEBPS/text/chapter%20C.xhtml",
+ "<html><body><p>Gamma text.</p></body></html>")
+ items = _read_epub_zipfile(path)
+
+ titles = [title for title, _ in items]
+ self.assertNotIn("nav", titles)
+ self.assertEqual(titles, ["chapterA", "chapterB", "chapter%20C"])
+
def _build_test_epub(path: Path, chapters=(("One", "First chapter text."),
("Two", "Second chapter text."))) -> None:
@@ -183,15 +228,20 @@ class EpubExtractionTests(unittest.TestCase):
@requires_epub
def test_epub_extraction_follows_spine_order(self):
+ # extract_text() only handles TXT and PDF; EPUB books (with their
+ # per-chapter structure) come through extract_sections().
+ from converter.extractors import extract_sections
+
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "book.epub"
_build_test_epub(path)
- text = extract_text(path)
+ sections = extract_sections(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."))
+ texts = [section.text for section in sections]
+ self.assertIn("First chapter text.", " ".join(texts))
+ self.assertIn("Second chapter text.", " ".join(texts))
+ self.assertLess(" ".join(texts).index("First chapter text."),
+ " ".join(texts).index("Second chapter text."))
class ExtractSectionsTests(unittest.TestCase):
diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py
index cbcdc8a..183e8aa 100644
--- a/app/tests/test_hub.py
+++ b/app/tests/test_hub.py
@@ -3565,12 +3565,13 @@ class SettingsTests(unittest.TestCase):
self.assertTrue(hub.common.update_config_value(
key, value, config_path=path))
text = path.read_text(encoding="utf-8")
+ # Strings render as proper Python literals (repr).
self.assertEqual(
text,
"# Default output options\n"
- 'AUDIO_FORMAT = "mp3"\n'
- 'AUDIO_BITRATE = "192k"\n'
- 'LANGUAGE = "Japanese"\n'
+ "AUDIO_FORMAT = 'mp3'\n"
+ "AUDIO_BITRATE = '192k'\n"
+ "LANGUAGE = 'Japanese'\n"
"\n"
"CHUNK_SIZE = 300 # words per request\n")
# The imported module mirrors the file immediately.
@@ -4117,10 +4118,11 @@ class SettingsTests(unittest.TestCase):
text = path.read_text(encoding="utf-8")
self.assertIn('AUDIO_FORMAT = "m4b"', text)
self.assertIn("CHUNK_SIZE = 300", text)
- # The settings-only fields are written back unchanged.
- self.assertIn('INPUT_DIR = "', text)
- self.assertIn('OUTPUT_DIR = "', text)
- self.assertIn('CLONE_WAV_DIR = "', text)
+ # The settings-only fields are written back (as repr string
+ # literals for the values the menu saved).
+ self.assertIn("INPUT_DIR = '/workspace/input'", text)
+ self.assertIn("OUTPUT_DIR = '/workspace/output'", text)
+ self.assertIn("CLONE_WAV_DIR = '/workspace/voices'", text)
self.assertIn("SPEED = 1.0", text)
self.assertIn("DEBUG = False", text)
# The running session also picked up the change in-memory.
@@ -4147,9 +4149,33 @@ class SettingsTests(unittest.TestCase):
("AUDIOCPP_API_URL", "http://127.0.0.1:8081")):
hub.common.update_config_value(key, value, config_path=path)
text = path.read_text(encoding="utf-8")
- self.assertIn('QWEN_API_URL = "http://127.0.0.1:7862"', text)
- self.assertIn('FASTER_API_URL = "http://127.0.0.1:8001"', text)
- self.assertIn('AUDIOCPP_API_URL = "http://127.0.0.1:8081"', text)
+ self.assertIn("QWEN_API_URL = 'http://127.0.0.1:7862'", text)
+ self.assertIn("FASTER_API_URL = 'http://127.0.0.1:8001'", text)
+ self.assertIn("AUDIOCPP_API_URL = 'http://127.0.0.1:8081'", text)
+
+ def test_update_config_value_escapes_quotes_and_backslashes(self):
+ # Strings containing quotes or backslashes must stay valid,
+ # unchanging Python: bare double-quote quoting would corrupt
+ # config.py (invalidating every later start) or silently alter
+ # the value once backslashes became escapes.
+ import tempfile
+ self._snapshot_settings()
+ with tempfile.TemporaryDirectory() as td:
+ path = Path(td) / "config.py"
+ path.write_text('INPUT_DIR = "input"\n', encoding="utf-8")
+ self.assertTrue(hub.common.update_config_value(
+ "INPUT_DIR", '/books/A "quoted" title\\', config_path=path))
+ text = path.read_text(encoding="utf-8")
+ compiled = compile(text, str(path), "exec")
+ scope = {}
+ exec(compiled, scope)
+ # The matcher must still find the (now weirdly quoted) value
+ # to update it again.
+ self.assertTrue(hub.common.update_config_value(
+ "INPUT_DIR", "plain", config_path=path))
+ self.assertIn("INPUT_DIR = 'plain'",
+ path.read_text(encoding="utf-8"))
+ self.assertEqual(scope["INPUT_DIR"], '/books/A "quoted" title\\')
class AudiocppServerConfigTests(unittest.TestCase):
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-10 13:06:55 +0000