aboutsummaryrefslogtreecommitdiff
path: root/app/tests
diff options
context:
space:
mode:
Diffstat (limited to 'app/tests')
-rw-r--r--app/tests/test_backends_audiocpp.py90
-rw-r--r--app/tests/test_hub.py213
2 files changed, 303 insertions, 0 deletions
diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py
index 9882ce1..e2b09d0 100644
--- a/app/tests/test_backends_audiocpp.py
+++ b/app/tests/test_backends_audiocpp.py
@@ -1058,5 +1058,95 @@ class NonInteractiveMainTests(unittest.TestCase):
self.assertEqual(ctx.exception.code, 2)
+class FetchServerEndpointsTests(unittest.TestCase):
+ """fetch_server_models / fetch_server_voices: live queries against a
+ running audiocpp_server (urlopen mocked)."""
+
+ @staticmethod
+ def _urlopen_responding(bodies, errors=None):
+ """A urlopen stub returning successive BODIES (bytes) or raising
+ successive ERRORS; records every requested URL."""
+ calls = []
+
+ def fake_urlopen(url, timeout=10):
+ calls.append(url)
+ if errors:
+ raise errors.pop(0)
+ body = bodies.pop(0)
+ context = MagicMock()
+ context.__enter__.return_value = context
+ context.__exit__.return_value = False
+ context.read.return_value = body
+ return context
+
+ return fake_urlopen, calls
+
+ def test_fetch_models_parses_id_family_task(self):
+ urlopen, calls = self._urlopen_responding([json.dumps({
+ "data": [{"id": "qwen", "family": "qwen3_tts", "task": "tts"},
+ {"id": "legacy"}],
+ }).encode("utf-8")])
+ with patch.object(make_server.urllib.request, "urlopen", urlopen):
+ models = make_server.fetch_server_models("http://127.0.0.1:8080")
+ # Missing fields mirror the converter's client: empty strings.
+ self.assertEqual(models, [
+ {"id": "qwen", "family": "qwen3_tts", "task": "tts"},
+ {"id": "legacy", "family": "", "task": ""},
+ ])
+ self.assertEqual(calls, ["http://127.0.0.1:8080/v1/models"])
+
+ def test_fetch_models_trailing_slash_url(self):
+ urlopen, calls = self._urlopen_responding(
+ [b'{"data": [{"id": "m"}]}'])
+ with patch.object(make_server.urllib.request, "urlopen", urlopen):
+ make_server.fetch_server_models("http://host:8080/")
+ self.assertEqual(calls, ["http://host:8080/v1/models"])
+
+ def test_fetch_models_connection_error_returns_none(self):
+ import urllib.error
+ urlopen, _ = self._urlopen_responding(
+ [], errors=[urllib.error.URLError("Connection refused")])
+ with patch.object(make_server.urllib.request, "urlopen", urlopen):
+ self.assertIsNone(
+ make_server.fetch_server_models("http://127.0.0.1:8080"))
+
+ def test_fetch_models_non_json_body_returns_none(self):
+ # A port answering TCP but not speaking audiocpp_server JSON.
+ urlopen, _ = self._urlopen_responding([b"<html>not json</html>"])
+ with patch.object(make_server.urllib.request, "urlopen", urlopen):
+ self.assertIsNone(
+ make_server.fetch_server_models("http://127.0.0.1:8080"))
+
+ def test_fetch_models_unexpected_document_yields_empty_list(self):
+ urlopen, _ = self._urlopen_responding([b'{"foo": 1}'])
+ with patch.object(make_server.urllib.request, "urlopen", urlopen):
+ self.assertEqual(
+ make_server.fetch_server_models("http://127.0.0.1:8080"), [])
+
+ def test_fetch_voices_parses_names_and_encodes_model(self):
+ urlopen, calls = self._urlopen_responding(
+ [b'{"voices": ["narrator", "obama"]}'])
+ with patch.object(make_server.urllib.request, "urlopen", urlopen):
+ voices = make_server.fetch_server_voices(
+ "http://127.0.0.1:8080", "qwen")
+ self.assertEqual(voices, ["narrator", "obama"])
+ self.assertEqual(calls,
+ ["http://127.0.0.1:8080/v1/audio/voices?model=qwen"])
+
+ def test_fetch_voices_error_returns_none(self):
+ import urllib.error
+ urlopen, _ = self._urlopen_responding(
+ [], errors=[urllib.error.URLError("boom")])
+ with patch.object(make_server.urllib.request, "urlopen", urlopen):
+ self.assertIsNone(
+ make_server.fetch_server_voices("http://h", "qwen"))
+
+ def test_fetch_voices_non_list_shape_returns_none(self):
+ urlopen, _ = self._urlopen_responding([b'{"voices": 5}'])
+ with patch.object(make_server.urllib.request, "urlopen", urlopen):
+ self.assertIsNone(
+ make_server.fetch_server_voices("http://h", "qwen"))
+
+
if __name__ == "__main__":
unittest.main()
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)."""
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-08 11:56:01 +0000