path: root/app/tests/test_hub.py

"""Tests for the TUI hub (ui/hub.py) menu and helpers.

The hub drives the same curses widgets as ui/tui.py, so these tests reuse
the fake curses/screen from test_tui to run the menu without a terminal.
"""

import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from backends import BackendInfo, BackendStatus, ServerSpec
from tests.test_tui import FakeCurses, FakeScreen
from ui import hub, tui


class _ScriptedTUI:
    """Stand-in for the tui widget module: answers each menu/line_edit/
    confirm call from a scripted answer list and records every prompt."""

    def __init__(self):
        self.script = []
        self.prompts = []
        self.options_seen = []
        self.flashes = []

    def _next(self, prompt, options=None):
        self.prompts.append(prompt)
        if options is not None:
            self.options_seen.append(options)
        return self.script.pop(0)

    def menu(self, stdscr, title, options, **kwargs):
        return self._next(title, options)

    def line_edit(self, stdscr, title, default, **kwargs):
        self.prompts.append(f"{title} [default: {default!r}]")
        return self.script.pop(0)

    def confirm(self, stdscr, question, **kwargs):
        return self._next(question)

    def flash(self, stdscr, text, kind="warn"):
        self.flashes.append(text)


class HubHelperTests(unittest.TestCase):
    """Pure helpers in hub.py (no curses)."""

    def test_is_float(self):
        self.assertTrue(hub._is_float("1.0"))
        self.assertTrue(hub._is_float("2"))
        self.assertFalse(hub._is_float("abc"))
        self.assertFalse(hub._is_float(""))

    def test_list_voices_from_dir(self):
        with __import__("tempfile").TemporaryDirectory() as td:
            d = Path(td)
            (d / "Narrator.wav").write_bytes(b"x")
            (d / "Alpha.WAV").write_bytes(b"x")
            (d / "notes.txt").write_bytes(b"x")
            voices = hub._list_voices(str(d))
        # Stems preserve case; sorting is case-insensitive.
        self.assertEqual(voices, ["Alpha", "Narrator"])

    def test_list_voices_missing_dir(self):
        self.assertEqual(hub._list_voices("/no/such/dir"), [])

    def test_status_mark(self):
        from backends import BackendStatus
        running = BackendStatus("k", "l", installed=True, configured=True,
                                running=True, managed=True)
        remote = BackendStatus("k", "l", installed=True, configured=True,
                               running=True)
        models = BackendStatus("k", "l", installed=True, configured=True,
                               running=True, managed=True,
                               running_models=["Base", "CustomVoice"])
        remote_models = BackendStatus("k", "l", installed=True,
                                      configured=True, running=True,
                                      running_models=["Base"])
        installed = BackendStatus("k", "l", installed=True,
                                  configured=False)
        none = BackendStatus("k", "l", installed=False, configured=False)
        # running beats installed (a server is up even if not configured);
        # only a backend that is neither installed nor running is dimmed.
        self.assertEqual(hub._status_mark(running),
                         ("running", "ok", "body"))
        # A server without a live recorded pid was started externally.
        self.assertEqual(hub._status_mark(remote),
                         ("running [remote]", "ok", "body"))
        # Multi-model backends name the models that answered.
        self.assertEqual(hub._status_mark(models),
                         ("running (Base, CustomVoice)", "ok", "body"))
        self.assertEqual(hub._status_mark(remote_models),
                         ("running [remote] (Base)", "ok", "body"))
        self.assertEqual(hub._status_mark(installed),
                         ("installed", "warn", "body"))
        self.assertEqual(hub._status_mark(none),
                         ("unavailable", "err", "dim"))
        self.assertEqual(hub._status_mark(None),
                         ("unavailable", "err", "dim"))


