diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_backends.py | 11 | ||||
| -rw-r--r-- | tests/test_backends_envs.py | 204 | ||||
| -rw-r--r-- | tests/test_backends_servers.py | 146 | ||||
| -rw-r--r-- | tests/test_hub.py | 145 |
4 files changed, 496 insertions, 10 deletions
diff --git a/tests/test_backends.py b/tests/test_backends.py index 8ee1be8..c0e8d4a 100644 --- a/tests/test_backends.py +++ b/tests/test_backends.py @@ -9,6 +9,13 @@ 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"]) @@ -138,6 +145,7 @@ class ServerRunningTests(unittest.TestCase): 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)) @@ -150,9 +158,10 @@ class ServerRunningTests(unittest.TestCase): server.close() def test_false_for_closed_port(self): - from backends import common # 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() diff --git a/tests/test_backends_envs.py b/tests/test_backends_envs.py new file mode 100644 index 0000000..cf4ecc6 --- /dev/null +++ b/tests/test_backends_envs.py @@ -0,0 +1,204 @@ +"""Tests for the managed Python environment (backends/envs.py).""" + +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +from backends import envs + + +class EnvPathTests(unittest.TestCase): + """Platform-aware path helpers (no venv actually created).""" + + def test_env_dir_under_envs_tts(self): + self.assertEqual(envs.ENV_DIR.name, "tts") + self.assertEqual(envs.ENV_DIR.parent.name, "envs") + + def test_env_python_posix(self): + with patch.object(envs, "_is_windows", return_value=False): + self.assertEqual(envs.env_python(), + envs.ENV_DIR / "bin" / "python") + + def test_env_python_windows(self): + with patch.object(envs, "_is_windows", return_value=True): + self.assertEqual(envs.env_python(), + envs.ENV_DIR / "Scripts" / "python.exe") + + def test_env_script_posix(self): + with patch.object(envs, "_is_windows", return_value=False): + self.assertEqual(envs.env_script("qwen-tts-demo"), + envs.ENV_DIR / "bin" / "qwen-tts-demo") + + def test_env_script_windows(self): + with patch.object(envs, "_is_windows", return_value=True): + self.assertEqual(envs.env_script("qwen-tts-demo"), + envs.ENV_DIR / "Scripts" / "qwen-tts-demo.exe") + + def test_env_exists_false_when_python_missing(self): + with patch.object(envs, "env_python", + return_value=Path("/no/such/path/python")): + self.assertFalse(envs.env_exists()) + + def test_is_managed_env_compares_resolved_executable(self): + fake_env_python = Path("/tmp/opencode/managed-env/bin/python") + with patch.object(envs, "env_python", return_value=fake_env_python), \ + patch.object(sys, "executable", str(fake_env_python)): + self.assertTrue(envs.is_managed_env()) + with patch.object(envs, "env_python", return_value=fake_env_python), \ + patch.object(sys, "executable", "/usr/bin/python3"): + self.assertFalse(envs.is_managed_env()) + + +class CreateEnvTests(unittest.TestCase): + def test_create_env_invokes_venv_module(self): + with patch.object(envs.common, "run_console_subprocess", + return_value=0) as run: + rc = envs.create_env() + self.assertEqual(rc, 0) + argv = run.call_args[0][0] + self.assertEqual(argv[0], sys.executable) + self.assertEqual(argv[1], "-m") + self.assertEqual(argv[2], "venv") + self.assertEqual(argv[3], str(envs.ENV_DIR)) + + def test_create_env_reports_remediation_on_failure(self): + with patch.object(envs.common, "run_console_subprocess", + return_value=1): + rc = envs.create_env() + self.assertEqual(rc, 1) + + +class PipInstallTests(unittest.TestCase): + def test_creates_env_first_when_missing(self): + calls = [] + + def fake_run(argv): + calls.append(list(argv)) + return 0 + + with patch.object(envs, "env_exists", return_value=False), \ + patch.object(envs, "create_env", return_value=0) as mk, \ + patch.object(envs.common, "run_console_subprocess", + side_effect=fake_run): + rc = envs.pip_install(["qwen-tts"]) + self.assertEqual(rc, 0) + mk.assert_called_once_with() + # The actual pip call targets the venv's python. + self.assertEqual(calls[0][0], str(envs.env_python())) + self.assertIn("pip", calls[0]) + self.assertIn("qwen-tts", calls[0]) + + def test_skips_create_when_env_exists(self): + with patch.object(envs, "env_exists", return_value=True), \ + patch.object(envs, "create_env") as mk, \ + patch.object(envs.common, "run_console_subprocess", + return_value=0): + envs.pip_install(["qwen-tts"]) + mk.assert_not_called() + + def test_returns_nonzero_when_create_fails(self): + with patch.object(envs, "env_exists", return_value=False), \ + patch.object(envs, "create_env", return_value=1), \ + patch.object(envs.common, "run_console_subprocess") as run: + rc = envs.pip_install(["qwen-tts"]) + self.assertEqual(rc, 1) + run.assert_not_called() + + +class ModuleAvailableTests(unittest.TestCase): + def test_false_when_env_missing(self): + with patch.object(envs, "env_exists", return_value=False): + self.assertFalse(envs.module_available("qwen_tts")) + + def test_true_when_subprocess_exits_zero(self): + import subprocess + fake = subprocess.CompletedProcess(args=["x"], returncode=0) + with patch.object(envs, "env_exists", return_value=True), \ + patch("subprocess.run", return_value=fake) as run: + self.assertTrue(envs.module_available("qwen_tts")) + argv = run.call_args[0][0] + self.assertEqual(argv[0], str(envs.env_python())) + self.assertIn("import qwen_tts", argv[2]) + + def test_false_when_subprocess_exits_nonzero(self): + import subprocess + fake = subprocess.CompletedProcess(args=["x"], returncode=1) + with patch.object(envs, "env_exists", return_value=True), \ + patch("subprocess.run", return_value=fake): + self.assertFalse(envs.module_available("qwen_tts")) + + def test_false_on_timeout(self): + import subprocess + with patch.object(envs, "env_exists", return_value=True), \ + patch("subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="x", timeout=1)): + self.assertFalse(envs.module_available("qwen_tts")) + + +class EnsureAppEnvTests(unittest.TestCase): + def test_creates_env_then_installs_when_marker_invalid(self): + with patch.object(envs, "env_exists", return_value=False), \ + patch.object(envs, "create_env", return_value=0), \ + patch.object(envs, "_marker_valid", return_value=False), \ + patch.object(envs, "install_requirements", return_value=0), \ + patch.object(envs, "_write_marker") as mk: + envs.ensure_app_env() + mk.assert_called_once_with() + + def test_raises_when_create_fails(self): + with patch.object(envs, "env_exists", return_value=False), \ + patch.object(envs, "create_env", return_value=1): + with self.assertRaises(RuntimeError): + envs.ensure_app_env() + + def test_raises_when_install_fails(self): + with patch.object(envs, "env_exists", return_value=True), \ + patch.object(envs, "_marker_valid", return_value=False), \ + patch.object(envs, "install_requirements", return_value=1): + with self.assertRaises(RuntimeError): + envs.ensure_app_env() + + def test_skips_install_when_marker_valid(self): + with patch.object(envs, "env_exists", return_value=True), \ + patch.object(envs, "_marker_valid", return_value=True), \ + patch.object(envs, "install_requirements") as mk: + envs.ensure_app_env() + mk.assert_not_called() + + +class BootstrapTests(unittest.TestCase): + def test_noop_when_already_managed(self): + with patch.object(envs, "is_managed_env", return_value=True), \ + patch.object(envs, "ensure_app_env") as mk, \ + patch("os.execv") as ex: + envs.bootstrap("/path/to/audiobook.py") + mk.assert_not_called() + ex.assert_not_called() + + def test_ensures_env_then_execvs(self): + with patch.object(envs, "is_managed_env", return_value=False), \ + patch.object(envs, "ensure_app_env") as mk_env, \ + patch("os.execv") as ex, \ + patch.object(sys, "argv", ["audiobook.py", "--backend", "qwen"]): + envs.bootstrap("/path/to/audiobook.py") + mk_env.assert_called_once_with() + py = str(envs.env_python()) + args = ex.call_args[0] + self.assertEqual(args[0], py) + self.assertEqual(args[1][0], py) + self.assertTrue(args[1][1].endswith("audiobook.py")) + self.assertEqual(args[1][2:], ["--backend", "qwen"]) + + def test_exits_when_ensure_raises(self): + with patch.object(envs, "is_managed_env", return_value=False), \ + patch.object(envs, "ensure_app_env", + side_effect=RuntimeError("boom")), \ + patch("os.execv") as ex, \ + self.assertRaises(SystemExit): + envs.bootstrap("/path/to/audiobook.py") + ex.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_backends_servers.py b/tests/test_backends_servers.py new file mode 100644 index 0000000..02b65e6 --- /dev/null +++ b/tests/test_backends_servers.py @@ -0,0 +1,146 @@ +"""Tests for the server lifecycle module (backends/servers.py).""" + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from backends import ServerSpec, servers + + +class StartTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + # A fake executable so Path(argv[0]).exists() passes. + self.exe = self.dir / "fake_server" + self.exe.write_bytes(b"#!/bin/sh\n") + self.spec = ServerSpec("test", "http://127.0.0.1:9999", + [str(self.exe), "--port", "9999"]) + + def tearDown(self): + self._tmp.cleanup() + + def test_returns_false_when_executable_missing(self): + spec = ServerSpec("nope", "http://127.0.0.1:1", ["/no/such/binary"]) + with patch.object(servers, "LOG_DIR", self.dir): + self.assertFalse(servers.start(spec)) + + def test_noop_when_already_running(self): + with patch.object(servers, "LOG_DIR", self.dir), \ + patch("backends.common.server_running", return_value=True), \ + patch("subprocess.Popen") as mk: + self.assertTrue(servers.start(self.spec)) + mk.assert_not_called() + + def test_happy_path_spawns_and_polls_until_ready(self): + proc = MagicMock() + proc.pid = 4242 + proc.poll.return_value = None # process still running + # server_running: False on the pre-check, True once inside the loop. + with patch.object(servers, "LOG_DIR", self.dir), \ + patch("subprocess.Popen", return_value=proc) as mk, \ + patch("backends.common.server_running", + side_effect=[False, True]), \ + patch("time.sleep"): + ok = servers.start(self.spec) + self.assertTrue(ok) + mk.assert_called_once() + # Pid file written. + self.assertEqual( + (self.dir / "test-server.pid").read_text(encoding="utf-8"), + "4242") + + def test_returns_false_when_process_exits_early(self): + proc = MagicMock() + proc.pid = 99 + proc.poll.return_value = 1 # exited with code 1 + 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) + self.assertFalse(ok) + # Pid file cleaned up after early exit. + self.assertFalse((self.dir / "test-server.pid").exists()) + + def test_returns_false_on_timeout(self): + proc = MagicMock() + proc.pid = 7 + proc.poll.return_value = None + # time.time: first call < deadline loop entry, then past deadline. + times = iter([0.0, float(servers.SERVER_START_TIMEOUT + 1)]) + 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"), \ + patch("time.time", side_effect=lambda: next(times)): + ok = servers.start(self.spec) + self.assertFalse(ok) + + +class StopTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def _write_pid(self, name, pid): + (self.dir / f"{name}-server.pid").write_text(str(pid), + encoding="utf-8") + + def test_returns_false_when_no_pid_file(self): + with patch.object(servers, "LOG_DIR", self.dir): + self.assertFalse(servers.stop("test")) + + def test_stops_alive_process_and_removes_pid_file(self): + self._write_pid("test", 1234) + with patch.object(servers, "LOG_DIR", self.dir), \ + patch.object(servers, "_pid_alive", return_value=True), \ + patch.object(servers, "_kill_pid", return_value=True) as mk: + ok = servers.stop("test") + self.assertTrue(ok) + mk.assert_called_once_with(1234) + self.assertFalse((self.dir / "test-server.pid").exists()) + + def test_already_dead_returns_true_and_cleans_pid_file(self): + self._write_pid("test", 1234) + with patch.object(servers, "LOG_DIR", self.dir), \ + patch.object(servers, "_pid_alive", return_value=False), \ + patch.object(servers, "_kill_pid") as mk: + ok = servers.stop("test") + self.assertTrue(ok) + mk.assert_not_called() + self.assertFalse((self.dir / "test-server.pid").exists()) + + def test_corrupt_pid_file_returns_false_and_cleans(self): + (self.dir / "test-server.pid").write_text("not-a-number", + encoding="utf-8") + with patch.object(servers, "LOG_DIR", self.dir): + self.assertFalse(servers.stop("test")) + self.assertFalse((self.dir / "test-server.pid").exists()) + + +class PidForTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_none_when_no_pid_file(self): + with patch.object(servers, "LOG_DIR", self.dir): + self.assertIsNone(servers.pid_for("test")) + + def test_returns_pid_from_file(self): + (self.dir / "test-server.pid").write_text("555\n", + encoding="utf-8") + with patch.object(servers, "LOG_DIR", self.dir): + self.assertEqual(servers.pid_for("test"), 555) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_hub.py b/tests/test_hub.py index ce9af43..21f795e 100644 --- a/tests/test_hub.py +++ b/tests/test_hub.py @@ -8,8 +8,9 @@ import unittest from pathlib import Path from unittest.mock import patch -from ui import hub, tui +from backends import BackendStatus, ServerSpec from tests.test_tui import FakeCurses, FakeScreen +from ui import hub, tui class HubHelperTests(unittest.TestCase): @@ -93,7 +94,7 @@ class HubMenuTests(unittest.TestCase): labels = [label for label, _ in captured["options"]] self.assertEqual(labels, ["Set up a backend...", "Quit"]) - def test_menu_has_all_four_when_one_installed(self): + def test_menu_has_all_five_when_one_installed(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): @@ -111,7 +112,7 @@ class HubMenuTests(unittest.TestCase): self.assertEqual( labels, ["Convert books...", "Set up a backend...", - "Configure a backend...", "Quit"]) + "Configure a backend...", "Server...", "Quit"]) # The status table is passed through, one row per backend. self.assertEqual(captured["rows"], [("qwen-tts", "installed", "warn", "body")]) @@ -137,9 +138,9 @@ class HubMenuTests(unittest.TestCase): [("audio.cpp", "unavailable", "err", "dim"), ("qwen-tts", "running", "ok", "body")]) - def test_menu_has_all_four_when_one_running_only(self): + def test_menu_has_all_five_when_one_running_only(self): # Running but not installed (an external server) still unlocks the - # Convert/Configure entries. + # Convert/Configure/Server entries. captured = {} def fake_menu(stdscr, title, options, **kwargs): @@ -156,14 +157,14 @@ class HubMenuTests(unittest.TestCase): self.assertEqual( labels, ["Convert books...", "Set up a backend...", - "Configure a backend...", "Quit"]) + "Configure a backend...", "Server...", "Quit"]) def test_convert_with_no_available_backend_offers_setup(self): # One installed-but-not-ready backend → Convert is offered. The # convert menu lists no available backend, so only "Set up a # backend..." is shown; Enter selects it → setup menu lists 3 # backends; Esc goes back → convert returns None → main menu loops. - # Then quit: main menu now has 4 options, Quit is the 4th (Down x3). + # Then quit: main menu now has 5 options, Quit is the 5th (Down x4). from backends import BackendInfo, BackendStatus none = BackendStatus("k", "l", installed=True, configured=False) infos = [BackendInfo("audiocpp", "audio.cpp", lambda: none, @@ -181,13 +182,139 @@ class HubMenuTests(unittest.TestCase): with patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub, "REGISTRY", infos): # Convert(Enter), setup-entry(Enter), Esc on setup menu, - # back at main menu -> Down x3 -> Enter (Quit). + # back at main menu -> Down x4 -> Enter (Quit). screen = FakeScreen(keys=[10, 10, 27, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, - FakeCurses.KEY_DOWN, 10]) + FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + 10]) result = hub._hub_menu(screen) self.assertIsNone(result) +class SelectSpecTests(unittest.TestCase): + """_select_spec: mode-aware server selection (qwen has two servers).""" + + def _qwen_status(self): + return BackendStatus( + "qwen", "qwen-tts", installed=True, configured=True, + servers=[ServerSpec("qwen-custom", "http://127.0.0.1:7860", []), + ServerSpec("qwen-clone", "http://127.0.0.1:7861", [])]) + + def test_qwen_custom_mode(self): + spec = hub._select_spec(self._qwen_status(), {"clone": None}) + self.assertEqual(spec.name, "qwen-custom") + + def test_qwen_clone_mode(self): + spec = hub._select_spec(self._qwen_status(), {"clone": "ref.wav"}) + self.assertEqual(spec.name, "qwen-clone") + + def test_audiocpp_returns_single_spec(self): + st = BackendStatus("audiocpp", "audio.cpp", installed=True, + configured=True, + servers=[ServerSpec("audiocpp", "http://x", [])]) + spec = hub._select_spec(st, {}) + self.assertEqual(spec.name, "audiocpp") + + def test_none_when_no_servers(self): + st = BackendStatus("qwen", "qwen-tts", installed=False, + configured=False) + self.assertIsNone(hub._select_spec(st, {})) + + +class RunConversionTests(unittest.TestCase): + """_run_conversion: autostart, hint-when-manual, and stop-after.""" + + def test_autostart_starts_server_then_converts(self): + spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) + status = BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, running=False, + servers=[spec]) + kwargs = {"autostart": "qwen-custom"} + with patch.object(hub, "detect_all", return_value=[status]), \ + patch.object(hub, "_find_spec", return_value=spec), \ + patch.object(hub.servers, "start", return_value=True) as mk_start, \ + patch.object(hub.audiobook, "convert", return_value=0) as mk_conv, \ + patch("builtins.input", return_value="n") as mk_input, \ + patch.object(hub.servers, "stop") as mk_stop: + hub._run_conversion("qwen", kwargs) + mk_start.assert_called_once_with(spec) + mk_conv.assert_called_once() + # User declined stopping → stop not called. + mk_stop.assert_not_called() + + def test_autostart_stop_when_user_says_yes(self): + spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) + status = BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, running=False, + servers=[spec]) + kwargs = {"autostart": "qwen-custom"} + with patch.object(hub, "detect_all", return_value=[status]), \ + patch.object(hub, "_find_spec", return_value=spec), \ + patch.object(hub.servers, "start", return_value=True), \ + patch.object(hub.audiobook, "convert", return_value=0), \ + patch("builtins.input", return_value="y"), \ + patch.object(hub.servers, "stop") as mk_stop: + hub._run_conversion("qwen", kwargs) + mk_stop.assert_called_once_with("qwen-custom") + + def test_autostart_aborts_when_server_fails(self): + spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) + status = BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, running=False, + launch_hint="hint cmd", servers=[spec]) + kwargs = {"autostart": "qwen-custom"} + with patch.object(hub, "detect_all", return_value=[status]), \ + patch.object(hub, "_find_spec", return_value=spec), \ + patch.object(hub.servers, "start", return_value=False), \ + patch.object(hub.audiobook, "convert") as mk_conv, \ + patch.object(hub.servers, "stop") as mk_stop: + hub._run_conversion("qwen", kwargs) + mk_conv.assert_not_called() + mk_stop.assert_not_called() + + def test_no_autostart_prints_hint_when_not_running(self): + status = BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, running=False, + launch_hint="the-hint") + with patch.object(hub, "detect_all", return_value=[status]), \ + patch.object(hub.audiobook, "convert", return_value=0) as mk_conv: + hub._run_conversion("qwen", {}) + mk_conv.assert_called_once() + + +class AddAutostartTests(unittest.TestCase): + """_add_autostart: offers to start the server when it isn't running.""" + + def setUp(self): + tui._THEME.clear() + self.curses = FakeCurses() + self._patcher = patch.dict("sys.modules", {"curses": self.curses}) + self._patcher.start() + self.addCleanup(self._patcher.stop) + self.addCleanup(tui._THEME.clear) + + def _status(self): + spec = ServerSpec("qwen-custom", "http://127.0.0.1:7860", ["x"]) + return BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, running=False, + servers=[spec]) + + def test_sets_autostart_when_user_confirms(self): + screen = FakeScreen(keys=[10]) # Enter = Yes + cmd = ("convert", "qwen", {"clone": None}) + with patch.object(hub, "detect_all", return_value=[self._status()]), \ + patch("backends.common.server_running", return_value=False): + hub._add_autostart(screen, cmd, [self._status()]) + self.assertEqual(cmd[2]["autostart"], "qwen-custom") + + def test_no_autostart_when_server_already_running(self): + screen = FakeScreen(keys=[10]) + cmd = ("convert", "qwen", {"clone": None}) + with patch.object(hub, "detect_all", return_value=[self._status()]), \ + patch("backends.common.server_running", return_value=True): + hub._add_autostart(screen, cmd, [self._status()]) + self.assertNotIn("autostart", cmd[2]) + + if __name__ == "__main__": unittest.main() |