aboutsummaryrefslogtreecommitdiff
path: root/app/tests/test_hub.py
diff options
context:
space:
mode:
authorhistoria <historiavg@proton.me>2026-08-24 06:04:38 -0400
committerhistoria <historiavg@proton.me>2026-08-24 06:04:38 -0400
commit4db8ea7a63107297450819d227497ebbb121ff38 (patch)
tree250b646195f98a4b4c42b04589c0ca8951d13fbb /app/tests/test_hub.py
parent4f3d16e9d532efd6a6bda283925efeb8c93b51c7 (diff)
downloadtts-audiobook-generator-4db8ea7a63107297450819d227497ebbb121ff38.tar.gz
fix: don't expect/use local config for remote servers
Diffstat (limited to 'app/tests/test_hub.py')
-rw-r--r--app/tests/test_hub.py213
1 files changed, 213 insertions, 0 deletions
diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py
index a009e3c..2fe0b2f 100644
--- a/app/tests/test_hub.py
+++ b/app/tests/test_hub.py
@@ -4,6 +4,8 @@ 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 json
+import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
@@ -13,6 +15,36 @@ from tests.test_tui import FakeCurses, FakeScreen
from ui import hub, tui
+class _ScriptedTUI:
+ """Stand-in for the tui widget module: answers each menu/line_edit/
+ confirm call from a scripted answer list and records every prompt."""
+
+ def __init__(self):
+ self.script = []
+ self.prompts = []
+ self.options_seen = []
+ self.flashes = []
+
+ def _next(self, prompt, options=None):
+ self.prompts.append(prompt)
+ if options is not None:
+ self.options_seen.append(options)
+ return self.script.pop(0)
+
+ def menu(self, stdscr, title, options, **kwargs):
+ return self._next(title, options)
+
+ def line_edit(self, stdscr, title, default, **kwargs):
+ self.prompts.append(f"{title} [default: {default!r}]")
+ return self.script.pop(0)
+
+ def confirm(self, stdscr, question, **kwargs):
+ return self._next(question)
+
+ def flash(self, stdscr, text, kind="warn"):
+ self.flashes.append(text)
+
+
class HubHelperTests(unittest.TestCase):
"""Pure helpers in hub.py (no curses)."""
@@ -385,6 +417,187 @@ class SubmenuStatusTableTests(unittest.TestCase):
[("Warning: ffmpeg not installed!", "err")])
+class ConvertFlowTests(unittest.TestCase):
+ """_convert_audiocpp / _convert_faster: local-config menus vs. live
+ queries against a running remote server."""
+
+ def setUp(self):
+ self.tui = _ScriptedTUI()
+ for name in ("menu", "line_edit", "confirm", "flash"):
+ patcher = patch.object(hub.tui, name, getattr(self.tui, name))
+ patcher.start()
+ self.addCleanup(patcher.stop)
+
+ def _answer_common_options(self):
+ # Output format, speed, single-file, chunk, debug.
+ self.tui.script += ["m4b", "1.5", False, False, False]
+
+ # ------------------------------------------------------------------
+ # audio.cpp: remote server (no local checkout / server.json)
+ # ------------------------------------------------------------------
+
+ def _patch_remote(self, models, voices=None):
+ """No local checkout; fetch helpers return MODELS/VOICES."""
+ checkout = patch.object(hub.audiocpp_backend, "find_local_checkout",
+ return_value=None)
+ fetched_models = patch.object(hub.audiocpp_backend,
+ "fetch_server_models",
+ lambda url: models)
+ fetched_voices = patch.object(hub.audiocpp_backend,
+ "fetch_server_voices",
+ lambda url, model_id: voices)
+ for patcher in (checkout, fetched_models, fetched_voices):
+ patcher.start()
+ self.addCleanup(patcher.stop)
+
+ def test_audiocpp_remote_queries_live_models_and_voices(self):
+ self._patch_remote(
+ [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}],
+ voices=["narrator"])
+ with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""):
+ self.tui.script += ["higgs", "narrator", ""]
+ self._answer_common_options()
+ cmd = hub._convert_audiocpp(None, [])
+ self.assertEqual(cmd[0], "convert")
+ self.assertEqual(cmd[1], hub.BACKEND_AUDIOCPP)
+ kwargs = cmd[2]
+ self.assertEqual(kwargs["model_id"], "higgs")
+ self.assertEqual(kwargs["voice"], "narrator")
+ self.assertIsNone(kwargs["instructions"])
+ # The model menu was fed from the live query.
+ self.assertEqual(self.tui.options_seen[0],
+ [("higgs (higgs_audio_tts, tts)", "higgs")])
+
+ def test_audiocpp_remote_qwen3_tts_offers_builtin_speaker_first(self):
+ self._patch_remote(
+ [{"id": "qwen", "family": "qwen3_tts", "task": "tts"}],
+ voices=["narrator"])
+ with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""):
+ self.tui.script += ["qwen", None, ""]
+ self._answer_common_options()
+ cmd = hub._convert_audiocpp(None, [])
+ self.assertIsNone(cmd[2]["voice"])
+ self.assertEqual(self.tui.options_seen[1],
+ [("(built-in speaker)", None), ("narrator", "narrator")])
+
+ def test_audiocpp_remote_missing_family_treated_as_qwen3_tts(self):
+ # Legacy servers omit family/task; the converter defaults them to
+ # qwen3_tts/tts and so must the menus (voice optional).
+ self._patch_remote([{"id": "legacy", "family": "", "task": ""}],
+ voices=[])
+ with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""):
+ # Empty server voices: no Voice menu, built-in speaker implied.
+ self.tui.script += ["legacy", ""]
+ self._answer_common_options()
+ cmd = hub._convert_audiocpp(None, [])
+ self.assertIsNotNone(cmd)
+ self.assertIsNone(cmd[2]["voice"])
+
+ def test_audiocpp_remote_vdes_needs_instructions_not_voice(self):
+ self._patch_remote(
+ [{"id": "design", "family": "qwen3_tts", "task": "vdes"}])
+ with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""):
+ self.tui.script += ["design", "A warm British narrator"]
+ self._answer_common_options()
+ cmd = hub._convert_audiocpp(None, [])
+ self.assertIsNone(cmd[2]["voice"])
+ self.assertEqual(cmd[2]["instructions"], "A warm British narrator")
+ # No voice prompt happened at all.
+ self.assertNotIn("Voice", [p for p in self.tui.prompts])
+
+ def test_audiocpp_remote_unreachable_models_flash_and_abort(self):
+ self._patch_remote(None) # endpoint did not answer valid JSON
+ cmd = hub._convert_audiocpp(None, [])
+ self.assertIsNone(cmd)
+ self.assertIn("Could not list models", self.tui.flashes[0])
+
+ def test_audiocpp_remote_empty_models_flash_and_abort(self):
+ self._patch_remote([])
+ cmd = hub._convert_audiocpp(None, [])
+ self.assertIsNone(cmd)
+ self.assertIn("hosts no model entries", self.tui.flashes[0])
+
+ def test_audiocpp_remote_no_server_voices_for_clone_model_aborts(self):
+ self._patch_remote(
+ [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}],
+ voices=[])
+ self.tui.script += ["higgs"]
+ cmd = hub._convert_audiocpp(None, [])
+ self.assertIsNone(cmd)
+ self.assertIn("lists none", self.tui.flashes[0])
+
+ def test_audiocpp_remote_failed_voices_query_aborts(self):
+ self._patch_remote(
+ [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}],
+ voices=None)
+ self.tui.script += ["higgs"]
+ cmd = hub._convert_audiocpp(None, [])
+ self.assertIsNone(cmd)
+ self.assertIn("Could not list voices", self.tui.flashes[0])
+
+ # ------------------------------------------------------------------
+ # audio.cpp: local managed setup keeps reading its server.json
+ # ------------------------------------------------------------------
+
+ def test_audiocpp_local_still_reads_server_json(self):
+ queried = []
+
+ def must_not_query(url):
+ queried.append(url)
+ raise AssertionError("live query on the local path")
+
+ with tempfile.TemporaryDirectory() as td:
+ root = Path(td)
+ (root / "server.json").write_text(json.dumps({
+ "models": [{"id": "qwen", "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), \
+ patch.object(hub.audiocpp_backend, "fetch_server_models",
+ must_not_query), \
+ patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""):
+ self.tui.script += ["qwen", "Narrator", ""]
+ self._answer_common_options()
+ cmd = hub._convert_audiocpp(None, [])
+ self.assertEqual(queried, [])
+ self.assertIsNotNone(cmd)
+ self.assertEqual(cmd[2]["model_id"], "qwen")
+ self.assertEqual(cmd[2]["voice"], "Narrator")
+
+ # ------------------------------------------------------------------
+ # faster: remote server (no local voices.json)
+ # ------------------------------------------------------------------
+
+ def test_faster_remote_prompts_for_a_voice_name(self):
+ with tempfile.TemporaryDirectory() as td:
+ with patch.object(hub.faster_backend, "_checkout",
+ return_value=Path(td)):
+ self.tui.script += ["obama"]
+ self._answer_common_options()
+ cmd = hub._convert_faster(None)
+ self.assertEqual(cmd[0], "convert")
+ self.assertEqual(cmd[1], "faster")
+ self.assertEqual(cmd[2]["voice"], "obama")
+ self.assertIn("Server-side voice", self.tui.prompts[0])
+
+ def test_faster_local_still_lists_voices_json(self):
+ with tempfile.TemporaryDirectory() as td:
+ checkout = Path(td)
+ (checkout / "voices.json").write_text(
+ json.dumps({"default": {}, "obama": {}}), encoding="utf-8")
+ with patch.object(hub.faster_backend, "_checkout",
+ return_value=checkout):
+ self.tui.script += ["obama"]
+ self._answer_common_options()
+ cmd = hub._convert_faster(None)
+ self.assertEqual(cmd[2]["voice"], "obama")
+ # The voice came from a menu over voices.json, not a text field.
+ self.assertIn("Select the voice to clone", self.tui.prompts[0])
+
+
class SelectSpecTests(unittest.TestCase):
"""_select_spec: mode-aware server selection (qwen has two servers)."""