class HubMenuTests(unittest.TestCase):
    """Drive _hub_menu with a fake screen (no terminal)."""

    def setUp(self):
        tui._THEME.clear()
        self.curses = FakeCurses()
        from unittest.mock import patch as _patch
        self._patcher = _patch.dict("sys.modules", {"curses": self.curses})
        self._patcher.start()
        self.addCleanup(self._patcher.stop)
        self.addCleanup(tui._THEME.clear)

    def _none_status(self, key="k", label="l"):
        from backends import BackendStatus
        return BackendStatus(key, label, installed=False, configured=False)

    def test_quit_returns_none_when_no_backend(self):
        # No backends installed/running: menu is [Set up, Settings, Quit].
        # Quit is the 3rd option (Down twice) then Enter.
        screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10])
        with patch.object(hub, "detect_all", return_value=[]):
            result = hub._hub_menu(screen)
        self.assertIsNone(result)

    def test_menu_has_only_setup_settings_and_quit_without_backends(self):
        # Capture the options handed to tui.menu: with nothing installed or
        # running, Convert/Configure must be absent.
        captured = {}

        def fake_menu(stdscr, title, options, **kwargs):
            captured["options"] = options
            return "quit"

        screen = FakeScreen()
        with patch.object(hub.tui, "menu", fake_menu), \
                patch.object(hub, "detect_all", return_value=[]):
            hub._hub_menu(screen)
        labels = [label for label, _ in captured["options"]]
        self.assertEqual(labels, ["Set up a backend", "Settings", "Quit"])

    def test_menu_has_all_six_when_one_installed(self):
        captured = {}

        def fake_menu(stdscr, title, options, **kwargs):
            captured["options"] = options
            captured["rows"] = kwargs.get("table_rows")
            return "quit"

        screen = FakeScreen()
        st = self._none_status("qwen", "qwen-tts")
        st.installed = True
        with patch.object(hub.tui, "menu", fake_menu), \
                patch.object(hub, "detect_all", return_value=[st]):
            hub._hub_menu(screen)
        labels = [label for label, _ in captured["options"]]
        self.assertEqual(
            labels,
            ["Convert books", "Set up a backend",
             "Configure a backend", "Start/Stop Backend Servers",
             "Settings", "Quit"])
        # The status table is passed through, one row per backend.
        self.assertEqual(captured["rows"],
                         [("qwen-tts", "installed", "warn", "body")])

    def test_table_dims_name_when_not_installed_and_not_running(self):
        captured = {}

        def fake_menu(stdscr, title, options, **kwargs):
            captured["rows"] = kwargs.get("table_rows")
            return "quit"

        screen = FakeScreen()
        dead = self._none_status("audiocpp", "audio.cpp")
        external = self._none_status("qwen", "qwen-tts")
        external.running = True
        with patch.object(hub.tui, "menu", fake_menu), \
                patch.object(hub, "detect_all",
                             return_value=[dead, external]):
            hub._hub_menu(screen)
        # Unusable backend: dim name. Running-but-not-installed stays
        # bright and is tagged remote (no pid file → not started by us).
        self.assertEqual(
            captured["rows"],
            [("audio.cpp", "unavailable", "err", "dim"),
             ("qwen-tts", "running [remote]", "ok", "body")])

    def test_menu_hides_configure_and_server_when_only_running(self):
        # Running but not installed (an external server) still unlocks
        # Convert — but Configure/Server need the backend on this machine.
        captured = {}

        def fake_menu(stdscr, title, options, **kwargs):
            captured["options"] = options
            return "quit"

        screen = FakeScreen()
        st = self._none_status("qwen", "qwen-tts")
        st.running = True
        with patch.object(hub.tui, "menu", fake_menu), \
                patch.object(hub, "detect_all", return_value=[st]):
            hub._hub_menu(screen)
        labels = [label for label, _ in captured["options"]]
        self.assertEqual(
            labels,
            ["Convert books", "Set up a backend", "Settings", "Quit"])

    def test_ffmpeg_warning_shown_when_missing(self):
        # ffmpeg not on PATH → a red notice is passed above the table.
        captured = {}

        def fake_menu(stdscr, title, options, **kwargs):
            captured["notice_lines"] = kwargs.get("notice_lines")
            return "quit"

        screen = FakeScreen()
        with patch.object(hub.tui, "menu", fake_menu), \
                patch.object(hub, "detect_all", return_value=[]), \
                patch.object(hub.shutil, "which", return_value=None):
            hub._hub_menu(screen)
        self.assertEqual(captured["notice_lines"],
                         [("Warning: ffmpeg not installed!", "err")])

    def test_ffmpeg_warning_hidden_when_installed(self):
        # ffmpeg on PATH → no notice is passed at all.
        captured = {}

        def fake_menu(stdscr, title, options, **kwargs):
            captured["notice_lines"] = kwargs.get("notice_lines")
            return "quit"

        screen = FakeScreen()
        with patch.object(hub.tui, "menu", fake_menu), \
                patch.object(hub, "detect_all", return_value=[]), \
                patch.object(hub.shutil, "which", return_value="/usr/bin/ffmpeg"):
            hub._hub_menu(screen)
        self.assertIsNone(captured["notice_lines"])

    def test_convert_with_no_available_backend_flashes(self):
        # Installed-but-not-ready backends → Convert is offered, but the
        # convert flow has nothing to list: it flashes a hint (no "Set up
        # a backend" detour anymore) and returns to the main menu. Then
        # quit: 6 main-menu options, Quit is the 6th (Down x5).
        from backends import BackendStatus
        statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True,
                                  configured=False),
                    BackendStatus("qwen", "qwen-tts", installed=True,
                                  configured=False),
                    BackendStatus("faster", "faster", installed=True,
                                  configured=False)]
        flashed = []

        def fake_flash(stdscr, text, kind="warn"):
            flashed.append(text)

        with patch.object(hub, "detect_all", return_value=statuses), \
                patch.object(hub.tui, "flash", fake_flash):
            # Convert(Enter) → flash → main menu; Down x5 -> Quit, Enter.
            screen = FakeScreen(keys=[10,
                                      FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN,
                                      FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN,
                                      FakeCurses.KEY_DOWN, 10])
            result = hub._hub_menu(screen)
        self.assertIsNone(result)
        self.assertEqual(len(flashed), 1)
        self.assertIn("No backend is ready", flashed[0])


