diff options
Diffstat (limited to 'app/tests/test_hub.py')
| -rw-r--r-- | app/tests/test_hub.py | 351 |
1 files changed, 326 insertions, 25 deletions
diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index ddd6ec0..485a37f 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -835,9 +835,9 @@ class ConvertFlowTests(unittest.TestCase): }) self.addCleanup(spec_cache.clear) - # Keys shared by every backend entry; a "-remote" backend's other - # option keys are namespaced under "<entry>." in the form dict - # (mirroring hub.py), so _form_values maps them automatically. + # Keys shared by every backend entry; every entry's other option keys + # are namespaced under "<entry>." in the form dict (mirroring hub.py, + # managed entries included), so _form_values maps them automatically. _COMMON_KEYS = frozenset(("backend", "single_file")) def _form_values(self, **overrides): @@ -850,8 +850,8 @@ class ConvertFlowTests(unittest.TestCase): values = {"single_file": False} values.update(overrides) backend = values.get("backend") or "" - if backend.endswith("-remote"): - prefix = f"{backend}." + prefix = f"{backend}." if backend else "" + if prefix: values = {(prefix + key if key not in self._COMMON_KEYS else key): value for key, value in values.items()} return values @@ -1979,10 +1979,11 @@ class ConvertFlowTests(unittest.TestCase): self.assertEqual(fields[0]["choices"], [("audio.cpp", "audiocpp"), ("audio.cpp [remote]", "audiocpp-remote")]) - # The two entries' fields are namespaced, so both carry their own - # values and picking one never leaks the other's into the run. + # The two entries' fields are namespaced (managed included), so + # both carry their own values and picking one never leaks the + # other's into the run. keys = [f["key"] for f in fields] - self.assertIn("model_id", keys) + self.assertIn("audiocpp.model_id", keys) self.assertIn("audiocpp-remote.model_id", keys) def test_managed_and_remote_entries_do_not_overwrite_each_other(self): @@ -2006,8 +2007,10 @@ class ConvertFlowTests(unittest.TestCase): # Managed selected: its picks must survive next to the # remote entry's same-shaped fields. self.tui.form_script.append({ - "backend": "audiocpp", "model_id": "qwen", - "audiocpp_voice": "", "instructions": "", + "backend": "audiocpp", + "audiocpp.model_id": "qwen", + "audiocpp.audiocpp_voice": "", + "audiocpp.instructions": "", "audiocpp-remote.model_id": "higgs", "audiocpp-remote.audiocpp_voice": "narrator", **common}) @@ -2022,7 +2025,7 @@ class ConvertFlowTests(unittest.TestCase): "backend": "audiocpp-remote", "audiocpp-remote.model_id": "higgs", "audiocpp-remote.audiocpp_voice": "narrator", - "model_id": "qwen", + "audiocpp.model_id": "qwen", **common}) remote_cmd = self._convert(None, statuses) self.assertIsNotNone(remote_cmd) @@ -2072,8 +2075,9 @@ class ConvertFlowTests(unittest.TestCase): self.assertEqual(cmd[2]["voice"], "Serena") fields = self.tui.forms_seen[0][1] self.assertEqual([f["key"] for f in fields], - ["backend", "mode", "speaker", "clone_dir", - "clone", "qwen_instructions", "single_file"]) + ["backend", "qwen.mode", "qwen.speaker", + "qwen.clone_dir", "qwen.clone", + "qwen.qwen_instructions", "single_file"]) mode_field = self._field("mode") # Model names are padded to the widest ("CustomVoice"/"VoiceDesign" # are 11 columns) plus a two-space gutter, so every (purpose) opens @@ -2354,9 +2358,10 @@ class ConvertFlowTests(unittest.TestCase): [("audio.cpp", "audiocpp"), ("qwen-tts", "qwen")]) self.assertEqual( [f["key"] for f in fields], - ["backend", "model_id", "audiocpp_voice", "instructions", - "request_options", "mode", "speaker", "clone_dir", - "clone", "qwen_instructions", "single_file"]) + ["backend", "audiocpp.model_id", "audiocpp.audiocpp_voice", + "audiocpp.instructions", "audiocpp.request_options", + "qwen.mode", "qwen.speaker", "qwen.clone_dir", + "qwen.clone", "qwen.qwen_instructions", "single_file"]) # The form opens on the configured default (audio.cpp): its fields # show, the other backend's hide. Instructions shows too (optional # style/delivery control even on the clone-only higgs entry), while @@ -2391,6 +2396,251 @@ class ConvertFlowTests(unittest.TestCase): "qwen_instructions"): self.assertFalse(self._field(key)["visible"](fields)) + # ------------------------------------------------------------------ + # SGLang-Omni: managed (installed models) and remote entries + # ------------------------------------------------------------------ + + def _patch_sglomni_installed(self, entries): + patcher = patch.object(hub.sglomni_backend, "installed_entries", + return_value=entries) + patcher.start() + self.addCleanup(patcher.stop) + + def test_sglomni_managed_speaker_model_sends_preset_voice(self): + entry = hub.sglomni_backend.entry_by_key("qwen3_tts_0_6b_customvoice") + self._patch_sglomni_installed([entry]) + with patch.object(hub.config, "AUDIO_FORMAT", "m4b"), \ + patch.object(hub.config, "LANGUAGE", "English"), \ + patch.object(hub.config, "SPEED", 1.0), \ + patch.object(hub.config, "DEBUG", False), \ + patch.object(hub.config, "STOP_SERVER_AND_EXIT", True): + self._mock_preflight() + self._answer_form(backend="sglomni", model_id=entry.key, + voice="Vivian", named_voice="", clone="", + clone_dir="/tmp", instructions="") + cmd = self._convert(None, [self._ready("sglomni", + "SGLang-Omni")]) + self.assertEqual(cmd[1], hub.BACKEND_SGLOMNI) + kwargs = cmd[2] + self.assertEqual(kwargs["model_id"], entry.key) + self.assertEqual(kwargs["voice"], "Vivian") + self.assertNotIn("clone", kwargs) + self.assertNotIn("api_url", kwargs) + + def test_sglomni_managed_clone_model_routes_the_reference(self): + entry = hub.sglomni_backend.entry_by_key("higgs_audio_v3_tts") + self._patch_sglomni_installed([entry]) + with patch.object(hub.config, "AUDIO_FORMAT", "m4b"), \ + patch.object(hub.config, "LANGUAGE", "English"), \ + patch.object(hub.config, "SPEED", 1.0), \ + patch.object(hub.config, "DEBUG", False), \ + patch.object(hub.config, "STOP_SERVER_AND_EXIT", True): + self._mock_preflight() + self._answer_form(backend="sglomni", model_id=entry.key, + voice="", named_voice="", + clone="/tmp/ref.wav", clone_dir="/tmp", + instructions="") + cmd = self._convert(None, [self._ready("sglomni", + "SGLang-Omni")]) + kwargs = cmd[2] + self.assertEqual(kwargs["clone"], "/tmp/ref.wav") + # A reference wins over any named voice: ref_audio drives the clone. + self.assertIsNone(kwargs["voice"]) + + def test_sglomni_managed_design_model_sends_instructions(self): + entry = hub.sglomni_backend.entry_by_key("qwen3_tts_1_7b_voicedesign") + self._patch_sglomni_installed([entry]) + with patch.object(hub.config, "AUDIO_FORMAT", "m4b"), \ + patch.object(hub.config, "LANGUAGE", "English"), \ + patch.object(hub.config, "SPEED", 1.0), \ + patch.object(hub.config, "DEBUG", False), \ + patch.object(hub.config, "STOP_SERVER_AND_EXIT", True): + self._mock_preflight() + self._answer_form(backend="sglomni", model_id=entry.key, + voice="", named_voice="", clone="", + clone_dir="/tmp", + instructions="A warm narrator.") + cmd = self._convert(None, [self._ready("sglomni", + "SGLang-Omni")]) + kwargs = cmd[2] + self.assertEqual(kwargs["instructions"], "A warm narrator.") + self.assertNotIn("clone", kwargs) + self.assertNotIn("voice", kwargs) + + def test_sglomni_remote_offers_the_hosted_model_and_uploaded_voices(self): + served = patch.object(hub.backend_probe, "sglomni_served_model", + return_value="bosonai/higgs-audio-v3-tts-4b") + voices = patch.object(hub.backend_probe, "sglomni_voice_names", + return_value=["narrator"]) + for patcher in (served, voices): + patcher.start() + self.addCleanup(patcher.stop) + with patch.object(hub.config, "AUDIO_FORMAT", "m4b"), \ + patch.object(hub.config, "LANGUAGE", "English"), \ + patch.object(hub.config, "SPEED", 1.0), \ + patch.object(hub.config, "DEBUG", False), \ + patch.object(hub.config, "STOP_SERVER_AND_EXIT", True): + self._mock_preflight() + self._answer_form(backend="sglomni-remote", + model_id="higgs_audio_v3_tts", + voice="", named_voice="narrator", + clone="", clone_dir="/tmp", instructions="") + cmd = self._convert( + None, [self._remote("sglomni", "SGLang-Omni", + url="http://sgl.local:8100")]) + kwargs = cmd[2] + self.assertEqual(kwargs["model_id"], "higgs_audio_v3_tts") + # An uploaded (named) voice rides the request's voice field. + self.assertEqual(kwargs["voice"], "narrator") + self.assertEqual(kwargs["api_url"], "http://sgl.local:8100") + + def test_sglomni_fields_do_not_shadow_audiocpp_fields(self): + # Regression: the merged Generate form keys every entry's fields + # under its backend key. With audio.cpp listed first, the shared + # unprefixed keys used to make audio.cpp's "model_id" shadow + # SGLang's: every SGLang model then inherited the FIRST installed + # model's capability (a clone model), so the preset-voice + # Voxtral TTS 4B showed "Voice to clone" instead of its preset + # Voice menu — and, reversed, an audio.cpp submission received + # SGLang's model_id/instructions values from the submit dict. + base = hub.sglomni_backend.entry_by_key("qwen3_tts_1_7b_base") + voxtral = hub.sglomni_backend.entry_by_key("voxtral_tts") + self._patch_sglomni_installed([base, voxtral]) + presets = patch.object( + hub.sglomni_backend, "preset_voices", + lambda entry: ["casual_male"] if entry.key == "voxtral_tts" + else list(entry.speakers or ())) + presets.start() + self.addCleanup(presets.stop) + statuses = [self._ready("audiocpp", "audio.cpp"), + self._ready("sglomni", "SGLang-Omni")] + with tempfile.TemporaryDirectory() as td: + root = Path(td) + (root / "server.json").write_text(json.dumps({ + "models": [{"id": "qwen", "family": "qwen3_tts", + "task": "tts"}], + }), encoding="utf-8") + with patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=root), \ + patch.object(hub.config, "AUDIO_FORMAT", "m4b"), \ + patch.object(hub.config, "LANGUAGE", "English"), \ + patch.object(hub.config, "SPEED", 1.0), \ + patch.object(hub.config, "DEBUG", False), \ + patch.object(hub.config, "STOP_SERVER_AND_EXIT", True): + self._mock_preflight() + # Voxtral picked: its run carries the preset voice, and in + # the form its Voice menu shows while the clone picker hides. + self._answer_form(backend="sglomni", model_id="voxtral_tts", + voice="casual_male", named_voice="", + clone="", clone_dir="/tmp", + instructions="") + cmd = self._convert(None, statuses) + fields = self.tui.forms_seen[-1][1] + # The captured fields carry the form's opening state (the + # audiocpp entry); point the pickers at the SGLang entry and + # the Voxtral model to assert its field visibility. + next(f for f in fields + if f["key"] == "backend")["value"] = "sglomni" + next(f for f in fields + if f["key"] == "sglomni.model_id")["value"] = \ + "voxtral_tts" + self.assertTrue(self._field("voice")["visible"](fields)) + self.assertIn( + "casual_male", + [label for label, _ in + self._field("voice")["choices"](fields)]) + self.assertFalse(self._field("clone")["visible"](fields)) + # An audio.cpp submission keeps its own picks: SGLang's + # same-named fields must not leak into its run. + with patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=root), \ + patch.object(hub.config, "AUDIO_FORMAT", "m4b"), \ + patch.object(hub.config, "LANGUAGE", "English"), \ + patch.object(hub.config, "SPEED", 1.0), \ + patch.object(hub.config, "DEBUG", False), \ + patch.object(hub.config, "STOP_SERVER_AND_EXIT", True): + self._mock_preflight() + self._answer_form(backend="audiocpp", model_id="qwen", + audiocpp_voice="", instructions="Style.", + voice="", named_voice="", clone="", + clone_dir="/tmp") + cmd = self._convert(None, statuses) + self.assertEqual(cmd[2]["model_id"], "qwen") + self.assertEqual(cmd[2]["instructions"], "Style.") + + def test_sglomni_every_catalog_model_drives_the_right_form(self): + """Every catalog entry shows the capability-matched voice fields. + + speaker -> the preset Voice menu; clone with a required reference + -> the clone picker only; clone that narrates without one -> the + default-voice pick alongside the clone picker; design -> the + Instructions box. + """ + + def fake_preset_voices(entry): + if entry.capability == "speaker": + return list(entry.speakers or ("casual_male",)) + return [] + + presets = patch.object(hub.sglomni_backend, "preset_voices", + fake_preset_voices) + presets.start() + self.addCleanup(presets.stop) + self._patch_sglomni_installed(list(hub.sglomni_backend.ENTRIES)) + with patch.object(hub.config, "AUDIO_FORMAT", "m4b"), \ + patch.object(hub.config, "LANGUAGE", "English"), \ + patch.object(hub.config, "SPEED", 1.0), \ + patch.object(hub.config, "DEBUG", False), \ + patch.object(hub.config, "STOP_SERVER_AND_EXIT", True): + self._mock_preflight() + for entry in hub.sglomni_backend.ENTRIES: + with self.subTest(entry=entry.key): + overrides = {"backend": "sglomni", + "model_id": entry.key, + "voice": "", "named_voice": "", + "clone": "", "clone_dir": "/tmp", + "instructions": ""} + if entry.capability == "speaker": + overrides["voice"] = \ + (entry.speakers or ("casual_male",))[0] + elif entry.capability == "clone": + overrides["clone"] = "/tmp/ref.wav" + else: # design + overrides["instructions"] = "A warm narrator." + self._answer_form(**overrides) + cmd = self._convert(None, [ + self._ready("sglomni", "SGLang-Omni")]) + self.assertIsNotNone(cmd) + kwargs = cmd[2] + self.assertEqual(kwargs["model_id"], entry.key) + fields = self.tui.forms_seen[-1][1] + next(f for f in fields + if f["key"] == "sglomni.model_id")["value"] = \ + entry.key + shown = {f["key"] for f in fields + if f["key"].startswith("sglomni.") + and f["visible"](fields)} + expected = {"sglomni.model_id"} + if entry.capability == "speaker": + expected.add("sglomni.voice") + self.assertEqual( + kwargs.get("voice"), + (entry.speakers or ("casual_male",))[0]) + self.assertNotIn("clone", kwargs) + elif entry.capability == "clone": + expected |= {"sglomni.clone_dir", "sglomni.clone"} + if not entry.requires_reference: + expected.add("sglomni.named_voice") + self.assertEqual(kwargs.get("clone"), "/tmp/ref.wav") + self.assertIsNone(kwargs.get("voice")) + else: # design + expected.add("sglomni.instructions") + self.assertEqual(kwargs.get("instructions"), + "A warm narrator.") + self.assertNotIn("voice", kwargs) + self.assertNotIn("clone", kwargs) + self.assertEqual(shown, expected) + class SelectSpecTests(unittest.TestCase): """_select_spec: single-server selection (qwen hosts one model at a time).""" @@ -2808,6 +3058,42 @@ class AddAutostartTests(unittest.TestCase): self.assertIsNone(hub._add_autostart(cmd, [self._status()])) self.assertEqual(cmd[2]["restart_server"], "qwen") + def test_sglomni_running_server_hosting_another_model_is_restarted(self): + # One model per server process: a managed sglomni server hosting + # Higgs while the run selected MOSS-TTS is restarted first. + entry = hub.sglomni_backend.entry_by_key("higgs_audio_v3_tts") + spec = ServerSpec("sglomni", "http://127.0.0.1:8100", ["x"]) + status = BackendStatus("sglomni", "SGLang-Omni", installed=True, + configured=True, running=True, + servers=[spec]) + cmd = ("convert", "sglomni", {"model_id": "moss_tts"}) + with patch.object(hub, "detect_all", return_value=[status]), \ + patch("backends.common.server_running", return_value=True), \ + patch.object(hub.servers, "alive", return_value=True), \ + patch.object(hub.sglomni_backend, "resolve_model", + return_value=entry), \ + patch.object(hub.backend_probe, "sglomni_served_model", + return_value="OpenMOSS-Team/MOSS-TTS-v1.5"): + self.assertIsNone(hub._add_autostart(cmd, [status])) + self.assertEqual(cmd[2]["restart_server"], "sglomni") + + def test_sglomni_running_server_hosting_the_wanted_model_is_kept(self): + entry = hub.sglomni_backend.entry_by_key("higgs_audio_v3_tts") + spec = ServerSpec("sglomni", "http://127.0.0.1:8100", ["x"]) + status = BackendStatus("sglomni", "SGLang-Omni", installed=True, + configured=True, running=True, + servers=[spec]) + cmd = ("convert", "sglomni", {"model_id": entry.key}) + with patch.object(hub, "detect_all", return_value=[status]), \ + patch("backends.common.server_running", return_value=True), \ + patch.object(hub.sglomni_backend, "resolve_model", + return_value=entry), \ + patch.object(hub.backend_probe, "sglomni_served_model", + return_value=entry.repo): + self.assertIsNone(hub._add_autostart(cmd, [status])) + self.assertNotIn("restart_server", cmd[2]) + self.assertNotIn("autostart", cmd[2]) + def test_foreign_server_with_wrong_model_refuses_the_run(self): cmd = ("convert", "qwen", {"clone": "/tmp/ref.wav"}) with patch.object(hub, "detect_all", return_value=[self._status()]), \ @@ -2953,9 +3239,11 @@ class SettingsTests(unittest.TestCase): "unload_models": True, "qwen_port": "7862", "faster_port": "8001", "audiocpp_port": "8081", + "sglomni_port": "8101", "audiocpp_remote_url": "10.0.0.5:8080", "faster_remote_url": "http://10.0.0.6:8000", - "qwen_remote_url": ""} + "qwen_remote_url": "", + "sglomni_remote_url": "10.0.0.7:8100"} with patch.object(hub.common, "update_config_value", fake_update), \ patch.object(hub, "_sync_audiocpp_server_port"): @@ -2980,7 +3268,11 @@ class SettingsTests(unittest.TestCase): "FASTER_REMOTE_URL": "http://10.0.0.6:8000", "AUDIOCPP_REMOTE_URL": - "http://10.0.0.5:8080"}) + "http://10.0.0.5:8080", + "SGLOMNI_API_URL": + "http://127.0.0.1:8101", + "SGLOMNI_REMOTE_URL": + "http://10.0.0.7:8100"}) # In-memory config is reloaded so this session sees the change, # and the converter module's folder globals follow the directories. self.assertEqual(hub.config.AUDIO_FORMAT, "ogg") @@ -3010,7 +3302,8 @@ class SettingsTests(unittest.TestCase): "stop_and_exit": True, "unload_models": True, "qwen_port": "7860", - "faster_port": "8000", "audiocpp_port": "8080"} + "faster_port": "8000", "audiocpp_port": "8080", + "sglomni_port": "8100"} with patch.object(hub.common, "update_config_value") as mk_update: with self.assertRaises(ValueError): hub._apply_settings({**base, "language": "Klingon"}) @@ -3114,7 +3407,8 @@ class SettingsTests(unittest.TestCase): "stop_and_exit": True, "unload_models": True, "qwen_port": "7860", - "faster_port": "8000", "audiocpp_port": "8080"} + "faster_port": "8000", "audiocpp_port": "8080", + "sglomni_port": "8100"} applied = [] @@ -3134,8 +3428,10 @@ class SettingsTests(unittest.TestCase): "speed", "debug", "stop_and_exit", "unload_models", "audiocpp_port", - "faster_port", "qwen_port", "audiocpp_remote_url", - "faster_remote_url", "qwen_remote_url"]) + "faster_port", "qwen_port", "sglomni_port", + "audiocpp_remote_url", + "faster_remote_url", "qwen_remote_url", + "sglomni_remote_url"]) kinds = {f["key"]: f["kind"] for f in captured["fields"]} self.assertEqual(kinds["audio_format"], "choice") self.assertEqual(kinds["audio_bitrate"], "text") @@ -3182,7 +3478,8 @@ class SettingsTests(unittest.TestCase): "unload_models": True, "qwen_port": "7860", "faster_port": "8000", - "audiocpp_port": "8080"}]) + "audiocpp_port": "8080", + "sglomni_port": "8100"}]) # Saving is silent: no confirmation flash either way. self.assertNotIn("flash", captured) @@ -3329,8 +3626,10 @@ class SettingsTests(unittest.TestCase): "AUDIOCPP_UNLOAD_MODELS", "QWEN_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL", + "SGLOMNI_API_URL", "QWEN_REMOTE_URL", - "FASTER_REMOTE_URL", "AUDIOCPP_REMOTE_URL")} + "FASTER_REMOTE_URL", "AUDIOCPP_REMOTE_URL", + "SGLOMNI_REMOTE_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) @@ -3354,7 +3653,9 @@ class SettingsTests(unittest.TestCase): 'AUDIOCPP_API_URL = "http://127.0.0.1:8080"\n' 'QWEN_REMOTE_URL = "http://127.0.0.1:7860"\n' 'FASTER_REMOTE_URL = "http://127.0.0.1:8000"\n' - 'AUDIOCPP_REMOTE_URL = "http://127.0.0.1:8080"\n', + 'AUDIOCPP_REMOTE_URL = "http://127.0.0.1:8080"\n' + 'SGLOMNI_API_URL = "http://127.0.0.1:8100"\n' + 'SGLOMNI_REMOTE_URL = "http://127.0.0.1:8100"\n', encoding="utf-8") with patch.object(hub.common, "CONFIG_PATH", path), \ patch.object(hub, "_sync_audiocpp_server_port"): |