diff options
Diffstat (limited to 'app/tests/test_hub.py')
| -rw-r--r-- | app/tests/test_hub.py | 288 |
1 files changed, 284 insertions, 4 deletions
diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index ef246ad..b1a91ed 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -10,7 +10,7 @@ import json import tempfile import unittest from pathlib import Path -from unittest.mock import patch +from unittest.mock import MagicMock, patch from backends import BackendInfo, BackendStatus, ServerSpec from converter.clients import audiocpp as audiocpp_client @@ -919,6 +919,18 @@ class ConvertFlowTests(unittest.TestCase): remote_urls=remote_urls, remote_models=list(remote_models or [])) + def _mock_preflight(self, book="book.txt"): + """Replace the real books-folder scan with a canned plan. + + Returns the mock so tests can assert how many per-model plans the + "All" flow computed (and with which name_tag/voice).""" + mk = MagicMock(return_value=([book], [(book, "planned")])) + patcher = patch.object(hub.AudiobookConverter, "preflight_overwrites", + mk) + patcher.start() + self.addCleanup(patcher.stop) + return mk + def test_audiocpp_remote_builds_one_form(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], @@ -996,8 +1008,12 @@ class ConvertFlowTests(unittest.TestCase): self.assertEqual(choices[1], ("a-much-longer-model-id".ljust(22) + " clone", "a-much-longer-model-id")) - self.assertEqual({label.index("clone") for label, _ in choices}, - {29}) + # Two configured models: the "All (multiple generation)" pick + # closes the menu, plain-text without capability columns. + self.assertEqual(choices[2], + ("All (multiple generation)", hub.AUDIOCPP_MODEL_ALL)) + self.assertEqual({label.index("clone") for label, _ in choices + if "clone" in label}, {29}) self.assertEqual({label.index("tts") for label, _ in choices if "tts" in label}, {24}) # A plain qwen3_tts entry (no CustomVoice in the id) is clone-only. @@ -1056,7 +1072,8 @@ class ConvertFlowTests(unittest.TestCase): choices = self._field("model_id")["choices"] base = "Qwen3-TTS-12Hz-1.7B-Base-GGUF" design = "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF" - labels = {value: label for label, value in choices} + labels = {value: label for label, value in choices + if value != hub.AUDIOCPP_MODEL_ALL} self.assertEqual(labels, {base: base.ljust(len(design)) + " clone", design: design.ljust(len(design)) @@ -1141,6 +1158,204 @@ class ConvertFlowTests(unittest.TestCase): model_field["on_change"](fields) self.assertEqual(voice_field["value"], "second") + # ------------------------------------------------------------------ + # "All (multiple generation)" model pick + # ------------------------------------------------------------------ + + def test_model_menu_all_option_only_with_multiple_models(self): + # "All (multiple generation)" closes the model menu only when more + # than one model is configured: a single-model server has nothing + # to compare. + self._patch_remote([{"id": "solo", "family": "higgs_audio_tts", + "task": "tts"}], voices=["narrator"]) + self._answer_form(backend="audiocpp-remote", model_id="solo", + audiocpp_voice="narrator", instructions="") + self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + self.assertEqual(self._field("model_id")["choices"], + [("solo tts clone", "solo")]) + self._patch_remote( + [{"id": "alpha", "family": "higgs_audio_tts", "task": "tts"}, + {"id": "beta", "family": "higgs_audio_tts", "task": "tts"}], + voices=["narrator"]) + self._answer_form(backend="audiocpp-remote", model_id="alpha", + audiocpp_voice="narrator", instructions="") + self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + choices = self._field("model_id")["choices"] + self.assertEqual(choices[-1], + ("All (multiple generation)", + hub.AUDIOCPP_MODEL_ALL)) + self.assertEqual(choices[0], ("alpha tts clone", "alpha")) + + def test_all_pick_maps_one_run_per_model_with_the_picked_clone_voice(self): + # The "All" pick produces no single model/voice: the run receives + # the configured model list and each model's voice, with the + # picked server-side clone voice shared by every clone-capable + # model — and each model's overwrite plan computed with its + # model-tagged name. + self._patch_remote( + [{"id": "alpha", "family": "higgs_audio_tts", "task": "tts"}, + {"id": "beta", "family": "chatterbox", "task": "tts"}], + voices=["narrator"]) + mk_pre = self._mock_preflight() + self._answer_form(backend="audiocpp-remote", + model_id=hub.AUDIOCPP_MODEL_ALL, + audiocpp_voice="narrator", instructions="") + cmd = self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + kwargs = cmd[2] + self.assertEqual(kwargs["model_ids"], ["alpha", "beta"]) + self.assertEqual(kwargs["model_voices"], + {"alpha": "narrator", "beta": "narrator"}) + self.assertNotIn("model_id", kwargs) + self.assertNotIn("voice", kwargs) + self.assertEqual(kwargs["api_url"], "http://audiocpp.local:8080") + self.assertEqual(mk_pre.call_count, 2) + self.assertEqual(mk_pre.call_args_list[0].kwargs["name_tag"], + "alpha") + self.assertEqual(mk_pre.call_args_list[1].kwargs["name_tag"], "beta") + self.assertEqual(kwargs["book_files"], ["book.txt"]) + self.assertEqual(kwargs["planned_by_model"], + {"alpha": [("book.txt", "planned")], + "beta": [("book.txt", "planned")]}) + self.assertNotIn("planned", kwargs) + + def test_all_voice_falls_back_per_capability(self): + # A CustomVoice entry cannot clone: it synthesizes with a built-in + # speaker (the pick when it names one, the first speaker when it + # names a clone voice); the Base entry cannot take a speaker name + # and clones with the picked server voice (the first when the pick + # names a speaker). + self._patch_remote( + [{"id": "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF", + "family": "qwen3_tts", "task": "tts"}, + {"id": "Qwen3-TTS-12Hz-1.7B-Base-GGUF", "family": "qwen3_tts", + "task": "tts"}], + voices=["narrator", "second"]) + self._mock_preflight() + self._answer_form( + backend="audiocpp-remote", model_id=hub.AUDIOCPP_MODEL_ALL, + audiocpp_voice="narrator", instructions="") + cmd = self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + self.assertEqual(cmd[2]["model_voices"], { + "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF": + hub.QWEN3_TTS_SPEAKERS[0], + "Qwen3-TTS-12Hz-1.7B-Base-GGUF": "narrator"}) + self._answer_form( + backend="audiocpp-remote", model_id=hub.AUDIOCPP_MODEL_ALL, + audiocpp_voice="Vivian", instructions="") + cmd = self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + self.assertEqual(cmd[2]["model_voices"], { + "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF": "Vivian", + "Qwen3-TTS-12Hz-1.7B-Base-GGUF": "narrator"}) + + def test_all_voice_union_offers_clone_voices_then_speakers(self): + # The Voice menu under "All" lists every model's clone voices + # first, then the built-in speakers while a speaker-capable model + # is configured; picking "All" keeps a voice the union offers. + self._patch_remote( + [{"id": "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF", + "family": "qwen3_tts", "task": "tts"}, + {"id": "alpha", "family": "chatterbox", "task": "tts"}], + voices=["narrator"]) + self._mock_preflight() + self._answer_form( + backend="audiocpp-remote", model_id="alpha", + audiocpp_voice="narrator", instructions="") + self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + fields = self.tui.forms_seen[0][1] + voice_field = self._field("audiocpp_voice") + model_field = self._field("model_id") + self.assertTrue(voice_field["visible"](fields)) + # Flip the Model pick to "All" (the form applies picks to the + # fields, firing on_change): the union lists every clone voice + # first, then the built-in speakers, and a clone pick survives. + voice_field["value"] = "narrator" + model_field["value"] = hub.AUDIOCPP_MODEL_ALL + model_field["on_change"](fields) + self.assertTrue(voice_field["visible"](fields)) + self.assertEqual(voice_field["choices"](fields), + [(v, v) for v in + ["narrator"] + list(hub.QWEN3_TTS_SPEAKERS)]) + self.assertEqual(voice_field["value"], "narrator") + + def test_all_refuses_voice_design_model_without_instructions(self): + # A vdes entry in the "All" run needs the Instructions text its + # voice comes from: Generate! refuses with a pointed message + # instead of failing the run (or silently skipping the model). + self._patch_remote( + [{"id": "alpha", "family": "higgs_audio_tts", "task": "tts"}, + {"id": "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF", + "family": "qwen3_tts", "task": "vdes"}], + voices=["narrator"]) + self._mock_preflight() + self._answer_form( + backend="audiocpp-remote", model_id=hub.AUDIOCPP_MODEL_ALL, + audiocpp_voice="narrator", instructions="") + self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + fields = self.tui.forms_seen[0][1] + voice_field = self._field("audiocpp_voice") + instructions_field = self._field("instructions") + # The form applies the "All" pick to the field before validating. + self._field("model_id")["value"] = hub.AUDIOCPP_MODEL_ALL + error = voice_field["validate"]("narrator") + self.assertIsNotNone(error) + self.assertIn("voice design", error) + self.assertIn("Instructions", error) + self.assertEqual(instructions_field["validate"](""), error) + self.assertIsNone(instructions_field["validate"]("A warm narrator")) + + def test_all_refuses_clone_only_model_without_voices(self): + # A clone-only family (chatterbox) whose server lists no voices + # cannot run in the "All" set: Generate! refuses naming the model + # (with a description the run would go — instruction voice). + self._patch_remote( + [{"id": "alpha", "family": "higgs_audio_tts", "task": "tts"}, + {"id": "beta", "family": "chatterbox", "task": "tts"}], + voices=[]) + self._mock_preflight() + self._answer_form( + backend="audiocpp-remote", model_id=hub.AUDIOCPP_MODEL_ALL, + audiocpp_voice="", instructions="") + self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + fields = self.tui.forms_seen[0][1] + voice_field = self._field("audiocpp_voice") + # The form applies the "All" pick to the field before validating. + self._field("model_id")["value"] = hub.AUDIOCPP_MODEL_ALL + error = voice_field["validate"]("") + self.assertIsNotNone(error) + self.assertIn("beta", error) + self.assertIn("Instructions", error) + # With a description the clone-only model designs its voice from + # it (instruction-voice mode): the run is accepted. + self._field("instructions")["value"] = "A warm narrator" + self.assertIsNone(voice_field["validate"]("")) + + def test_all_pick_works_on_the_managed_entry(self): + # The managed entry plans the same "All" run from server.json: + # voice_dir stems feed the union and the picked clone voice is + # shared by both clone-capable models. + with tempfile.TemporaryDirectory() as td: + root = Path(td) + (root / "server.json").write_text(json.dumps({ + "models": [{"id": "qwen-1_7b", "family": "qwen3_tts", + "task": "tts"}, + {"id": "qwen-0_6b", "family": "qwen3_tts", + "task": "tts"}], + "voice_dir": str(root), + }), encoding="utf-8") + (root / "Narrator.wav").write_bytes(b"x") + with patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=root): + self._mock_preflight() + self._answer_form(backend="audiocpp", + model_id=hub.AUDIOCPP_MODEL_ALL, + audiocpp_voice="Narrator", instructions="") + cmd = self._convert(None, + [self._ready("audiocpp", "audio.cpp")]) + kwargs = cmd[2] + self.assertEqual(kwargs["model_ids"], ["qwen-1_7b", "qwen-0_6b"]) + self.assertEqual(kwargs["model_voices"], + {"qwen-1_7b": "Narrator", "qwen-0_6b": "Narrator"}) + def test_audiocpp_local_model_switch_keeps_the_picked_voice(self): # The managed entry's voice list is shared by every model in # server.json, so switching models keeps the picked voice. @@ -2359,6 +2574,71 @@ class PreflightTests(unittest.TestCase): with self.assertRaises(hub._BackToForm): confirm("overwrite?", True) + # -- "All (multiple generation)": one plan per model ---------------- + + def _all_cmd(self): + return ("convert", "audiocpp", { + "model_ids": ["m1", "m2"], + "model_voices": {"m1": "narrator", "m2": None}, + "output_format": "mp3", "clone": None}) + + def test_all_run_plans_each_model_and_stashes_planned_by_model(self): + stdscr = object() + cmd = self._all_cmd() + with patch.object(hub.AudiobookConverter, "preflight_overwrites", + side_effect=[(["book.txt"], + [("book.txt", "book_m1_narrator")]), + (["book.txt"], + [("book.txt", "book_m2_designed")])]) \ + as mk_pre: + self.assertTrue(hub._preflight(stdscr, cmd)) + self.assertEqual(mk_pre.call_count, 2) + first, second = mk_pre.call_args_list + # Each model plans with its own voice (so its narrator tag — and + # therefore its overwrite questions — match the real run) and its + # model-tagged output name. + self.assertEqual(first.kwargs["voice"], "narrator") + self.assertEqual(first.kwargs["voice_mode"], + hub.voice_mode_for("audiocpp", "narrator", + None, None)) + self.assertEqual(first.kwargs["name_tag"], "m1") + self.assertIsNone(second.kwargs["voice"]) + self.assertEqual(second.kwargs["voice_mode"], + hub.voice_mode_for("audiocpp", None, None, None)) + self.assertEqual(second.kwargs["name_tag"], "m2") + self.assertEqual(cmd[2]["book_files"], ["book.txt"]) + self.assertEqual(cmd[2]["planned_by_model"], + {"m1": [("book.txt", "book_m1_narrator")], + "m2": [("book.txt", "book_m2_designed")]}) + self.assertNotIn("planned", cmd[2]) + + def test_all_run_no_books_flashes_and_returns_false(self): + stdscr = object() + with patch.object(hub.AudiobookConverter, "preflight_overwrites", + return_value=([], [])), \ + patch.object(hub.tui, "flash") as mk_flash: + self.assertFalse(hub._preflight(stdscr, self._all_cmd())) + mk_flash.assert_called_once() + + def test_all_run_all_skipped_flashes_and_returns_false(self): + stdscr = object() + with patch.object(hub.AudiobookConverter, "preflight_overwrites", + return_value=(["book.txt"], [])), \ + patch.object(hub.tui, "flash") as mk_flash: + self.assertFalse(hub._preflight(stdscr, self._all_cmd())) + mk_flash.assert_called_once() + + def test_all_run_confirm_esc_raises_back_to_form(self): + stdscr = object() + with patch.object(hub.AudiobookConverter, "preflight_overwrites", + return_value=(["book.txt"], + [("book.txt", "x")])) as mk_pre: + hub._preflight(stdscr, self._all_cmd()) + confirm = mk_pre.call_args.kwargs["confirm"] + with patch.object(hub.tui, "confirm", return_value=hub._CANCEL): + with self.assertRaises(hub._BackToForm): + confirm("overwrite?", True) + class DispatchConversionTests(unittest.TestCase): """_Hub._run_conversion: builds the config and runs the run view.""" |