class SubmenuStatusTableTests(unittest.TestCase):
    """First picker screen of every flow repeats the backend status table.

    Entries themselves stay clean: setup lists bare labels, and the
    Start/Stop menu offers only installed backends.
    """

    def _capture_menu(self, captured):
        def fake_menu(stdscr, title, options, **kwargs):
            captured["title"] = title
            captured["options"] = options
            captured.update(kwargs)
            return hub._GO_BACK  # Esc: back out immediately

        return fake_menu

    def test_setup_menu_lists_bare_labels_and_status_table(self):
        captured = {}
        infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0),
                 BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)]
        statuses = [
            BackendStatus("audiocpp", "audio.cpp", installed=True,
                          configured=True),
            BackendStatus("qwen", "qwen-tts", installed=False,
                          configured=False, running=True),
        ]
        with patch.object(hub, "REGISTRY", infos), \
                patch.object(hub.tui, "menu",
                             self._capture_menu(captured)), \
                patch.object(hub.shutil, "which",
                             return_value="/usr/bin/ffmpeg"):
            result = hub._setup_menu(None, statuses)
        self.assertIsNone(result)
        # No inline "(running)"-style suffix on the entries anymore...
        self.assertEqual([label for label, _ in captured["options"]],
                         ["audio.cpp", "qwen-tts"])
        # ...the shared status table carries the states instead.
        self.assertEqual(captured["table_title"], "Backend status")
        self.assertEqual(
            captured["table_rows"],
            [("audio.cpp", "installed", "warn", "body"),
             ("qwen-tts", "running [remote]", "ok", "body")])
        self.assertIsNone(captured["notice_lines"])

    def test_convert_menu_shows_status_table(self):
        captured = {}
        st = BackendStatus("qwen", "qwen-tts", installed=True,
                           configured=True)
        with patch.object(hub.tui, "menu", self._capture_menu(captured)), \
                patch.object(hub.shutil, "which", return_value="/x"):
            result = hub._convert_menu(None, [st])
        self.assertIsNone(result)
        self.assertEqual(captured["title"], "Convert books with...")
        # Only convertible backends are listed — no "Set up a backend"
        # detour inside the Convert flow.
        self.assertEqual([label for label, _ in captured["options"]],
                         ["qwen-tts"])
        self.assertEqual(
            captured["table_rows"], [("qwen-tts", "installed", "warn",
                                      "body")])

    def test_convert_with_nothing_ready_flashes_instead_of_menu(self):
        # Installed-but-unconfigured → nothing convertible: a hint flash
        # replaces the old fallback menu entirely.
        flashed = []
        menus = []

        def fake_menu(*args, **kwargs):
            menus.append((args, kwargs))
            return hub._GO_BACK

        def fake_flash(stdscr, text, kind="warn"):
            flashed.append(text)

        st = BackendStatus("qwen", "qwen-tts", installed=True,
                           configured=False)
        with patch.object(hub.tui, "menu", fake_menu), \
                patch.object(hub.tui, "flash", fake_flash):
            result = hub._convert_menu(None, [st])
        self.assertIsNone(result)
        self.assertEqual(menus, [])
        self.assertIn("No backend is ready", flashed[0])

    def test_configure_menu_shows_status_table(self):
        captured = {}
        infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)]
        statuses = [BackendStatus("qwen", "qwen-tts", installed=True,
                                  configured=True)]
        with patch.object(hub, "REGISTRY", infos), \
                patch.object(hub.tui, "menu",
                             self._capture_menu(captured)), \
                patch.object(hub.shutil, "which", return_value="/x"):
            result = hub._configure_menu(None, statuses)
        self.assertIsNone(result)
        self.assertEqual(captured["table_title"], "Backend status")
        self.assertEqual(
            captured["table_rows"], [("qwen-tts", "installed", "warn",
                                      "body")])

    def test_server_menu_lists_only_installed_backends(self):
        captured = {}
        installed = BackendStatus("audiocpp", "audio.cpp", installed=True,
                                  configured=True)
        remote = BackendStatus("qwen", "qwen-tts", installed=False,
                               configured=False, running=True)
        gone = BackendStatus("faster", "faster-qwen3-tts", installed=False,
                             configured=False)
        with patch.object(hub.tui, "menu",
                          self._capture_menu(captured)), \
                patch.object(hub.shutil, "which", return_value="/x"):
            result = hub._server_menu(None, [installed, remote, gone])
        self.assertIsNone(result)
        # Only the installed backend is offered; a running external server
        # (remote) can't be stopped from here and must not appear.
        self.assertEqual([label for label, _ in captured["options"]],
                         ["audio.cpp"])
        # The status table still shows all three, states included.
        self.assertEqual([row[0] for row in captured["table_rows"]],
                         ["audio.cpp", "qwen-tts", "faster-qwen3-tts"])

    def test_server_menu_flashes_when_nothing_installed(self):
        flashed = []

        def fake_flash(stdscr, text, kind="warn"):
            flashed.append(text)

        remote = BackendStatus("qwen", "qwen-tts", installed=False,
                               configured=False, running=True)
        with patch.object(hub.tui, "flash", fake_flash):
            result = hub._server_menu(None, [remote])
        self.assertIsNone(result)
        self.assertEqual(len(flashed), 1)
        self.assertIn("No backend is installed", flashed[0])

    def test_submenu_repeats_ffmpeg_warning(self):
        captured = {}
        infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)]
        statuses = [BackendStatus("qwen", "qwen-tts", installed=True,
                                  configured=True)]
        with patch.object(hub, "REGISTRY", infos), \
                patch.object(hub.tui, "menu",
                             self._capture_menu(captured)), \
                patch.object(hub.shutil, "which", return_value=None):
            hub._setup_menu(None, statuses)
        self.assertEqual(captured["notice_lines"],
                         [("Warning: ffmpeg not installed!", "err")])


