diff options
Diffstat (limited to 'app/tests')
| -rw-r--r-- | app/tests/test_backends.py | 105 | ||||
| -rw-r--r-- | app/tests/test_backends_probe.py | 104 | ||||
| -rw-r--r-- | app/tests/test_hub.py | 273 | ||||
| -rw-r--r-- | app/tests/test_tts.py | 51 |
4 files changed, 456 insertions, 77 deletions
diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py index ac4c8ef..2cb5a95 100644 --- a/app/tests/test_backends.py +++ b/app/tests/test_backends.py @@ -73,15 +73,18 @@ class DetectAllTests(unittest.TestCase): self.assertFalse(status.running) self.assertIn("audiocpp_server", status.launch_hint) - def test_audiocpp_running_when_server_probe_succeeds(self): + def test_audiocpp_running_when_remote_server_identified(self): from backends import audiocpp with patch.object(audiocpp, "find_local_checkout", return_value=None), \ - patch("backends.common.server_running", return_value=True): + patch.object(audiocpp.probe, "identify_server", + return_value="audiocpp"): status = audiocpp.detect() - # Not installed (no checkout) but an external server is up. + # Not installed (no checkout) but a remote server answers. self.assertFalse(status.installed) self.assertTrue(status.running) + self.assertTrue(status.remote) + self.assertIn("audiocpp", status.remote_urls) def test_qwen_status_reflects_install(self): from backends import qwen @@ -98,33 +101,37 @@ class DetectAllTests(unittest.TestCase): self.assertFalse(status.installed) self.assertFalse(status.configured) - def test_qwen_running_when_either_port_is_up(self): - # Either the CustomVoice port or the Base port counts as running, - # and the status names which model answered. Probes: CustomVoice - # (QWEN_API_URL) first, then Base (CLONE_API_URL). + def test_qwen_running_when_either_remote_url_is_up(self): + # Either the CustomVoice or the Base remote URL answering counts as + # running, and the status names which model answered. Probes: Base + # (CLONE_REMOTE_URL) first, then CustomVoice (QWEN_REMOTE_URL). from backends import qwen with patch.object(qwen, "_is_installed", return_value=False), \ - patch("backends.common.server_running", - side_effect=[True, False]): + patch.object(qwen.probe, "identify_server", + side_effect=[None, "qwen-custom"]): status = qwen.detect() self.assertTrue(status.running) + self.assertTrue(status.remote) + self.assertEqual(status.remote_models, ["CustomVoice"]) self.assertEqual(status.running_models, ["CustomVoice"]) with patch.object(qwen, "_is_installed", return_value=False), \ - patch("backends.common.server_running", - side_effect=[False, True]): + patch.object(qwen.probe, "identify_server", + side_effect=["qwen-clone", None]): status = qwen.detect() self.assertTrue(status.running) + self.assertEqual(status.remote_models, ["Base"]) self.assertEqual(status.running_models, ["Base"]) def test_qwen_running_models_names_both_ports(self): - # Both ports up → both models, Base first (the hub renders + # Both remote URLs up → both models, Base first (the hub renders # "running (Base, CustomVoice)"). from backends import qwen with patch.object(qwen, "_is_installed", return_value=False), \ - patch("backends.common.server_running", - side_effect=[True, True]): + patch.object(qwen.probe, "identify_server", + side_effect=["qwen-clone", "qwen-custom"]): status = qwen.detect() self.assertTrue(status.running) + self.assertEqual(status.remote_models, ["Base", "CustomVoice"]) self.assertEqual(status.running_models, ["Base", "CustomVoice"]) def test_qwen_detect_marks_our_server_as_managed(self): @@ -169,13 +176,16 @@ class DetectAllTests(unittest.TestCase): self.assertFalse(status.running) self.assertIn("openai_server.py", status.launch_hint) - def test_faster_running_when_server_probe_succeeds(self): + def test_faster_running_when_remote_server_identified(self): from backends import faster with patch.object(faster, "_is_installed", return_value=False), \ patch.object(faster, "_is_cloned", return_value=False), \ - patch("backends.common.server_running", return_value=True): + patch.object(faster.probe, "identify_server", + return_value="faster"): status = faster.detect() self.assertTrue(status.running) + self.assertTrue(status.remote) + self.assertIn("faster", status.remote_urls) class ServerRunningTests(unittest.TestCase): @@ -212,5 +222,68 @@ class ServerRunningTests(unittest.TestCase): self.assertFalse(common.server_running("")) +class RemoteUrlTests(unittest.TestCase): + """backends.common.normalize_remote_url: host:port / URL -> http(s)://.""" + + def test_bare_host_port_gets_http_scheme(self): + from backends import common + self.assertEqual(common.normalize_remote_url("10.0.0.5:8080"), + "http://10.0.0.5:8080") + + def test_full_url_preserved(self): + from backends import common + self.assertEqual(common.normalize_remote_url( + "https://10.0.0.5:8443/path"), "https://10.0.0.5:8443/path") + + def test_empty_means_disabled(self): + from backends import common + self.assertEqual(common.normalize_remote_url(""), "") + self.assertEqual(common.normalize_remote_url(" "), "") + + def test_whitespace_stripped(self): + from backends import common + self.assertEqual(common.normalize_remote_url(" 10.0.0.5:8080 "), + "http://10.0.0.5:8080") + + def test_invalid_rejected(self): + from backends import common + for value in ("http://", "not a url", "10.0.0.5:notaport", "://"): + with self.assertRaises(ValueError, msg=value): + common.normalize_remote_url(value) + + +class RemoteSuppressionTests(unittest.TestCase): + """A server this tool started must not also be reported as remote.""" + + def test_audiocpp_own_server_suppresses_remote(self): + from backends import audiocpp + from backends import servers as servers_mod + with tempfile.TemporaryDirectory() as td: + root = Path(td) + checkout = root / "audio.cpp" + checkout.mkdir() + (checkout / "model_specs").mkdir() + (checkout / "build" / "linux-cuda-release" / "bin").mkdir( + parents=True) + (checkout / "build" / "linux-cuda-release" / "bin" + / "audiocpp_server").write_bytes(b"x") + (checkout / "server.json").write_text('{"models":[]}', + encoding="utf-8") + (Path(td) / "audiocpp-server.pid").write_text( + "4242", encoding="utf-8") + with patch.object(audiocpp, "find_local_checkout", + return_value=checkout), \ + patch.object(servers_mod, "LOG_DIR", Path(td)), \ + patch.object(servers_mod, "_pid_alive", + return_value=True), \ + patch.object(audiocpp.probe, "identify_server", + return_value="audiocpp"): + status = audiocpp.detect() + self.assertTrue(status.managed) + self.assertTrue(status.running) + self.assertFalse(status.remote) + self.assertEqual(status.remote_urls, {}) + + if __name__ == "__main__": unittest.main() diff --git a/app/tests/test_backends_probe.py b/app/tests/test_backends_probe.py new file mode 100644 index 0000000..08f9fd9 --- /dev/null +++ b/app/tests/test_backends_probe.py @@ -0,0 +1,104 @@ +"""Tests for backends.probe: identifying which backend answers at a URL.""" + +import json +import unittest +from unittest.mock import patch + +from backends import probe + + +class _FakeResponse: + def __init__(self, payload): + self._payload = payload + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self): + return json.dumps(self._payload).encode("utf-8") + + +class IdentifyServerTests(unittest.TestCase): + def _patch_http(self, routes): + """routes: URL path -> JSON payload dict (missing = HTTP error).""" + import urllib.parse + + def fake_urlopen(url, timeout=None): + path = urllib.parse.urlsplit(url).path + payload = routes.get(path) + if payload is None: + raise OSError("HTTP 404") + return _FakeResponse(payload) + + return patch.object(probe.urllib.request, "urlopen", + side_effect=fake_urlopen) + + def test_audiocpp_identified(self): + routes = {"/health": {"status": "ok"}, + "/v1/models": {"data": [{"id": "qwen", "family": "qwen3_tts", + "task": "tts"}]}} + with patch.object(probe.common, "server_running", return_value=True), \ + self._patch_http(routes): + self.assertEqual(probe.identify_server("http://127.0.0.1:8080"), + "audiocpp") + + def test_faster_identified(self): + with patch.object(probe.common, "server_running", return_value=True), \ + self._patch_http({"/health": {"model_loaded": True}}): + self.assertEqual(probe.identify_server("http://127.0.0.1:8000"), + "faster") + + def test_qwen_custom_and_clone_identified(self): + with patch.object(probe.common, "server_running", return_value=True), \ + self._patch_http({"/info": {"named_endpoints": + {"/run_instruct": {}}}}): + self.assertEqual(probe.identify_server("http://x:7860"), + "qwen-custom") + with patch.object(probe.common, "server_running", return_value=True), \ + self._patch_http({"/info": {"named_endpoints": + {"/run_voice_clone": {}}}}): + self.assertEqual(probe.identify_server("http://x:7861"), + "qwen-clone") + + def test_unreachable_returns_none(self): + with patch.object(probe.common, "server_running", return_value=False): + self.assertIsNone(probe.identify_server("http://127.0.0.1:8080")) + + def test_health_ok_without_models_catalog_is_not_audiocpp(self): + # A service that answers {"status": "ok"} but not /v1/models is not + # recognized as audio.cpp. + with patch.object(probe.common, "server_running", return_value=True), \ + self._patch_http({"/health": {"status": "ok"}}): + self.assertIsNone(probe.identify_server("http://x")) + + def test_empty_url_returns_none(self): + self.assertIsNone(probe.identify_server("")) + self.assertIsNone(probe.identify_server(None)) + + +class SameEndpointTests(unittest.TestCase): + def test_same_host_port(self): + self.assertTrue(probe.same_endpoint("http://127.0.0.1:8080", + "http://127.0.0.1:8080/")) + + def test_scheme_ignored(self): + self.assertTrue(probe.same_endpoint("http://h:8080", "https://h:8080")) + + def test_different_port(self): + self.assertFalse(probe.same_endpoint("http://127.0.0.1:8080", + "http://127.0.0.1:8000")) + + def test_different_host(self): + self.assertFalse(probe.same_endpoint("http://127.0.0.1:8080", + "http://10.0.0.5:8080")) + + def test_empty_url(self): + self.assertFalse(probe.same_endpoint("", "http://h:8080")) + self.assertFalse(probe.same_endpoint("http://h:8080", "")) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index f40b72d..725f4f6 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -75,29 +75,33 @@ class HubHelperTests(unittest.TestCase): def test_status_mark(self): from backends import BackendStatus - running = BackendStatus("k", "l", installed=True, configured=True, - running=True, managed=True) + local = BackendStatus("k", "l", installed=True, configured=True, + running=True, managed=True) remote = BackendStatus("k", "l", installed=True, configured=True, - running=True) + running=True, remote=True) + both = BackendStatus("k", "l", installed=True, configured=True, + running=True, managed=True, remote=True) models = BackendStatus("k", "l", installed=True, configured=True, running=True, managed=True, running_models=["Base", "CustomVoice"]) remote_models = BackendStatus("k", "l", installed=True, configured=True, running=True, - running_models=["Base"]) + remote=True, running_models=["Base"]) installed = BackendStatus("k", "l", installed=True, configured=False) none = BackendStatus("k", "l", installed=False, configured=False) # running beats installed (a server is up even if not configured); # only a backend that is neither installed nor running is dimmed. - self.assertEqual(hub._status_mark(running), - ("running", "ok", "body")) - # A server without a live recorded pid was started externally. + self.assertEqual(hub._status_mark(local), + ("running [local]", "ok", "body")) + # A server this tool did not start, found at the remote URL. self.assertEqual(hub._status_mark(remote), ("running [remote]", "ok", "body")) + self.assertEqual(hub._status_mark(both), + ("running [local, remote]", "ok", "body")) # Multi-model backends name the models that answered. self.assertEqual(hub._status_mark(models), - ("running (Base, CustomVoice)", "ok", "body")) + ("running [local] (Base, CustomVoice)", "ok", "body")) self.assertEqual(hub._status_mark(remote_models), ("running [remote] (Base)", "ok", "body")) self.assertEqual(hub._status_mark(installed), @@ -183,12 +187,13 @@ class HubMenuTests(unittest.TestCase): dead = self._none_status("audiocpp", "audio.cpp") external = self._none_status("qwen", "qwen-tts") external.running = True + external.remote = True with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[dead, external]): hub._hub_menu(screen) # Unusable backend: dim name. Running-but-not-installed stays - # bright and is tagged remote (no pid file → not started by us). + # bright and is tagged remote (found at its remote URL). self.assertEqual( captured["rows"], [("audio.cpp", "unavailable", "err", "dim"), @@ -308,7 +313,7 @@ class SubmenuStatusTableTests(unittest.TestCase): BackendStatus("audiocpp", "audio.cpp", installed=True, configured=True), BackendStatus("qwen", "qwen-tts", installed=False, - configured=False, running=True), + configured=False, running=True, remote=True), ] with patch.object(hub, "REGISTRY", infos), \ patch.object(hub.tui, "menu", @@ -468,35 +473,45 @@ class ConvertFlowTests(unittest.TestCase): # ------------------------------------------------------------------ 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) + """Fetch helpers return MODELS/VOICES for a remote audio.cpp server.""" 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): + for patcher in (fetched_models, fetched_voices): patcher.start() self.addCleanup(patcher.stop) + def _remote(self, key, label, spec_name=None, url=None, + remote_urls=None, remote_models=None): + """A running remote backend status (not installed on this machine).""" + if remote_urls is None: + remote_urls = {spec_name or key: + url or f"http://{key}.local:8080"} + return BackendStatus(key, label, installed=False, configured=False, + running=True, remote=True, + remote_urls=remote_urls, + remote_models=list(remote_models or [])) + def test_audiocpp_remote_builds_one_form(self): self._patch_remote( [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): - self._answer_form(backend="audiocpp", model_id="higgs", + self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="", speed="1.5") - cmd = hub._convert_menu(None, - [self._ready("audiocpp", "audio.cpp")]) + cmd = hub._convert_menu( + None, [self._remote("audiocpp", "audio.cpp")]) 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"]) + self.assertEqual(kwargs["api_url"], "http://audiocpp.local:8080") self.assertEqual(kwargs["output_format"], "m4b") self.assertEqual(kwargs["speed"], 1.5) self.assertFalse(kwargs["single_file"]) @@ -511,8 +526,9 @@ class ConvertFlowTests(unittest.TestCase): "single_file", "debug"]) self.assertEqual(form_kwargs["buttons"], ("Generate!", "Cancel")) self.assertTrue(form_kwargs["start_on_buttons"]) - # The backend field offers the ready backend. - self.assertEqual(fields[0]["choices"], [("audio.cpp", "audiocpp")]) + # The backend field offers the remote entry under a [remote] label. + self.assertEqual(fields[0]["choices"], + [("audio.cpp [remote]", "audiocpp-remote")]) # The model menu was fed from the live query (label, id). self.assertEqual(self._field("model_id")["choices"], [("higgs (higgs_audio_tts, tts)", "higgs")]) @@ -522,11 +538,11 @@ class ConvertFlowTests(unittest.TestCase): [{"id": "qwen", "family": "qwen3_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): - self._answer_form(backend="audiocpp", model_id="qwen", + self._answer_form(backend="audiocpp-remote", model_id="qwen", audiocpp_voice="(built-in speaker)", instructions="") - cmd = hub._convert_menu(None, - [self._ready("audiocpp", "audio.cpp")]) + cmd = hub._convert_menu( + None, [self._remote("audiocpp", "audio.cpp")]) # The sentinel maps to "no voice" (built-in speaker). self.assertIsNone(cmd[2]["voice"]) fields = self.tui.forms_seen[0][1] @@ -542,11 +558,11 @@ class ConvertFlowTests(unittest.TestCase): self._patch_remote([{"id": "legacy", "family": "", "task": ""}], voices=[]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): - self._answer_form(backend="audiocpp", model_id="legacy", + self._answer_form(backend="audiocpp-remote", model_id="legacy", audiocpp_voice="(built-in speaker)", instructions="") - cmd = hub._convert_menu(None, - [self._ready("audiocpp", "audio.cpp")]) + cmd = hub._convert_menu( + None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNotNone(cmd) self.assertIsNone(cmd[2]["voice"]) @@ -554,11 +570,11 @@ class ConvertFlowTests(unittest.TestCase): self._patch_remote( [{"id": "design", "family": "qwen3_tts", "task": "vdes"}]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): - self._answer_form(backend="audiocpp", model_id="design", + self._answer_form(backend="audiocpp-remote", model_id="design", audiocpp_voice=None, instructions="A warm British narrator") - cmd = hub._convert_menu(None, - [self._ready("audiocpp", "audio.cpp")]) + cmd = hub._convert_menu( + None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNone(cmd[2]["voice"]) self.assertEqual(cmd[2]["instructions"], "A warm British narrator") fields = self.tui.forms_seen[0][1] @@ -574,25 +590,25 @@ class ConvertFlowTests(unittest.TestCase): [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=["narrator"]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): - self._answer_form(backend="audiocpp", model_id="higgs", + self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="") hub._convert_menu(None, - [self._ready("audiocpp", "audio.cpp")]) + [self._remote("audiocpp", "audio.cpp")]) voice_field = self._field("audiocpp_voice") self.assertIsNotNone(voice_field["validate"]("")) self.assertIsNone(voice_field["validate"]("narrator")) def test_audiocpp_remote_unreachable_models_flash_and_abort(self): self._patch_remote(None) # endpoint did not answer valid JSON - cmd = hub._convert_menu(None, - [self._ready("audiocpp", "audio.cpp")]) + cmd = hub._convert_menu( + None, [self._remote("audiocpp", "audio.cpp")]) 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_menu(None, - [self._ready("audiocpp", "audio.cpp")]) + cmd = hub._convert_menu( + None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNone(cmd) self.assertIn("hosts no model entries", self.tui.flashes[0]) @@ -603,10 +619,10 @@ class ConvertFlowTests(unittest.TestCase): [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], voices=[]) with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): - self._answer_form(backend="audiocpp", model_id="higgs", + self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="", instructions="") - cmd = hub._convert_menu(None, - [self._ready("audiocpp", "audio.cpp")]) + cmd = hub._convert_menu( + None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNotNone(cmd) self.assertIsNone(cmd[2]["voice"]) fields = self.tui.forms_seen[0][1] @@ -646,6 +662,50 @@ class ConvertFlowTests(unittest.TestCase): self.assertEqual(cmd[2]["model_id"], "qwen") self.assertEqual(cmd[2]["voice"], "Narrator") + def test_managed_and_remote_both_offered(self): + # A ready managed audio.cpp (server.json) AND a running remote + # audio.cpp: both entries appear. The managed entry reads server.json + # (no api_url), the remote entry live-queries (api_url set). + self._patch_remote( + [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], + voices=["narrator"]) + 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, "AUDIOCPP_INSTRUCTIONS", ""): + self._answer_form(backend="audiocpp", model_id="qwen", + audiocpp_voice="(built-in speaker)", + instructions="") + cmd = hub._convert_menu(None, [ + self._ready("audiocpp", "audio.cpp"), + self._remote("audiocpp", "audio.cpp")]) + self.assertEqual(cmd[0], "convert") + self.assertEqual(cmd[1], hub.BACKEND_AUDIOCPP) + # Managed entry: no api_url override (uses the configured local URL). + self.assertNotIn("api_url", cmd[2]) + self.assertEqual(cmd[2]["model_id"], "qwen") + fields = self.tui.forms_seen[0][1] + self.assertEqual(fields[0]["choices"], + [("audio.cpp", "audiocpp"), + ("audio.cpp [remote]", "audiocpp-remote")]) + + def test_audiocpp_remote_mapper_adds_api_url(self): + self._patch_remote( + [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], + voices=["narrator"]) + with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): + self._answer_form(backend="audiocpp-remote", model_id="higgs", + audiocpp_voice="narrator", instructions="") + cmd = hub._convert_menu( + None, [self._remote("audiocpp", "audio.cpp", + url="http://10.0.0.5:8080")]) + self.assertEqual(cmd[2]["api_url"], "http://10.0.0.5:8080") + # ------------------------------------------------------------------ # common fields: output format, speed, single-file, debug # ------------------------------------------------------------------ @@ -740,6 +800,36 @@ class ConvertFlowTests(unittest.TestCase): self.assertEqual(voice_field["choices"], [("default", "default"), ("obama", "obama")]) + def test_faster_remote_uses_text_voice_and_api_url(self): + st = self._remote("faster", "faster-qwen3-tts", + url="http://10.0.0.5:8000") + self._answer_form(backend="faster-remote", faster_voice="obama") + cmd = hub._convert_menu(None, [st]) + self.assertEqual(cmd[0], "convert") + self.assertEqual(cmd[1], "faster") + self.assertEqual(cmd[2]["voice"], "obama") + self.assertEqual(cmd[2]["api_url"], "http://10.0.0.5:8000") + self.assertEqual(self._field("faster_voice")["kind"], "text") + + 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. + st = self._remote( + "qwen", "qwen-tts", + remote_urls={"qwen-clone": "http://10.0.0.5:7861"}, + remote_models=["Base"]) + with patch.object(hub.qwen_backend, "QWEN_SPEAKERS", ["Vivian"]), \ + patch.object(hub.config, "SPEAKER", "Vivian"): + self._answer_form(backend="qwen-remote", mode="clone", + speaker="Vivian", clone="/tmp/ref.wav") + cmd = hub._convert_menu(None, [st]) + self.assertEqual(cmd[1], hub.BACKEND_QWEN) + self.assertEqual(cmd[2]["clone"], "/tmp/ref.wav") + self.assertEqual(cmd[2]["api_url"], "http://10.0.0.5:7861") + fields = self.tui.forms_seen[0][1] + self.assertEqual(self._field("mode")["choices"], + [("Clone from a .wav file", "clone")]) + # ------------------------------------------------------------------ # multiple backends: the Backend picker gates which options show # ------------------------------------------------------------------ @@ -747,15 +837,20 @@ class ConvertFlowTests(unittest.TestCase): def test_multiple_backends_gate_options_on_backend_value(self): # Two ready backends: the form leads with a Backend picker and the # per-backend fields are hidden/shown by its value. - self._patch_remote( - [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}], - voices=["narrator"]) - with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): - self._answer_form(backend="qwen", mode="custom", speaker="Vivian", - clone="") - cmd = hub._convert_menu(None, [ - self._ready("audiocpp", "audio.cpp"), - self._ready("qwen", "qwen-tts")]) + with tempfile.TemporaryDirectory() as td: + root = Path(td) + (root / "server.json").write_text(json.dumps({ + "models": [{"id": "higgs", "family": "higgs_audio_tts", + "task": "tts"}], + }), encoding="utf-8") + with patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=root), \ + patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): + self._answer_form(backend="qwen", mode="custom", + speaker="Vivian", clone="") + cmd = hub._convert_menu(None, [ + self._ready("audiocpp", "audio.cpp"), + self._ready("qwen", "qwen-tts")]) self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], hub.BACKEND_QWEN) fields = self.tui.forms_seen[0][1] @@ -882,6 +977,33 @@ class RunConversionTests(unittest.TestCase): hub._run_conversion("qwen", {}) mk_conv.assert_called_once() + def test_remote_conversion_skips_setup_checks(self): + # A remote conversion targets an external server: no autostart, no + # "not fully set up" warning, no launch hint — just convert. + with patch.object(hub, "detect_all", return_value=[]) as mk_detect, \ + patch.object(hub.audiobook, "convert", return_value=0) as mk_conv: + hub._run_conversion("audiocpp", {"api_url": "http://10.0.0.5:8080"}) + mk_conv.assert_called_once_with( + backend="audiocpp", api_url="http://10.0.0.5:8080") + # The remote path never re-detects or touches managed-instance state. + mk_detect.assert_not_called() + + def test_managed_conversion_warns_when_port_occupied_by_other_server(self): + spec = ServerSpec("audiocpp", "http://127.0.0.1:8080", ["x"]) + status = BackendStatus("audiocpp", "audio.cpp", installed=True, + configured=True, running=False, + servers=[spec]) + with patch.object(hub, "detect_all", return_value=[status]), \ + patch.object(hub, "_select_spec", return_value=spec), \ + patch("backends.common.server_running", return_value=True), \ + patch.object(hub.servers, "alive", return_value=False), \ + patch.object(hub.audiobook, "convert", return_value=0) as mk_conv, \ + patch("builtins.print") as mk_print: + hub._run_conversion("audiocpp", {}) + mk_conv.assert_called_once() + printed = " ".join(str(call.args[0]) for call in mk_print.call_args_list) + self.assertIn("did not start", printed) + class AddAutostartTests(unittest.TestCase): """_add_autostart: always starts the server when it isn't running.""" @@ -906,6 +1028,13 @@ class AddAutostartTests(unittest.TestCase): hub._add_autostart(cmd, [self._status()]) self.assertNotIn("autostart", cmd[2]) + def test_no_autostart_for_remote_conversion(self): + # A remote conversion (api_url set) never autostarts: the server is + # external to this tool, so there is nothing to start/stop here. + cmd = ("convert", "audiocpp", {"api_url": "http://10.0.0.5:8080"}) + hub._add_autostart(cmd, []) + self.assertNotIn("autostart", cmd[2]) + class SettingsTests(unittest.TestCase): """Settings menu: field collection, validation, config.py writing.""" @@ -955,17 +1084,24 @@ class SettingsTests(unittest.TestCase): original = {name: getattr(hub.config, name) for name in ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "QWEN_API_URL", "CLONE_API_URL", - "FASTER_API_URL", "AUDIOCPP_API_URL")} + "FASTER_API_URL", "AUDIOCPP_API_URL", + "QWEN_REMOTE_URL", "CLONE_REMOTE_URL", + "FASTER_REMOTE_URL", "AUDIOCPP_REMOTE_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) values = {"audio_format": "ogg", "audio_bitrate": " 192k ", "language": "en", "chunk_size": "300", "qwen_custom_port": "7862", "qwen_clone_port": "7863", - "faster_port": "8001", "audiocpp_port": "8081"} + "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": ""} with patch.object(hub, "_write_config", fake_write), \ patch.object(hub, "_sync_audiocpp_server_port"): hub._apply_settings(values) - # Values are trimmed and language normalized to a display name. + # Values are trimmed and language normalized to a display name; + # remote URLs are normalized to full http(s) URLs (empty = off). self.assertEqual(written, {"AUDIO_FORMAT": "ogg", "AUDIO_BITRATE": "192k", "LANGUAGE": "English", @@ -974,7 +1110,13 @@ class SettingsTests(unittest.TestCase): "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"}) + "http://127.0.0.1:8081", + "QWEN_REMOTE_URL": "", + "CLONE_REMOTE_URL": "", + "FASTER_REMOTE_URL": + "http://10.0.0.6:8000", + "AUDIOCPP_REMOTE_URL": + "http://10.0.0.5:8080"}) # In-memory config is reloaded so this session sees the change. self.assertEqual(hub.config.AUDIO_FORMAT, "ogg") self.assertEqual(hub.config.AUDIO_BITRATE, "192k") @@ -982,12 +1124,16 @@ class SettingsTests(unittest.TestCase): self.assertEqual(hub.config.CHUNK_SIZE, 300) self.assertEqual(hub.config.QWEN_API_URL, "http://127.0.0.1:7862") self.assertEqual(hub.config.FASTER_API_URL, "http://127.0.0.1:8001") + self.assertEqual(hub.config.AUDIOCPP_REMOTE_URL, + "http://10.0.0.5:8080") def test_apply_settings_rejects_bad_values(self): original = {name: getattr(hub.config, name) for name in ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "QWEN_API_URL", "CLONE_API_URL", - "FASTER_API_URL", "AUDIOCPP_API_URL")} + "FASTER_API_URL", "AUDIOCPP_API_URL", + "QWEN_REMOTE_URL", "CLONE_REMOTE_URL", + "FASTER_REMOTE_URL", "AUDIOCPP_REMOTE_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) base = {"audio_format": "m4b", "audio_bitrate": "128k", @@ -1001,6 +1147,9 @@ class SettingsTests(unittest.TestCase): hub._apply_settings({**base, "chunk_size": "0"}) with self.assertRaises(ValueError): hub._apply_settings({**base, "audiocpp_port": "70000"}) + with self.assertRaises(ValueError): + hub._apply_settings({**base, + "audiocpp_remote_url": "not a url"}) mk_write.assert_not_called() def test_field_validators(self): @@ -1044,18 +1193,24 @@ class SettingsTests(unittest.TestCase): self.assertEqual([f["key"] for f in captured["fields"]], ["audio_format", "audio_bitrate", "language", "chunk_size", "audiocpp_port", "faster_port", - "qwen_custom_port", "qwen_clone_port"]) + "qwen_custom_port", "qwen_clone_port", + "audiocpp_remote_url", "faster_remote_url", + "qwen_custom_remote_url", "qwen_clone_remote_url"]) kinds = {f["key"]: f["kind"] for f in captured["fields"]} self.assertEqual(kinds["audio_format"], "choice") self.assertEqual(kinds["audio_bitrate"], "text") self.assertEqual(kinds["audiocpp_port"], "text") + 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["audiocpp_remote_url"], + "audio.cpp remote URL") self.assertNotIn("(clone)", " ".join(labels.values())) - # The ports section note hangs off the first port field so it - # renders between the output settings and the ports. + # The ports section note hangs off the first port field, the remote + # section note off the first remote URL field. notes = {f["key"]: f.get("note") for f in captured["fields"]} self.assertTrue(notes["audiocpp_port"]) + self.assertTrue(notes["audiocpp_remote_url"]) self.assertIsNone(notes["audio_format"]) self.assertIsNone(notes["qwen_custom_port"]) self.assertEqual(applied, [{"audio_format": "ogg", @@ -1094,7 +1249,9 @@ class SettingsTests(unittest.TestCase): original = {name: getattr(hub.config, name) for name in ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE", "CHUNK_SIZE", "QWEN_API_URL", "CLONE_API_URL", - "FASTER_API_URL", "AUDIOCPP_API_URL")} + "FASTER_API_URL", "AUDIOCPP_API_URL", + "QWEN_REMOTE_URL", "CLONE_REMOTE_URL", + "FASTER_REMOTE_URL", "AUDIOCPP_REMOTE_URL")} self.addCleanup(lambda: [setattr(hub.config, name, value) for name, value in original.items()]) @@ -1110,7 +1267,11 @@ class SettingsTests(unittest.TestCase): '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', + '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") with patch.object(hub.config, "__file__", str(path)): # Down to Chunk size, Enter -> editor, Ctrl-U + '300', diff --git a/app/tests/test_tts.py b/app/tests/test_tts.py index f99e257..0026702 100644 --- a/app/tests/test_tts.py +++ b/app/tests/test_tts.py @@ -84,6 +84,19 @@ class QwenTTSClientLanguageTests(unittest.TestCase): QwenTTSClient(language="klingon") mock_connect.assert_not_called() + def test_api_url_override_stored(self): + client = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM, + api_url="http://10.0.0.5:7860") + self.assertEqual(client.api_url, "http://10.0.0.5:7860") + + def test_api_url_override_used_by_connect(self): + with patch.object(QwenTTSClient, "_init_client") as mk_init: + client = QwenTTSClient.__new__(QwenTTSClient) + client.voice_mode = tts.VOICE_MODE_CUSTOM + client.api_url = "http://10.0.0.5:7860" + client._connect() + mk_init.assert_called_once_with("http://10.0.0.5:7860", clone=False) + class SeedResolutionTests(unittest.TestCase): """CONSTANT_SEED: one seed per run, reused for every request, so the @@ -1391,7 +1404,7 @@ class BackendWiringTests(unittest.TestCase): patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, backend=tts.BACKEND_FASTER, voice="narrator") - mock_faster.assert_called_once_with(voice="narrator") + mock_faster.assert_called_once_with(voice="narrator", api_url=None) mock_qwen.assert_not_called() mock_audiocpp.assert_not_called() @@ -1405,7 +1418,8 @@ class BackendWiringTests(unittest.TestCase): mock_audiocpp.assert_called_once_with(voice="narrator", language="Japanese", model_id=None, instructions=None, - request_options={}) + request_options={}, + api_url=None) mock_faster.assert_not_called() mock_qwen.assert_not_called() @@ -1416,7 +1430,8 @@ class BackendWiringTests(unittest.TestCase): mock_audiocpp.assert_called_once_with(voice=None, language=config.LANGUAGE, model_id=None, instructions=None, - request_options={}) + request_options={}, + api_url=None) def test_audiocpp_backend_model_id_is_wired_through(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: @@ -1426,7 +1441,7 @@ class BackendWiringTests(unittest.TestCase): mock_audiocpp.assert_called_once_with( voice="narrator", language=config.LANGUAGE, model_id="higgs", instructions=None, - request_options={}) + request_options={}, api_url=None) def test_audiocpp_backend_instructions_and_options_are_wired_through(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: @@ -1439,7 +1454,8 @@ class BackendWiringTests(unittest.TestCase): voice=None, language=config.LANGUAGE, model_id=None, instructions="A warm adult narrator", - request_options={"emotion": "neutral", "speed": "1.1"}) + request_options={"emotion": "neutral", "speed": "1.1"}, + api_url=None) def test_qwen_backend_uses_qwen_client(self): with patch("converter.converter.FasterTTSClient") as mock_faster, \ @@ -1457,6 +1473,31 @@ class BackendWiringTests(unittest.TestCase): AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, backend=tts.BACKEND_QWEN) + def test_api_url_override_reaches_each_client(self): + # A remote conversion threads api_url through to the selected client. + with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: + AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, + backend=tts.BACKEND_AUDIOCPP, voice="narrator", + api_url="http://10.0.0.5:8080") + mock_audiocpp.assert_called_once_with( + voice="narrator", language=config.LANGUAGE, model_id=None, + instructions=None, request_options={}, + api_url="http://10.0.0.5:8080") + with patch("converter.converter.FasterTTSClient") as mock_faster: + AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, + backend=tts.BACKEND_FASTER, voice="narrator", + api_url="http://10.0.0.5:8000") + mock_faster.assert_called_once_with(voice="narrator", + api_url="http://10.0.0.5:8000") + with patch("converter.converter.QwenTTSClient") as mock_qwen: + AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, + backend=tts.BACKEND_QWEN, + api_url="http://10.0.0.5:7860") + mock_qwen.assert_called_once_with( + voice_mode=tts.VOICE_MODE_CUSTOM, voice_clone_ref_audio=None, + voice_clone_ref_text=None, skip_transcription=False, + language=config.LANGUAGE, api_url="http://10.0.0.5:7860") + def test_audiocpp_clone_mode_does_not_require_reference(self): # Cloning is server-side for the audiocpp backend, so the # clone-mode voice can be selected without local reference audio. |