path: root/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 unittest
from pathlib import Path
from unittest.mock import patch

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


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)
        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"))
        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, Quit]. Quit is the
        # 2nd option (Down once) then Enter.
        screen = FakeScreen(keys=[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_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...", "Quit"])

    def test_menu_has_all_five_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...", "Server...", "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.
        self.assertEqual(
            captured["rows"],
            [("audio.cpp", "unavailable", "err", "dim"),
             ("qwen-tts", "running", "ok", "body")])

    def test_menu_has_all_five_when_one_running_only(self):
        # Running but not installed (an external server) still unlocks the
        # Convert/Configure/Server entries.
        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...",
             "Configure a backend...", "Server...", "Quit"])

    def test_convert_with_no_available_backend_offers_setup(self):
        # One installed-but-not-ready backend → Convert is offered. The
        # convert menu lists no available backend, so only "Set up a
        # backend..." is shown; Enter selects it → setup menu lists 3
        # backends; Esc goes back → convert returns None → main menu loops.
        # Then quit: main menu now has 5 options, Quit is the 5th (Down x4).
        from backends import BackendInfo, BackendStatus
        none = BackendStatus("k", "l", installed=True, configured=False)
        infos = [BackendInfo("audiocpp", "audio.cpp", lambda: none,
                             lambda: 0),
                 BackendInfo("qwen", "qwen-tts", lambda: none, lambda: 0),
                 BackendInfo("faster", "faster", lambda: none, lambda: 0)]
        # installed=True so the main menu shows Convert; but ready/running
        # is False so the convert menu's available list is empty.
        statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True,
                                  configured=False),
                    BackendStatus("qwen", "qwen-tts", installed=True,
                                  configured=False),
                    BackendStatus("faster", "faster", installed=True,
                                  configured=False)]
        with patch.object(hub, "detect_all", return_value=statuses), \
                patch.object(hub, "REGISTRY", infos):
            # Convert(Enter), setup-entry(Enter), Esc on setup menu,
            # back at main menu -> Down x4 -> Enter (Quit).
            screen = FakeScreen(keys=[10, 10, 27,
                                       FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN,
                                       FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN,
                                        10])
            result = hub._hub_menu(screen)
        self.assertIsNone(result)


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


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