class ConvertFlowTests(unittest.TestCase):
    """_convert_audiocpp / _convert_faster: local-config menus vs. live
    queries against a running remote server."""

    def setUp(self):
        self.tui = _ScriptedTUI()
        for name in ("menu", "line_edit", "confirm", "flash"):
            patcher = patch.object(hub.tui, name, getattr(self.tui, name))
            patcher.start()
            self.addCleanup(patcher.stop)

    def _answer_common_options(self):
        # Output format, speed, single-file, chunk, debug.
        self.tui.script += ["m4b", "1.5", False, False, False]

    # ------------------------------------------------------------------
    # audio.cpp: remote server (no local checkout / server.json)
    # ------------------------------------------------------------------

    def _patch_remote(self, models, voices=None):
        """No local checkout; fetch helpers return MODELS/VOICES."""
        checkout = patch.object(hub.audiocpp_backend, "find_local_checkout",
                                return_value=None)
        fetched_models = patch.object(hub.audiocpp_backend,
                                      "fetch_server_models",
                                      lambda url: models)
        fetched_voices = patch.object(hub.audiocpp_backend,
                                      "fetch_server_voices",
                                      lambda url, model_id: voices)
        for patcher in (checkout, fetched_models, fetched_voices):
            patcher.start()
            self.addCleanup(patcher.stop)

    def test_audiocpp_remote_queries_live_models_and_voices(self):
        self._patch_remote(
            [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}],
            voices=["narrator"])
        with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""):
            self.tui.script += ["higgs", "narrator", ""]
            self._answer_common_options()
            cmd = hub._convert_audiocpp(None, [])
        self.assertEqual(cmd[0], "convert")
        self.assertEqual(cmd[1], hub.BACKEND_AUDIOCPP)
        kwargs = cmd[2]
        self.assertEqual(kwargs["model_id"], "higgs")
        self.assertEqual(kwargs["voice"], "narrator")
        self.assertIsNone(kwargs["instructions"])
        # The model menu was fed from the live query.
        self.assertEqual(self.tui.options_seen[0],
                         [("higgs  (higgs_audio_tts, tts)", "higgs")])

    def test_audiocpp_remote_qwen3_tts_offers_builtin_speaker_first(self):
        self._patch_remote(
            [{"id": "qwen", "family": "qwen3_tts", "task": "tts"}],
            voices=["narrator"])
        with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""):
            self.tui.script += ["qwen", None, ""]
            self._answer_common_options()
            cmd = hub._convert_audiocpp(None, [])
        self.assertIsNone(cmd[2]["voice"])
        self.assertEqual(self.tui.options_seen[1],
                         [("(built-in speaker)", None), ("narrator", "narrator")])

    def test_audiocpp_remote_missing_family_treated_as_qwen3_tts(self):
        # Legacy servers omit family/task; the converter defaults them to
        # qwen3_tts/tts and so must the menus (voice optional).
        self._patch_remote([{"id": "legacy", "family": "", "task": ""}],
                           voices=[])
        with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""):
            # Empty server voices: no Voice menu, built-in speaker implied.
            self.tui.script += ["legacy", ""]
            self._answer_common_options()
            cmd = hub._convert_audiocpp(None, [])
        self.assertIsNotNone(cmd)
        self.assertIsNone(cmd[2]["voice"])

    def test_audiocpp_remote_vdes_needs_instructions_not_voice(self):
        self._patch_remote(
            [{"id": "design", "family": "qwen3_tts", "task": "vdes"}])
        with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""):
            self.tui.script += ["design", "A warm British narrator"]
            self._answer_common_options()
            cmd = hub._convert_audiocpp(None, [])
        self.assertIsNone(cmd[2]["voice"])
        self.assertEqual(cmd[2]["instructions"], "A warm British narrator")
        # No voice prompt happened at all.
        self.assertNotIn("Voice", [p for p in self.tui.prompts])

    def test_audiocpp_remote_unreachable_models_flash_and_abort(self):
        self._patch_remote(None)  # endpoint did not answer valid JSON
        cmd = hub._convert_audiocpp(None, [])
        self.assertIsNone(cmd)
        self.assertIn("Could not list models", self.tui.flashes[0])

    def test_audiocpp_remote_empty_models_flash_and_abort(self):
        self._patch_remote([])
        cmd = hub._convert_audiocpp(None, [])
        self.assertIsNone(cmd)
        self.assertIn("hosts no model entries", self.tui.flashes[0])

    def test_audiocpp_remote_no_server_voices_for_clone_model_aborts(self):
        self._patch_remote(
            [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}],
            voices=[])
        self.tui.script += ["higgs"]
        cmd = hub._convert_audiocpp(None, [])
        self.assertIsNone(cmd)
        self.assertIn("lists none", self.tui.flashes[0])

    def test_audiocpp_remote_failed_voices_query_aborts(self):
        self._patch_remote(
            [{"id": "higgs", "family": "higgs_audio_tts", "task": "tts"}],
            voices=None)
        self.tui.script += ["higgs"]
        cmd = hub._convert_audiocpp(None, [])
        self.assertIsNone(cmd)
        self.assertIn("Could not list voices", self.tui.flashes[0])

    # ------------------------------------------------------------------
    # audio.cpp: local managed setup keeps reading its server.json
    # ------------------------------------------------------------------

    def test_audiocpp_local_still_reads_server_json(self):
        queried = []

        def must_not_query(url):
            queried.append(url)
            raise AssertionError("live query on the local path")

        with tempfile.TemporaryDirectory() as td:
            root = Path(td)
            (root / "server.json").write_text(json.dumps({
                "models": [{"id": "qwen", "family": "qwen3_tts",
                            "task": "tts"}],
                "voice_dir": str(root),
            }), encoding="utf-8")
            (root / "Narrator.wav").write_bytes(b"x")
            with patch.object(hub.audiocpp_backend, "find_local_checkout",
                              return_value=root), \
                    patch.object(hub.audiocpp_backend, "fetch_server_models",
                                 must_not_query), \
                    patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""):
                self.tui.script += ["qwen", "Narrator", ""]
                self._answer_common_options()
                cmd = hub._convert_audiocpp(None, [])
        self.assertEqual(queried, [])
        self.assertIsNotNone(cmd)
        self.assertEqual(cmd[2]["model_id"], "qwen")
        self.assertEqual(cmd[2]["voice"], "Narrator")

    # ------------------------------------------------------------------
    # faster: remote server (no local voices.json)
    # ------------------------------------------------------------------

    def test_faster_remote_prompts_for_a_voice_name(self):
        with tempfile.TemporaryDirectory() as td:
            with patch.object(hub.faster_backend, "_checkout",
                              return_value=Path(td)):
                self.tui.script += ["obama"]
                self._answer_common_options()
                cmd = hub._convert_faster(None)
        self.assertEqual(cmd[0], "convert")
        self.assertEqual(cmd[1], "faster")
        self.assertEqual(cmd[2]["voice"], "obama")
        self.assertIn("Server-side voice", self.tui.prompts[0])

    def test_faster_local_still_lists_voices_json(self):
        with tempfile.TemporaryDirectory() as td:
            checkout = Path(td)
            (checkout / "voices.json").write_text(
                json.dumps({"default": {}, "obama": {}}), encoding="utf-8")
            with patch.object(hub.faster_backend, "_checkout",
                              return_value=checkout):
                self.tui.script += ["obama"]
                self._answer_common_options()
                cmd = hub._convert_faster(None)
        self.assertEqual(cmd[2]["voice"], "obama")
        # The voice came from a menu over voices.json, not a text field.
        self.assertIn("Select the voice to clone", self.tui.prompts[0])


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


