aboutsummaryrefslogtreecommitdiff
path: root/tests/test_backends.py
diff options
context:
space:
mode:
authorhistoria <historiavg@proton.me>2026-08-24 02:59:26 -0400
committerhistoria <historiavg@proton.me>2026-08-24 02:59:26 -0400
commitf00249db9d1ea051d29aa1bcca869fc4b88e83eb (patch)
treea75f076fac1b63e0b4bf2eb8f54affbcc681a891 /tests/test_backends.py
parent9dd4f9595be3b1d76a3a07dc3eca90cfaf8a3f97 (diff)
downloadtts-audiobook-generator-f00249db9d1ea051d29aa1bcca869fc4b88e83eb.tar.gz
refactor: add app directory, dir structure change
Diffstat (limited to 'tests/test_backends.py')
-rw-r--r--tests/test_backends.py178
1 files changed, 0 insertions, 178 deletions
diff --git a/tests/test_backends.py b/tests/test_backends.py
deleted file mode 100644
index c0e8d4a..0000000
--- a/tests/test_backends.py
+++ /dev/null
@@ -1,178 +0,0 @@
-"""Tests for the backends package registry and detection aggregation."""
-
-import tempfile
-import unittest
-from pathlib import Path
-from unittest.mock import patch
-
-from backends import REGISTRY, detect_all, get
-
-
-class RegistryTests(unittest.TestCase):
- def setUp(self):
- # The registry is built lazily on first access (the backend modules
- # pull in converter.tts and its deps, which are only available inside
- # the managed venv). Trigger the build so these tests don't depend on
- # another test class having called detect_all() first.
- get("audiocpp")
-
- def test_registry_has_the_three_backends(self):
- keys = [info.key for info in REGISTRY]
- self.assertEqual(keys, ["audiocpp", "qwen", "faster"])
-
- def test_every_entry_has_detect_and_setup_tui(self):
- for info in REGISTRY:
- self.assertTrue(callable(info.detect), info.key)
- self.assertTrue(callable(info.setup_tui), info.key)
- self.assertIsInstance(info.configure_actions, list)
- for action in info.configure_actions:
- self.assertTrue(callable(action.run))
-
- def test_get_returns_entry_by_key(self):
- self.assertIs(get("audiocpp").key, "audiocpp")
- self.assertIsNone(get("nonexistent"))
-
-
-class DetectAllTests(unittest.TestCase):
- def test_detect_all_returns_one_status_per_backend(self):
- with patch("backends.common.server_running", return_value=False):
- statuses = detect_all()
- self.assertEqual([s.key for s in statuses],
- ["audiocpp", "qwen", "faster"])
- for s in statuses:
- self.assertIn(s.key, ("audiocpp", "qwen", "faster"))
- # ready requires both installed and configured; on a clean
- # machine none are ready.
- if s.ready:
- self.assertTrue(s.installed and s.configured)
- # running is always probed; patched False here so a dev machine
- # running a real server can't flake the test.
- self.assertFalse(s.running)
-
- def test_audiocpp_status_when_cloned_built_configured(self):
- 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")
- from backends import audiocpp
- with patch.object(audiocpp, "find_local_checkout",
- return_value=checkout), \
- patch("backends.common.server_running",
- return_value=False):
- status = audiocpp.detect()
- self.assertTrue(status.installed)
- self.assertTrue(status.configured)
- self.assertTrue(status.ready)
- self.assertFalse(status.running)
- self.assertIn("audiocpp_server", status.launch_hint)
-
- def test_audiocpp_running_when_server_probe_succeeds(self):
- from backends import audiocpp
- with patch.object(audiocpp, "find_local_checkout",
- return_value=None), \
- patch("backends.common.server_running", return_value=True):
- status = audiocpp.detect()
- # Not installed (no checkout) but an external server is up.
- self.assertFalse(status.installed)
- self.assertTrue(status.running)
-
- def test_qwen_status_reflects_install(self):
- from backends import qwen
- with patch.object(qwen, "_is_installed", return_value=True), \
- patch("backends.common.server_running", return_value=False):
- status = qwen.detect()
- self.assertTrue(status.installed)
- self.assertTrue(status.configured)
- self.assertFalse(status.running)
- self.assertIn("qwen-tts-demo", status.launch_hint)
- with patch.object(qwen, "_is_installed", return_value=False), \
- patch("backends.common.server_running", return_value=False):
- status = qwen.detect()
- 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.
- from backends import qwen
- with patch.object(qwen, "_is_installed", return_value=False), \
- patch("backends.common.server_running",
- side_effect=[True, False]):
- status = qwen.detect()
- self.assertTrue(status.running)
- with patch.object(qwen, "_is_installed", return_value=False), \
- patch("backends.common.server_running",
- side_effect=[False, True]):
- status = qwen.detect()
- self.assertTrue(status.running)
-
- def test_faster_status_reflects_install_clone_voices(self):
- from backends import faster
- with tempfile.TemporaryDirectory() as td:
- checkout = Path(td) / "faster-qwen3-tts"
- (checkout / "examples").mkdir(parents=True)
- (checkout / "examples" / "openai_server.py").write_text("x")
- (checkout / "voices.json").write_text('{"default":{}}',
- encoding="utf-8")
- with patch.object(faster, "_is_installed", return_value=True), \
- patch.object(faster, "_checkout",
- return_value=checkout), \
- patch("backends.common.server_running",
- return_value=False):
- status = faster.detect()
- self.assertTrue(status.installed)
- self.assertTrue(status.configured)
- self.assertFalse(status.running)
- self.assertIn("openai_server.py", status.launch_hint)
-
- def test_faster_running_when_server_probe_succeeds(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):
- status = faster.detect()
- self.assertTrue(status.running)
-
-
-class ServerRunningTests(unittest.TestCase):
- """backends.common.server_running: TCP probe against a real socket."""
-
- def test_true_for_open_port(self):
- import socket
-
- from backends import common
- server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- server.bind(("127.0.0.1", 0))
- server.listen(1)
- host, port = server.getsockname()
- url = f"http://127.0.0.1:{port}"
- try:
- self.assertTrue(common.server_running(url))
- finally:
- server.close()
-
- def test_false_for_closed_port(self):
- # Pick an unused port by opening + closing a socket, then probe it.
- import socket
-
- from backends import common
- s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- s.bind(("127.0.0.1", 0))
- _, port = s.getsockname()
- s.close()
- self.assertFalse(common.server_running(f"http://127.0.0.1:{port}"))
-
- def test_false_for_invalid_url(self):
- from backends import common
- self.assertFalse(common.server_running("not a url"))
- self.assertFalse(common.server_running(""))
-
-
-if __name__ == "__main__":
- unittest.main()