"""Tests for the faster-qwen3-tts backend setup module (backends/faster.py).""" import json import sys import tempfile import threading import unittest import contextlib from pathlib import Path from unittest.mock import patch from backends import faster as make_voices class FindWavFilesTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.folder = Path(self._tmp.name) def tearDown(self): self._tmp.cleanup() def _touch(self, name): path = self.folder / name path.write_bytes(b"x") return path def test_finds_only_wavs_case_insensitive(self): self._touch("b.wav") self._touch("a.WAV") self._touch("notes.txt") (self.folder / "sub").mkdir() (self.folder / "sub" / "c.wav").write_bytes(b"x") names = [path.name for path in make_voices.find_wav_files(self.folder)] self.assertEqual(names, ["a.WAV", "b.wav"]) def test_sorted_alphabetically_case_insensitive(self): for name in ("Zed.wav", "alpha.wav", "Beta.wav"): self._touch(name) names = [path.name for path in make_voices.find_wav_files(self.folder)] self.assertEqual(names, ["alpha.wav", "Beta.wav", "Zed.wav"]) def test_empty_directory_returns_empty_list(self): self.assertEqual(make_voices.find_wav_files(self.folder), []) class BuildVoicesTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.folder = Path(self._tmp.name) self.narrator = self.folder / "narrator.wav" self.narrator.write_bytes(b"x") self.other = self.folder / "other.wav" self.other.write_bytes(b"x") def tearDown(self): self._tmp.cleanup() def test_voices_named_after_basenames_with_absolute_paths(self): transcripts = {str(self.narrator): "First transcript.", str(self.other): "Second transcript."} with patch.object(make_voices, "transcribe_reference_audio", side_effect=lambda path, model_name="base": transcripts[path]): voices = make_voices.build_voices([self.narrator, self.other], "English", "base") self.assertEqual(list(voices), ["narrator", "other"]) self.assertEqual(voices["narrator"]["ref_text"], "First transcript.") self.assertEqual(voices["narrator"]["language"], "English") self.assertTrue(Path(voices["narrator"]["ref_audio"]).is_absolute()) self.assertEqual(Path(voices["narrator"]["ref_audio"]), self.narrator.resolve()) def test_failed_transcription_keeps_entry_with_empty_text(self): with patch.object(make_voices, "transcribe_reference_audio", return_value=None): voices = make_voices.build_voices([self.narrator], "English", "base") self.assertEqual(voices["narrator"]["ref_text"], "") def test_whisper_model_name_is_passed_through(self): with patch.object(make_voices, "transcribe_reference_audio", return_value="text") as mock_transcribe: make_voices.build_voices([self.narrator], "English", "large-v3") self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], "large-v3") class LoadVoicesTests(unittest.TestCase): """load_voices: read voices.json, or {} when unusable.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.dir = Path(self._tmp.name) self.path = self.dir / "voices.json" def tearDown(self): self._tmp.cleanup() def test_reads_dict_document(self): self.path.write_text(json.dumps({"narrator": {"ref_text": "hi"}}), encoding="utf-8") self.assertEqual(make_voices.load_voices(self.path), {"narrator": {"ref_text": "hi"}}) def test_missing_file_returns_empty(self): self.assertEqual(make_voices.load_voices(self.path), {}) def test_unreadable_json_returns_empty(self): self.path.write_text("not json", encoding="utf-8") self.assertEqual(make_voices.load_voices(self.path), {}) def test_non_dict_document_returns_empty(self): self.path.write_text("[1, 2]", encoding="utf-8") self.assertEqual(make_voices.load_voices(self.path), {}) class DecideFasterTranscriptionTests(unittest.TestCase): """_decide_faster_transcription: the re-transcribe plan questions.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.folder = Path(self._tmp.name) self.narrator = self.folder / "narrator.wav" self.narrator.write_bytes(b"x") self.new_voice = self.folder / "new.wav" self.new_voice.write_bytes(b"x") def tearDown(self): self._tmp.cleanup() def test_new_voices_default_to_missing_mode(self): confirm = lambda q, default=True: True # noqa: E731 plan = make_voices._decide_faster_transcription( [self.narrator, self.new_voice], {"narrator": {"ref_text": "old"}}, confirm) self.assertEqual(plan["mode"], "missing") self.assertEqual([w.name for w in plan["missing"]], ["new.wav"]) def test_declining_new_voices_transcribes_all(self): confirm = lambda q, default=True: False # noqa: E731 plan = make_voices._decide_faster_transcription( [self.narrator, self.new_voice], {"narrator": {"ref_text": "old"}}, confirm) self.assertEqual(plan["mode"], "all") def test_no_new_voices_offers_retranscribe_default_no(self): confirm = lambda q, default=True: default # noqa: E731 plan = make_voices._decide_faster_transcription( [self.narrator], {"narrator": {"ref_text": "old"}}, confirm) self.assertEqual(plan["mode"], "keep") def test_no_new_voices_accepted_retranscribes_all(self): confirm = lambda q, default=True: True # noqa: E731 plan = make_voices._decide_faster_transcription( [self.narrator], {"narrator": {"ref_text": "old"}}, confirm) self.assertEqual(plan["mode"], "all") def test_cancel_returns_none(self): confirm = lambda q, default=True: None # noqa: E731 plan = make_voices._decide_faster_transcription( [self.narrator], {"narrator": {"ref_text": "old"}}, confirm) self.assertIsNone(plan) class MainTests(unittest.TestCase): """The flag-only (non-TUI) path through main(), end to end.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.folder = Path(self._tmp.name) (self.folder / "narrator.wav").write_bytes(b"x") (self.folder / "alpha.wav").write_bytes(b"x") self.output = self.folder / "voices.json" # Avoid touching the real converter/config.py and pip/git. patcher = patch.object(make_voices.common, "update_config_value", return_value=False) patcher.start() self.addCleanup(patcher.stop) patcher = patch.object(make_voices, "_interactive", return_value=False) patcher.start() self.addCleanup(patcher.stop) def tearDown(self): self._tmp.cleanup() def _run(self, argv): with patch.object(sys, "argv", ["backends/faster.py"] + argv), \ patch.object(make_voices, "transcribe_reference_audio", return_value="hello"): return make_voices.main() def test_writes_json_with_alphabetical_voice_order(self): exit_code = self._run([str(self.folder), "--output", str(self.output), "--skip-install", "--skip-clone"]) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(list(data), ["alpha", "narrator"]) self.assertEqual(data["alpha"]["ref_text"], "hello") self.assertEqual(data["alpha"]["language"], "English") def test_custom_output_path(self): custom = Path(self._tmp.name) / "custom.json" exit_code = self._run([str(self.folder), "--output", str(custom), "--skip-install", "--skip-clone"]) self.assertEqual(exit_code, 0) 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: self._run([str(self.folder), "--output", str(self.output), "--language", "klingon", "--skip-install", "--skip-clone"]) self.assertEqual(ctx.exception.code, 2) mock_transcribe.assert_not_called() def test_missing_input_dir_errors(self): with self.assertRaises(SystemExit) as ctx: self._run([str(self.folder / "nope"), "--output", str(self.output), "--skip-install", "--skip-clone"]) self.assertEqual(ctx.exception.code, 2) def test_no_wav_files_returns_error(self): empty = Path(tempfile.mkdtemp()) try: exit_code = self._run([str(empty), "--output", str(empty / "voices.json"), "--skip-install", "--skip-clone"]) self.assertEqual(exit_code, 1) finally: empty.rmdir() def test_existing_output_declined_keeps_file(self): self.output.write_text('{"old": true}', encoding="utf-8") with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe: exit_code = self._run([str(self.folder), "--output", str(self.output), "--skip-install", "--skip-clone"]) self.assertEqual(exit_code, 1) mock_transcribe.assert_not_called() self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), {"old": True}) def test_force_overwrites_without_prompt(self): self.output.write_text('{"old": true}', encoding="utf-8") exit_code = self._run([str(self.folder), "--output", str(self.output), "--force", "--skip-install", "--skip-clone"]) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(list(data), ["alpha", "narrator"]) if __name__ == "__main__": unittest.main() class SetupScreenTests(unittest.TestCase): """setup_screen: the wizard run on the hub's screen, setup tail via the in-TUI task view.""" def test_abort_returns_one_without_executing(self): with patch.object(make_voices, "_wizard", return_value=None) as mk_wizard, \ patch.object(make_voices, "_execute_steps") as mk_steps: rc = make_voices.setup_screen(None) self.assertEqual(rc, 1) mk_wizard.assert_called_once() mk_steps.assert_not_called() def test_success_runs_the_tail_in_the_task_view(self): settings = {"wav_dir": Path("/x")} steps = [make_voices.taskview.TaskStep("t", lambda emit, cancel: 0)] with patch.object(make_voices, "_wizard", return_value=settings), \ patch.object(make_voices, "_execute_steps", return_value=steps) as mk_steps, \ patch.object(make_voices.taskview, "run_steps", return_value=0) as mk_run: rc = make_voices.setup_screen(None) self.assertEqual(rc, 0) mk_steps.assert_called_once() self.assertIs(mk_steps.call_args[0][0], settings) mk_run.assert_called_once() self.assertEqual(mk_run.call_args[0][2], steps) class UninstallTests(unittest.TestCase): """uninstall: stop the server, pip-uninstall, delete the checkout.""" def test_pips_and_removes_checkout(self): with tempfile.TemporaryDirectory() as td: checkout = Path(td) / "faster-qwen3-tts" 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: rc = make_voices.uninstall(emit="EMIT") self.assertEqual(rc, 0) mk_stop.assert_called_once_with("faster") # The task view's emit is forwarded so pip never touches the terminal. mk_pip.assert_called_once_with(["faster-qwen3-tts"], emit="EMIT") self.assertFalse(checkout.exists()) 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, "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, "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) mk_stop.assert_called_once_with("faster") mk_pip.assert_not_called() def test_cancel_before_delete_keeps_checkout(self): # Cancel between phases: pip runs to completion, but a pending # cancellation stops the checkout deletion from starting. with tempfile.TemporaryDirectory() as td: checkout = Path(td) / "faster-qwen3-tts" checkout.mkdir() cancel = threading.Event() 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): rc = make_voices.uninstall(cancel=cancel) self.assertEqual(rc, 130) self.assertTrue(checkout.exists())