aboutsummaryrefslogtreecommitdiff
path: root/app/tests
diff options
context:
space:
mode:
Diffstat (limited to 'app/tests')
-rw-r--r--app/tests/test_backends.py13
-rw-r--r--app/tests/test_backends_audiocpp.py122
-rw-r--r--app/tests/test_backends_servers.py102
-rw-r--r--app/tests/test_converter_progress.py183
-rw-r--r--app/tests/test_hub.py191
-rw-r--r--app/tests/test_runview.py210
6 files changed, 741 insertions, 80 deletions
diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py
index 2cb5a95..acee6b6 100644
--- a/app/tests/test_backends.py
+++ b/app/tests/test_backends.py
@@ -5,7 +5,18 @@ import unittest
from pathlib import Path
from unittest.mock import patch
-from backends import REGISTRY, detect_all, get
+from backends import REGISTRY, ServerSpec, detect_all, format_launch_hint, get
+
+
+class FormatLaunchHintTests(unittest.TestCase):
+ def test_plain_specs_join_argv(self):
+ specs = [ServerSpec("a", "http://x", ["cmd", "--flag"])]
+ self.assertEqual(format_launch_hint(specs), "cmd --flag")
+
+ def test_cwd_prefixes_the_command(self):
+ specs = [ServerSpec("a", "http://x", ["cmd"], cwd=Path("/opt/audio.cpp"))]
+ self.assertEqual(format_launch_hint(specs),
+ "cd /opt/audio.cpp && cmd")
class RegistryTests(unittest.TestCase):
diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py
index e2b09d0..563ed78 100644
--- a/app/tests/test_backends_audiocpp.py
+++ b/app/tests/test_backends_audiocpp.py
@@ -1148,5 +1148,127 @@ class FetchServerEndpointsTests(unittest.TestCase):
make_server.fetch_server_voices("http://h", "qwen"))
+class MissingModelEntriesTests(unittest.TestCase):
+ """missing_model_entries: server.json paths vs. files on disk."""
+
+ def setUp(self):
+ self._tmp = tempfile.TemporaryDirectory()
+ self.dir = Path(self._tmp.name)
+
+ def tearDown(self):
+ self._tmp.cleanup()
+
+ def _server_json(self, models):
+ path = self.dir / "server.json"
+ path.write_text(json.dumps({"models": models}), encoding="utf-8")
+ return path
+
+ def test_relative_path_resolves_against_config_dir(self):
+ (self.dir / "models" / "present").mkdir(parents=True)
+ (self.dir / "models" / "present" / "m.gguf").write_bytes(b"x")
+ path = self._server_json([
+ {"id": "a", "path": "models/present"},
+ {"id": "b", "path": "models/absent"},
+ ])
+ missing = make_server.missing_model_entries(path)
+ self.assertEqual([m["id"] for m in missing], ["b"])
+
+ def test_empty_directory_counts_as_missing(self):
+ (self.dir / "models" / "empty").mkdir(parents=True)
+ path = self._server_json([{"id": "a", "path": "models/empty"}])
+ self.assertEqual(len(make_server.missing_model_entries(path)), 1)
+
+ def test_absolute_paths_honored(self):
+ target = self.dir / "absolute"
+ target.mkdir()
+ (target / "m.gguf").write_bytes(b"x")
+ path = self._server_json([{"id": "a", "path": str(target)}])
+ self.assertEqual(make_server.missing_model_entries(path), [])
+
+ def test_unreadable_json_returns_empty(self):
+ path = self.dir / "server.json"
+ path.write_text("not json", encoding="utf-8")
+ self.assertEqual(make_server.missing_model_entries(path), [])
+
+ def test_no_models_returns_empty(self):
+ path = self._server_json([])
+ self.assertEqual(make_server.missing_model_entries(path), [])
+
+
+class ModelInstallHintsTests(unittest.TestCase):
+ """model_install_hints: maps missing paths to the install command."""
+
+ def test_maps_path_to_install_id_via_catalog(self):
+ import tempfile
+ with tempfile.TemporaryDirectory() as td:
+ checkout = Path(td)
+ specs = checkout / "model_specs"
+ specs.mkdir()
+ (specs / "qwen3_tts.json").write_text(json.dumps({
+ "family": "qwen3_tts", "category": "tts",
+ "tasks": ["tts"],
+ "packages": [{
+ "id": "qwen3_tts_0_6b_base_q8_0", "format": "gguf",
+ "target_directory": "Qwen3-TTS-12Hz-0.6B-Base-GGUF",
+ }],
+ }), encoding="utf-8")
+ missing = [{"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}]
+ hints = make_server.model_install_hints(checkout, missing)
+ self.assertEqual(len(hints), 1)
+ self.assertIn("qwen3_tts_0_6b_base_q8_0", hints[0])
+
+ def test_unmapped_path_names_the_path(self):
+ import tempfile
+ with tempfile.TemporaryDirectory() as td:
+ checkout = Path(td)
+ (checkout / "model_specs").mkdir()
+ hints = make_server.model_install_hints(
+ checkout, [{"id": "x", "rel": "models/nope"}])
+ self.assertIn("models/nope", hints[0])
+ self.assertNotIn("install", hints[0])
+
+
+class DetectServerSpecTests(unittest.TestCase):
+ """detect(): the server spec carries the checkout cwd + identity."""
+
+ def _checkout(self):
+ tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(tmp.cleanup)
+ checkout = Path(tmp.name)
+ (checkout / "model_specs").mkdir()
+ build = checkout / "build" / "linux-cuda-release" / "bin"
+ build.mkdir(parents=True)
+ (build / "audiocpp_server").write_bytes(b"x")
+ (checkout / "server.json").write_text(json.dumps({
+ "models": [{"id": "qwen", "family": "qwen3_tts",
+ "path": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}],
+ }), encoding="utf-8")
+ return checkout
+
+ def test_spec_has_cwd_and_identity(self):
+ checkout = self._checkout()
+ with patch.object(make_server, "find_local_checkout",
+ return_value=checkout), \
+ patch.object(make_server, "_detect_remote",
+ return_value=(False, {})):
+ status = make_server.detect()
+ self.assertEqual(len(status.servers), 1)
+ spec = status.servers[0]
+ self.assertEqual(spec.cwd, checkout)
+ self.assertEqual(spec.identity, "audiocpp")
+ self.assertIn("--config", spec.argv)
+
+ def test_models_missing_flag_and_details(self):
+ checkout = self._checkout()
+ with patch.object(make_server, "find_local_checkout",
+ return_value=checkout), \
+ patch.object(make_server, "_detect_remote",
+ return_value=(False, {})):
+ status = make_server.detect()
+ self.assertTrue(status.models_missing)
+ self.assertTrue(any("not downloaded" in line
+ for line in status.details))
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/app/tests/test_backends_servers.py b/app/tests/test_backends_servers.py
index 987ff24..201d5b6 100644
--- a/app/tests/test_backends_servers.py
+++ b/app/tests/test_backends_servers.py
@@ -78,6 +78,108 @@ class StartTests(unittest.TestCase):
ok = servers.start(self.spec)
self.assertFalse(ok)
+ def _boot_proc(self):
+ proc = MagicMock()
+ proc.pid = 4242
+ proc.poll.return_value = None
+ return proc
+
+ def test_cwd_passed_to_popen(self):
+ """A spec with a cwd spawns the server in that working directory.
+
+ audio.cpp resolves model_specs/<family>.json relative to its
+ process working directory, so the hub must start it from the
+ checkout.
+ """
+ spec = ServerSpec("test", "http://127.0.0.1:9999",
+ [str(self.exe)], cwd=Path("/opt/audio.cpp"))
+ proc = self._boot_proc()
+ 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(spec)
+ self.assertTrue(ok)
+ kwargs = mk.call_args.kwargs
+ self.assertEqual(kwargs.get("cwd"), "/opt/audio.cpp")
+
+ def test_identity_spec_waits_for_http_identity(self):
+ """Readiness needs the server to answer HTTP as its identity.
+
+ A TCP-accepting but still-booting server (lazy model load, slow
+ listen-before-serve) must not count as ready.
+ """
+ spec = ServerSpec("test", "http://127.0.0.1:9999",
+ [str(self.exe)], identity="audiocpp")
+ proc = self._boot_proc()
+ with patch.object(servers, "LOG_DIR", self.dir), \
+ patch("subprocess.Popen", return_value=proc), \
+ patch("backends.common.server_running", return_value=True), \
+ patch.object(servers.probe, "identify_server",
+ side_effect=[None, None, "audiocpp"]), \
+ patch("time.sleep"):
+ ok = servers.start(spec)
+ self.assertTrue(ok)
+
+ def test_faster_identity_requires_model_loaded(self):
+ """The faster identity additionally waits for /health model_loaded."""
+ spec = ServerSpec("test", "http://127.0.0.1:9999",
+ [str(self.exe)], identity="faster")
+ proc = self._boot_proc()
+ with patch.object(servers, "LOG_DIR", self.dir), \
+ patch("subprocess.Popen", return_value=proc), \
+ patch("backends.common.server_running", return_value=True), \
+ patch.object(servers.probe, "identify_server",
+ return_value="faster"), \
+ patch.object(servers.probe, "faster_model_loaded",
+ side_effect=[False, True]), \
+ patch("time.sleep"):
+ ok = servers.start(spec)
+ self.assertTrue(ok)
+
+ def test_progress_receives_boot_events(self):
+ events = []
+ proc = self._boot_proc()
+ with patch.object(servers, "LOG_DIR", self.dir), \
+ patch("subprocess.Popen", return_value=proc), \
+ patch("backends.common.server_running",
+ side_effect=[False, True]), \
+ patch("time.sleep"):
+ ok = servers.start(self.spec, progress=events.append)
+ self.assertTrue(ok)
+ kinds = [event["kind"] for event in events]
+ self.assertEqual(kinds, ["starting", "ready"])
+ self.assertEqual(events[0]["pid"], 4242)
+ self.assertIn("--port", events[0]["argv"])
+
+ def test_cancel_aborts_boot_kills_process_and_reports(self):
+ import threading
+ cancel = threading.Event()
+ cancel.set()
+ proc = self._boot_proc()
+ with patch.object(servers, "LOG_DIR", self.dir), \
+ patch("subprocess.Popen", return_value=proc), \
+ patch("backends.common.server_running", return_value=False), \
+ patch.object(servers, "_kill_pid", return_value=True) as mk, \
+ patch("time.sleep"):
+ ok = servers.start(self.spec, cancel=cancel)
+ self.assertFalse(ok)
+ mk.assert_called_once_with(4242)
+ self.assertFalse((self.dir / "test-server.pid").exists())
+
+ def test_console_progress_prints_events(self):
+ import io
+ from contextlib import redirect_stdout
+ buf = io.StringIO()
+ with redirect_stdout(buf):
+ servers._console_progress({"kind": "running", "name": "test",
+ "url": "http://127.0.0.1:9999"})
+ servers._console_progress({"kind": "error", "message": "boom"})
+ out = buf.getvalue()
+ self.assertIn("already running", out)
+ self.assertIn("boom", out)
+
class StopTests(unittest.TestCase):
def setUp(self):
diff --git a/app/tests/test_converter_progress.py b/app/tests/test_converter_progress.py
new file mode 100644
index 0000000..1041173
--- /dev/null
+++ b/app/tests/test_converter_progress.py
@@ -0,0 +1,183 @@
+"""Tests for the converter's progress-event and cancellation plumbing.
+
+These exercise the wiring the TUI run view relies on: a ``progress``
+callback receiving book/chunk/done events, a ``cancel`` (threading.Event)
+aborting the run between chunks (raising ConversionCancelled), and the
+injectable ``confirm`` hook on the overwrite prompt.
+"""
+
+import io
+import tempfile
+import threading
+import unittest
+from contextlib import redirect_stdout
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from converter import config, tts
+from converter import converter as converter_mod
+from converter.converter import (
+ AudiobookConverter,
+ ConversionCancelled,
+ prompt_overwrite,
+ voice_mode_for,
+)
+
+
+class VoiceModeForTests(unittest.TestCase):
+ def test_faster_always_clones(self):
+ self.assertEqual(voice_mode_for(tts.BACKEND_FASTER),
+ tts.VOICE_MODE_CLONE)
+
+ def test_audiocpp_voice_clones(self):
+ self.assertEqual(voice_mode_for(tts.BACKEND_AUDIOCPP, voice="narrator"),
+ tts.VOICE_MODE_CLONE)
+
+ def test_audiocpp_no_voice_is_custom(self):
+ self.assertEqual(voice_mode_for(tts.BACKEND_AUDIOCPP),
+ tts.VOICE_MODE_CUSTOM)
+
+ def test_qwen_clone_wav_clones(self):
+ self.assertEqual(voice_mode_for(tts.BACKEND_QWEN, clone="x.wav"),
+ tts.VOICE_MODE_CLONE)
+
+ def test_qwen_no_clone_is_custom(self):
+ self.assertEqual(voice_mode_for(tts.BACKEND_QWEN),
+ tts.VOICE_MODE_CUSTOM)
+
+
+class PromptOverwriteConfirmTests(unittest.TestCase):
+ def test_confirm_callback_receives_message_and_default(self):
+ calls = []
+ result = prompt_overwrite([Path("out.mp3")], "out",
+ confirm=lambda m, d: calls.append((m, d)) or False)
+ self.assertFalse(result)
+ self.assertEqual(len(calls), 1)
+ self.assertTrue(calls[0][1]) # default yes
+ self.assertIn("out.mp3", calls[0][0])
+
+
+class _ConvertFixture:
+ """A real AudiobookConverter whose TTS client is stubbed."""
+
+ def __init__(self, test_case):
+ self.test = test_case
+ self._books_tmp = tempfile.TemporaryDirectory()
+ self._output_tmp = tempfile.TemporaryDirectory()
+ self._orig = (converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER)
+ converter_mod.BOOKS_FOLDER = Path(self._books_tmp.name)
+ converter_mod.AUDIOBOOKS_FOLDER = Path(self._output_tmp.name)
+ (converter_mod.BOOKS_FOLDER / "book.txt").write_text(
+ "one two three four five", encoding="utf-8")
+ # The stub returns a path that does not exist on disk, so the
+ # final assembly (and cover art) is patched out of the run() path.
+ self._patchers = [
+ patch.object(converter_mod.audio, "combine_chunks",
+ return_value=True),
+ patch.object(converter_mod.audio, "combine_chapters_to_m4b",
+ return_value=True),
+ patch.object(converter_mod.cover, "generate_cover",
+ return_value=None),
+ ]
+ for patcher in self._patchers:
+ patcher.start()
+ test_case.addCleanup(self.cleanup)
+
+ def cleanup(self):
+ converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = self._orig
+ for patcher in self._patchers:
+ patcher.stop()
+ self._books_tmp.cleanup()
+ self._output_tmp.cleanup()
+
+ def build(self, progress=None, cancel=None):
+ # Patch the TTS client construction so the real constructor runs
+ # (exercising the progress/cancel wiring) without dialing a server.
+ with patch.object(converter_mod, "QwenTTSClient",
+ return_value=MagicMock()):
+ converter = AudiobookConverter(
+ voice_mode=tts.VOICE_MODE_CUSTOM, backend=tts.BACKEND_QWEN,
+ output_format="mp3", language="English",
+ progress=progress, cancel=cancel)
+ converter.tts.process_chunk_with_retry.return_value = "chunk_0001.wav"
+ converter._book_files = [converter_mod.BOOKS_FOLDER / "book.txt"]
+ converter._planned = [(converter_mod.BOOKS_FOLDER / "book.txt",
+ "book_Vivian")]
+ return converter
+
+
+class ProgressEventTests(unittest.TestCase):
+ def setUp(self):
+ self.fixture = _ConvertFixture(self)
+
+ def test_run_emits_book_chunks_done(self):
+ events = []
+ converter = self.fixture.build(progress=events.append)
+ ok = converter.run()
+ self.assertTrue(ok)
+ kinds = [event["kind"] for event in events]
+ self.assertEqual(kinds, ["book", "chunks", "chunk_done",
+ "book_done", "done"])
+ self.assertEqual(events[0]["name"], "book.txt")
+ self.assertEqual(events[-1]["ok"], 1)
+
+ def test_run_suppresses_console_prints_when_progress_set(self):
+ buf = io.StringIO()
+ converter = self.fixture.build(progress=lambda e: None)
+ with redirect_stdout(buf):
+ converter.run()
+ # The banner/summary/chunk prints are replaced by events.
+ out = buf.getvalue()
+ self.assertNotIn("CONVERSION SUMMARY", out)
+ self.assertNotIn("PROCESSING", out)
+ self.assertNotIn("completed", out)
+
+ def test_chunk_failed_sets_error_state(self):
+ events = []
+ converter = self.fixture.build(progress=events.append)
+ converter.tts.process_chunk_with_retry.return_value = None
+ converter.run()
+ self.assertIn("chunk_failed",
+ [event["kind"] for event in events])
+ self.assertEqual(events[-1]["kind"], "done")
+ self.assertEqual(events[-1]["ok"], 0)
+
+
+class CancelTests(unittest.TestCase):
+ def setUp(self):
+ self.fixture = _ConvertFixture(self)
+
+ def test_cancel_between_chunks_aborts_and_emits_cancelled(self):
+ events = []
+ cancel = threading.Event()
+ converter = self.fixture.build(progress=events.append, cancel=cancel)
+ # Cancel as the first chunk completes; the next chunk's pre-check
+ # must raise ConversionCancelled before requesting it.
+ def generate(chunk_num, text):
+ cancel.set()
+ return "chunk_0001.wav"
+
+ converter.tts.process_chunk_with_retry.side_effect = generate
+ with patch.object(converter_mod, "chunking") as mk_chunking:
+ mk_chunking.split_into_chunks.return_value = [
+ "one two", "three four", "five"]
+ converter.run()
+ kinds = [event["kind"] for event in events]
+ self.assertIn("cancelled", kinds)
+ self.assertEqual(events[-1]["kind"], "done")
+ self.assertTrue(events[-1]["cancelled"])
+
+ def test_check_cancelled_raises_when_event_set(self):
+ cancel = threading.Event()
+ cancel.set()
+ converter = self.fixture.build(cancel=cancel)
+ with self.assertRaises(ConversionCancelled):
+ converter._check_cancelled()
+
+ def test_check_cancelled_silent_when_not_set(self):
+ converter = self.fixture.build(cancel=threading.Event())
+ converter._check_cancelled() # no raise
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py
index a0545cc..5f91a61 100644
--- a/app/tests/test_hub.py
+++ b/app/tests/test_hub.py
@@ -917,92 +917,125 @@ class SelectSpecTests(unittest.TestCase):
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()
-
- def test_remote_conversion_skips_setup_checks(self):
- # A remote conversion targets an external server: no autostart, no
- # "not fully set up" warning, no launch hint — just convert.
- with patch.object(hub, "detect_all", return_value=[]) as mk_detect, \
- patch.object(hub.audiobook, "convert", return_value=0) as mk_conv:
- hub._run_conversion("audiocpp", {"api_url": "http://10.0.0.5:8080"})
- mk_conv.assert_called_once_with(
- backend="audiocpp", api_url="http://10.0.0.5:8080")
+class PrepareRunConfigTests(unittest.TestCase):
+ """_prepare_run_config: the run view's inputs from the accepted form."""
+
+ def _spec(self, name="qwen-custom", url="http://127.0.0.1:7860"):
+ return ServerSpec(name, url, ["x"])
+
+ def test_remote_targets_the_api_url(self):
+ with patch.object(hub, "detect_all", return_value=[]) as mk_detect:
+ cfg = hub._prepare_run_config(
+ "audiocpp", {"api_url": "http://10.0.0.5:8080"})
+ self.assertEqual(cfg.server_url, "http://10.0.0.5:8080")
+ self.assertIsNone(cfg.autostart_spec)
+ self.assertEqual(cfg.server_identity, "audiocpp")
+ self.assertIn("remote", cfg.backend_label)
# The remote path never re-detects or touches managed-instance state.
mk_detect.assert_not_called()
- def test_managed_conversion_warns_when_port_occupied_by_other_server(self):
+ def test_autostart_sets_the_spec_and_pops_the_flag(self):
+ spec = self._spec()
+ kwargs = {"autostart": "qwen-custom"}
+ with patch.object(hub, "detect_all", return_value=[]), \
+ patch.object(hub, "_find_spec", return_value=spec):
+ cfg = hub._prepare_run_config("qwen", kwargs)
+ self.assertIs(cfg.autostart_spec, spec)
+ self.assertEqual(cfg.server_name, "qwen-custom")
+ self.assertNotIn("autostart", kwargs)
+
+ def test_autostart_with_missing_spec_continues_with_notice(self):
+ kwargs = {"autostart": "gone"}
+ with patch.object(hub, "detect_all", return_value=[]), \
+ patch.object(hub, "_find_spec", return_value=None):
+ cfg = hub._prepare_run_config("qwen", kwargs)
+ self.assertIsNone(cfg.autostart_spec)
+ self.assertIn("gone", cfg.notice)
+
+ def test_managed_conversion_flags_foreign_server_on_port(self):
spec = ServerSpec("audiocpp", "http://127.0.0.1:8080", ["x"])
status = BackendStatus("audiocpp", "audio.cpp", installed=True,
- configured=True, running=False,
- servers=[spec])
+ configured=True, servers=[spec])
with patch.object(hub, "detect_all", return_value=[status]), \
- patch.object(hub, "_select_spec", return_value=spec), \
patch("backends.common.server_running", return_value=True), \
- patch.object(hub.servers, "alive", return_value=False), \
- patch.object(hub.audiobook, "convert", return_value=0) as mk_conv, \
- patch("builtins.print") as mk_print:
- hub._run_conversion("audiocpp", {})
- mk_conv.assert_called_once()
- printed = " ".join(str(call.args[0]) for call in mk_print.call_args_list)
- self.assertIn("did not start", printed)
+ patch.object(hub.servers, "alive", return_value=False):
+ cfg = hub._prepare_run_config("audiocpp", {})
+ self.assertEqual(cfg.server_url, spec.url)
+ self.assertIsNone(cfg.autostart_spec)
+ self.assertIn("did not start", cfg.notice)
+
+ def test_managed_not_running_sets_url_without_autostart(self):
+ spec = self._spec()
+ status = BackendStatus("qwen", "qwen-tts", installed=True,
+ configured=True, servers=[spec])
+ with patch.object(hub, "detect_all", return_value=[status]), \
+ patch("backends.common.server_running", return_value=False):
+ cfg = hub._prepare_run_config("qwen", {"clone": None})
+ self.assertEqual(cfg.server_url, spec.url)
+ self.assertIsNone(cfg.autostart_spec)
+
+
+class PreflightTests(unittest.TestCase):
+ """_preflight: overwrite prompts run in the TUI, plan stashed in kwargs."""
+
+ def _cmd(self):
+ return ("convert", "qwen", {"clone": None, "output_format": "mp3"})
+
+ def test_books_stashed_on_kwargs(self):
+ stdscr = object()
+ cmd = self._cmd()
+ with patch.object(hub.AudiobookConverter, "preflight_overwrites",
+ return_value=(["book.txt"], [("book.txt", "x")])) \
+ as mk_pre:
+ self.assertTrue(hub._preflight(stdscr, cmd))
+ self.assertEqual(cmd[2]["book_files"], ["book.txt"])
+ self.assertEqual(cmd[2]["planned"], [("book.txt", "x")])
+ # The confirm callback passed to preflight is a TUI yes/no.
+ confirm = mk_pre.call_args.kwargs["confirm"]
+ with patch.object(hub.tui, "confirm", return_value=True) as mk_confirm:
+ self.assertTrue(confirm("overwrite?", True))
+ mk_confirm.assert_called_once()
+
+ def test_nothing_to_convert_flashes_and_returns_false(self):
+ stdscr = object()
+ with patch.object(hub.AudiobookConverter, "preflight_overwrites",
+ return_value=([], [])), \
+ patch.object(hub.tui, "flash") as mk_flash:
+ self.assertFalse(hub._preflight(stdscr, self._cmd()))
+ mk_flash.assert_called_once()
+
+ def test_all_skipped_flashes_and_returns_false(self):
+ stdscr = object()
+ with patch.object(hub.AudiobookConverter, "preflight_overwrites",
+ return_value=(["book.txt"], [])), \
+ patch.object(hub.tui, "flash") as mk_flash:
+ self.assertFalse(hub._preflight(stdscr, self._cmd()))
+ mk_flash.assert_called_once()
+
+
+class DispatchConversionTests(unittest.TestCase):
+ """_dispatch_conversion: builds the config and runs the run view."""
+
+ def test_runs_run_view_inside_curses(self):
+ from tests.test_tui import FakeScreen
+
+ class FakeView:
+ def __init__(self, scr, config):
+ self.config = config
+ def run(self):
+ pass
+
+ made = []
+ with patch.object(hub, "_prepare_run_config",
+ return_value=hub.runview.RunConfig(
+ backend="qwen", backend_label="qwen-tts",
+ kwargs={}, book_files=[], planned=[])) as mk_cfg, \
+ patch("curses.wrapper",
+ side_effect=lambda cb: cb(FakeScreen())) as mk_wrapper, \
+ patch.object(hub.runview, "RunView", FakeView):
+ hub._dispatch_conversion("qwen", {})
+ mk_cfg.assert_called_once()
+ mk_wrapper.assert_called_once()
class AddAutostartTests(unittest.TestCase):
diff --git a/app/tests/test_runview.py b/app/tests/test_runview.py
new file mode 100644
index 0000000..73cae5f
--- /dev/null
+++ b/app/tests/test_runview.py
@@ -0,0 +1,210 @@
+"""Tests for the run view (ui/runview.py) — the conversion status screen.
+
+The view is driven the same way as the other TUI widgets: the fake curses
+module and recording screen from test_tui stand in for a terminal, the
+worker/monitor threads are stubbed, and events are fed through the view's
+own queue to exercise state transitions, rendering, and the Esc/q
+cancel → stop-server flow.
+"""
+
+import sys
+import unittest
+from unittest.mock import patch
+
+from tests.test_tui import FakeCurses, FakeScreen
+from ui import runview
+
+
+def _config(**overrides):
+ kwargs = dict(backend="audiocpp", backend_label="audio.cpp",
+ kwargs={}, book_files=["book.txt"], planned=["book.txt"],
+ server_name="audiocpp",
+ server_url="http://127.0.0.1:8080",
+ server_identity="audiocpp")
+ kwargs.update(overrides)
+ return runview.RunConfig(**kwargs)
+
+
+class _FakeTui:
+ """Stand-in for the curses module (installed into sys.modules)."""
+
+ def setUp(self):
+ self.curses = FakeCurses()
+ patcher = patch.dict(sys.modules, {"curses": self.curses})
+ patcher.start()
+ self.addCleanup(patcher.stop)
+ runview.tui._THEME.clear()
+ self.addCleanup(runview.tui._THEME.clear)
+
+ def make_view(self, keys=(), width=80, height=24, **cfg):
+ screen = FakeScreen(keys=keys, width=width, height=height)
+ # Patch the thread targets at the class level BEFORE construction so
+ # __init__'s Thread(target=self._worker_main) binds the stub.
+ with patch.object(runview.RunView, "_worker_main", lambda self: None), \
+ patch.object(runview.RunView, "_monitor_main",
+ lambda self: None):
+ view = runview.RunView(screen, _config(**cfg),
+ clock=lambda: 1000.0)
+ return view, screen
+
+
+class FormatTests(_FakeTui, unittest.TestCase):
+ def test_format_elapsed(self):
+ self.assertEqual(runview._format_elapsed(0), "0:00")
+ self.assertEqual(runview._format_elapsed(65), "1:05")
+ self.assertEqual(runview._format_elapsed(3661), "1:01:01")
+
+ def test_fit_truncates_with_tilde(self):
+ self.assertEqual(runview._fit("hello", 3), "he~")
+ self.assertEqual(runview._fit("hi", 10), "hi")
+
+ def test_wrap_wraps_on_word_boundaries(self):
+ self.assertEqual(runview._wrap("aaaa bbbb cccc dddd", 12),
+ ["aaaa bbbb", "cccc dddd"])
+
+
+class StateTransitionTests(_FakeTui, unittest.TestCase):
+ def test_boot_flow_starting_to_ready(self):
+ view, _ = self.make_view()
+ view.handle_event({"kind": "starting", "name": "audiocpp",
+ "pid": 1, "log_path": "/tmp/x.log"})
+ self.assertEqual(view.phase, "boot")
+ self.assertEqual(view.server, "starting")
+ self.assertTrue(view.started_server)
+ view.handle_event({"kind": "ready", "name": "audiocpp",
+ "url": "http://x"})
+ self.assertEqual(view.server, "ready")
+
+ def test_chunk_progress_updates(self):
+ view, _ = self.make_view()
+ view.handle_event({"kind": "book", "index": 1, "total": 2,
+ "name": "book.txt"})
+ self.assertEqual(view.phase, "convert")
+ view.handle_event({"kind": "chunks", "total": 10})
+ view.handle_event({"kind": "chunk_done", "chunk": 4, "total": 10})
+ self.assertEqual(view.chunk_done, 4)
+ self.assertEqual(view.chunk_total, 10)
+
+ def test_done_all_books_is_terminal(self):
+ view, _ = self.make_view()
+ view.handle_event({"kind": "book", "index": 1, "total": 1,
+ "name": "b"})
+ view.handle_event({"kind": "book_done", "name": "b", "ok": True})
+ view.handle_event({"kind": "done", "ok": 1, "total": 1})
+ self.assertEqual(view.phase, "done")
+
+ def test_chunk_failure_leads_to_error(self):
+ view, _ = self.make_view()
+ view.handle_event({"kind": "book", "index": 1, "total": 1,
+ "name": "b"})
+ view.handle_event({"kind": "chunk_failed", "chunk": 3, "total": 5})
+ view.handle_event({"kind": "book_done", "name": "b", "ok": False})
+ view.handle_event({"kind": "done", "ok": 0, "total": 1})
+ self.assertEqual(view.phase, "error")
+ self.assertTrue(view.error_message)
+
+ def test_server_exit_during_boot_is_error(self):
+ view, _ = self.make_view()
+ view.handle_event({"kind": "starting", "name": "audiocpp"})
+ view.handle_event({"kind": "exited", "name": "audiocpp",
+ "returncode": 1, "log_tail": ["boom"]})
+ self.assertEqual(view.phase, "error")
+ self.assertEqual(view.server, "error")
+ self.assertEqual(view.log_tail, ["boom"])
+
+ def test_server_down_during_convert(self):
+ view, _ = self.make_view()
+ view.handle_event({"kind": "book", "index": 1, "total": 1,
+ "name": "b"})
+ view.handle_event({"kind": "server_down"})
+ self.assertEqual(view.server, "down")
+
+
+class RenderTests(_FakeTui, unittest.TestCase):
+ def _strings(self, screen):
+ return " ".join(text for _, _, text, _ in screen.strings)
+
+ def test_boot_screen_shows_server_and_status(self):
+ view, screen = self.make_view()
+ view.handle_event({"kind": "starting", "name": "audiocpp",
+ "pid": 1, "log_path": "/tmp/x.log"})
+ view.render()
+ text = self._strings(screen)
+ self.assertIn("Server", text)
+ self.assertIn("audio.cpp", text)
+ self.assertIn("Status", text)
+ self.assertIn("starting", text)
+ self.assertIn("Esc or q: cancel", text)
+
+ def test_summary_screen_after_done(self):
+ view, screen = self.make_view()
+ view.handle_event({"kind": "book", "index": 1, "total": 1,
+ "name": "book.txt"})
+ view.handle_event({"kind": "book_done", "name": "book.txt",
+ "ok": True})
+ view.handle_event({"kind": "done", "ok": 1, "total": 1})
+ view.render()
+ text = self._strings(screen)
+ self.assertIn("completed", text)
+ self.assertIn("book.txt", text)
+ self.assertIn("press any key", text)
+
+ def test_error_screen_shows_detail_and_log(self):
+ view, screen = self.make_view(log_path="/tmp/audiobook.log")
+ view.handle_event({"kind": "book", "index": 1, "total": 1,
+ "name": "b"})
+ view.handle_event({"kind": "chunk_failed", "chunk": 1, "total": 3})
+ view.handle_event({"kind": "book_done", "name": "b", "ok": False})
+ view.handle_event({"kind": "done", "ok": 0, "total": 1})
+ view.render()
+ text = self._strings(screen)
+ self.assertIn("failed", text)
+ self.assertIn("/tmp/audiobook.log", text)
+
+
+class RunLoopTests(_FakeTui, unittest.TestCase):
+ def test_terminal_screen_key_returns(self):
+ view, screen = self.make_view(keys=[ord("x")])
+ view._queue.put({"kind": "done", "ok": 1, "total": 1})
+ view.run()
+ # Returned to the menu without touching the server stop prompt
+ # (started_server is False).
+ self.assertEqual(view.phase, "done")
+
+ def test_esc_cancels_and_confirms_stop_server(self):
+ confirm_answers = [True, True] # cancel? yes; stop server? yes
+ with patch.object(runview.tui, "confirm",
+ side_effect=confirm_answers), \
+ patch.object(runview.servers, "alive", return_value=True), \
+ patch.object(runview.servers, "stop") as mk_stop:
+ view, screen = self.make_view(keys=[27, ord("x")],
+ autostart_spec="SPEC")
+ view.started_server = True
+ view.run()
+ mk_stop.assert_called_once_with("audiocpp")
+
+ def test_esc_decline_cancel_keeps_running(self):
+ # First Esc: "cancel?" answered No → the run continues; a second
+ # key then exits via a terminal state the test feeds.
+ with patch.object(runview.tui, "confirm",
+ side_effect=[False]):
+ view, screen = self.make_view(keys=[27])
+ # Feeds a done event after the declined cancel so run() can exit.
+ view._queue.put({"kind": "done", "ok": 1, "total": 1})
+ screen.keys.append(ord("x"))
+ view.run()
+ self.assertEqual(view.phase, "done")
+
+ def test_stop_server_not_asked_when_dead(self):
+ with patch.object(runview.tui, "confirm", return_value=True), \
+ patch.object(runview.servers, "alive", return_value=False), \
+ patch.object(runview.servers, "stop") as mk_stop:
+ view, screen = self.make_view(keys=[27, ord("x")],
+ autostart_spec="SPEC")
+ view.started_server = True
+ view.run()
+ mk_stop.assert_not_called()
+
+
+if __name__ == "__main__":
+ unittest.main()
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-08 13:25:13 +0000