diff options
Diffstat (limited to 'app/tests')
| -rw-r--r-- | app/tests/test_audiobook_cli.py | 16 | ||||
| -rw-r--r-- | app/tests/test_backends.py | 8 | ||||
| -rw-r--r-- | app/tests/test_backends_envs.py | 4 | ||||
| -rw-r--r-- | app/tests/test_backends_managed.py | 95 | ||||
| -rw-r--r-- | app/tests/test_backends_servers.py | 52 | ||||
| -rw-r--r-- | app/tests/test_backends_sglomni.py | 828 | ||||
| -rw-r--r-- | app/tests/test_hub.py | 351 | ||||
| -rw-r--r-- | app/tests/test_runview.py | 81 | ||||
| -rw-r--r-- | app/tests/test_tts.py | 40 | ||||
| -rw-r--r-- | app/tests/test_tts_sglomni.py | 364 | ||||
| -rw-r--r-- | app/tests/test_tui.py | 19 |
11 files changed, 1824 insertions, 34 deletions
diff --git a/app/tests/test_audiobook_cli.py b/app/tests/test_audiobook_cli.py index 8076c8e..ab13107 100644 --- a/app/tests/test_audiobook_cli.py +++ b/app/tests/test_audiobook_cli.py @@ -629,7 +629,7 @@ class ManagedServerWiringTests(unittest.TestCase): server.shutdown.side_effect = lambda: events.append("shutdown") ensure = MagicMock(return_value=server) - def _ensure(backend, voice_mode): + def _ensure(backend, voice_mode, model=None): events.append(("ensure", backend, voice_mode)) return server ensure.side_effect = _ensure @@ -657,6 +657,20 @@ class ManagedServerWiringTests(unittest.TestCase): self.assertEqual(events, [("ensure", "audiocpp", "custom_voice"), "shutdown"]) + def test_not_ok_boot_records_the_server_log_pointer(self): + # The dated run log the failure pointers name must not stay empty + # when the run stops at a failed boot. + with patch.object(audiobook.logging, "error") as mk_log: + code, _, _, _, _ = self._convert(server_ok=False) + self.assertEqual(code, 1) + mk_log.assert_called_once() + self.assertIn("failed to start", mk_log.call_args.args[0]) + + def test_ok_boot_logs_no_failure(self): + with patch.object(audiobook.logging, "error") as mk_log: + self._convert() + mk_log.assert_not_called() + def test_shutdown_runs_when_the_conversion_fails(self): code, _, _, events, _ = self._convert( run_raises=RuntimeError("server unreachable")) diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py index 7a39880..e8ad1eb 100644 --- a/app/tests/test_backends.py +++ b/app/tests/test_backends.py @@ -37,9 +37,9 @@ class RegistryTests(unittest.TestCase): # another test class having called detect_all() first. get("audiocpp") - def test_registry_has_the_three_backends(self): + def test_registry_has_every_backend(self): keys = [info.key for info in REGISTRY] - self.assertEqual(keys, ["audiocpp", "qwen", "faster"]) + self.assertEqual(keys, ["audiocpp", "qwen", "faster", "sglomni"]) def test_every_entry_has_detect_setup_and_uninstall(self): for info in REGISTRY: @@ -65,9 +65,9 @@ class DetectAllTests(unittest.TestCase): with patch("backends.common.server_running", return_value=False): statuses = detect_all() self.assertEqual([s.key for s in statuses], - ["audiocpp", "qwen", "faster"]) + ["audiocpp", "qwen", "faster", "sglomni"]) for s in statuses: - self.assertIn(s.key, ("audiocpp", "qwen", "faster")) + self.assertIn(s.key, ("audiocpp", "qwen", "faster", "sglomni")) # ready requires both installed and configured; on a clean # machine none are ready. if s.ready: diff --git a/app/tests/test_backends_envs.py b/app/tests/test_backends_envs.py index d3b35c1..e38f009 100644 --- a/app/tests/test_backends_envs.py +++ b/app/tests/test_backends_envs.py @@ -134,7 +134,7 @@ class PipInstallTests(unittest.TestCase): side_effect=fake_run): rc = envs.pip_install(["qwen-tts"]) self.assertEqual(rc, 0) - mk.assert_called_once_with(None) + mk.assert_called_once_with(None, None) # The actual pip call targets the venv's python. self.assertEqual(calls[0][0], str(envs.env_python())) self.assertIn("pip", calls[0]) @@ -155,7 +155,7 @@ class PipInstallTests(unittest.TestCase): self.assertEqual(rc, 0) # Both create-if-missing and pip itself are scoped to the qwen env; # the app env is never touched. - mk.assert_called_once_with(envs.QWEN_ENV_DIR) + mk.assert_called_once_with(envs.QWEN_ENV_DIR, None) self.assertEqual(calls[0][0], str(envs.env_python(envs.QWEN_ENV_DIR))) diff --git a/app/tests/test_backends_managed.py b/app/tests/test_backends_managed.py index cd1f03a..e0bb074 100644 --- a/app/tests/test_backends_managed.py +++ b/app/tests/test_backends_managed.py @@ -18,8 +18,8 @@ from backends.managed import ManagedServer, ensure_running from backends.probe import (IDENTITY_AUDIOCPP, IDENTITY_QWEN_CLONE, IDENTITY_QWEN_CUSTOM, IDENTITY_QWEN_DESIGN) from converter.clients import (BACKEND_AUDIOCPP, BACKEND_QWEN, - VOICE_MODE_CLONE, VOICE_MODE_CUSTOM, - VOICE_MODE_DESIGN) + BACKEND_SGLOMNI, VOICE_MODE_CLONE, + VOICE_MODE_CUSTOM, VOICE_MODE_DESIGN) def _spec(name="audiocpp", url="http://127.0.0.1:8080", identity=None): @@ -269,6 +269,97 @@ class QwenEnsureRunningTests(unittest.TestCase): mk_start.assert_not_called() +class SglomniEnsureRunningTests(unittest.TestCase): + """sglomni hosts one model per server: the running-model check is + keyed on the served HuggingFace repo id (the qwen rules again).""" + + def _detect_sglomni(self): + from backends.sglomni import status as sg_status + from backends.sglomni.catalog import entry_by_key + with patch("backends.sglomni.gpu.compute_capability", + return_value=None): + spec = sg_status.build_spec(entry_by_key("zonos2")) + return _status([spec], installed=True, label="SGLang-Omni") + + def _run(self, model): + out = io.StringIO() + with redirect_stdout(out): + result = ensure_running(BACKEND_SGLOMNI, VOICE_MODE_CLONE, + model=model) + return result, out.getvalue() + + def test_spec_aims_at_the_model_the_run_selected(self): + with patch("backends.detect", return_value=self._detect_sglomni()), \ + patch("backends.sglomni.models.model_installed", + return_value=True), \ + patch("backends.sglomni.gpu.compute_capability", + return_value=None), \ + patch("backends.common.server_running", + return_value=False), \ + patch.object(servers, "start", + return_value=True) as mk_start: + result, _ = self._run("zonos2") + spec = mk_start.call_args.args[0] + self.assertIn("Zyphra/zonos2", spec.argv) + self.assertNotIn("--config", spec.argv) + self.assertTrue(result.started) + + def test_fp8_fallback_prints_a_note_and_boots_the_bf16_config(self): + with patch("backends.detect", return_value=self._detect_sglomni()), \ + patch("backends.sglomni.models.model_installed", + return_value=True), \ + patch("backends.sglomni.gpu.compute_capability", + return_value=(8, 6)), \ + patch("backends.common.server_running", + return_value=False), \ + patch.object(servers, "start", + return_value=True) as mk_start: + result, output = self._run("zonos2") + self.assertIn("bf16", output) + self.assertIn("--config", mk_start.call_args.args[0].argv) + self.assertTrue(result.started) + + def test_managed_server_hosting_another_model_is_rebooted(self): + with patch("backends.detect", return_value=self._detect_sglomni()), \ + patch("backends.sglomni.models.model_installed", + return_value=True), \ + patch("backends.sglomni.gpu.compute_capability", + return_value=None), \ + patch("backends.common.server_running", + return_value=True), \ + patch("backends.probe.sglomni_served_model", + return_value="Qwen/Qwen3-TTS-12Hz-1.7B-Base"), \ + patch.object(servers, "alive", return_value=True), \ + patch.object(servers, "start", + return_value=True) as mk_start, \ + patch.object(servers, "stop") as mk_stop: + result, output = self._run("zonos2") + self.assertIn("restarting", output) + mk_stop.assert_called_once_with("sglomni") + self.assertIn("Zyphra/zonos2", + mk_start.call_args.args[0].argv) + self.assertTrue(result.started) + + def test_foreign_server_hosting_another_model_refuses_the_run(self): + with patch("backends.detect", return_value=self._detect_sglomni()), \ + patch("backends.sglomni.models.model_installed", + return_value=True), \ + patch("backends.sglomni.gpu.compute_capability", + return_value=None), \ + patch("backends.common.server_running", + return_value=True), \ + patch("backends.probe.sglomni_served_model", + return_value="Qwen/Qwen3-TTS-12Hz-1.7B-Base"), \ + patch.object(servers, "alive", return_value=False), \ + patch.object(servers, "start") as mk_start, \ + patch.object(servers, "stop") as mk_stop: + result, output = self._run("zonos2") + self.assertFalse(result.ok) + self.assertIn("this run needs Zyphra/zonos2", output) + mk_start.assert_not_called() + mk_stop.assert_not_called() + + class ManagedModuleSmokeTests(unittest.TestCase): """Import-surface sanity for the module the CLI wires in.""" diff --git a/app/tests/test_backends_servers.py b/app/tests/test_backends_servers.py index 61897d4..4065a34 100644 --- a/app/tests/test_backends_servers.py +++ b/app/tests/test_backends_servers.py @@ -1,9 +1,11 @@ """Tests for the server lifecycle module (backends/servers.py).""" +import io import os import signal import tempfile import unittest +from contextlib import redirect_stdout from pathlib import Path from unittest.mock import MagicMock, patch @@ -118,6 +120,56 @@ class StartTests(unittest.TestCase): # Pid file cleaned up after early exit. self.assertFalse((self.dir / "test-server.pid").exists()) + def test_exited_event_carries_a_known_crash_hint(self): + """The exited event's log tail is scanned for known signatures.""" + (self.dir / "test-server.log").write_text( + "triton.compiler.errors.CompilationError:\n" + 'ValueError("type fp8e4nv not supported in this architecture. ' + 'The supported fp8 dtypes are")\n', encoding="utf-8") + proc = MagicMock() + proc.pid = 99 + proc.poll.return_value = 1 + events = [] + with patch.object(servers, "LOG_DIR", self.dir), \ + patch("subprocess.Popen", return_value=proc), \ + patch("backends.common.server_running", return_value=False), \ + patch("time.sleep"): + ok = servers.start(self.spec, progress=events.append) + self.assertFalse(ok) + exited = next(e for e in events if e.get("kind") == "exited") + self.assertIn("FP8", exited["hint"]) + self.assertIn("8.9", exited["hint"]) + + def test_exited_event_has_no_hint_for_unknown_crashes(self): + proc = MagicMock() + proc.pid = 99 + proc.poll.return_value = 1 + events = [] + with patch.object(servers, "LOG_DIR", self.dir), \ + patch("subprocess.Popen", return_value=proc), \ + patch("backends.common.server_running", return_value=False), \ + patch("time.sleep"): + servers.start(self.spec, progress=events.append) + exited = next(e for e in events if e.get("kind") == "exited") + self.assertIsNone(exited["hint"]) + + def test_boot_hint_reads_the_log_tail(self): + self.assertIsNone(servers._boot_hint(["everything fine"])) + self.assertIn("FP8", servers._boot_hint( + ["x", 'ValueError("type fp8e4nv not supported in this ' + 'architecture")', "y"])) + self.assertIsNone(servers._boot_hint([])) + + def test_console_progress_prints_the_hint(self): + out = io.StringIO() + with redirect_stdout(out): + servers._console_progress({ + "kind": "exited", "name": "test", "returncode": 1, + "log_tail": ["boom"], "hint": "FP8 needs compute " + "capability 8.9+"}) + self.assertIn("hint: FP8 needs compute capability 8.9+", + out.getvalue()) + def test_returns_false_on_timeout(self): proc = MagicMock() proc.pid = 7 diff --git a/app/tests/test_backends_sglomni.py b/app/tests/test_backends_sglomni.py new file mode 100644 index 0000000..a11aa92 --- /dev/null +++ b/app/tests/test_backends_sglomni.py @@ -0,0 +1,828 @@ +"""Tests for the SGLang-Omni backend package (backends/sglomni).""" + +import io +import json +import urllib.error +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from backends import envs, probe, servers +from backends.sglomni import catalog, constants, models, pythonenv, status +from backends.sglomni.catalog import CAPABILITY_CLONE, CAPABILITY_DESIGN, \ + CAPABILITY_SPEAKER, ENTRIES, config_path, entry_by_key, entry_by_repo, \ + install_tree_families +from backends.sglomni.pythonenv import SGLOMNI_ENV + + +class CatalogTests(unittest.TestCase): + """The model catalog is the single source of hosting/voice facts.""" + + def test_unique_keys_and_repos(self): + keys = [entry.key for entry in ENTRIES] + repos = [entry.repo for entry in ENTRIES] + self.assertEqual(len(keys), len(set(keys))) + self.assertEqual(len(repos), len(set(repos))) + + def test_capabilities_are_known(self): + for entry in ENTRIES: + self.assertIn(entry.capability, + (CAPABILITY_SPEAKER, CAPABILITY_CLONE, + CAPABILITY_DESIGN)) + + def test_vendored_config_files_exist(self): + for entry in ENTRIES: + path = config_path(entry) + if entry.config is None: + self.assertIsNone(path) + else: + self.assertTrue(path.is_file(), f"missing {path}") + + def test_config_declares_the_entry_repo(self): + # The vendored yaml pins model_path — it must match the entry's + # repo, or the server would host something else than the run + # selected (the client's connect check would refuse it). The + # files are flat `key: value` documents, parsed by hand here. + for entry in ENTRIES: + path = config_path(entry) + if path is None: + continue + data = {} + for line in path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("#") or ":" not in line: + continue + key, _, value = line.partition(":") + data[key.strip()] = value.strip() + self.assertEqual(data.get("model_path"), entry.repo, + f"stale config for {entry.key}") + + def test_clone_capability_matches_reference_requirement(self): + # Only clone models carry a reference requirement. + for entry in ENTRIES: + if entry.requires_reference: + self.assertEqual(entry.capability, CAPABILITY_CLONE, + entry.key) + + def test_entry_lookup_by_key_and_repo(self): + entry = ENTRIES[0] + self.assertIs(entry_by_key(entry.key), entry) + self.assertIs(entry_by_repo(entry.repo), entry) + self.assertIsNone(entry_by_key("nope")) + self.assertIsNone(entry_by_repo("nope")) + + def test_install_tree_covers_every_entry(self): + families = install_tree_families(list(ENTRIES)) + covered = [option["key"] for family in families + for option in family["options"]] + self.assertEqual(sorted(covered), + sorted(entry.key for entry in ENTRIES)) + + def test_install_tree_has_no_detail_line(self): + # The install screen's status line under the Confirm/Back buttons + # would only repeat the catalog keys under the cursor — the tree + # carries no detail at all, so no status line is drawn. + families = install_tree_families(list(ENTRIES)) + self.assertTrue(families) + for family in families: + self.assertNotIn("detail", family) + + def test_install_tree_filters_unavailable(self): + some = [ENTRIES[0]] + families = install_tree_families(some) + covered = [option["key"] for family in families + for option in family["options"]] + self.assertEqual(covered, [ENTRIES[0].key]) + + +class ModelInstallStateTests(unittest.TestCase): + """Install state reads the shared HuggingFace hub cache layout.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.cache = Path(self._tmp.name) + patcher = patch.object(models, "_hf_cache_dir", return_value=self.cache) + patcher.start() + self.addCleanup(patcher.stop) + self.addCleanup(self._tmp.cleanup) + + def _seed_repo(self, repo): + directory = self.cache / ("models--" + repo.replace("/", "--")) + (directory / "refs").mkdir(parents=True) + (directory / "refs" / "main").write_text("hash\n") + (directory / "snapshots" / "abc").mkdir(parents=True) + (directory / "snapshots" / "abc" / "weights.safetensors") \ + .write_bytes(b"x") + return directory + + def test_model_installed_needs_refs_and_snapshots(self): + entry = entry_by_key("higgs_audio_v3_tts") + self.assertFalse(models.model_installed(entry)) + self._seed_repo(entry.repo) + self.assertTrue(models.model_installed(entry)) + + def test_installed_entries_in_catalog_order(self): + first, second = ENTRIES[0], ENTRIES[4] + self._seed_repo(second.repo) + self._seed_repo(first.repo) + keys = models.installed_keys() + self.assertEqual(keys, [first.key, second.key]) + + def test_delete_model_weights_removes_only_targeted_repos(self): + entry = ENTRIES[0] + other = ENTRIES[1] + self._seed_repo(entry.repo) + self._seed_repo(other.repo) + removed = models.delete_model_weights([entry]) + self.assertEqual(removed, 1) + self.assertFalse(models.model_installed(entry)) + self.assertTrue(models.model_installed(other)) + + def test_preset_voices_from_voice_embedding(self): + # Voxtral-style: preset voices ship as voice_embedding/*.pt in the + # downloaded snapshot. + entry = entry_by_key("voxtral_tts") + directory = self._seed_repo(entry.repo) + (directory / "snapshots" / "abc" / "voice_embedding").mkdir() + (directory / "snapshots" / "abc" / "voice_embedding" / "casual_male.pt") \ + .write_bytes(b"x") + (directory / "snapshots" / "abc" / "voice_embedding" / "default.pt") \ + .write_bytes(b"x") + self.assertEqual(models.preset_voices(entry), + ["casual_male", "default"]) + + def test_preset_voices_from_catalog_table(self): + entry = entry_by_key("qwen3_tts_0_6b_customvoice") + self.assertTrue(entry.speakers) + self.assertEqual(models.preset_voices(entry), list(entry.speakers)) + + def test_resolve_model_unknown_key_raises(self): + with self.assertRaises(RuntimeError) as ctx: + models.resolve_model("nope") + self.assertIn("Unknown sglang-omni model", str(ctx.exception)) + + def test_resolve_model_requires_installed_weights(self): + with self.assertRaises(RuntimeError) as ctx: + models.resolve_model("higgs_audio_v3_tts") + self.assertIn("not downloaded", str(ctx.exception)) + + def test_resolve_model_auto_selects_the_single_install(self): + entry = ENTRIES[0] + self._seed_repo(entry.repo) + self.assertIs(models.resolve_model(None), entry) + self.assertIs(models.resolve_model(entry.key), entry) + + def test_resolve_model_needs_a_pick_with_several_installs(self): + self._seed_repo(ENTRIES[0].repo) + self._seed_repo(ENTRIES[1].repo) + with self.assertRaises(RuntimeError) as ctx: + models.resolve_model(None) + self.assertIn("--model", str(ctx.exception)) + + +class PythonEnvTests(unittest.TestCase): + """Interpreter selection for the version-pinned venv.""" + + def test_env_compatible_needs_310_to_312(self): + with patch.object(pythonenv, "env_version", return_value=(3, 12)): + self.assertTrue(pythonenv.env_compatible()) + with patch.object(pythonenv, "env_version", return_value=(3, 13)): + self.assertFalse(pythonenv.env_compatible()) + with patch.object(pythonenv, "env_version", return_value=None): + self.assertFalse(pythonenv.env_compatible()) + + def test_prepare_env_noop_on_compatible_venv(self): + with patch.object(pythonenv, "env_compatible", return_value=True), \ + patch.object(envs, "create_env") as mock_create, \ + patch.object(envs, "provision_env_with_uv") as mock_uv: + self.assertEqual(pythonenv.prepare_env(), 0) + mock_create.assert_not_called() + mock_uv.assert_not_called() + + def test_prepare_env_uses_compatible_system_interpreter(self): + interpreter = Path("/usr/bin/python3.12") + with patch.object(pythonenv, "env_compatible", return_value=False), \ + patch.object(envs, "env_exists", return_value=False), \ + patch.object(envs, "compatible_interpreter", + return_value=interpreter) as mock_find, \ + patch.object(envs, "create_env", + return_value=0) as mock_create: + self.assertEqual(pythonenv.prepare_env(), 0) + mock_find.assert_called_once() + mock_create.assert_called_once_with(SGLOMNI_ENV, interpreter) + + def test_prepare_env_falls_back_to_uv(self): + with patch.object(pythonenv, "env_compatible", return_value=False), \ + patch.object(envs, "env_exists", return_value=False), \ + patch.object(envs, "compatible_interpreter", + return_value=None), \ + patch.object(envs, "ensure_uv", return_value=0) as mock_uv_install, \ + patch.object(envs, "provision_env_with_uv", + return_value=0) as mock_uv: + self.assertEqual(pythonenv.prepare_env(), 0) + mock_uv_install.assert_called_once() + mock_uv.assert_called_once() + + def test_prepare_env_reports_uv_failure(self): + with patch.object(pythonenv, "env_compatible", return_value=False), \ + patch.object(envs, "env_exists", return_value=False), \ + patch.object(envs, "compatible_interpreter", + return_value=None), \ + patch.object(envs, "ensure_uv", return_value=1): + self.assertEqual(pythonenv.prepare_env(), 1) + + +class BuildSpecTests(unittest.TestCase): + """The managed ServerSpec: sgl-omni serve with model/config/port.""" + + def test_spec_hosts_the_entry_repo_with_config(self): + entry = entry_by_key("qwen3_tts_0_6b_customvoice") + spec = status.build_spec(entry) + self.assertEqual(spec.name, constants.SERVER_NAME) + self.assertEqual(spec.identity, probe.IDENTITY_SGLOMNI) + self.assertEqual(spec.start_timeout, constants.SERVER_START_TIMEOUT) + script, serve, flag, repo, cfg_flag, cfg, port_flag, port = spec.argv + self.assertEqual(serve, "serve") + self.assertEqual(flag, "--model-path") + self.assertEqual(repo, entry.repo) + self.assertEqual(cfg_flag, "--config") + self.assertEqual(Path(cfg), config_path(entry)) + self.assertEqual(port_flag, "--port") + self.assertIn(port, str(spec.url)) + + def test_spec_omits_config_when_none_needed(self): + entry = entry_by_key("higgs_audio_v3_tts") + spec = status.build_spec(entry) + self.assertNotIn("--config", spec.argv) + + +class GpuCapabilityTests(unittest.TestCase): + """The nvidia-smi-backed GPU facts are best-effort and cached.""" + + def setUp(self): + gpu_module = status.gpu + gpu_module._query.cache_clear() + self.addCleanup(gpu_module._query.cache_clear) + + def _nvidia_smi(self, *, stdout="", returncode=0, installed=True): + def fake_run(argv, **_kwargs): + if not installed: + raise FileNotFoundError("nvidia-smi") + return SimpleNamespace(returncode=returncode, stdout=stdout) + return fake_run + + def test_parses_name_and_compute_capability(self): + with patch.object(status.gpu.shutil, "which", return_value="/x"), \ + patch.object(status.gpu.subprocess, "run", + side_effect=self._nvidia_smi( + stdout="NVIDIA GeForce RTX 3090, 8.6\n")): + self.assertEqual(status.gpu.compute_capability(), (8, 6)) + self.assertEqual(status.gpu.describe(), + "NVIDIA GeForce RTX 3090 (compute capability 8.6)") + + def test_none_when_nvidia_smi_missing(self): + with patch.object(status.gpu.shutil, "which", return_value=None): + self.assertIsNone(status.gpu.compute_capability()) + self.assertIsNone(status.gpu.describe()) + + def test_none_when_the_query_fails_or_is_garbage(self): + for kwargs in (dict(returncode=1), + dict(stdout=""), + dict(stdout="name only\n")): + with self.subTest(stdout=kwargs.get("stdout")): + with patch.object(status.gpu.shutil, "which", + return_value="/x"), \ + patch.object(status.gpu.subprocess, "run", + side_effect=self._nvidia_smi(**kwargs)): + self.assertIsNone(status.gpu.compute_capability()) + + +class Fp8FallbackTests(unittest.TestCase): + """FP8-only pipelines fall back to a vendored bf16 config on old GPUs.""" + + ZONOS2 = "zonos2" + + def _fallback(self, capability): + return patch("backends.sglomni.gpu.compute_capability", + return_value=capability) + + def test_fallback_config_declares_the_repo_and_disables_fp8(self): + entry = entry_by_key("zonos2") + path = catalog.fallback_config_path(entry) + self.assertIsNotNone(path) + self.assertTrue(path.is_file(), f"missing {path}") + text = path.read_text(encoding="utf-8") + self.assertIn(f"model_path: {entry.repo}", text) + self.assertRegex(text, r"fp8:\s*false") + # bf16 weights need a bigger static pool than the builder's 0.5 + # default (24 GB card: >=0.64 for any KV cache at all). + self.assertRegex(text, r"mem_fraction_static:\s*0\.70") + + def test_only_fp8_models_carry_a_fallback(self): + for entry in ENTRIES: + if entry.fp8_moe: + self.assertIsNotNone(entry.fp8_min_compute_capability, + entry.key) + self.assertIsNotNone(entry.bf16_config, entry.key) + else: + self.assertIsNone(catalog.fallback_config_path(entry)) + + def test_fallback_needed_below_the_capability_floor(self): + entry = entry_by_key("zonos2") + with patch("backends.sglomni.gpu.compute_capability", + return_value=(8, 6)): + self.assertTrue(status.needs_fp8_fallback(entry)) + self.assertEqual(status.launch_config_path(entry), + catalog.fallback_config_path(entry)) + note = status.gpu_fallback_note(entry) + self.assertIn("bf16", note) + self.assertIn("8.9", note) + + def test_no_fallback_at_or_above_the_capability(self): + entry = entry_by_key("zonos2") + for capability in ((8, 9), (9, 0), (10, 0)): + with self.subTest(capability=capability): + with patch("backends.sglomni.gpu.compute_capability", + return_value=capability): + self.assertFalse(status.needs_fp8_fallback(entry)) + self.assertIsNone(status.gpu_fallback_note(entry)) + self.assertEqual(status.launch_config_path(entry), + config_path(entry)) + + def test_no_fallback_without_an_answerable_gpu(self): + # A GPU this tool cannot read keeps upstream defaults instead of + # second-guessing the host. + entry = entry_by_key("zonos2") + with patch("backends.sglomni.gpu.compute_capability", + return_value=None): + self.assertFalse(status.needs_fp8_fallback(entry)) + self.assertIsNone(status.gpu_fallback_note(entry)) + self.assertEqual(status.launch_config_path(entry), + config_path(entry)) + + def test_non_fp8_models_never_fall_back(self): + for entry in ENTRIES: + if entry.key == "zonos2": + continue + with patch("backends.sglomni.gpu.compute_capability", + return_value=(1, 0)): + self.assertFalse(status.needs_fp8_fallback(entry)) + + def test_spec_launches_the_bf16_config_on_an_old_gpu(self): + entry = entry_by_key("zonos2") + with patch("backends.sglomni.gpu.compute_capability", + return_value=(8, 6)): + spec = status.build_spec(entry) + self.assertIn("--config", spec.argv) + self.assertEqual(Path(spec.argv[spec.argv.index("--config") + 1]), + catalog.fallback_config_path(entry)) + + def test_spec_keeps_the_default_pipeline_on_modern_gpus(self): + entry = entry_by_key("zonos2") + with patch("backends.sglomni.gpu.compute_capability", + return_value=(9, 0)): + spec = status.build_spec(entry) + self.assertNotIn("--config", spec.argv) + + def test_detect_tags_a_fallback_model(self): + entry = entry_by_key("zonos2") + with patch("backends.sglomni.status._is_installed", + return_value=True), \ + patch("backends.sglomni.status.installed_entries", + return_value=[entry]), \ + patch("backends.sglomni.gpu.compute_capability", + return_value=(8, 6)), \ + patch.object(servers, "manages", return_value=False), \ + patch.object(status, "_detect_remote", + return_value=([], {})): + st = status.detect() + self.assertIn("zonos2 (bf16 fallback)", + next(line for line in st.details + if line.startswith("models: "))) + + def test_install_prints_the_fallback_note(self): + out = io.StringIO() + with patch("backends.sglomni.models.prepare_env", return_value=0), \ + patch("backends.common.pip_install", return_value=0), \ + patch("backends.sglomni.models._hf_download_prefix", + return_value=["hf"]), \ + patch("backends.common.run_console_subprocess", + return_value=0), \ + patch("backends.sglomni.gpu.compute_capability", + return_value=(8, 6)), \ + redirect_stdout(out): + rc = models.install_model("zonos2") + self.assertEqual(rc, 0) + self.assertIn("bf16", out.getvalue()) + + +class DetectTests(unittest.TestCase): + """detect() reports install state, models, and the running model.""" + + def _detect(self, *, script=False, module=False, entries=()): + entry = entries[0] if entries else None + spec = [status.build_spec(entry)] if entry else [] + with patch("backends.sglomni.status._is_installed", + return_value=script or module), \ + patch("backends.sglomni.status.installed_entries", + return_value=list(entries)), \ + patch.object(servers, "manages", return_value=False), \ + patch.object(status, "_detect_remote", + return_value=([], {})): + return status.detect(), spec + + def test_not_installed(self): + st, _spec = self._detect() + self.assertFalse(st.installed) + self.assertFalse(st.configured) + self.assertFalse(st.running) + self.assertEqual(st.servers, []) + self.assertEqual(st.partial, "") + + def test_installed_without_models_is_partial(self): + st, _spec = self._detect(script=True) + self.assertTrue(st.installed) + self.assertFalse(st.configured) + self.assertEqual(st.partial, "installed (no models)") + + def test_configured_reports_a_spec_and_ready(self): + entry = entry_by_key("higgs_audio_v3_tts") + st, _spec = self._detect(script=True, entries=[entry]) + self.assertTrue(st.configured) + self.assertTrue(st.ready) + self.assertEqual(len(st.servers), 1) + self.assertIn(entry.repo, st.servers[0].argv) + + +class ProbeIdentityTests(unittest.TestCase): + """A healthy sglang-omni /health identifies the sglomni backend.""" + + def _urlopen_returning(self, payloads): + calls = {"index": 0} + + def fake_urlopen(url, timeout=3.0): + if calls["index"] >= len(payloads): + # Past the scripted payloads (e.g. the gradio fallback + # probe): behave like a 404 — urlopen raises, _get_json + # maps that to None. + calls["index"] += 1 + raise urllib.error.URLError("HTTP 404") + payload = payloads[calls["index"]] + calls["index"] += 1 + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = json.dumps(payload).encode("utf-8") + return response + + return fake_urlopen, calls + + def test_healthy_server_with_stages_identifies_sglomni(self): + health = {"status": "healthy", "running": True, + "stages": ["preprocessing", "tts_generation", "vocoder"]} + fake_urlopen, _calls = self._urlopen_returning([health]) + with patch.object(probe.common, "server_running", return_value=True), \ + patch("backends.probe.urllib.request.urlopen", fake_urlopen): + self.assertEqual(probe.identify_server("http://127.0.0.1:8100"), + probe.IDENTITY_SGLOMNI) + + def test_unhealthy_server_is_not_sglomni(self): + health = {"status": "unhealthy", "running": False, "stages": []} + fake_urlopen, _calls = self._urlopen_returning([health]) + with patch.object(probe.common, "server_running", return_value=True), \ + patch("backends.probe.urllib.request.urlopen", fake_urlopen): + self.assertIsNone(probe.identify_server("http://127.0.0.1:8100")) + + def test_served_model_read_from_v1_models(self): + payload = {"object": "list", "data": [ + {"id": "bosonai/higgs-audio-v3-tts-4b", "root": + "bosonai/higgs-audio-v3-tts-4b"}]} + fake_urlopen, _calls = self._urlopen_returning([payload]) + with patch("backends.probe.urllib.request.urlopen", fake_urlopen): + self.assertEqual( + probe.sglomni_served_model("http://127.0.0.1:8100"), + "bosonai/higgs-audio-v3-tts-4b") + + def test_served_model_none_on_garbage(self): + fake_urlopen, _calls = self._urlopen_returning([{"data": []}]) + with patch("backends.probe.urllib.request.urlopen", fake_urlopen): + self.assertIsNone( + probe.sglomni_served_model("http://127.0.0.1:8100")) + + def test_voice_names_read_from_uploaded_voices(self): + payload = {"uploaded_voice_names": ["narrator", "second narrator"]} + fake_urlopen, _calls = self._urlopen_returning([payload]) + with patch("backends.probe.urllib.request.urlopen", fake_urlopen): + self.assertEqual( + probe.sglomni_voice_names("http://127.0.0.1:8100"), + ["narrator", "second narrator"]) + + +class ModelsScreenTests(unittest.TestCase): + """models_screen: the audio.cpp-style checkbox tree driving steps. + + The tree is scripted (like the qwen models_screen tests): each + fake render records its arguments and returns the next scripted + answer; the task-view run executes its steps inline so delegation to + install/uninstall_model is observable. + """ + + FAMILY_INDEX = {option["key"]: index + for index, family in + enumerate(install_tree_families(list(ENTRIES))) + for option in family["options"]} + + def _screen(self, answers, *, installed=(), package=True, confirm=True, + extra=()): + """Run models_screen with scripted tree answers; record calls. + + Returns ``(rc, trees, confirms, flashes, runs)``: trees holds one + (title, kwargs) per render, confirms every uninstall question, + flashes every (text, kind), and runs each (title, step titles) + while executing its steps' work inline. + """ + import contextlib + + from backends.sglomni import wizard + choices = list(answers) + trees, confirms, flashes, runs = [], [], [], [] + + def fake_tree(stdscr, title, families, **kwargs): + trees.append((title, kwargs)) + return choices.pop(0) + + def fake_confirm(scr, question, **kwargs): + confirms.append((question, kwargs)) + return confirm + + def fake_flash(scr, text, kind="warn"): + flashes.append((text, kind)) + + def fake_run(scr, title, steps, **kwargs): + runs.append((title, [step.title for step in steps])) + for step in steps: + step.work(None, None) + return 0 + + patches = [ + patch.object(wizard, "_is_installed", return_value=package), + patch.object(models, "installed_keys", + return_value=list(installed)), + patch.object(wizard.tui, "checkbox_tree", fake_tree), + patch.object(wizard.tui, "confirm", fake_confirm), + patch.object(wizard.tui, "flash", fake_flash), + patch.object(wizard.taskview, "run_steps", fake_run), + *extra, + ] + with contextlib.ExitStack() as stack: + for ctx in patches: + stack.enter_context(ctx) + rc = wizard.models_screen(None) + return rc, trees, confirms, flashes, runs + + def test_tree_is_the_audio_cpp_modify_flow(self): + from backends.sglomni import wizard + first = ENTRIES[0] + rc, trees, _confirms, _flashes, _runs = self._screen( + [wizard._GO_BACK], installed=(first.key,)) + self.assertEqual(rc, 0) + title, kwargs = trees[0] + self.assertEqual(title, "Select SGLang-Omni Models") + # The installed model starts checked (a modify list), Confirm is + # pre-focused, and an empty selection is a valid answer. + self.assertEqual(kwargs["checked"], + {(self.FAMILY_INDEX[first.key], first.key)}) + self.assertTrue(kwargs["start_on_buttons"]) + self.assertTrue(kwargs["allow_empty"]) + self.assertIs(kwargs["back_value"], wizard._GO_BACK) + + def test_checking_a_model_runs_one_install_step(self): + from backends.sglomni import wizard + entry = ENTRIES[0] + requested = [] + + def capture(key, *, emit=None, cancel=None): + requested.append(key) + return 0 + + picked = [(self.FAMILY_INDEX[entry.key], entry.key)] + rc, _trees, confirms, flashes, runs = self._screen( + [picked, wizard._GO_BACK], + extra=[patch.object(models, "install_model", + side_effect=capture)]) + self.assertEqual(rc, 0) + self.assertEqual(requested, [entry.key]) + self.assertEqual(runs, [("Configure SGLang-Omni", + [f"Install {entry.label}"])]) + self.assertEqual(confirms, []) + self.assertEqual(flashes[-1], + ("SGLang-Omni models updated: 1 installed.", "ok")) + + def test_unchecking_confirms_then_deletes_the_weights(self): + from backends.sglomni import wizard + entry = ENTRIES[0] + rc, _trees, confirms, flashes, runs = self._screen( + [[], wizard._GO_BACK], installed=(entry.key,)) + self.assertEqual(rc, 0) + # An empty selection is accepted (allow_empty) and uninstalls + # everything installed: one confirm, one removal step. + question, kwargs = confirms[0] + self.assertEqual(question, "Remove cached weights for 1 model?") + self.assertIn(entry.label, kwargs["body"]) + self.assertEqual(runs, [("Configure SGLang-Omni", + [f"Delete {entry.label} weights"])]) + self.assertEqual(flashes[-1], + ("SGLang-Omni models updated: 1 removed.", "ok")) + + def test_uninstall_step_stops_nothing_and_deletes_real_weights(self): + # The removal step is uninstall_model itself: with a redirected + # HF cache the seeded weight directory is really deleted. + from backends.sglomni import wizard + entry = ENTRIES[0] + with tempfile.TemporaryDirectory() as td: + directory = Path(td) / ("models--" + + entry.repo.replace("/", "--")) + (directory / "refs").mkdir(parents=True) + (directory / "refs" / "main").write_text("hash\n") + (directory / "snapshots" / "abc").mkdir(parents=True) + (directory / "snapshots" / "abc" / "weights.safetensors") \ + .write_bytes(b"x") + rc, _trees, _confirms, _flashes, _runs = self._screen( + [[], wizard._GO_BACK], installed=(entry.key,), + extra=[ + patch.object(models, "_hf_cache_dir", + return_value=Path(td)), + patch.object(models, "_managed_running_repo", + return_value=None), + ]) + self.assertEqual(rc, 0) + self.assertFalse(directory.exists()) + + def test_declining_the_uninstall_confirm_runs_nothing(self): + from backends.sglomni import wizard + entry = ENTRIES[0] + rc, trees, confirms, flashes, runs = self._screen( + [[], wizard._GO_BACK], installed=(entry.key,), confirm=False) + self.assertEqual(rc, 0) + # The decline re-opens the tree (second render), no work happens. + self.assertEqual(len(trees), 2) + self.assertEqual(len(confirms), 1) + self.assertEqual(runs, []) + self.assertEqual(flashes, []) + + def test_install_without_package_flashes_guidance_instead(self): + from backends.sglomni import wizard + entry = ENTRIES[0] + picked = [(self.FAMILY_INDEX[entry.key], entry.key)] + rc, _trees, confirms, flashes, runs = self._screen( + [picked, wizard._GO_BACK], package=False) + self.assertEqual(rc, 0) + self.assertEqual(runs, []) + self.assertEqual(confirms, []) + self.assertEqual(flashes, [( + "Install the SGLang-Omni backend first " + "(Configure Backends > Install Backend).", "warn")]) + + def test_unchanged_selection_re_opens_the_tree(self): + from backends.sglomni import wizard + entry = ENTRIES[0] + picked = [(self.FAMILY_INDEX[entry.key], entry.key)] + rc, trees, _confirms, flashes, runs = self._screen( + [picked, wizard._GO_BACK], installed=(entry.key,)) + self.assertEqual(rc, 0) + self.assertEqual(len(trees), 2) + self.assertEqual(runs, []) + self.assertEqual(flashes, []) + + def test_mixed_selection_removes_before_downloading(self): + # Unchecking the installed model and checking another in one + # confirm: a single run whose removal step precedes the download. + from backends.sglomni import wizard + gone, added = ENTRIES[0], ENTRIES[1] + picked = [(self.FAMILY_INDEX[added.key], added.key)] + rc, _trees, confirms, flashes, runs = self._screen( + [picked, wizard._GO_BACK], installed=(gone.key,)) + self.assertEqual(rc, 0) + self.assertEqual(len(confirms), 1) + self.assertEqual(runs, [( + "Configure SGLang-Omni", + [f"Delete {gone.label} weights", f"Install {added.label}"])]) + self.assertEqual( + flashes[-1], + ("SGLang-Omni models updated: 1 installed, 1 removed.", "ok")) + + +class SetupWizardTests(unittest.TestCase): + """_wizard: the setup tree reconciles models like the Configure screen.""" + + FAMILY_INDEX = {option["key"]: index + for index, family in + enumerate(install_tree_families(list(ENTRIES))) + for option in family["options"]} + + def _wizard(self, answers, *, installed=(), package=True, confirm=True): + """Run _wizard with scripted tree answers; return (settings, trees, + confirms).""" + from backends.sglomni import wizard + trees, confirms = [], [] + + def fake_tree(stdscr, title, families, **kwargs): + trees.append((title, kwargs)) + return answers.pop(0) + + def fake_confirm(scr, question, **kwargs): + confirms.append(question) + return confirm + + args = wizard.build_parser().parse_args([]) + with patch.object(wizard, "_preflight", return_value=[]), \ + patch.object(wizard, "_gpu_warning", return_value=None), \ + patch.object(wizard, "_is_installed", + return_value=package), \ + patch.object(models, "installed_keys", + return_value=list(installed)), \ + patch.object(wizard.tui, "checkbox_tree", fake_tree), \ + patch.object(wizard.tui, "confirm", fake_confirm), \ + patch.object(wizard.tui, "flash", lambda *a, **k: None): + settings = wizard._wizard(None, args) + return settings, trees, confirms + + def test_esc_aborts(self): + from backends.sglomni import wizard + settings, _trees, confirms = self._wizard([wizard._GO_BACK]) + self.assertIsNone(settings) + self.assertEqual(confirms, []) + + def test_modify_flow_installs_new_and_keeps_installed(self): + from backends.sglomni import wizard + first, second = ENTRIES[0], ENTRIES[1] + # The installed model stays checked (kept as-is); the new one is + # added — the diff installs the new one only. + picked = [(self.FAMILY_INDEX[first.key], first.key), + (self.FAMILY_INDEX[second.key], second.key)] + settings, _trees, confirms = self._wizard( + [picked], installed=(first.key,)) + self.assertEqual(settings["keys"], [second.key]) + self.assertEqual(settings["uninstall_keys"], []) + self.assertEqual(confirms, []) + + def test_unchecking_requires_a_confirm_then_uninstalls(self): + from backends.sglomni import wizard + first = ENTRIES[0] + settings, _trees, confirms = self._wizard([[]], + installed=(first.key,)) + self.assertEqual(settings["keys"], []) + self.assertEqual(settings["uninstall_keys"], [first.key]) + self.assertEqual(confirms, ["Remove cached weights for 1 model?"]) + + def test_declined_confirm_re_opens_the_tree(self): + from backends.sglomni import wizard + first = ENTRIES[0] + settings, trees, confirms = self._wizard( + [[], wizard._GO_BACK], installed=(first.key,), confirm=False) + self.assertIsNone(settings) + self.assertEqual(len(trees), 2) + self.assertEqual(len(confirms), 1) + + def test_empty_tree_installs_the_package_only(self): + from backends.sglomni import wizard + settings, _trees, confirms = self._wizard([[]]) + self.assertEqual(settings["keys"], []) + self.assertEqual(settings["uninstall_keys"], []) + self.assertEqual(confirms, []) + + def test_steps_remove_before_downloading(self): + from backends.sglomni import wizard + gone, added = ENTRIES[0], ENTRIES[1] + steps = wizard._execute_steps({ + "do_python": False, "do_install": False, + "uninstall_keys": [gone.key], "keys": [added.key]}) + self.assertEqual([step.title for step in steps], + [f"Delete {gone.label} weights", + f"Install {added.label}"]) + + def test_run_tui_drives_the_wizard_in_a_curses_session(self): + # The standalone CLI wraps the wizard in its own curses session + # (passing a real screen through) and runs the model work as a + # console tail afterwards. + from backends.sglomni import wizard + settings = {"keys": [], "uninstall_keys": [], "do_python": False, + "do_install": False} + screens = [] + with patch.object(wizard, "_wizard", + side_effect=lambda scr, args: + screens.append(scr) or settings), \ + patch.object(wizard, "_execute", return_value=7) as execute, \ + patch("curses.wrapper", + side_effect=lambda fn: fn("SCREEN")): + rc = wizard.run_tui(wizard.build_parser().parse_args([])) + self.assertEqual(rc, 7) + self.assertEqual(screens, ["SCREEN"]) + execute.assert_called_once_with(settings) + + +if __name__ == "__main__": + unittest.main() 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"): diff --git a/app/tests/test_runview.py b/app/tests/test_runview.py index 0d99a6a..7f79053 100644 --- a/app/tests/test_runview.py +++ b/app/tests/test_runview.py @@ -237,6 +237,46 @@ class StateTransitionTests(_FakeTui, unittest.TestCase): self.assertEqual(view.server, "error") self.assertEqual(view.log_tail, ["boom"]) + def test_server_exit_keeps_a_known_crash_hint(self): + view, _ = self.make_view() + view.handle_event({"kind": "exited", "name": "sglomni", + "returncode": 1, "log_tail": ["fp8..."], + "hint": "FP8 needs compute capability 8.9+"}) + self.assertEqual(view.phase, "error") + self.assertEqual(view.boot_hint, + "FP8 needs compute capability 8.9+") + + def test_boot_failure_is_recorded_in_the_dated_log(self): + # A failed boot never reaches the converter, so without this the + # dated log the failure pointers name would stay blank. + with tempfile.TemporaryDirectory() as tmp: + log_path = os.path.join(tmp, "audiobook_test.log") + view, _ = self.make_view(log_path=log_path) + view.handle_event({"kind": "starting", "name": "sglomni", + "log_path": "/tmp/sglomni-server.log"}) + view.handle_event({"kind": "exited", "name": "sglomni", + "returncode": 1, "log_tail": ["boom"], + "hint": "FP8 needs compute capability 8.9+"}) + with open(log_path, encoding="utf-8") as logf: + text = logf.read() + self.assertIn("ERROR - server exited with code 1", text) + self.assertIn("WARNING - hint: FP8 needs compute capability 8.9+", + text) + self.assertIn("the server's own output is in /tmp/sglomni-server.log", + text) + + def test_boot_timeout_is_recorded_without_optional_detail(self): + with tempfile.TemporaryDirectory() as tmp: + log_path = os.path.join(tmp, "audiobook_test.log") + view, _ = self.make_view(log_path=log_path) + view.handle_event({"kind": "timeout", "name": "sglomni", + "seconds": 1200, "log_tail": []}) + with open(log_path, encoding="utf-8") as logf: + text = logf.read() + self.assertIn("ERROR - server did not become ready in time", text) + self.assertNotIn("hint:", text) + self.assertNotIn("the server's own output is in", text) + def test_server_down_during_convert(self): view, _ = self.make_view() view.handle_event({"kind": "book", "index": 1, "total": 1, @@ -397,6 +437,25 @@ class RenderTests(_FakeTui, unittest.TestCase): self.assertIn("not responding", text) self.assertIn("the server is not responding", text) + def test_error_summary_draws_the_boot_hint(self): + view, screen = self.make_view() + view.handle_event({"kind": "exited", "name": "sglomni", + "returncode": 1, "log_tail": [], + "hint": "FP8 needs compute capability 8.9+"}) + view.render() + self.assertIn("FP8 needs compute capability 8.9+", + self._strings(screen)) + + def test_error_screen_names_the_server_log(self): + view, screen = self.make_view() + view.handle_event({"kind": "starting", "name": "sglomni", + "log_path": "/tmp/sglomni-server.log"}) + view.handle_event({"kind": "exited", "name": "sglomni", + "returncode": 1, "log_tail": ["boom"]}) + view.render() + self.assertIn("server log: /tmp/sglomni-server.log", + self._strings(screen)) + def test_summary_screen_after_done(self): view, screen = self.make_view() view.handle_event({"kind": "book", "index": 1, "total": 1, @@ -645,6 +704,28 @@ class RunLoopTests(_FakeTui, unittest.TestCase): self.assertIn("Full details in the log file: /tmp/runs/a.log", mk_notice.call_args[0][0]) + def test_stop_and_exit_boot_failure_names_reason_hint_and_server_log(self): + # A run that dies in the boot phase must not summarize as a bare + # "No books were converted": the reason, the known-crash hint, + # and the server's own log path all land in the summary. + with patch.object(runview.servers, "stop"), \ + patch.object(runview.common, + "record_post_tui_notice") as mk_notice: + view, screen = self.make_view([], stop_and_exit=True, + log_path="/tmp/runs/a.log") + view._queue.put({"kind": "starting", "name": "sglomni", + "log_path": "/tmp/sglomni-server.log"}) + view._queue.put({"kind": "exited", "name": "sglomni", + "returncode": 1, "log_tail": [], + "hint": "FP8 needs compute capability 8.9+"}) + view.run() + text = mk_notice.call_args[0][0] + self.assertIn("No books were converted", text) + self.assertIn("Failure: server exited with code 1", text) + self.assertIn("hint: FP8 needs compute capability 8.9+", text) + self.assertIn("server log: /tmp/sglomni-server.log", text) + self.assertIn("Full details in the log file: /tmp/runs/a.log", text) + class WorkerTests(_FakeTui, unittest.TestCase): """The worker thread's handoff into audiobook.convert.""" diff --git a/app/tests/test_tts.py b/app/tests/test_tts.py index 39b407d..7170262 100644 --- a/app/tests/test_tts.py +++ b/app/tests/test_tts.py @@ -30,6 +30,7 @@ from converter.clients import ( BACKEND_AUDIOCPP, BACKEND_FASTER, BACKEND_QWEN, + BACKEND_SGLOMNI, LANGUAGE_CHOICES, LANGUAGE_ISO_CODES, MODEL_SIZE, @@ -2496,6 +2497,19 @@ class BackendWiringTests(unittest.TestCase): backend=BACKEND_AUDIOCPP, voice=voice, instructions=instructions) + def _sglomni_converter(self, model="qwen3_tts_1_7b_base", clone=None, + instructions=None): + from backends.sglomni.catalog import entry_by_key + api_url = "http://127.0.0.1:8100" + with patch("converter.converter.SgOmniTTSClient") as client: + client.return_value.entry = entry_by_key(model) + client.return_value.api_url = api_url + return AudiobookConverter( + voice_mode=VOICE_MODE_CLONE if clone else + (VOICE_MODE_DESIGN if instructions else VOICE_MODE_CUSTOM), + voice_clone_ref_audio=clone, backend=BACKEND_SGLOMNI, + model_id=model, instructions=instructions, api_url=api_url) + def test_narrator_tag_uses_faster_voice_name(self): converter = self._faster_converter(voice="male_richard_poe") self.assertEqual(converter._narrator_tag(), "male_richard_poe") @@ -2536,6 +2550,32 @@ class BackendWiringTests(unittest.TestCase): converter._print_banner() self.assertIn("higgs_audio_tts", buffer.getvalue()) + def test_sglomni_banner_prints_model_and_resolves_model_id(self): + # Regression: the banner read self.model_id, which __init__ never + # stored — every sglomni run crashed there after a good connect. + converter = self._sglomni_converter(model="zonos2", + clone="voices/ref.wav") + self.assertEqual(converter.model_id, "zonos2") + buffer = io.StringIO() + with redirect_stdout(buffer): + converter._print_banner() # must not raise + output = buffer.getvalue() + self.assertIn("ZONOS2", output) + self.assertIn("Zyphra/zonos2", output) + self.assertIn("voice cloning from a reference clip", output) + + def test_sglomni_wiring_resolves_and_stores_the_model_key(self): + from backends.sglomni.catalog import entry_by_key + api_url = "http://127.0.0.1:8100" + with patch("converter.converter.SgOmniTTSClient") as client: + AudiobookConverter(voice_mode=VOICE_MODE_CLONE, + voice_clone_ref_audio="voices/ref.wav", + backend=BACKEND_SGLOMNI, model_id="zonos2", + api_url=api_url) + self.assertEqual( + client.call_args.kwargs["model"], "zonos2") + self.assertEqual(entry_by_key("zonos2").repo, "Zyphra/zonos2") + def test_non_faster_narrator_tag_unchanged(self): with tempfile.TemporaryDirectory() as tmp: ref = Path(tmp) / "ref.wav" diff --git a/app/tests/test_tts_sglomni.py b/app/tests/test_tts_sglomni.py new file mode 100644 index 0000000..2dee364 --- /dev/null +++ b/app/tests/test_tts_sglomni.py @@ -0,0 +1,364 @@ +"""Tests for the SGLang-Omni TTS client (converter/clients/sglomni.py).""" + +import base64 +import io +import json +import tempfile +import unittest +import urllib.error +import wave +from pathlib import Path +from unittest.mock import MagicMock, patch + +from converter import config +from converter.clients import SgOmniTTSClient +from converter.clients.base import NonRetryableTTSError +from converter.clients.sglomni import _data_url, _is_loopback +from converter.clients.speakers import QWEN3_TTS_SPEAKERS + + +class CatalogConsistencyTests(unittest.TestCase): + """The backend catalog's vendored facts match the converter's.""" + + def test_customvoice_speakers_match_the_qwen_table(self): + from backends.sglomni.catalog import QWEN_CUSTOMVOICE_SPEAKERS + self.assertEqual(QWEN_CUSTOMVOICE_SPEAKERS, QWEN3_TTS_SPEAKERS) + +_DUMMY_CHUNKS = Path(tempfile.gettempdir()) / "audiobook_sglomni_test_chunks" + + +def _make_wav() -> bytes: + """A real minimal RIFF/WAVE file (what a server response looks like).""" + buffer = io.BytesIO() + with wave.open(buffer, "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(24000) + wav_file.writeframes(b"\x01\x00" * 16) + return buffer.getvalue() + + +_WAV_BYTES = _make_wav() +_WAV_FRAMES = b"\x01\x00" * 16 + + +class LoopbackTests(unittest.TestCase): + def test_loopback_hosts(self): + self.assertTrue(_is_loopback("http://127.0.0.1:8100")) + self.assertTrue(_is_loopback("http://localhost:8100")) + self.assertFalse(_is_loopback("http://10.20.30.40:8100")) + + def test_data_url_carries_mime_and_bytes(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "ref.wav" + path.write_bytes(b"abc") + url = _data_url(path) + self.assertTrue(url.startswith("data:audio/wav;base64,")) + self.assertEqual( + base64.b64decode(url.partition(";base64,")[2]), b"abc") + + +class ConnectInputTests(unittest.TestCase): + """Capability-driven validation before any HTTP is attempted.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.ref = Path(self._tmp.name) / "narrator.wav" + self.ref.write_bytes(b"abc") + self.addCleanup(self._tmp.cleanup) + + def _client(self, model="higgs_audio_v3_tts", **kwargs): + # Bypass _connect (HTTP) — these tests cover the input checks. + with patch.object(SgOmniTTSClient, "_connect"): + return SgOmniTTSClient(_DUMMY_CHUNKS, model=model, **kwargs) + + def test_unknown_model_raises(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(model="nope") + self.assertIn("Unknown SGLang-Omni model", str(ctx.exception)) + + def test_design_model_requires_instructions(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(model="qwen3_tts_1_7b_voicedesign") + self.assertIn("--instructions", str(ctx.exception)) + + def test_reference_required_model_refuses_to_connect_without_one(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(model="qwen3_tts_1_7b_base") + self.assertIn("requires reference audio", str(ctx.exception)) + + def test_missing_reference_file_raises(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(model="higgs_audio_v3_tts", + ref_audio=str(Path(self._tmp.name) / "gone.wav")) + self.assertIn("Reference audio not found", str(ctx.exception)) + + def test_clone_capable_model_allows_text_only(self): + client = self._client(model="higgs_audio_v3_tts") + self.assertIsNone(client.ref_audio) + + def test_speaker_model_ignores_the_clone_reference(self): + client = self._client(model="qwen3_tts_0_6b_customvoice", + ref_audio=str(self.ref)) + self.assertIsNone(client.ref_audio) + + def test_seed_only_sent_for_models_that_accept_it(self): + with patch("converter.clients.sglomni.resolve_request_seed", + return_value=42): + client = self._client(model="qwen3_tts_1_7b_base", + ref_audio=str(self.ref)) + self.assertEqual(client._seed, 42) + client = self._client(model="higgs_audio_v3_tts") + self.assertIsNone(client._seed) + + def test_negative_seed_is_not_sent(self): + with patch("converter.clients.sglomni.resolve_request_seed", + return_value=-1): + client = self._client(model="qwen3_tts_1_7b_base", + ref_audio=str(self.ref)) + self.assertIsNone(client._seed) + + +class ConnectHealthTests(unittest.TestCase): + """_connect gates on /health and the hosted model.""" + + def _response(self, payload): + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = json.dumps(payload).encode("utf-8") + return response + + def _connect(self, payloads, **kwargs): + # urlopen is called once per _get_json call, in order. + responses = [self._response(payload) for payload in payloads] + with patch("converter.clients.sglomni.urllib.request.urlopen", + side_effect=responses): + with patch.object(SgOmniTTSClient, "_resolve_reference_text"): + return SgOmniTTSClient(_DUMMY_CHUNKS, + model="higgs_audio_v3_tts", **kwargs) + + def test_unreachable_server_raises_with_guidance(self): + import urllib.error + with patch("converter.clients.sglomni.urllib.request.urlopen", + side_effect=urllib.error.URLError("refused")): + with self.assertRaises(RuntimeError) as ctx: + SgOmniTTSClient(_DUMMY_CHUNKS, model="higgs_audio_v3_tts") + self.assertIn("not reachable", str(ctx.exception)) + self.assertIn("sgl-omni", str(ctx.exception)) + + def test_booting_server_raises(self): + with self.assertRaises(RuntimeError) as ctx: + self._connect([{"status": "unhealthy"}]) + self.assertIn("not healthy", str(ctx.exception)) + + def test_foreign_hosted_model_raises_with_both_names(self): + with self.assertRaises(RuntimeError) as ctx: + self._connect([ + {"status": "healthy", "stages": []}, + {"data": [{"id": "Zyphra/zonos2"}]}, + ]) + message = str(ctx.exception) + self.assertIn("Zyphra/zonos2", message) + self.assertIn("bosonai/higgs-audio-v3-tts-4b", message) + + def test_matching_model_connects(self): + client = self._connect([ + {"status": "healthy", "stages": []}, + {"data": [{"id": "bosonai/higgs-audio-v3-tts-4b"}]}, + ]) + self.assertEqual(client.entry.repo, "bosonai/higgs-audio-v3-tts-4b") + + +class PayloadTests(unittest.TestCase): + """The /v1/audio/speech request shape per voice capability.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.ref = Path(self._tmp.name) / "narrator.wav" + self.ref.write_bytes(b"abc") + self.addCleanup(self._tmp.cleanup) + + def _make_client(self, model, **kwargs): + client = SgOmniTTSClient.__new__(SgOmniTTSClient) + from backends.sglomni.catalog import entry_by_key + client.entry = entry_by_key(model) + client.api_url = "http://127.0.0.1:8100" + client.voice = kwargs.get("voice") + if "ref_audio" in kwargs: + kwargs["ref_audio"] = str(self.ref) + client.ref_audio = kwargs.get("ref_audio") + client.ref_text = kwargs.get("ref_text", "") + client.instructions = kwargs.get("instructions", "") + client.language = "English" + client._seed = None + return client + + def test_speaker_payload_sends_the_preset_name(self): + client = self._make_client("qwen3_tts_0_6b_customvoice", + voice="Vivian") + payload = client._request_payload("Hello.") + self.assertEqual(payload["voice"], "Vivian") + self.assertEqual(payload["model"], + "Qwen/Qwen3-TTS-12Hz-0.6B-CustomVoice") + self.assertEqual(payload["response_format"], "wav") + self.assertNotIn("ref_audio", payload) + self.assertNotIn("task_type", payload) + + def test_speaker_without_voice_uses_the_server_default(self): + client = self._make_client("voxtral_tts") + self.assertEqual(client._request_payload("Hello.")["voice"], + "default") + + def test_design_payload_sends_task_type_and_instructions(self): + client = self._make_client("qwen3_tts_1_7b_voicedesign", + instructions="A warm narrator.") + payload = client._request_payload("Hello.") + self.assertEqual(payload["task_type"], "VoiceDesign") + self.assertEqual(payload["instructions"], "A warm narrator.") + + def test_clone_payload_sends_reference_path_on_loopback(self): + client = self._make_client("higgs_audio_v3_tts", + ref_audio=str(self.ref), + ref_text="A transcript.") + payload = client._request_payload("Hello.") + self.assertEqual(payload["ref_audio"], str(self.ref.resolve())) + self.assertEqual(payload["ref_text"], "A transcript.") + + def test_clone_payload_inlines_audio_for_remote_servers(self): + client = self._make_client("higgs_audio_v3_tts", + ref_audio=str(self.ref)) + client.api_url = "http://10.20.30.40:8100" + payload = client._request_payload("Hello.") + self.assertTrue(payload["ref_audio"].startswith( + "data:audio/wav;base64,")) + self.assertEqual( + base64.b64decode(payload["ref_audio"].partition(";base64,")[2]), + b"abc") + self.assertNotIn("ref_text", payload) + + def test_clone_without_reference_sends_no_reference_fields(self): + client = self._make_client("higgs_audio_v3_tts") + payload = client._request_payload("Hello.") + self.assertNotIn("ref_audio", payload) + self.assertEqual(payload["voice"], "default") + + def test_seed_included_when_resolved(self): + client = self._make_client("qwen3_tts_1_7b_base", + ref_audio="x.wav") + client._seed = 7 + self.assertEqual(client._request_payload("Hello.")["seed"], 7) + + +class RequestErrorTests(unittest.TestCase): + """OpenAI-style error envelopes decide retryability.""" + + def _make_client(self): + return SgOmniTTSClient.__new__(SgOmniTTSClient) + + def test_bad_request_envelope_is_not_retryable(self): + client = self._make_client() + detail = json.dumps({"error": { + "message": "voice 'nope' not found", + "type": "BadRequestError", "code": 400}}) + error = client._request_error(400, detail) + self.assertIsInstance(error, NonRetryableTTSError) + self.assertIn("voice 'nope' not found", str(error)) + + def test_server_error_is_retryable(self): + client = self._make_client() + error = client._request_error(503, "overloaded") + self.assertNotIsInstance(error, NonRetryableTTSError) + + def test_non_json_4xx_is_not_retryable(self): + client = self._make_client() + error = client._request_error(422, "plain text rejection") + self.assertIsInstance(error, NonRetryableTTSError) + + +class GenerateChunkTests(unittest.TestCase): + """Chunk generation: WAV output, sub-chunking, bookkeeping.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._sleep = patch("converter.clients.base.time.sleep") + self._sleep.start() + self.addCleanup(self._sleep.stop) + self.addCleanup(self._tmp.cleanup) + + def _make_client(self): + client = SgOmniTTSClient.__new__(SgOmniTTSClient) + from backends.sglomni.catalog import entry_by_key + client.entry = entry_by_key("higgs_audio_v3_tts") + client.chunks_dir = Path(self._tmp.name) + client.api_url = "http://127.0.0.1:8100" + client.voice = None + client.ref_audio = None + client.ref_text = "" + client.instructions = "" + client.language = "English" + client._seed = None + return client + + def _read_wav(self, path): + with wave.open(str(path), "rb") as wav_file: + return wav_file.readframes(wav_file.getnframes()) + + def test_generate_chunk_writes_the_wav_response(self): + client = self._make_client() + with patch.object(client, "_request_wav", return_value=_WAV_BYTES): + result = client.generate_chunk("Hello world.", 1) + self.assertIsNotNone(result) + path = Path(result) + self.assertEqual(path.name, "chunk_0001.wav") + self.assertEqual(self._read_wav(path), _WAV_FRAMES) + + def test_long_text_is_subchunked_and_concatenated(self): + client = self._make_client() + text = " ".join(f"word{i}" for i in range(24)) + responses = [_WAV_BYTES, _WAV_BYTES, _WAV_BYTES] + with patch.object(config, "CHUNK_SIZE", 10), \ + patch.object(client, "_request_wav", + side_effect=responses) as mock_wav, \ + patch("converter.clients.sglomni.concat_audio_files") as mock_concat: + result = client.generate_chunk(text, 1) + # 24 words at CHUNK_SIZE 10 -> three sub-requests (10/10/4). + self.assertEqual(mock_wav.call_count, 3) + self.assertIsNotNone(result) + mock_concat.assert_called_once() + args = mock_concat.call_args[0] + self.assertEqual(len(args[0]), 3) + self.assertEqual(args[1], Path(result)) + + def test_single_subchunk_skips_concatenation(self): + client = self._make_client() + with patch.object(client, "_request_wav", return_value=_WAV_BYTES), \ + patch("converter.clients.sglomni.concat_audio_files") as mock_concat: + client.generate_chunk("Hello.", 1) + mock_concat.assert_not_called() + + def test_empty_text_fails_the_chunk(self): + client = self._make_client() + with patch.object(client, "_request_wav") as mock_wav: + self.assertIsNone(client.generate_chunk(" ", 1)) + mock_wav.assert_not_called() + + def test_request_failure_fails_the_chunk_attempt(self): + client = self._make_client() + with patch.object(client, "_request_wav", + side_effect=RuntimeError("down")) as mock_wav: + self.assertIsNone(client.generate_chunk("Hello.", 1)) + self.assertEqual(mock_wav.call_count, 1) + + def test_stale_chunk_files_are_removed(self): + stale = Path(self._tmp.name) / "chunk_0001.mp3" + stale.write_bytes(b"old") + client = self._make_client() + with patch.object(client, "_request_wav", return_value=_WAV_BYTES): + client.generate_chunk("Hello.", 1) + remaining = sorted(path.name for path in + Path(self._tmp.name).glob("chunk_0001.*")) + self.assertEqual(remaining, ["chunk_0001.wav"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/tests/test_tui.py b/app/tests/test_tui.py index e3383cb..ea2dd79 100644 --- a/app/tests/test_tui.py +++ b/app/tests/test_tui.py @@ -1303,6 +1303,25 @@ class CheckboxTreeTests(TuiTestCase): start_on_buttons=True) self.assertEqual(picked, [(0, "pkg-a")]) + def test_allow_empty_confirms_with_nothing_checked(self): + # allow_empty=True: Confirm on an empty tree returns [] instead + # of flashing — a meaningful answer for pickers where unchecking + # means removing. (Tab first: the focus starts on the rows.) + screen = FakeScreen(keys=[9, 10]) + self.assertEqual( + tui.checkbox_tree(screen, "Pick models", self.FAMILIES, + allow_empty=True), []) + + def test_allow_empty_accepts_a_fully_unchecked_tree(self): + # The modify flow: a pre-checked option is unchecked (Down Down + # Space), then Confirm accepts the now-empty selection. + screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + ord(" "), 9, 10]) + picked = tui.checkbox_tree( + screen, "Pick models", self.FAMILIES, + checked={(0, "pkg-b")}, allow_empty=True) + self.assertEqual(picked, []) + def test_prechecked_options_draw_as_checked(self): screen = FakeScreen(keys=[9, 10]) tui.checkbox_tree(screen, "Pick models", self.FAMILIES, |