class SettingsTests(unittest.TestCase):
    """Settings menu: field collection, validation, config.py writing."""

    def test_write_config_preserves_comments_and_other_lines(self):
        import tempfile
        with tempfile.TemporaryDirectory() as td:
            path = Path(td) / "config.py"
            path.write_text(
                "# Default output options\n"
                'AUDIO_FORMAT = "m4b"\n'
                'AUDIO_BITRATE = "128k"\n'
                'LANGUAGE = "English"\n'
                "\n"
                "CHUNK_SIZE = 250  # words per request\n",
                encoding="utf-8")
            with patch.object(hub.config, "__file__", str(path)):
                hub._write_config({"AUDIO_FORMAT": "mp3",
                                   "AUDIO_BITRATE": "192k",
                                   "LANGUAGE": "Japanese",
                                   "CHUNK_SIZE": 300})
            text = path.read_text(encoding="utf-8")
        self.assertEqual(
            text,
            "# Default output options\n"
            'AUDIO_FORMAT = "mp3"\n'
            'AUDIO_BITRATE = "192k"\n'
            'LANGUAGE = "Japanese"\n'
            "\n"
            "CHUNK_SIZE = 300  # words per request\n")

    def test_write_config_missing_key_raises(self):
        import tempfile
        with tempfile.TemporaryDirectory() as td:
            path = Path(td) / "config.py"
            path.write_text("X = 1\n", encoding="utf-8")
            with patch.object(hub.config, "__file__", str(path)):
                with self.assertRaises(ValueError):
                    hub._write_config({"AUDIO_FORMAT": "mp3"})

    def test_apply_settings_writes_and_reloads_in_memory(self):
        written = {}

        def fake_write(updates):
            written.update(updates)

        original = {name: getattr(hub.config, name) for name in
                    ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE",
                     "CHUNK_SIZE", "QWEN_API_URL", "CLONE_API_URL",
                     "FASTER_API_URL", "AUDIOCPP_API_URL")}
        self.addCleanup(lambda: [setattr(hub.config, name, value)
                                 for name, value in original.items()])
        values = {"audio_format": "ogg", "audio_bitrate": " 192k ",
                  "language": "en", "chunk_size": "300",
                  "qwen_custom_port": "7862", "qwen_clone_port": "7863",
                  "faster_port": "8001", "audiocpp_port": "8081"}
        with patch.object(hub, "_write_config", fake_write), \
                patch.object(hub, "_sync_audiocpp_server_port"):
            hub._apply_settings(values)
        # Values are trimmed and language normalized to a display name.
        self.assertEqual(written, {"AUDIO_FORMAT": "ogg",
                                   "AUDIO_BITRATE": "192k",
                                   "LANGUAGE": "English",
                                   "CHUNK_SIZE": 300,
                                   "QWEN_API_URL": "http://127.0.0.1:7862",
                                   "CLONE_API_URL": "http://127.0.0.1:7863",
                                   "FASTER_API_URL": "http://127.0.0.1:8001",
                                   "AUDIOCPP_API_URL":
                                       "http://127.0.0.1:8081"})
        # In-memory config is reloaded so this session sees the change.
        self.assertEqual(hub.config.AUDIO_FORMAT, "ogg")
        self.assertEqual(hub.config.AUDIO_BITRATE, "192k")
        self.assertEqual(hub.config.LANGUAGE, "English")
        self.assertEqual(hub.config.CHUNK_SIZE, 300)
        self.assertEqual(hub.config.QWEN_API_URL, "http://127.0.0.1:7862")
        self.assertEqual(hub.config.FASTER_API_URL, "http://127.0.0.1:8001")

    def test_apply_settings_rejects_bad_values(self):
        original = {name: getattr(hub.config, name) for name in
                    ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE",
                     "CHUNK_SIZE", "QWEN_API_URL", "CLONE_API_URL",
                     "FASTER_API_URL", "AUDIOCPP_API_URL")}
        self.addCleanup(lambda: [setattr(hub.config, name, value)
                                 for name, value in original.items()])
        base = {"audio_format": "m4b", "audio_bitrate": "128k",
                "language": "English", "chunk_size": "250",
                "qwen_custom_port": "7860", "qwen_clone_port": "7861",
                "faster_port": "8000", "audiocpp_port": "8080"}
        with patch.object(hub, "_write_config") as mk_write:
            with self.assertRaises(ValueError):
                hub._apply_settings({**base, "language": "Klingon"})
            with self.assertRaises(ValueError):
                hub._apply_settings({**base, "chunk_size": "0"})
            with self.assertRaises(ValueError):
                hub._apply_settings({**base, "audiocpp_port": "70000"})
            mk_write.assert_not_called()

    def test_field_validators(self):
        self.assertIsNone(hub._validate_bitrate("128k"))
        self.assertIsNotNone(hub._validate_bitrate("   "))
        self.assertIsNone(hub._validate_language("English"))
        self.assertIsNone(hub._validate_language("en"))
        self.assertIsNotNone(hub._validate_language("Klingon"))
        self.assertIsNone(hub._validate_chunk_size("250"))
        self.assertIsNotNone(hub._validate_chunk_size("abc"))
        self.assertIsNotNone(hub._validate_chunk_size("0"))
        self.assertIsNone(hub._validate_port("8080"))
        self.assertIsNone(hub._validate_port("1"))
        self.assertIsNone(hub._validate_port("65535"))
        self.assertIsNotNone(hub._validate_port("0"))
        self.assertIsNotNone(hub._validate_port("70000"))
        self.assertIsNotNone(hub._validate_port("abc"))

    def test_settings_menu_builds_form_and_saves(self):
        captured = {}

        def fake_form(stdscr, title, fields, back_value=None):
            captured["fields"] = fields
            return {"audio_format": "ogg", "audio_bitrate": "192k",
                    "language": "English", "chunk_size": "300",
                    "qwen_custom_port": "7860", "qwen_clone_port": "7861",
                    "faster_port": "8000", "audiocpp_port": "8080"}

        applied = []

        def fake_apply(values):
            applied.append(values)

        def fake_flash(stdscr, text, kind="warn"):
            captured["flash"] = (text, kind)

        with patch.object(hub.tui, "form", fake_form), \
                patch.object(hub, "_apply_settings", fake_apply), \
                patch.object(hub.tui, "flash", fake_flash):
            hub._settings_menu(None)
        self.assertEqual([f["key"] for f in captured["fields"]],
                         ["audio_format", "audio_bitrate", "language",
                          "chunk_size", "audiocpp_port", "faster_port",
                          "qwen_custom_port", "qwen_clone_port"])
        kinds = {f["key"]: f["kind"] for f in captured["fields"]}
        self.assertEqual(kinds["audio_format"], "choice")
        self.assertEqual(kinds["audio_bitrate"], "text")
        self.assertEqual(kinds["audiocpp_port"], "text")
        labels = {f["key"]: f["label"] for f in captured["fields"]}
        self.assertEqual(labels["qwen_clone_port"], "qwen-tts Base port")
        self.assertNotIn("(clone)", " ".join(labels.values()))
        # The ports section note hangs off the first port field so it
        # renders between the output settings and the ports.
        notes = {f["key"]: f.get("note") for f in captured["fields"]}
        self.assertTrue(notes["audiocpp_port"])
        self.assertIsNone(notes["audio_format"])
        self.assertIsNone(notes["qwen_custom_port"])
        self.assertEqual(applied, [{"audio_format": "ogg",
                                    "audio_bitrate": "192k",
                                    "language": "English",
                                    "chunk_size": "300",
                                    "qwen_custom_port": "7860",
                                    "qwen_clone_port": "7861",
                                    "faster_port": "8000",
                                    "audiocpp_port": "8080"}])
        self.assertEqual(captured["flash"], ("Settings saved.", "ok"))

    def test_settings_menu_cancel_does_not_apply(self):
        def fake_form(stdscr, title, fields, back_value=None):
            return back_value  # user pressed Cancel

        applied = []

        def fake_apply(values):
            applied.append(values)

        with patch.object(hub.tui, "form", fake_form), \
                patch.object(hub, "_apply_settings", fake_apply):
            hub._settings_menu(None)
        self.assertEqual(applied, [])

    def test_settings_menu_writes_config_end_to_end(self):
        import tempfile
        tui._THEME.clear()
        self.addCleanup(tui._THEME.clear)
        curses = FakeCurses()
        patcher = patch.dict("sys.modules", {"curses": curses})
        patcher.start()
        self.addCleanup(patcher.stop)

        original = {name: getattr(hub.config, name) for name in
                    ("AUDIO_FORMAT", "AUDIO_BITRATE", "LANGUAGE",
                     "CHUNK_SIZE", "QWEN_API_URL", "CLONE_API_URL",
                     "FASTER_API_URL", "AUDIOCPP_API_URL")}
        self.addCleanup(lambda: [setattr(hub.config, name, value)
                                 for name, value in original.items()])

        with tempfile.TemporaryDirectory() as td:
            path = Path(td) / "config.py"
            path.write_text(
                "# Default output options\n"
                'AUDIO_FORMAT = "m4b"\n'
                'AUDIO_BITRATE = "128k"\n'
                'LANGUAGE = "English"\n'
                "\n"
                "CHUNK_SIZE = 250\n"
                'QWEN_API_URL = "http://127.0.0.1:7860"\n'
                'CLONE_API_URL = "http://127.0.0.1:7861"\n'
                'FASTER_API_URL = "http://127.0.0.1:8000"\n'
                'AUDIOCPP_API_URL = "http://127.0.0.1:8080"\n',
                encoding="utf-8")
            with patch.object(hub.config, "__file__", str(path)):
                # Down to Chunk size, Enter -> editor, Ctrl-U + '300',
                # Enter; Tab -> Save, Enter; a key dismisses the flash.
                screen = FakeScreen(keys=[
                    FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN,
                    FakeCurses.KEY_DOWN, 10, 21, ord("3"), ord("0"),
                    ord("0"), 10, 9, 10, 10])
                hub._settings_menu(screen)
            text = path.read_text(encoding="utf-8")
        self.assertIn('AUDIO_FORMAT = "m4b"', text)
        self.assertIn("CHUNK_SIZE = 300", text)
        # The running session also picked up the change in-memory.
        self.assertEqual(hub.config.CHUNK_SIZE, 300)

    def test_settings_menu_updates_backend_ports(self):
        import tempfile
        original = {name: getattr(hub.config, name) for name in
                    ("QWEN_API_URL", "CLONE_API_URL",
                     "FASTER_API_URL", "AUDIOCPP_API_URL")}
        self.addCleanup(lambda: [setattr(hub.config, name, value)
                                 for name, value in original.items()])
        with tempfile.TemporaryDirectory() as td:
            path = Path(td) / "config.py"
            path.write_text(
                'QWEN_API_URL = "http://127.0.0.1:7860"\n'
                'CLONE_API_URL = "http://127.0.0.1:7861"\n'
                'FASTER_API_URL = "http://127.0.0.1:8000"\n'
                'AUDIOCPP_API_URL = "http://127.0.0.1:8080"\n',
                encoding="utf-8")
            with patch.object(hub.config, "__file__", str(path)), \
                    patch.object(hub, "_sync_audiocpp_server_port"):
                hub._write_config({
                    "QWEN_API_URL": "http://127.0.0.1:7862",
                    "CLONE_API_URL": "http://127.0.0.1:7863",
                    "FASTER_API_URL": "http://127.0.0.1:8001",
                    "AUDIOCPP_API_URL": "http://127.0.0.1:8081",
                })
            text = path.read_text(encoding="utf-8")
        self.assertIn('QWEN_API_URL = "http://127.0.0.1:7862"', text)
        self.assertIn('CLONE_API_URL = "http://127.0.0.1:7863"', text)
        self.assertIn('FASTER_API_URL = "http://127.0.0.1:8001"', text)
        self.assertIn('AUDIOCPP_API_URL = "http://127.0.0.1:8081"', text)


