diff options
Diffstat (limited to 'app/tests')
| -rw-r--r-- | app/tests/test_audio.py | 30 | ||||
| -rw-r--r-- | app/tests/test_backends_audiocpp.py | 2 | ||||
| -rw-r--r-- | app/tests/test_backends_servers.py | 37 | ||||
| -rw-r--r-- | app/tests/test_chunking.py | 170 | ||||
| -rw-r--r-- | app/tests/test_cleaning.py | 61 | ||||
| -rw-r--r-- | app/tests/test_converter.py | 37 | ||||
| -rw-r--r-- | app/tests/test_converter_progress.py | 11 | ||||
| -rw-r--r-- | app/tests/test_extractors.py | 60 | ||||
| -rw-r--r-- | app/tests/test_hub.py | 113 | ||||
| -rw-r--r-- | app/tests/test_instruction_capabilities.py | 419 |
10 files changed, 904 insertions, 36 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 2904e40..5c6f60b 100644 --- a/app/tests/test_chunking.py +++ b/app/tests/test_chunking.py @@ -113,6 +113,176 @@ class SplitIntoChunksTests(unittest.TestCase): self.assertNotIn("1, 000", joined) self.assertNotIn("12: 30", joined) + def test_smart_strategy_distributes_sentences(self): + # The smart strategy leaves whole sentences intact: a chunk + # crossing the target (85% of the limit) ends at the next clean + # boundary early; legacy keeps packing until the limit. + sentences = " ".join( + " ".join(f"Sw{j}" for j in range(n)) + "." + for n in (9, 8, 3, 9, 8, 3)) + smart = split_into_chunks(sentences, max_words=20, smart=True) + legacy = split_into_chunks(sentences, max_words=20, smart=False) + self.assertEqual([len(c.split()) for c in smart], [17, 20, 3]) + self.assertEqual([len(c.split()) for c in legacy], [20, 20]) + self.assertEqual(self._content(smart), + self._content(sentences.split(" "))) + + def test_dialogue_tag_stays_with_its_quote(self): + # A closing quote followed by a lowercase attribution ("she + # said.") is one sentence unit: the chunk never separates them. + text = "“Come here!” she said. “Are you coming?” He did not answer." + chunks = split_into_chunks(text, max_words=10, smart=True) + self.assertIn("“Come here!” she said.", chunks[0]) + self._assert_bounded(chunks, 10) + self._assert_content(text, chunks) + + def test_abbreviations_and_initials_do_not_split(self): + text = ("Dr. Smith met J. K. R. at the U.S. border with No. 3 " + "shortly. He continued onward.") + chunks = split_into_chunks(text, max_words=50, smart=True) + self.assertEqual(len(chunks), 1) + self.assertIn("Dr. Smith", chunks[0]) + self.assertIn("J. K. R.", chunks[0]) + self.assertIn("U.S.", chunks[0]) + self._assert_content(text, chunks) + + def test_decimals_are_never_boundaries(self): + text = "Pi is 3.14159 and the toll was 1,000,000 miles. Next one." + chunks = split_into_chunks(text, max_words=12, smart=True) + joined = " ".join(chunks) + self.assertIn("3.14159", joined) + self.assertIn("1,000,000", joined) + self._assert_bounded(chunks, 12) + self._assert_content(text, chunks) + + def test_unclosed_quote_still_respects_the_limit(self): + # A quote with no closing mark must not grow chunks or silence + # later boundaries; units carry the open state and stay bounded. + text = "“No closing quote follows. " + "filler words here. " * 20 + chunks = split_into_chunks(text, max_words=10, smart=True) + self._assert_bounded(chunks, 10) + self._assert_content(text, chunks) + + def test_paragraph_boundary_resets_an_open_quote(self): + # Blank lines reset the quotation state: an unclosed quote in + # one paragraph cannot trap later paragraphs inside the quote. + text = ("“Still open. filler words inside the quote here.\n\n" + "New paragraph. Words outside any quote now, plenty.") + chunks = split_into_chunks(text, max_words=10, smart=True) + self._assert_bounded(chunks, 10) + self._assert_content(text, chunks) + second = [c for c in chunks if "New paragraph." in c] + self.assertTrue(second) + # The paragraph's text starts a fresh chunk: the open-quote + # prefix did not swallow it. + self.assertTrue( + any(chunk.startswith("New paragraph.") or + "“Still open." not in chunk + for chunk in chunks)) + + def test_quote_retreat_keeps_short_quotes_whole(self): + # When the next sentence would overflow, the break retreats to + # the last boundary that ended outside a quotation — before the + # quote — so a short quotation lands in one request. + text = ("Plain opening words here to fill. " + "“A short quote. With two sentences. Inside.” " + "More trailing prose follows this quote. And more.") + chunks = split_into_chunks(text, max_words=14, smart=True) + self._assert_bounded(chunks, 14) + self._assert_content(text, chunks) + quote_chunks = [c for c in chunks if "short quote" in c] + self.assertTrue(quote_chunks) + self.assertIn("Inside.", quote_chunks[0]) + + def test_oversized_single_tokens_respect_char_cap(self): + # Degenerate text (punctuation-free run): bounded by the hard + # character cap, with every character preserved. + text = "a" * 5000 + chunks = split_into_chunks(text, max_words=250, smart=True) + self.assertGreater(len(chunks), 1) + 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. + text = "。".join(["字" * 15 for _ in range(30)]) + "。" + chunks = split_into_chunks(text, max_words=40, smart=True) + self._assert_bounded(chunks, 40) + self._assert_content(text, chunks) + joined = " ".join(chunks).split() + self.assertTrue(all(len(j) <= 33 for j in joined)) + + def test_legacy_and_smart_matching_smart_negation(self): + # smart=False preserves the legacy splitting exactly. + sentences = " ".join( + " ".join(f"Sw{j}" for j in range(9)) + "." for _ in range(10)) + with patch.object(config, "CHUNK_SIZE", 25): + legacy_default = split_into_chunks(sentences, smart=False) + self.assertEqual([len(c.split()) for c in legacy_default], + [18] * 5) + self.assertEqual(self._content(legacy_default), + self._content(sentences.split())) + + # -- helpers --------------------------------------------------------- + + @staticmethod + def _content(chunks_or_words): + """Non-whitespace content of the pieces (or of a token list).""" + pieces = (chunks_or_words if isinstance(chunks_or_words, str) + else " ".join(chunks_or_words)) + return "".join(pieces.split()) + + def _assert_content(self, text, chunks): + self.assertEqual(self._content(chunks), self._content(text)) + + def _assert_bounded(self, chunks, max_words): + from converter.chunking import _count_words + for chunk in chunks: + self.assertLessEqual(_count_words(chunk), max_words, + chunk[:40]) + if __name__ == "__main__": unittest.main() 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 85ae6ca..3795fbc 100644 --- a/app/tests/test_converter.py +++ b/app/tests/test_converter.py @@ -585,6 +585,24 @@ class ChunkClampPromptTests(unittest.TestCase): with patch.object(config, "CHUNK_SIZE", 80): self.assertFalse(chunk_clamp_needed(self._entry())) + def test_per_run_clamp_caps_the_chapter_chunks(self): + # The popup's clamp caps the chapter chunks themselves, so each + # progress unit and debug dump spans one generation request. + converter = AudiobookConverter.__new__(AudiobookConverter) + converter.chunk_size = 4 + with patch.object(config, "CHUNK_SIZE", 250): + chunks = converter._chapter_chunks(" ".join(f"Sw{i} ." for i in range(12))) + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(c.split()) <= 4 for c in chunks)) + + def test_chapter_chunks_follow_chunk_size_without_a_clamp(self): + converter = AudiobookConverter.__new__(AudiobookConverter) + self.assertIsNone(converter.chunk_size) + with patch.object(config, "CHUNK_SIZE", 8): + chunks = converter._chapter_chunks(" ".join(f"Sw{i} ." for i in range(20))) + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(c.split()) <= 8 for c in chunks)) + def test_prompt_answers(self): entry = self._entry() with patch("builtins.input", return_value=""): @@ -691,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_converter_progress.py b/app/tests/test_converter_progress.py index 0a44725..7fae974 100644 --- a/app/tests/test_converter_progress.py +++ b/app/tests/test_converter_progress.py @@ -53,14 +53,19 @@ class VoiceModeForTests(unittest.TestCase): VOICE_MODE_CUSTOM) def test_qwen_instructions_design(self): - # Qwen: instructions alone select the VoiceDesign model, taking - # precedence over a clone reference. + # Qwen: instructions alone select the VoiceDesign model. With a + # clone reference or a built-in speaker, the reference/speaker + # wins and the instructions only ride along (a directed run on + # models that support it). self.assertEqual(voice_mode_for(BACKEND_QWEN, instructions="A warm narrator"), VOICE_MODE_DESIGN) self.assertEqual(voice_mode_for(BACKEND_QWEN, clone="x.wav", instructions="A warm narrator"), - VOICE_MODE_DESIGN) + VOICE_MODE_CLONE) + self.assertEqual(voice_mode_for(BACKEND_QWEN, voice="Vivian", + instructions="A warm narrator"), + VOICE_MODE_CUSTOM) self.assertEqual(voice_mode_for(BACKEND_QWEN, instructions=" "), VOICE_MODE_CUSTOM) 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%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 84796ee..183e8aa 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -2873,7 +2873,10 @@ class ConvertFlowTests(unittest.TestCase): elif entry.capability == "clone": if entry.requires_reference: overrides["voice"] = str(ref) - else: # design + if entry.supports_instructions \ + or entry.capability == "design": + # Supported entries show the optional + # delivery/style field; design requires it. overrides["instructions"] = "A warm narrator." self._answer_form(**overrides) if hub.converter_mod.chunk_clamp_needed(entry): @@ -2924,10 +2927,15 @@ class ConvertFlowTests(unittest.TestCase): else: # design expected = {"sglomni.model_id", "sglomni.instructions"} - self.assertEqual(kwargs.get("instructions"), - "A warm narrator.") self.assertNotIn("voice", kwargs) self.assertNotIn("clone", kwargs) + if entry.supports_instructions: + # Design requires it; supported entries forward + # it as an optional delivery/style control. + expected = set(expected) | \ + {"sglomni.instructions"} + self.assertEqual(kwargs.get("instructions"), + "A warm narrator.") self.assertEqual(shown, expected) @@ -3521,7 +3529,8 @@ class SettingsTests(unittest.TestCase): # Keys _apply_settings persists; every test that triggers a real or # fake config write restores these afterwards. _SETTING_KEYS = ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", - "CHUNK_SIZE", "INPUT_DIR", "OUTPUT_DIR", + "CHUNK_SIZE", "SMART_CHUNKING", "INPUT_DIR", + "OUTPUT_DIR", "CLONE_WAV_DIR", "SPEED", "DEBUG", "STOP_SERVER_AND_EXIT", "AUDIOCPP_UNLOAD_MODELS", @@ -3556,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. @@ -3596,6 +3606,7 @@ class SettingsTests(unittest.TestCase): original_folders[1]) values = {"audio_format": "ogg", "audio_bitrate": " 192k ", "language": "en", "chunk_size": "300", + "smart_chunking": True, "input_dir": " /books ", "output_dir": "/audiobooks", "clone_wav_dir": " /refs/wavs ", "speed": "1.5", "debug": True, @@ -3618,6 +3629,7 @@ class SettingsTests(unittest.TestCase): "AUDIO_BITRATE": "192k", "LANGUAGE": "English", "CHUNK_SIZE": 300, + "SMART_CHUNKING": True, "INPUT_DIR": "/books", "OUTPUT_DIR": "/audiobooks", "CLONE_WAV_DIR": "/refs/wavs", @@ -3644,6 +3656,7 @@ class SettingsTests(unittest.TestCase): self.assertEqual(hub.config.AUDIO_BITRATE, "192k") self.assertEqual(hub.config.LANGUAGE, "English") self.assertEqual(hub.config.CHUNK_SIZE, 300) + self.assertEqual(hub.config.SMART_CHUNKING, True) self.assertEqual(hub.config.INPUT_DIR, "/books") self.assertEqual(hub.config.OUTPUT_DIR, "/audiobooks") self.assertEqual(hub.config.CLONE_WAV_DIR, "/refs/wavs") @@ -3663,6 +3676,7 @@ class SettingsTests(unittest.TestCase): self._snapshot_settings() base = {"audio_format": "m4b", "audio_bitrate": "128k", "language": "English", "chunk_size": "250", + "smart_chunking": True, "input_dir": "./input", "output_dir": "./output", "clone_wav_dir": "./voices", "speed": "1.0", "debug": False, @@ -3693,9 +3707,46 @@ class SettingsTests(unittest.TestCase): "audiocpp_remote_url": "not a url"}) mk_update.assert_not_called() + def test_smart_chunking_toggle_persists(self): + self._snapshot_settings() + + def fake_update(key, value, config_path=None): + setattr(hub.config, key, value) + return True + + values = {"audio_format": "m4b", "audio_bitrate": "128k", + "language": "English", "chunk_size": "250", + "smart_chunking": False, + "input_dir": "./input", "output_dir": "./output", + "clone_wav_dir": "./voices", + "speed": "1.0", "debug": False, + "stop_and_exit": True, "unload_models": True, + "qwen_port": "7860", + "faster_port": "8000", "audiocpp_port": "8080", + "sglomni_port": "8100"} + with patch.object(hub.common, "update_config_value", fake_update), \ + patch.object(hub, "_sync_audiocpp_server_port"): + hub._apply_settings(values) + self.assertEqual(hub.config.SMART_CHUNKING, False) + + values["smart_chunking"] = True + with patch.object(hub.common, "update_config_value", fake_update), \ + patch.object(hub, "_sync_audiocpp_server_port"): + hub._apply_settings(values) + self.assertEqual(hub.config.SMART_CHUNKING, True) + + def test_smart_chunking_field_defaults_on_with_help(self): + fields = hub._settings_fields() + field = next(f for f in fields if f["key"] == "smart_chunking") + self.assertEqual(field["kind"], "bool") + self.assertEqual(field["value"], hub.config.SMART_CHUNKING) + self.assertTrue(field["help"]) + self.assertTrue(field["label"], "Smart chunking") + def test_field_validators(self): self.assertIsNone(hub._validate_bitrate("128k")) self.assertIsNotNone(hub._validate_bitrate(" ")) + self.assertIsNotNone(hub._validate_bitrate(" ")) self.assertIsNone(hub._validate_language("English")) self.assertIsNone(hub._validate_language("en")) self.assertIsNotNone(hub._validate_language("Klingon")) @@ -3786,6 +3837,7 @@ class SettingsTests(unittest.TestCase): captured["fields"] = fields return {"audio_format": "ogg", "audio_bitrate": "192k", "language": "English", "chunk_size": "300", + "smart_chunking": True, "input_dir": "/books", "output_dir": "/audiobooks", "clone_wav_dir": "/refs/wavs", "speed": "1.0", "debug": False, @@ -3809,7 +3861,8 @@ class SettingsTests(unittest.TestCase): hub._Hub(None).screen_settings() self.assertEqual([f["key"] for f in captured["fields"]], ["audio_format", "audio_bitrate", "language", - "chunk_size", "input_dir", "output_dir", + "chunk_size", "smart_chunking", "input_dir", + "output_dir", "clone_wav_dir", "speed", "debug", "stop_and_exit", "unload_models", @@ -3859,6 +3912,7 @@ class SettingsTests(unittest.TestCase): "audio_bitrate": "192k", "language": "English", "chunk_size": "300", + "smart_chunking": True, "input_dir": "/books", "output_dir": "/audiobooks", "clone_wav_dir": "/refs/wavs", @@ -4012,7 +4066,8 @@ class SettingsTests(unittest.TestCase): original = {name: getattr(hub.config, name) for name in ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", - "CHUNK_SIZE", "INPUT_DIR", "OUTPUT_DIR", + "CHUNK_SIZE", "SMART_CHUNKING", "INPUT_DIR", + "OUTPUT_DIR", "CLONE_WAV_DIR", "SPEED", "DEBUG", "STOP_SERVER_AND_EXIT", "AUDIOCPP_UNLOAD_MODELS", @@ -4034,6 +4089,7 @@ class SettingsTests(unittest.TestCase): 'LANGUAGE = "English"\n' "\n" "CHUNK_SIZE = 250\n" + "SMART_CHUNKING = True\n" 'INPUT_DIR = "./input"\n' 'OUTPUT_DIR = "./output"\n' 'CLONE_WAV_DIR = "./voices"\n' @@ -4062,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. @@ -4092,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): diff --git a/app/tests/test_instruction_capabilities.py b/app/tests/test_instruction_capabilities.py new file mode 100644 index 0000000..440ee08 --- /dev/null +++ b/app/tests/test_instruction_capabilities.py @@ -0,0 +1,419 @@ +"""Instruction-support and guidance regressions across the TTS backends. + +Covers the capabilities the models actually implement (verified against +each backend's serving code) and what the clients send for them: +Breeze-TTS 2's recommended guidance strength with instructions, the +audio.cpp Qwen3-TTS variant split (CustomVoice reads instructions, the +Base cloner does not), the SGLang models that consume a separate style +instruction alongside their voice conditioning, and the Qwen demo's +CustomVoice instruction parameter. +""" + +import io +import json +import tempfile +import unittest +import wave +from pathlib import Path +from unittest.mock import MagicMock, patch + +from converter.clients import ( + BACKEND_QWEN, AudioCppTTSClient, QwenTTSClient, + VOICE_MODE_CLONE, VOICE_MODE_CUSTOM, VOICE_MODE_DESIGN, +) +from converter.clients.audiocpp import ( + AUDIOCPP_FAMILY_BREEZE_TTS, + AUDIOCPP_FAMILY_PROFILES, + AUDIOCPP_VOICE_OPTIONAL, + audiocpp_entry_supports_instructions, + audiocpp_family_voice_policy, +) + + +_WAV_BYTES = b"RIFF\x18\x00\x00\x00WAVEfmt \x10\x00\x00\x00" + + +# --------------------------------------------------------------------------- +# Pure helpers +# --------------------------------------------------------------------------- + +class VoicePolicyKnownFamiliesTests(unittest.TestCase): + """Families whose verified policy must survive a stale local spec.""" + + def test_breeze_is_tts_plus_clone_even_without_a_local_spec(self): + # Remote Breeze entries against older local checkouts carry no + # breeze_tts spec at all: the fallback keeps instructions-only + # voice direction connectable instead of demanding a reference. + self.assertEqual( + audiocpp_family_voice_policy(AUDIOCPP_FAMILY_BREEZE_TTS), + AUDIOCPP_VOICE_OPTIONAL) + + def test_vibevoice_accepts_reference_audio_despite_its_spec(self): + # vibevoice.json declares only "tts", but the implementation + # accepts reference audio: a mixed tts+clone family, so a picked + # voice must not be silently dropped. + self.assertEqual(audiocpp_family_voice_policy("vibevoice"), + AUDIOCPP_VOICE_OPTIONAL) + + +class EntryInstructionSupportTests(unittest.TestCase): + """audiocpp_entry_supports_instructions: True/False/None per entry.""" + + def test_breeze_supports_instructions(self): + self.assertIs( + audiocpp_entry_supports_instructions( + AUDIOCPP_FAMILY_BREEZE_TTS, "tts", "Breeze-TTS-2-GGUF"), + True) + + def test_qwen_customvoice_and_design_support_instructions(self): + self.assertIs( + audiocpp_entry_supports_instructions( + "qwen3_tts", "tts", "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"), + True) + self.assertIs( + audiocpp_entry_supports_instructions( + "qwen3_tts", "vdes", "Qwen3-TTS-12Hz-1.7B-VoiceDesign"), + True) + + def test_qwen_base_cloner_provably_does_not(self): + self.assertIs( + audiocpp_entry_supports_instructions( + "qwen3_tts", "tts", "Qwen3-TTS-12Hz-1.7B-Base-GGUF"), + False) + + def test_unknown_families_are_unknown_not_unsupported(self): + self.assertIsNone( + audiocpp_entry_supports_instructions("chatterbox", + "tts", "x")) + + +# --------------------------------------------------------------------------- +# audio.cpp request payloads (instructed Breeze runs) +# --------------------------------------------------------------------------- + +def _breeze_client(instructions=None, request_options=None, + voice="narrator", seed=-1): + """A fully-initialized Breeze client (no HTTP machinery touched).""" + with patch.object(AudioCppTTSClient, "_connect"): + client = AudioCppTTSClient( + Path("."), voice=voice, instructions=instructions, + request_options=request_options) + client.api_url = "http://127.0.0.1:8080" + client.model_id = "Breeze-TTS-2-GGUF" + client.family = AUDIOCPP_FAMILY_BREEZE_TTS + client.task = "tts" + client.profile = AUDIOCPP_FAMILY_PROFILES[AUDIOCPP_FAMILY_BREEZE_TTS] + client.design_mode = False + client.instruction_voice = False + client.plain_mode = False + client.preset_mode = True + client.speaker_mode = False + client._seed = seed + client._resolve_auto_guidance() + return client + + +def _captured_payload(client): + """The JSON body _request_wav sends, via a stubbed urlopen.""" + response = MagicMock() + response.read.return_value = _WAV_BYTES + response.__enter__ = lambda self: response + response.__exit__ = lambda self, *exc: None + with patch("converter.clients.audiocpp.urllib.request.urlopen") \ + as urlopen: + urlopen.return_value = response + client._request_wav("Hello there.") + request = urlopen.call_args[0][0] + return json.loads(request.data.decode("utf-8")) + + +class BreezeGuidanceDefaultTests(unittest.TestCase): + """Breeze guidance: recommended 4 with instructions, otherwise none.""" + + def test_instructed_clone_carries_guidance_4_and_the_instruction(self): + client = _breeze_client(instructions="Screaming, crazed, yelling") + self.assertEqual(client._auto_guidance_scale, 4.0) + payload = _captured_payload(client) + self.assertEqual(payload["guidance_scale"], 4.0) + self.assertEqual(payload["voice"], "narrator") + self.assertEqual( + payload["options"], + {"instruction": "Screaming, crazed, yelling"}) + self.assertNotIn("instructions", payload) + + def test_option_instruction_also_gets_the_guidance_default(self): + client = _breeze_client(request_options={ + "instruction": "Read slowly and warmly."}) + self.assertEqual(client._auto_guidance_scale, 4.0) + payload = _captured_payload(client) + self.assertEqual(payload["guidance_scale"], 4.0) + self.assertEqual(payload["options"]["instruction"], + "Read slowly and warmly.") + + def test_explicit_guidance_option_is_preserved(self): + client = _breeze_client( + instructions="Screaming, crazed, yelling", + request_options={"guidance_scale": "2.5"}) + self.assertIsNone(client._auto_guidance_scale) + payload = _captured_payload(client) + self.assertNotIn("guidance_scale", payload) + self.assertEqual(payload["options"]["guidance_scale"], "2.5") + + def test_guidance_0_override_still_counts_as_explicit(self): + # 0 selects the instruction-free branch: a deliberate setting. + client = _breeze_client( + instructions="Screaming", + request_options={"guidance_scale": "0"}) + payload = _captured_payload(client) + self.assertNotIn("guidance_scale", payload) + + def test_plain_clone_without_instructions_uses_the_backend_default(self): + client = _breeze_client() + self.assertIsNone(client._auto_guidance_scale) + payload = _captured_payload(client) + self.assertNotIn("guidance_scale", payload) + self.assertNotIn("options", payload) + self.assertEqual(payload["voice"], "narrator") + + +class InstructionConflictTests(unittest.TestCase): + """Two different instruction sources are refused before connecting.""" + + def test_conflicting_instructions_and_option_raise_without_a_server(self): + with self.assertRaises(RuntimeError) as ctx: + _breeze_client(instructions="calm narration", + request_options={"instruction": "screaming"}) + self.assertIn("Two conflicting instructions", + str(ctx.exception)) + + def test_identical_instructions_from_both_sources_are_accepted(self): + client = _breeze_client( + instructions="calm narration", + request_options={"instruction": "calm narration"}) + self.assertEqual(client.instructions, "calm narration") + + def test_option_only_instruction_is_folded_into_the_reports(self): + client = _breeze_client( + request_options={"instruction": "calm narration"}) + self.assertEqual(client.instructions, "calm narration") + + +class SeedPrecisionTests(unittest.TestCase): + """Full-range uint64 seeds travel as decimal strings (audio.cpp docs).""" + + def test_seed_above_2_pow_53_is_sent_as_a_string(self): + seed = 2 ** 53 + 3 # beyond the exact JSON-number integer range + client = _breeze_client(seed=seed) + payload = _captured_payload(client) + self.assertEqual(payload["seed"], str(seed)) + + def test_ordinary_seeds_stay_numbers(self): + client = _breeze_client(seed=42) + payload = _captured_payload(client) + self.assertEqual(payload["seed"], 42) + + +# --------------------------------------------------------------------------- +# SGLang-Omni: instructions on supported pipelines +# --------------------------------------------------------------------------- + +class SgOmniInstructionTests(unittest.TestCase): + """instructions reach the payload only where the serving code reads it.""" + + _tmp_dir = None + _REF = None + + @classmethod + def setUpClass(cls): + buffer = io.BytesIO() + with wave.open(buffer, "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(24000) + wav_file.writeframes(b"\x01\x00" * 16) + cls._tmp_dir = tempfile.TemporaryDirectory() + cls._REF = Path(cls._tmp_dir.name) / "narrator.wav" + cls._REF.write_bytes(buffer.getvalue()) + cls.addClassCleanup(cls._tmp_dir.cleanup) + + def _client(self, model, **kwargs): + from converter.clients import SgOmniTTSClient + with patch.object(SgOmniTTSClient, "_connect"): + client = SgOmniTTSClient( + Path("."), model=model, ref_audio=str(self._REF), + ref_text="Hello transcript.", instructions="screaming", + **kwargs) + entry = client.entry + payload = client._request_payload("Hello there.") + return entry, payload + + def test_qwen_base_clone_carries_ref_and_instruction(self): + entry, payload = self._client("qwen3_tts_1_7b_base") + self.assertIn("ref_audio", payload) + self.assertNotEqual(payload.get("task_type"), "VoiceDesign") + self.assertEqual(payload["instructions"], "screaming") + + def test_moss_clone_carries_ref_and_instruction(self): + entry, payload = self._client("moss_tts") + self.assertIn("ref_audio", payload) + self.assertEqual(payload["instructions"], "screaming") + + def test_customvoice_speaker_with_instruction(self): + entry, payload = self._client("qwen3_tts_0_6b_customvoice", + voice="Vivian") + self.assertEqual(payload["voice"], "Vivian") + self.assertEqual(payload["instructions"], "screaming") + self.assertNotIn("task_type", payload) + + def test_design_remains_voice_design_with_instruction(self): + from converter.clients import SgOmniTTSClient + with patch.object(SgOmniTTSClient, "_connect"): + client = SgOmniTTSClient( + Path("."), model="qwen3_tts_1_7b_voicedesign", + instructions="a warm narrator") + payload = client._request_payload("Hello there.") + self.assertEqual(payload["task_type"], "VoiceDesign") + self.assertEqual(payload["instructions"], "a warm narrator") + self.assertNotIn("ref_audio", payload) + + def test_unsupported_model_refuses_instructions_at_connect(self): + with self.assertRaises(RuntimeError) as ctx: + self._client("higgs_audio_v3_tts") + self.assertIn("does not consume style instructions", + str(ctx.exception)) + + +# --------------------------------------------------------------------------- +# Qwen demo: CustomVoice instruction parameter +# --------------------------------------------------------------------------- + +class QwenCustomVoiceInstructionTests(unittest.TestCase): + """The run_instruct endpoint takes an ``instruct`` delivery control.""" + + def test_run_instruct_sends_instruct_alongside_the_speaker(self): + client = QwenTTSClient.__new__(QwenTTSClient) + client.voice_mode = VOICE_MODE_CUSTOM + client.speaker = "Vivian" + client.language = "Auto" + client.instructions = "screaming, crazed" + client._seed = -1 + client.client = MagicMock() + client._resolve_api_name = lambda *names: names[0] + client._endpoint_accepts_param = MagicMock(return_value=True) + client._generate_custom_voice("Hello there.") + predict = client.client.predict + predict.assert_called_once_with( + text="Hello there.", lang_disp="Auto", + spk_disp="Vivian", instruct="screaming, crazed", + api_name="/run_instruct") + + def test_custom_voice_without_instructions_is_unchanged(self): + client = QwenTTSClient.__new__(QwenTTSClient) + client.voice_mode = VOICE_MODE_CUSTOM + client.speaker = "Vivian" + client.language = "Auto" + client.instructions = "" + client._seed = -1 + client.client = MagicMock() + client._resolve_api_name = lambda *names: names[0] + client._endpoint_accepts_param = MagicMock(return_value=True) + client._generate_custom_voice("Hello there.") + _, kwargs = client.client.predict.call_args + self.assertNotIn("instruct", kwargs) + + +# --------------------------------------------------------------------------- +# audiobook voice-mode routing (qwen) +# --------------------------------------------------------------------------- + +class CatalogInstructionFlagsTests(unittest.TestCase): + """Only the verified pipelines carry supports_instructions.""" + + def test_catalog_marks_only_the_verified_pipelines(self): + from backends.sglomni.catalog import ENTRIES + supported = {"qwen3_tts_0_6b_customvoice", "qwen3_tts_0_6b_base", + "qwen3_tts_1_7b_base", "qwen3_tts_1_7b_voicedesign", + "moss_tts", "moss_tts_local"} + for entry in ENTRIES: + with self.subTest(entry=entry.key): + self.assertEqual(entry.supports_instructions, + entry.key in supported) + + +class GradioPrefixProbeTests(unittest.TestCase): + """Qwen demos under modern Gradio sit behind /gradio_api.""" + + def _identify(self, modern_payload, legacy_payload=None): + import backends.probe as probe + seen = [] + + def fake_get_json(url, timeout): + seen.append(url) + if url == "http://x/gradio_api/info": + return modern_payload + if url == "http://x/info": + return legacy_payload + return None + + with patch.object(probe, "_get_json", side_effect=fake_get_json): + with patch.object(probe.common, "server_running", + return_value=True): + identity = probe._identify_gradio("http://x", 1.0) + return identity, seen + + def test_modern_prefix_is_probed_first_and_identifies(self): + payload = {"named_endpoints": {"/run_instruct": {}}} + identity, seen = self._identify(payload) + self.assertEqual(identity, probe_identity("qwen-custom")) + self.assertEqual(seen, ["http://x/gradio_api/info"]) + + def test_legacy_info_still_identifies_older_gradio(self): + payload = {"named_endpoints": {"/run_voice_clone": {}}} + identity, seen = self._identify(None, payload) + self.assertEqual(identity, probe_identity("qwen-clone")) + self.assertEqual(seen, ["http://x/gradio_api/info", + "http://x/info"]) + + def test_neither_prefix_answers_none(self): + identity, _ = self._identify(None) + self.assertIsNone(identity) + + +def probe_identity(name): + """The probe's IDENTITY_* constant for a backend NAME (local import).""" + import backends.probe as probe + return {"qwen-custom": probe.IDENTITY_QWEN_CUSTOM, + "qwen-clone": probe.IDENTITY_QWEN_CLONE, + }[name] + + +# --------------------------------------------------------------------------- +# audiobook voice-mode routing (qwen) +# --------------------------------------------------------------------------- + +class QwenVoiceModeRoutingTests(unittest.TestCase): + """speaker + instructions is a directed CustomVoice run, not Design.""" + + def test_voice_mode_for_qwen_combinations(self): + from converter.converter import voice_mode_for + cases = [ + (dict(voice=None, clone=None, instructions=None), + VOICE_MODE_CUSTOM), + (dict(voice=None, clone=None, instructions="screaming"), + VOICE_MODE_DESIGN), + (dict(voice="Vivian", clone=None, instructions="screaming"), + VOICE_MODE_CUSTOM), + (dict(voice=None, clone="ref.wav", instructions=None), + VOICE_MODE_CLONE), + ] + for kwargs, expected in cases: + with self.subTest(**kwargs): + self.assertEqual( + voice_mode_for(BACKEND_QWEN, voice=kwargs["voice"], + clone=kwargs["clone"], + instructions=kwargs["instructions"]), + expected) + + +if __name__ == "__main__": + unittest.main() |