path: root/app/tests/test_backends_faster.py

"""Tests for the faster-qwen3-tts backend setup module (backends/faster.py)."""

import json
import sys
import tempfile
import threading
import unittest
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 PlanForTests(unittest.TestCase):
    """_plan_for: execution plans for the wizard's transcription toggle."""

    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_missing_plan_carries_only_unlisted_wavs(self):
        plan = make_voices._plan_for("missing", self.folder,
                                     {"narrator": {"ref_text": "old"}})
        self.assertEqual(plan["mode"], "missing")
        self.assertEqual([w.name for w in plan["missing"]], ["new.wav"])
        self.assertEqual(plan["existing"], {"narrator": {"ref_text": "old"}})

    def test_all_plan_names_the_mode(self):
        plan = make_voices._plan_for("all", self.folder,
                                     {"narrator": {"ref_text": "old"}})
        self.assertEqual(plan["mode"], "all")
        self.assertEqual(plan["missing"], [])


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.setup, "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"])


class WizardFormTests(unittest.TestCase):
    """The faster wizard: one combined form instead of a screen chain."""

    def _args(self, *extra):
        return make_voices.build_parser().parse_args(list(extra))

    def _wavs(self):
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        folder = Path(tmp.name)
        (folder / "narrator.wav").write_bytes(b"x")
        return folder

    def test_fresh_run_asks_one_form_without_transcription_choice(self):
        folder = self._wavs()
        captured = {}

        def fake_form(stdscr, title, fields, **kwargs):
            captured["title"] = title
            captured["keys"] = [f["key"] for f in fields]
            by_key = {f["key"]: f for f in fields}
            # The directory browser alerts on the .wavs it lists.
            self.assertIs(by_key["wav_dir"]["info"],
                          make_voices.common.wav_dir_info)
            self.assertIs(by_key["wav_dir"]["preview"],
                          make_voices.common.wav_dir_preview)
            by_key["wav_dir"]["value"] = folder
            return {f["key"]: f["value"] for f in fields}

        with patch.object(make_voices, "_is_installed", return_value=True), \
                patch.object(make_voices, "_is_cloned", return_value=True), \
                patch.object(make_voices.tui, "form",
                             side_effect=fake_form) as mk_form:
            settings = make_voices._wizard(
                None, self._args("--output", str(folder / "voices.json"),
                                 "--skip-install", "--skip-clone"))
        self.assertIsNotNone(settings)
        self.assertEqual(mk_form.call_count, 1)
        self.assertEqual(captured["keys"],
                         ["wav_dir", "language", "whisper_model"])
        self.assertEqual(settings["wav_dir"], folder)
        # Nothing was configured before, so everything is transcribed and
        # no keep/new-only choice exists.
        self.assertEqual(settings["plan"]["mode"], "all")

    def test_modify_run_offers_transcription_toggle(self):
        folder = self._wavs()
        (folder / "new.wav").write_bytes(b"x")  # a voice not in voices.json
        output = folder / "voices.json"
        existing = {"narrator": {
            "ref_audio": str(folder / "narrator.wav"),
            "ref_text": "old transcript", "language": "English"}}
        output.write_text(json.dumps(existing), encoding="utf-8")
        captured = {}

        def fake_form(stdscr, title, fields, **kwargs):
            captured["keys"] = [f["key"] for f in fields]
            by_key = {f["key"]: f for f in fields}
            self.assertIn("transcription", by_key)
            # A plain in-place toggle (no popup): fixed choices, new-only.
            row = by_key["transcription"]
            self.assertEqual(row["kind"], "toggle")
            self.assertEqual(row["value"], "missing")
            self.assertEqual(
                row["choices"],
                [("Transcribe new voices", "missing"),
                 ("Re-transcribe all voices", "all")])
            result = {f["key"]: f["value"] for f in fields}
            result["transcription"] = "missing"
            return result

        with patch.object(make_voices, "_is_installed", return_value=True), \
                patch.object(make_voices, "_is_cloned", return_value=True), \
                patch.object(make_voices.tui, "form",
                             side_effect=fake_form):
            settings = make_voices._wizard(
                None, self._args("--output", str(output),
                                 "--skip-install", "--skip-clone"))
        self.assertIsNotNone(settings)
        self.assertEqual(captured["keys"],
                         ["wav_dir", "language", "whisper_model",
                          "transcription"])
        self.assertEqual(settings["plan"]["mode"], "missing")
        self.assertEqual([w.name for w in settings["plan"]["missing"]],
                         ["new.wav"])
        self.assertEqual(settings["wav_dir"], folder)

    def test_modify_run_toggled_all_retranscribes_everything(self):
        folder = self._wavs()
        (folder / "new.wav").write_bytes(b"x")
        output = folder / "voices.json"
        existing = {"narrator": {
            "ref_audio": str(folder / "narrator.wav"),
            "ref_text": "old transcript", "language": "English"}}
        output.write_text(json.dumps(existing), encoding="utf-8")

        def fake_form(stdscr, title, fields, **kwargs):
            result = {f["key"]: f["value"] for f in fields}
            result["transcription"] = "all"
            return result

        with patch.object(make_voices, "_is_installed", return_value=True), \
                patch.object(make_voices, "_is_cloned", return_value=True), \
                patch.object(make_voices.tui, "form",
                             side_effect=fake_form):
            settings = make_voices._wizard(
                None, self._args("--output", str(output),
                                 "--skip-install", "--skip-clone"))
        self.assertIsNotNone(settings)
        self.assertEqual(settings["plan"]["mode"], "all")

    def test_cancel_aborts(self):
        with patch.object(make_voices, "_is_installed", return_value=True), \
                patch.object(make_voices, "_is_cloned", return_value=True), \
                patch.object(make_voices.tui, "form",
                             side_effect=lambda *a, **k: k["back_value"]):
            settings = make_voices._wizard(
                None, self._args("--output", "/tmp/x.json",
                                 "--skip-install", "--skip-clone"))
        self.assertIsNone(settings)

    def test_port_flag_removed(self):
        parser = make_voices.build_parser()
        with self.assertRaises(SystemExit):
            parser.parse_args(["--port", "8000"])


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,
        # and the package comes out of the faster backend's own venv.
        mk_pip.assert_called_once_with(["faster-qwen3-tts"], emit="EMIT",
                                       env_dir=make_voices.FASTER_ENV)
        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,
                                       env_dir=make_voices.FASTER_ENV)

    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())