class AudiocppServerConfigTests(unittest.TestCase):
    """update_server_config_port: rewriting the checkout's server.json."""

    def _make_checkout(self, td, port=8080):
        from backends import audiocpp as audiocpp_backend
        import json
        checkout = Path(td) / "audio.cpp"
        checkout.mkdir()
        server_json = checkout / "server.json"
        server_json.write_text(
            json.dumps({"host": "127.0.0.1", "port": port,
                        "models": [{"id": "qwen"}]}, indent=2),
            encoding="utf-8")
        return audiocpp_backend, checkout, server_json

    def test_rewrites_existing_server_json_port(self):
        import tempfile
        import json
        with tempfile.TemporaryDirectory() as td:
            mod, checkout, server_json = self._make_checkout(td, port=8080)
            with patch.object(mod, "find_local_checkout",
                              return_value=checkout):
                self.assertTrue(mod.update_server_config_port(9090))
            data = json.loads(server_json.read_text(encoding="utf-8"))
        self.assertEqual(data["port"], 9090)
        # Other keys are preserved.
        self.assertEqual(data["host"], "127.0.0.1")
        self.assertEqual(data["models"], [{"id": "qwen"}])

    def test_noop_when_port_unchanged(self):
        import tempfile
        with tempfile.TemporaryDirectory() as td:
            mod, checkout, server_json = self._make_checkout(td, port=8080)
            before = server_json.read_text(encoding="utf-8")
            with patch.object(mod, "find_local_checkout",
                              return_value=checkout):
                self.assertTrue(mod.update_server_config_port(8080))
            self.assertEqual(server_json.read_text(encoding="utf-8"), before)

    def test_false_when_no_checkout(self):
        from backends import audiocpp as audiocpp_backend
        with patch.object(audiocpp_backend, "find_local_checkout",
                          return_value=None):
            self.assertFalse(audiocpp_backend.update_server_config_port(9090))


if __name__ == "__main__":
    unittest.main()