diff options
Diffstat (limited to 'app/tests/test_hub.py')
| -rw-r--r-- | app/tests/test_hub.py | 188 |
1 files changed, 133 insertions, 55 deletions
diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index 48cef0b..44c63fd 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -1376,13 +1376,15 @@ class ConvertFlowTests(unittest.TestCase): self.assertTrue(single["visible"](fields)) # ------------------------------------------------------------------ - # qwen: speaker or clone + # qwen: model picker (Base / CustomVoice / VoiceDesign) # ------------------------------------------------------------------ def test_qwen_builds_speaker_and_clone_form(self): with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian", "Serena"]), \ - patch.object(hub.config, "SPEAKER", "Vivian"): + patch.object(hub.config, "SPEAKER", "Vivian"), \ + patch.object(hub.qwen_backend.config, "QWEN_MODEL", + "CustomVoice"): self._answer_form(backend="qwen", mode="custom", speaker="Serena", clone="") # The fake mirrors the real update_config_value contract: @@ -1399,29 +1401,56 @@ class ConvertFlowTests(unittest.TestCase): self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], hub.BACKEND_QWEN) self.assertIsNone(cmd[2]["clone"]) + self.assertIsNone(cmd[2].get("instructions")) self.assertEqual(speaker_in_memory, "Serena") fields = self.tui.forms_seen[0][1] self.assertEqual([f["key"] for f in fields], ["backend", "mode", "speaker", "clone", + "qwen_instructions", "output_format", "language", "speed", "single_file", "debug", "stop_and_exit"]) mode_field = self._field("mode") self.assertEqual(mode_field["choices"], - [("Built-in speaker", "custom"), - ("Clone from a .wav file", "clone")]) + [("CustomVoice (built-in voices)", "custom"), + ("Base (voice cloning)", "clone"), + ("VoiceDesign (design)", "design")]) speaker_field = self._field("speaker") clone_field = self._field("clone") - # Speaker shows in custom mode; the .wav path shows in clone mode. + design_field = self._field("qwen_instructions") + # Speaker shows in custom mode; the .wav path in clone mode and the + # instruction in design mode. self.assertTrue(speaker_field["visible"](fields)) self.assertFalse(clone_field["visible"](fields)) + self.assertFalse(design_field["visible"](fields)) mode_field["value"] = "clone" self.assertFalse(speaker_field["visible"](fields)) self.assertTrue(clone_field["visible"](fields)) + mode_field["value"] = "design" + self.assertFalse(speaker_field["visible"](fields)) + self.assertFalse(clone_field["visible"](fields)) + self.assertTrue(design_field["visible"](fields)) - def test_qwen_clone_mode_passes_path_and_keeps_speaker(self): + def test_qwen_design_mode_passes_instructions_and_persists_model(self): with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ - patch.object(hub.config, "SPEAKER", "Vivian"): + patch.object(hub.config, "SPEAKER", "Vivian"), \ + patch.object(hub.qwen_backend.config, "QWEN_MODEL", + "CustomVoice"): + self._answer_form(backend="qwen", mode="design", + qwen_instructions="A warm narrator") + with patch.object(hub.common, "update_config_value") as mk_update: + cmd = self._convert(None, + [self._ready("qwen", "qwen-tts")]) + self.assertEqual(cmd[2]["clone"], None) + self.assertEqual(cmd[2]["instructions"], "A warm narrator") + # The model switch is persisted (CustomVoice -> VoiceDesign). + mk_update.assert_called_once_with("QWEN_MODEL", "VoiceDesign") + + def test_qwen_clone_mode_passes_path_and_persists_model(self): + with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ + patch.object(hub.config, "SPEAKER", "Vivian"), \ + patch.object(hub.qwen_backend.config, "QWEN_MODEL", + "CustomVoice"): self._answer_form(backend="qwen", mode="clone", speaker="Vivian", clone="/tmp/ref.wav") with patch.object(hub.common, "update_config_value") as mk_update: @@ -1429,6 +1458,22 @@ class ConvertFlowTests(unittest.TestCase): [self._ready("qwen", "qwen-tts")]) self.assertEqual(cmd[2]["clone"], "/tmp/ref.wav") # Clone mode does not touch the global speaker. + keys = [c.args[0] for c in mk_update.call_args_list] + self.assertNotIn("SPEAKER", keys) + # ...but remembers the switch to the Base model. + self.assertEqual(keys, ["QWEN_MODEL"]) + self.assertEqual(mk_update.call_args.args[1], "Base") + + def test_qwen_same_model_run_persists_nothing_new(self): + with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ + patch.object(hub.config, "SPEAKER", "Vivian"), \ + patch.object(hub.qwen_backend.config, "QWEN_MODEL", + "CustomVoice"): + self._answer_form(backend="qwen", mode="custom", speaker="Vivian") + with patch.object(hub.common, "update_config_value") as mk_update: + cmd = self._convert(None, + [self._ready("qwen", "qwen-tts")]) + self.assertIsNotNone(cmd) mk_update.assert_not_called() # ------------------------------------------------------------------ @@ -1513,10 +1558,10 @@ class ConvertFlowTests(unittest.TestCase): def test_qwen_remote_limited_modes_and_api_url(self): # A remote qwen with only the Base (clone) demo answering: the form - # offers only clone mode and targets the clone remote URL. + # offers only the Base model and targets the single remote URL. st = self._remote( "qwen", "qwen-tts", - remote_urls={"qwen-clone": "http://10.0.0.5:7861"}, + remote_urls={"qwen": "http://10.0.0.5:7861"}, remote_models=["Base"]) with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ patch.object(hub.config, "SPEAKER", "Vivian"): @@ -1527,7 +1572,7 @@ class ConvertFlowTests(unittest.TestCase): self.assertEqual(cmd[2]["clone"], "/tmp/ref.wav") self.assertEqual(cmd[2]["api_url"], "http://10.0.0.5:7861") self.assertEqual(self._field("mode")["choices"], - [("Clone from a .wav file", "clone")]) + [("Base (voice cloning)", "clone")]) # ------------------------------------------------------------------ # multiple backends: the Backend picker gates which options show @@ -1560,6 +1605,7 @@ class ConvertFlowTests(unittest.TestCase): [f["key"] for f in fields], ["backend", "model_id", "audiocpp_voice", "instructions", "request_options", "mode", "speaker", "clone", + "qwen_instructions", "output_format", "language", "speed", "single_file", "debug", "stop_and_exit"]) # The form opens on the configured default (audio.cpp): its fields @@ -1572,7 +1618,7 @@ class ConvertFlowTests(unittest.TestCase): self.assertFalse(self._field("request_options")["visible"](fields)) # Language shows for every backend except faster entries. self.assertTrue(self._field("language")["visible"](fields)) - for key in ("mode", "speaker", "clone"): + for key in ("mode", "speaker", "clone", "qwen_instructions"): self.assertFalse(self._field(key)["visible"](fields)) # Picking qwen in the Backend field swaps which options show. fields[0]["value"] = "qwen" @@ -1591,26 +1637,23 @@ class ConvertFlowTests(unittest.TestCase): fields[0]["value"] = "audiocpp" for key in ("model_id", "audiocpp_voice"): self.assertTrue(self._field(key)["visible"](fields)) - for key in ("mode", "speaker", "clone"): + for key in ("mode", "speaker", "clone", "qwen_instructions"): self.assertFalse(self._field(key)["visible"](fields)) class SelectSpecTests(unittest.TestCase): - """_select_spec: mode-aware server selection (qwen has two servers).""" + """_select_spec: single-server selection (qwen hosts one model at a time).""" def _qwen_status(self): return BackendStatus( "qwen", "qwen-tts", installed=True, configured=True, - servers=[ServerSpec("qwen-custom", "http://127.0.0.1:7860", []), - ServerSpec("qwen-clone", "http://127.0.0.1:7861", [])]) + servers=[ServerSpec("qwen", "http://127.0.0.1:7860", [])]) - def test_qwen_custom_mode(self): + def test_qwen_returns_the_single_spec(self): spec = hub._select_spec(self._qwen_status(), {"clone": None}) - self.assertEqual(spec.name, "qwen-custom") - - def test_qwen_clone_mode(self): + self.assertEqual(spec.name, "qwen") spec = hub._select_spec(self._qwen_status(), {"clone": "ref.wav"}) - self.assertEqual(spec.name, "qwen-clone") + self.assertEqual(spec.name, "qwen") def test_audiocpp_returns_single_spec(self): st = BackendStatus("audiocpp", "audio.cpp", installed=True, @@ -1628,7 +1671,7 @@ class SelectSpecTests(unittest.TestCase): class PrepareRunConfigTests(unittest.TestCase): """_prepare_run_config: the run view's inputs from the accepted form.""" - def _spec(self, name="qwen-custom", url="http://127.0.0.1:7860"): + def _spec(self, name="qwen", url="http://127.0.0.1:7860"): return ServerSpec(name, url, ["x"]) def test_remote_targets_the_api_url(self): @@ -1644,14 +1687,32 @@ class PrepareRunConfigTests(unittest.TestCase): def test_autostart_sets_the_spec_and_pops_the_flag(self): spec = self._spec() - kwargs = {"autostart": "qwen-custom"} + kwargs = {"autostart": "qwen"} with patch.object(hub, "detect_all", return_value=[]), \ patch.object(hub, "_find_spec", return_value=spec): cfg = hub._prepare_run_config("qwen", kwargs) self.assertIs(cfg.autostart_spec, spec) - self.assertEqual(cfg.server_name, "qwen-custom") + self.assertFalse(cfg.restart_first) + self.assertEqual(cfg.server_name, "qwen") self.assertNotIn("autostart", kwargs) + def test_restart_first_stops_and_boots_before_converting(self): + # A running managed server hosting another model than this run + # selected: the recorded spec boots again after a stop. + spec = self._spec() + status = BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, servers=[spec]) + kwargs = {"restart_server": "qwen"} + 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): + cfg = hub._prepare_run_config("qwen", kwargs) + self.assertIs(cfg.autostart_spec, spec) + self.assertTrue(cfg.restart_first) + self.assertNotIn("restart_server", kwargs) + self.assertEqual(cfg.server_url, spec.url) + def test_stop_and_exit_travels_on_the_config_not_the_kwargs(self): # The run-view toggle is not a converter kwarg: it moves onto the # config (and defaults to off when the form did not send it). @@ -1846,7 +1907,7 @@ class AddAutostartTests(unittest.TestCase): """_add_autostart: always starts the server when it isn't running.""" def _status(self): - spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) + spec = ServerSpec("qwen", "http://127.0.0.1:7860", ["x"]) return BackendStatus("qwen", "qwen-tts", installed=True, configured=True, running=False, servers=[spec]) @@ -1855,15 +1916,43 @@ class AddAutostartTests(unittest.TestCase): cmd = ("convert", "qwen", {"clone": None}) with patch.object(hub, "detect_all", return_value=[self._status()]), \ patch("backends.common.server_running", return_value=False): - hub._add_autostart(cmd, [self._status()]) - self.assertEqual(cmd[2]["autostart"], "qwen-custom") + self.assertIsNone(hub._add_autostart(cmd, [self._status()])) + self.assertEqual(cmd[2]["autostart"], "qwen") - def test_no_autostart_when_server_already_running(self): + def test_no_autostart_when_server_already_running_same_model(self): cmd = ("convert", "qwen", {"clone": None}) with patch.object(hub, "detect_all", return_value=[self._status()]), \ - patch("backends.common.server_running", return_value=True): - hub._add_autostart(cmd, [self._status()]) + patch("backends.common.server_running", return_value=True), \ + patch.object(hub.backend_probe, "identify_server", + return_value="qwen-custom"): + self.assertIsNone(hub._add_autostart(cmd, [self._status()])) + self.assertNotIn("autostart", cmd[2]) + + def test_running_server_hosting_another_model_is_restarted(self): + # The managed qwen server hosts CustomVoice but the run selected + # the Base (clone) model: restart (stop + boot) before converting. + cmd = ("convert", "qwen", {"clone": "/tmp/ref.wav"}) + with patch.object(hub, "detect_all", return_value=[self._status()]), \ + patch("backends.common.server_running", return_value=True), \ + patch.object(hub.servers, "alive", return_value=True), \ + patch.object(hub.backend_probe, "identify_server", + return_value="qwen-custom"): + self.assertIsNone(hub._add_autostart(cmd, [self._status()])) + self.assertEqual(cmd[2]["restart_server"], "qwen") + + 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()]), \ + patch("backends.common.server_running", return_value=True), \ + patch.object(hub.servers, "alive", return_value=False), \ + patch.object(hub.backend_probe, "identify_server", + return_value="qwen-custom"): + message = hub._add_autostart(cmd, [self._status()]) + self.assertIsNotNone(message) + self.assertIn("CustomVoice", message) + self.assertIn("Base", message) self.assertNotIn("autostart", cmd[2]) + self.assertNotIn("restart_server", cmd[2]) def test_no_autostart_for_remote_conversion(self): # A remote conversion (api_url set) never autostarts: the server is @@ -1881,9 +1970,9 @@ class SettingsTests(unittest.TestCase): _SETTING_KEYS = ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "STOP_SERVER_AND_EXIT", "AUDIOCPP_UNLOAD_MODELS", - "QWEN_API_URL", "CLONE_API_URL", + "QWEN_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL", - "QWEN_REMOTE_URL", "CLONE_REMOTE_URL", + "QWEN_REMOTE_URL", "FASTER_REMOTE_URL", "AUDIOCPP_REMOTE_URL") def _snapshot_settings(self): @@ -1946,12 +2035,11 @@ class SettingsTests(unittest.TestCase): "language": "en", "chunk_size": "300", "stop_and_exit": False, "unload_models": True, - "qwen_custom_port": "7862", "qwen_clone_port": "7863", + "qwen_port": "7862", "faster_port": "8001", "audiocpp_port": "8081", "audiocpp_remote_url": "10.0.0.5:8080", "faster_remote_url": "http://10.0.0.6:8000", - "qwen_custom_remote_url": "", - "qwen_clone_remote_url": ""} + "qwen_remote_url": ""} with patch.object(hub.common, "update_config_value", fake_update), \ patch.object(hub, "_sync_audiocpp_server_port"): @@ -1965,12 +2053,10 @@ class SettingsTests(unittest.TestCase): "STOP_SERVER_AND_EXIT": False, "AUDIOCPP_UNLOAD_MODELS": True, "QWEN_API_URL": "http://127.0.0.1:7862", - "CLONE_API_URL": "http://127.0.0.1:7863", "FASTER_API_URL": "http://127.0.0.1:8001", "AUDIOCPP_API_URL": "http://127.0.0.1:8081", "QWEN_REMOTE_URL": "", - "CLONE_REMOTE_URL": "", "FASTER_REMOTE_URL": "http://10.0.0.6:8000", "AUDIOCPP_REMOTE_URL": @@ -1993,7 +2079,7 @@ class SettingsTests(unittest.TestCase): "language": "English", "chunk_size": "250", "stop_and_exit": True, "unload_models": True, - "qwen_custom_port": "7860", "qwen_clone_port": "7861", + "qwen_port": "7860", "faster_port": "8000", "audiocpp_port": "8080"} with patch.object(hub.common, "update_config_value") as mk_update: with self.assertRaises(ValueError): @@ -2032,7 +2118,7 @@ class SettingsTests(unittest.TestCase): "language": "English", "chunk_size": "300", "stop_and_exit": True, "unload_models": True, - "qwen_custom_port": "7860", "qwen_clone_port": "7861", + "qwen_port": "7860", "faster_port": "8000", "audiocpp_port": "8080"} applied = [] @@ -2051,10 +2137,8 @@ class SettingsTests(unittest.TestCase): ["audio_format", "audio_bitrate", "language", "chunk_size", "stop_and_exit", "unload_models", "audiocpp_port", - "faster_port", "qwen_custom_port", - "qwen_clone_port", "audiocpp_remote_url", - "faster_remote_url", "qwen_custom_remote_url", - "qwen_clone_remote_url"]) + "faster_port", "qwen_port", "audiocpp_remote_url", + "faster_remote_url", "qwen_remote_url"]) kinds = {f["key"]: f["kind"] for f in captured["fields"]} self.assertEqual(kinds["audio_format"], "choice") self.assertEqual(kinds["audio_bitrate"], "text") @@ -2063,7 +2147,7 @@ class SettingsTests(unittest.TestCase): self.assertEqual(kinds["unload_models"], "bool") self.assertEqual(kinds["audiocpp_remote_url"], "text") labels = {f["key"]: f["label"] for f in captured["fields"]} - self.assertEqual(labels["qwen_clone_port"], "qwen-tts Base port") + self.assertEqual(labels["qwen_port"], "qwen-tts port") self.assertEqual(labels["audiocpp_remote_url"], "audio.cpp remote URL") self.assertNotIn("(clone)", " ".join(labels.values())) @@ -2073,15 +2157,14 @@ class SettingsTests(unittest.TestCase): self.assertTrue(notes["audiocpp_port"]) self.assertTrue(notes["audiocpp_remote_url"]) self.assertIsNone(notes["audio_format"]) - self.assertIsNone(notes["qwen_custom_port"]) + self.assertIsNone(notes["qwen_port"]) self.assertEqual(applied, [{"audio_format": "ogg", "audio_bitrate": "192k", "language": "English", "chunk_size": "300", "stop_and_exit": True, "unload_models": True, - "qwen_custom_port": "7860", - "qwen_clone_port": "7861", + "qwen_port": "7860", "faster_port": "8000", "audiocpp_port": "8080"}]) # Saving is silent: no confirmation flash either way. @@ -2193,7 +2276,7 @@ class SettingsTests(unittest.TestCase): return {"audio_format": "m4b", "audio_bitrate": "128k", "language": "English", "chunk_size": "250", "stop_and_exit": True, "unload_models": True, - "qwen_custom_port": "7860", "qwen_clone_port": "7861", + "qwen_port": "7860", "faster_port": "8000", "audiocpp_port": "8080"} applied = [] @@ -2225,9 +2308,9 @@ class SettingsTests(unittest.TestCase): ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "STOP_SERVER_AND_EXIT", "AUDIOCPP_UNLOAD_MODELS", - "QWEN_API_URL", "CLONE_API_URL", + "QWEN_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL", - "QWEN_REMOTE_URL", "CLONE_REMOTE_URL", + "QWEN_REMOTE_URL", "FASTER_REMOTE_URL", "AUDIOCPP_REMOTE_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) @@ -2244,11 +2327,9 @@ class SettingsTests(unittest.TestCase): "STOP_SERVER_AND_EXIT = True\n" "AUDIOCPP_UNLOAD_MODELS = True\n" 'QWEN_API_URL = "http://127.0.0.1:7860"\n' - 'CLONE_API_URL = "http://127.0.0.1:7861"\n' 'FASTER_API_URL = "http://127.0.0.1:8000"\n' 'AUDIOCPP_API_URL = "http://127.0.0.1:8080"\n' 'QWEN_REMOTE_URL = "http://127.0.0.1:7860"\n' - 'CLONE_REMOTE_URL = "http://127.0.0.1:7861"\n' 'FASTER_REMOTE_URL = "http://127.0.0.1:8000"\n' 'AUDIOCPP_REMOTE_URL = "http://127.0.0.1:8080"\n', encoding="utf-8") @@ -2270,7 +2351,7 @@ class SettingsTests(unittest.TestCase): def test_settings_menu_updates_backend_ports(self): import tempfile original = {name: getattr(hub.config, name) for name in - ("QWEN_API_URL", "CLONE_API_URL", + ("QWEN_API_URL", "FASTER_API_URL", "AUDIOCPP_API_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) @@ -2278,20 +2359,17 @@ class SettingsTests(unittest.TestCase): path = Path(td) / "config.py" path.write_text( 'QWEN_API_URL = "http://127.0.0.1:7860"\n' - 'CLONE_API_URL = "http://127.0.0.1:7861"\n' 'FASTER_API_URL = "http://127.0.0.1:8000"\n' 'AUDIOCPP_API_URL = "http://127.0.0.1:8080"\n', encoding="utf-8") self._snapshot_settings() for key, value in ( ("QWEN_API_URL", "http://127.0.0.1:7862"), - ("CLONE_API_URL", "http://127.0.0.1:7863"), ("FASTER_API_URL", "http://127.0.0.1:8001"), ("AUDIOCPP_API_URL", "http://127.0.0.1:8081")): hub.common.update_config_value(key, value, config_path=path) text = path.read_text(encoding="utf-8") self.assertIn('QWEN_API_URL = "http://127.0.0.1:7862"', text) - self.assertIn('CLONE_API_URL = "http://127.0.0.1:7863"', text) self.assertIn('FASTER_API_URL = "http://127.0.0.1:8001"', text) self.assertIn('AUDIOCPP_API_URL = "http://127.0.0.1:8081"', text) |