diff options
Diffstat (limited to 'app/tests/test_hub.py')
| -rw-r--r-- | app/tests/test_hub.py | 158 |
1 files changed, 153 insertions, 5 deletions
diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index a9d4a4b..4186c92 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -113,7 +113,7 @@ class HubHelperTests(unittest.TestCase): self.assertEqual(hub._status_mark(remote_models), ("running [remote] (Base)", "ok", "body")) self.assertEqual(hub._status_mark(installed), - ("installed", "warn", "body")) + ("installed", "ok", "body")) self.assertEqual(hub._status_mark(none), ("unavailable", "err", "dim")) self.assertEqual(hub._status_mark(None), @@ -213,7 +213,7 @@ class HubMenuTests(unittest.TestCase): "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")]) + [("qwen-tts", "installed", "ok", "body")]) def test_table_dims_name_when_not_installed_and_not_running(self): captured = {} @@ -378,7 +378,7 @@ class SubmenuStatusTableTests(unittest.TestCase): self.assertEqual(captured["table_title"], "Backend status") self.assertEqual( captured["table_rows"], - [("qwen-tts", "installed", "warn", "body"), + [("qwen-tts", "installed", "ok", "body"), ("faster-qwen3-tts", "running [remote]", "ok", "body")]) self.assertIsNone(captured["notice_lines"]) @@ -398,6 +398,32 @@ class SubmenuStatusTableTests(unittest.TestCase): self.assertEqual([label for label, _ in captured["options"]], ["Install Backend"]) + def test_configure_backends_menu_lists_update_between_install_uninstall(self): + captured = {} + infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, + update=lambda **kw: 0), + BackendInfo("faster", "faster-qwen3-tts", lambda: None, + lambda: 0, update=lambda **kw: 0)] + statuses = [ + BackendStatus("qwen", "qwen-tts", installed=True, + configured=True), + BackendStatus("faster", "faster-qwen3-tts", installed=False, + configured=False), + ] + with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub, "detect_all", return_value=statuses), \ + patch.object(hub.tui, "menu", + self._capture_menu(captured)), \ + patch.object(hub.shutil, "which", return_value="/x"): + result = hub._Hub(None).screen_configure() + self.assertIs(result, tui.Wizard.BACK) + # Update sits between Install and Uninstall; it is offered once for + # the whole set of installed backends (faster has nothing on disk + # and so contributes nothing). + self.assertEqual([label for label, _ in captured["options"]], + ["Install Backend", "Update backends", + "Uninstall Backend"]) + def test_configure_backends_menu_audiocpp_model_actions(self): captured = {} infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0)] @@ -541,7 +567,7 @@ class SubmenuStatusTableTests(unittest.TestCase): self.assertEqual(self._labels(captured["options"]), ["Configure qwen-tts", "Uninstall Backend"]) self.assertEqual(captured["table_rows"], - [("qwen-tts", "installed", "warn", "body")]) + [("qwen-tts", "installed", "ok", "body")]) def test_selecting_qwen_runs_its_configure_screen_not_setup(self): ran = [] @@ -653,7 +679,7 @@ class SubmenuStatusTableTests(unittest.TestCase): self.assertIs(result, tui.Wizard.BACK) self.assertEqual(captured["table_title"], "Backend status") self.assertEqual( - captured["table_rows"], [("qwen-tts", "installed", "warn", + captured["table_rows"], [("qwen-tts", "installed", "ok", "body")]) def test_server_menu_lists_only_installed_backends(self): @@ -2762,6 +2788,128 @@ class ConfigureBackendsDispatchTests(unittest.TestCase): self.assertEqual(len(flashes), 1) self.assertEqual(flashes[0][1], "err") + def test_update_backends_action_runs_one_step_per_installed_backend(self): + calls = [] + + def make_update(name): + def update(*, emit=None, cancel=None): + calls.append((name, emit, cancel)) + return 0 + return update + + # faster is installed in statuses but has no update action → one + # step fewer; audiocpp's on-disk check routes through the checkout. + infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, + lambda: 0, update=make_update("audiocpp")), + BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, + update=make_update("qwen")), + BackendInfo("faster", "faster-qwen3-tts", lambda: None, + lambda: 0)] + statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True, + configured=True), + BackendStatus("qwen", "qwen-tts", installed=True, + configured=True), + BackendStatus("faster", "faster-qwen3-tts", + installed=True, configured=True)] + patch_flash, flashes = self._capture_flashes() + with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub, "detect_all", return_value=statuses), \ + patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=Path("/co")), \ + patch.object(hub.taskview, "run_steps", + return_value=0) as mk_run, \ + patch_flash: + hub._update_backends_action(None) + # One task-view run titled "Update backends", one step per + # updatable backend in registry order; executing a step + # forwards emit/cancel to that backend's update. + mk_run.assert_called_once() + self.assertEqual(mk_run.call_args[0][0], None) + self.assertEqual(mk_run.call_args[0][1], "Update backends") + steps = mk_run.call_args[0][2] + self.assertEqual([step.title for step in steps], + ["Update audio.cpp", "Update qwen-tts"]) + def emit(line): + pass + steps[1].work(emit, "CANCEL") + self.assertEqual(calls, [("qwen", emit, "CANCEL")]) + self.assertEqual(flashes[-1][1], "ok") + + def test_update_backends_action_flashes_error_when_something_failed(self): + infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, + update=lambda **kw: 0)] + statuses = [BackendStatus("qwen", "qwen-tts", installed=True, + configured=True)] + patch_flash, flashes = self._capture_flashes() + with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub, "detect_all", return_value=statuses), \ + patch.object(hub.taskview, "run_steps", return_value=1), \ + patch_flash: + hub._update_backends_action(None) + self.assertEqual(flashes[-1][1], "err") + self.assertIn("did not complete", flashes[-1][0]) + + def test_update_backends_action_flashes_warn_when_cancelled(self): + infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, + update=lambda **kw: 0)] + statuses = [BackendStatus("qwen", "qwen-tts", installed=True, + configured=True)] + patch_flash, flashes = self._capture_flashes() + with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub, "detect_all", return_value=statuses), \ + patch.object(hub.taskview, "run_steps", return_value=130), \ + patch_flash: + hub._update_backends_action(None) + self.assertEqual(flashes[-1][1], "warn") + self.assertIn("cancelled", flashes[-1][0]) + + def test_update_backends_action_without_targets_flashes_a_hint(self): + # An installed backend without an update action (and nothing else + # installed): the entry never shows, but a direct call still + # explains itself instead of running an empty task view. + infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)] + statuses = [BackendStatus("qwen", "qwen-tts", installed=True, + configured=True)] + patch_flash, flashes = self._capture_flashes() + with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub, "detect_all", return_value=statuses), \ + patch.object(hub.taskview, "run_steps") as mk_run, \ + patch_flash: + hub._update_backends_action(None) + mk_run.assert_not_called() + self.assertEqual(flashes, + [("No installed backend supports updating.", + "warn")]) + + def test_selecting_update_runs_the_action_and_reshows_the_menu(self): + titles = [] + + def fake_menu(stdscr, title, options, **kwargs): + titles.append(title) + return "update" if len(titles) == 1 else tui.Wizard.BACK + + ran = [] + invalidated = [] + info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0, + update=lambda **kw: 0) + statuses = [BackendStatus("qwen", "qwen-tts", installed=True, + configured=True)] + with patch.object(hub.tui, "menu", fake_menu), \ + patch.object(hub, "detect_all", return_value=statuses), \ + patch.object(hub, "REGISTRY", [info]), \ + patch.object(hub, "_update_backends_action", + side_effect=lambda scr: ran.append("update")), \ + patch.object(hub, "invalidate_detect_cache", + side_effect=lambda: invalidated.append(True)), \ + patch.object(hub.shutil, "which", return_value="/x"): + result = hub._Hub(None).screen_configure() + self.assertIs(result, tui.Wizard.BACK) + self.assertEqual(ran, ["update"]) + self.assertEqual(invalidated, [True]) + # An inline action: the same menu re-shows (second title) with a + # freshly detected status table. + self.assertEqual(titles, ["Configure backends", "Configure backends"]) + def test_pick_backend_install_lists_uninstalled_only(self): captured = {} |