aboutsummaryrefslogtreecommitdiff
path: root/app/tests
diff options
context:
space:
mode:
authorhistoria <historiavg@proton.me>2026-08-26 00:21:28 -0400
committerhistoria <historiavg@proton.me>2026-08-26 00:21:28 -0400
commit7cf7d2f7849936e3c84ee431433c2e8bcb6706ac (patch)
tree80d753cbd35929187ffc717a45f7bc3e9cbdbb1e /app/tests
parent4c3e78c39c81d2997e88b84e1b5a25b244e870e4 (diff)
downloadtts-audiobook-generator-7cf7d2f7849936e3c84ee431433c2e8bcb6706ac.tar.gz
fix: harden server start/stop guards
Diffstat (limited to 'app/tests')
-rw-r--r--app/tests/test_backends.py25
-rw-r--r--app/tests/test_backends_faster.py39
-rw-r--r--app/tests/test_backends_servers.py17
3 files changed, 74 insertions, 7 deletions
diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py
index 8ea6a3f..1f146e5 100644
--- a/app/tests/test_backends.py
+++ b/app/tests/test_backends.py
@@ -332,7 +332,9 @@ class QwenUninstallTests(unittest.TestCase):
def test_stops_servers_and_pips(self):
from backends import qwen
- with patch.object(qwen.servers, "stop") as mk_stop, \
+ # Pid files exist for both managed servers, so stop runs.
+ with patch.object(qwen.servers, "pid_for", return_value=1234), \
+ patch.object(qwen.servers, "stop") as mk_stop, \
patch.object(qwen.common, "pip_uninstall",
return_value=0) as mk_pip:
rc = qwen.uninstall(emit="EMIT")
@@ -342,23 +344,36 @@ class QwenUninstallTests(unittest.TestCase):
# The task view's emit is forwarded so pip never touches the terminal.
mk_pip.assert_called_once_with([qwen.QWEN_PIP_PKG], emit="EMIT")
+ def test_skips_stop_when_no_server_was_started(self):
+ # No pid files: stop() is not called (no "not started by this
+ # tool" noise during an uninstall).
+ from backends import qwen
+ with patch.object(qwen.servers, "pid_for", return_value=None), \
+ patch.object(qwen.servers, "stop") as mk_stop, \
+ patch.object(qwen.common, "pip_uninstall", return_value=0):
+ rc = qwen.uninstall()
+ self.assertEqual(rc, 0)
+ mk_stop.assert_not_called()
+
def test_cancel_before_pip_skips_uninstall(self):
import threading
from backends import qwen
cancel = threading.Event()
cancel.set()
- with patch.object(qwen.servers, "stop") as mk_stop, \
+ with patch.object(qwen.servers, "pid_for", return_value=1234), \
+ patch.object(qwen.servers, "stop") as mk_stop, \
patch.object(qwen.common, "pip_uninstall") as mk_pip:
rc = qwen.uninstall(cancel=cancel)
self.assertEqual(rc, 130)
self.assertEqual(mk_stop.call_count, 2)
mk_pip.assert_not_called()
- def test_pip_failure_warns_but_still_succeeds(self):
+ def test_pip_failure_propagates_the_exit_code(self):
from backends import qwen
- with patch.object(qwen.servers, "stop"), \
+ with patch.object(qwen.servers, "pid_for", return_value=1234), \
+ patch.object(qwen.servers, "stop"), \
patch.object(qwen.common, "pip_uninstall",
return_value=1):
rc = qwen.uninstall()
- self.assertEqual(rc, 0)
+ self.assertEqual(rc, 1)
diff --git a/app/tests/test_backends_faster.py b/app/tests/test_backends_faster.py
index 7e27b47..55d617c 100644
--- a/app/tests/test_backends_faster.py
+++ b/app/tests/test_backends_faster.py
@@ -203,6 +203,31 @@ class MainTests(unittest.TestCase):
self.assertTrue(custom.exists())
self.assertFalse(self.output.exists())
+ def test_fresh_install_defaults_voices_json_into_the_checkout(self):
+ # Regression: on a fresh machine the clone runs as part of this
+ # same setup run, so voices.json must be written where detect()
+ # and the server launch read it (the checkout) — not the wav dir.
+ checkout = Path(self._tmp.name) / "faster-qwen3-tts"
+
+ def fake_clone(url, target, emit=None, cancel=None):
+ checkout.mkdir(parents=True, exist_ok=True) # what git would do
+ return 0
+
+ with patch.object(make_voices, "_is_installed", return_value=False), \
+ patch.object(make_voices, "_checkout",
+ return_value=checkout), \
+ patch.object(make_voices.common, "pip_install",
+ return_value=0), \
+ patch.object(make_voices.common, "git_clone",
+ side_effect=fake_clone) as mk_clone:
+ exit_code = self._run([str(self.folder)])
+ self.assertEqual(exit_code, 0)
+ mk_clone.assert_called_once()
+ voices = json.loads(
+ (checkout / "voices.json").read_text(encoding="utf-8"))
+ self.assertEqual(list(voices), ["alpha", "narrator"])
+ self.assertFalse((self.folder / "voices.json").exists())
+
def test_invalid_language_errors_before_work(self):
with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe:
with self.assertRaises(SystemExit) as ctx:
@@ -288,6 +313,8 @@ class UninstallTests(unittest.TestCase):
checkout.mkdir()
with patch.object(make_voices, "_checkout",
return_value=checkout), \
+ patch.object(make_voices.servers, "pid_for",
+ return_value=1234), \
patch.object(make_voices.servers, "stop") as mk_stop, \
patch.object(make_voices.common, "pip_uninstall",
return_value=0) as mk_pip:
@@ -301,17 +328,23 @@ class UninstallTests(unittest.TestCase):
def test_no_checkout_still_uninstalls_the_package(self):
with patch.object(make_voices, "_checkout",
return_value=Path("/no/such/dir")), \
- patch.object(make_voices.servers, "stop"), \
+ patch.object(make_voices.servers, "pid_for",
+ return_value=None), \
+ patch.object(make_voices.servers, "stop") as mk_stop, \
patch.object(make_voices.common, "pip_uninstall",
return_value=0) as mk_pip:
rc = make_voices.uninstall()
self.assertEqual(rc, 0)
+ # No pid file: no stop attempt (and no noise about it).
+ mk_stop.assert_not_called()
mk_pip.assert_called_once_with(["faster-qwen3-tts"], emit=None)
def test_cancel_before_pip_skips_everything_after_stopping(self):
cancel = threading.Event()
cancel.set()
- with patch.object(make_voices.servers, "stop") as mk_stop, \
+ with patch.object(make_voices.servers, "pid_for",
+ return_value=1234), \
+ patch.object(make_voices.servers, "stop") as mk_stop, \
patch.object(make_voices.common, "pip_uninstall") as mk_pip:
rc = make_voices.uninstall(cancel=cancel)
self.assertEqual(rc, 130)
@@ -328,6 +361,8 @@ class UninstallTests(unittest.TestCase):
cancel.set()
with patch.object(make_voices, "_checkout",
return_value=checkout), \
+ patch.object(make_voices.servers, "pid_for",
+ return_value=1234), \
patch.object(make_voices.servers, "stop"), \
patch.object(make_voices.common, "pip_uninstall",
return_value=0):
diff --git a/app/tests/test_backends_servers.py b/app/tests/test_backends_servers.py
index 8eaf53f..ab05eed 100644
--- a/app/tests/test_backends_servers.py
+++ b/app/tests/test_backends_servers.py
@@ -35,6 +35,23 @@ class StartTests(unittest.TestCase):
self.assertTrue(servers.start(self.spec))
mk.assert_not_called()
+ def test_refuses_to_double_start_while_previous_boot_is_alive(self):
+ """A live pid file blocks a second spawn of the same server.
+
+ A previous ``start`` whose server is still booting must not be
+ orphaned by a duplicate process on the same port.
+ """
+ pid_file = self.dir / "test-server.pid"
+ pid_file.write_text("4242", encoding="utf-8")
+ with patch.object(servers, "LOG_DIR", self.dir), \
+ patch.object(servers, "_pid_alive", return_value=True), \
+ patch("subprocess.Popen") as mk, \
+ patch("backends.common.server_running", return_value=False):
+ ok = servers.start(self.spec)
+ self.assertFalse(ok)
+ mk.assert_not_called()
+ self.assertTrue(pid_file.exists())
+
def test_happy_path_spawns_and_polls_until_ready(self):
proc = MagicMock()
proc.pid = 4242