class UpdateTests(unittest.TestCase):
    """update: stop the server, pip install -U, refresh the checkout."""

    def test_pip_upgrade_and_checkout_update(self):
        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_install",
                             return_value=0) as mk_pip, \
                patch.object(make_voices, "_is_cloned",
                             return_value=True), \
                patch.object(make_voices, "_checkout",
                             return_value=Path("/co")), \
                patch.object(make_voices.common, "git_update",
                             return_value=0) as mk_git:
            rc = make_voices.update(emit="EMIT")
        self.assertEqual(rc, 0)
        mk_stop.assert_called_once_with("faster")
        # The task view's emit is forwarded, the install is an upgrade,
        # and the package lands in the faster backend's own venv.
        mk_pip.assert_called_once_with([make_voices.FASTER_PIP_PKG],
                                       emit="EMIT", cancel=None,
                                       env_dir=make_voices.FASTER_ENV,
                                       upgrade=True)
        mk_git.assert_called_once_with(Path("/co"), emit="EMIT", cancel=None)

    def test_no_checkout_updates_the_package_only(self):
        with 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_install",
                             return_value=0) as mk_pip, \
                patch.object(make_voices, "_is_cloned",
                             return_value=False), \
                patch.object(make_voices.common, "git_update") as mk_git:
            rc = make_voices.update()
        self.assertEqual(rc, 0)
        mk_stop.assert_not_called()
        mk_git.assert_not_called()
        mk_pip.assert_called_once()

    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_install") as mk_pip:
            rc = make_voices.update(cancel=cancel)
        self.assertEqual(rc, 130)
        mk_stop.assert_called_once_with("faster")
        mk_pip.assert_not_called()

    def test_cancel_after_pip_skips_the_checkout(self):
        cancel = threading.Event()
        cancel.set()
        with patch.object(make_voices.servers, "pid_for",
                          return_value=None), \
                patch.object(make_voices.common, "pip_install",
                             return_value=0), \
                patch.object(make_voices, "_is_cloned",
                             return_value=True), \
                patch.object(make_voices.common, "git_update") as mk_git:
            rc = make_voices.update(cancel=cancel)
        self.assertEqual(rc, 130)
        mk_git.assert_not_called()

    def test_checkout_failure_propagates_after_a_successful_pip(self):
        with patch.object(make_voices.servers, "pid_for",
                          return_value=None), \
                patch.object(make_voices.common, "pip_install",
                             return_value=0), \
                patch.object(make_voices, "_is_cloned",
                             return_value=True), \
                patch.object(make_voices, "_checkout",
                             return_value=Path("/co")), \
                patch.object(make_voices.common, "git_update",
                             return_value=3) as mk_git:
            rc = make_voices.update()
        self.assertEqual(rc, 3)
        mk_git.assert_called_once()

    def test_pip_failure_still_updates_the_checkout(self):
        with patch.object(make_voices.servers, "pid_for",
                          return_value=None), \
                patch.object(make_voices.common, "pip_install",
                             return_value=1), \
                patch.object(make_voices, "_is_cloned",
                             return_value=True), \
                patch.object(make_voices, "_checkout",
                             return_value=Path("/co")), \
                patch.object(make_voices.common, "git_update",
                             return_value=0):
            rc = make_voices.update()
        self.assertEqual(rc, 1)