From 4db8ea7a63107297450819d227497ebbb121ff38 Mon Sep 17 00:00:00 2001 From: historia Date: Mon, 24 Aug 2026 06:04:38 -0400 Subject: fix: don't expect/use local config for remote servers --- app/tests/test_backends_audiocpp.py | 90 +++++++++++++++ app/tests/test_hub.py | 213 ++++++++++++++++++++++++++++++++++++ 2 files changed, 303 insertions(+) (limited to 'app/tests') 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"not json"]) + 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).""" -- cgit v1.2.3