diff options
Diffstat (limited to 'app/tests')
| -rw-r--r-- | app/tests/test_backends.py | 55 | ||||
| -rw-r--r-- | app/tests/test_backends_audiocpp.py | 230 | ||||
| -rw-r--r-- | app/tests/test_backends_common.py | 120 | ||||
| -rw-r--r-- | app/tests/test_backends_envs.py | 28 | ||||
| -rw-r--r-- | app/tests/test_backends_faster.py | 97 | ||||
| -rw-r--r-- | app/tests/test_hub.py | 158 |
6 files changed, 683 insertions, 5 deletions
diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py index 7fdcce6..d4534fb 100644 --- a/app/tests/test_backends.py +++ b/app/tests/test_backends.py @@ -47,6 +47,10 @@ class RegistryTests(unittest.TestCase): self.assertTrue(callable(info.setup_screen), info.key) self.assertTrue(callable(info.uninstall), info.key) + def test_every_entry_has_an_update_action(self): + for info in REGISTRY: + self.assertTrue(callable(info.update), info.key) + def test_get_returns_entry_by_key(self): self.assertIs(get("audiocpp").key, "audiocpp") self.assertIsNone(get("nonexistent")) @@ -868,3 +872,54 @@ class QwenUninstallTests(unittest.TestCase): return_value=1): rc = qwen.uninstall() self.assertEqual(rc, 1) + + +class QwenUpdateTests(unittest.TestCase): + """qwen.update: stop the single server, then pip install -U the package.""" + + def test_stops_server_and_pip_upgrades_into_the_qwen_env(self): + from backends import qwen + with patch.object(qwen.servers, "pid_for", return_value=1234), \ + patch.object(qwen.servers, "stop") as mk_stop, \ + patch.object(qwen.common, "pip_install", + return_value=0) as mk_pip: + rc = qwen.update(emit="EMIT") + self.assertEqual(rc, 0) + self.assertEqual([c.args[0] for c in mk_stop.call_args_list], + ["qwen"]) + # The task view's emit is forwarded, the install is an upgrade, + # and the package lands in the qwen backend's own venv. + mk_pip.assert_called_once_with([qwen.QWEN_PIP_PKG], emit="EMIT", + cancel=None, env_dir=qwen.QWEN_ENV, + upgrade=True) + + def test_skips_stop_when_no_server_was_started(self): + from backends import qwen + with patch.object(qwen.servers, "pid_for", return_value=None), \ + patch.object(qwen.servers, "stop") as mk_stop, \ + patch.object(qwen.common, "pip_install", return_value=0): + rc = qwen.update() + self.assertEqual(rc, 0) + mk_stop.assert_not_called() + + def test_cancel_before_pip_skips_the_upgrade(self): + import threading + + from backends import qwen + cancel = threading.Event() + cancel.set() + with patch.object(qwen.servers, "pid_for", return_value=1234), \ + patch.object(qwen.servers, "stop") as mk_stop, \ + patch.object(qwen.common, "pip_install") as mk_pip: + rc = qwen.update(cancel=cancel) + self.assertEqual(rc, 130) + self.assertEqual(mk_stop.call_count, 1) + mk_pip.assert_not_called() + + def test_pip_failure_propagates_the_exit_code(self): + from backends import qwen + with patch.object(qwen.servers, "pid_for", return_value=None), \ + patch.object(qwen.servers, "stop"), \ + patch.object(qwen.common, "pip_install", return_value=1): + rc = qwen.update() + self.assertEqual(rc, 1) diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py index ab405aa..3e97b82 100644 --- a/app/tests/test_backends_audiocpp.py +++ b/app/tests/test_backends_audiocpp.py @@ -1277,6 +1277,236 @@ class BuildAudiocppTests(unittest.TestCase): self.assertIn("No build script found", notices[0]) +class AudiocppUpdateTests(unittest.TestCase): + """update: stop the server, refresh the checkout, rebuild when stale. + + The rebuild fires when the checkout moved OR the on-disk binary is + missing/older than HEAD's commit time (an interrupted earlier build). + """ + + COMMIT_TIME = 1_000_000 + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = Path(self._td.name) / "audio.cpp" + self.checkout.mkdir() + self.addCleanup(common.drain_post_tui_notices) + + def tearDown(self): + self._td.cleanup() + + def _make_binary(self, backend="cuda"): + bin_dir = self.checkout / "build" / f"linux-{backend}-release" / "bin" + bin_dir.mkdir(parents=True, exist_ok=True) + binary = bin_dir / "audiocpp_server" + binary.write_bytes(b"x") + return binary + + def _patch_decision(self, heads, binary, *, commit_time=COMMIT_TIME): + """Patch git state + a built binary for BACKEND ("cuda" default). + + Returns the (mocks) (build, git_update) pair for assertions. + BINARY None means no binary on disk (a present binary is stamped + newer than COMMIT_TIME — stamp it differently after calling this + to simulate staleness); COMMIT_TIME None means the commit-time + probe cannot be answered. + """ + if binary is not None and commit_time is not None: + os.utime(binary, (commit_time + 100,) * 2) + return patch.object(common, "git_head", side_effect=heads), \ + patch.object(common, "git_commit_time", + return_value=commit_time), \ + patch.object(make_server.build, "load_server_config", + return_value={"backend": "cuda"}) + + def test_no_checkout_is_a_reported_noop(self): + with patch.object(make_server.build, "find_local_checkout", + return_value=None), \ + patch.object(make_server.build.servers, "pid_for", + return_value=None), \ + patch.object(common, "git_update") as mk_git: + rc = make_server.build.update() + self.assertEqual(rc, 0) + mk_git.assert_not_called() + + def test_stops_server_then_skips_rebuild_for_a_fresh_binary(self): + binary = self._make_binary() + patches = self._patch_decision(["a", "a"], binary) + with patch.object(make_server.build, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server.build.servers, "pid_for", + return_value=1234), \ + patch.object(make_server.build.servers, "stop") as mk_stop, \ + patches[0], patches[1], patches[2], \ + patch.object(common, "git_update", + return_value=0) as mk_git, \ + patch.object(make_server.build, "build_audiocpp") as mk_build: + rc = make_server.build.update(emit="EMIT") + self.assertEqual(rc, 0) + mk_stop.assert_called_once_with("audiocpp") + mk_git.assert_called_once_with(self.checkout, emit="EMIT", + cancel=None) + # HEAD did not move and the binary is newer than HEAD's commit: + # the binary still matches the sources. + mk_build.assert_not_called() + + def test_moved_head_rebuilds_even_with_a_fresh_binary(self): + binary = self._make_binary() + patches = self._patch_decision(["a", "b"], binary) + with patch.object(make_server.build, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server.build.servers, "pid_for", + return_value=None), \ + patches[0], patches[1], patches[2], \ + patch.object(common, "git_update", return_value=0), \ + patch.object(make_server.build, "build_audiocpp", + return_value=0) as mk_build: + rc = make_server.build.update(emit="EMIT") + self.assertEqual(rc, 0) + mk_build.assert_called_once_with(self.checkout, "cuda", + emit="EMIT", cancel=None) + + def test_moved_head_falls_back_to_the_detected_backend(self): + patches = self._patch_decision(["a", "b"], None) + with patch.object(make_server.build, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server.build.servers, "pid_for", + return_value=None), \ + patches[0], patches[1], \ + patch.object(make_server.build, "load_server_config", + return_value={}), \ + patch.object(make_server.build, "detect_backend", + return_value="vulkan"), \ + patch.object(common, "git_update", return_value=0), \ + patch.object(make_server.build, "build_audiocpp", + return_value=0) as mk_build: + rc = make_server.build.update() + self.assertEqual(rc, 0) + mk_build.assert_called_once_with(self.checkout, "vulkan", + emit=None, cancel=None) + + def test_no_known_backend_skips_the_rebuild(self): + with patch.object(make_server.build, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server.build.servers, "pid_for", + return_value=None), \ + patch.object(common, "git_head", side_effect=["a", "b"]), \ + patch.object(common, "git_update", return_value=0), \ + patch.object(make_server.build, "load_server_config", + return_value={}), \ + patch.object(make_server.build, "detect_backend", + return_value=None), \ + patch.object(make_server.build, "build_audiocpp") as mk_build: + rc = make_server.build.update() + self.assertEqual(rc, 0) + mk_build.assert_not_called() + + def test_stale_binary_rebuilds_without_head_movement(self): + # The cancelled-rebuild scenario: sources already at HEAD, the old + # binary predates the new commit → the next update rebuilds. + binary = self._make_binary() + patches = self._patch_decision(["a", "a"], binary) + os.utime(binary, (self.COMMIT_TIME - 100,) * 2) + with patch.object(make_server.build, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server.build.servers, "pid_for", + return_value=None), \ + patches[0], patches[1], patches[2], \ + patch.object(common, "git_update", return_value=0), \ + patch.object(make_server.build, "build_audiocpp", + return_value=0) as mk_build: + rc = make_server.build.update() + self.assertEqual(rc, 0) + mk_build.assert_called_once_with(self.checkout, "cuda", + emit=None, cancel=None) + + def test_missing_binary_rebuilds_without_head_movement(self): + patches = self._patch_decision(["a", "a"], None) + with patch.object(make_server.build, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server.build.servers, "pid_for", + return_value=None), \ + patches[0], patches[1], patches[2], \ + patch.object(common, "git_update", return_value=0), \ + patch.object(make_server.build, "build_audiocpp", + return_value=0) as mk_build: + rc = make_server.build.update() + self.assertEqual(rc, 0) + mk_build.assert_called_once_with(self.checkout, "cuda", + emit=None, cancel=None) + + def test_unknown_commit_time_rebuilds_without_head_movement(self): + binary = self._make_binary() + patches = self._patch_decision(["a", "a"], binary, commit_time=None) + with patch.object(make_server.build, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server.build.servers, "pid_for", + return_value=None), \ + patches[0], patches[1], patches[2], \ + patch.object(common, "git_update", return_value=0), \ + patch.object(make_server.build, "build_audiocpp", + return_value=0) as mk_build: + rc = make_server.build.update() + self.assertEqual(rc, 0) + mk_build.assert_called_once_with(self.checkout, "cuda", + emit=None, cancel=None) + + def test_checkout_failure_skips_the_rebuild(self): + with patch.object(make_server.build, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server.build.servers, "pid_for", + return_value=None), \ + patch.object(common, "git_update", + return_value=128) as mk_git, \ + patch.object(make_server.build, "build_audiocpp") as mk_build: + rc = make_server.build.update() + self.assertEqual(rc, 128) + mk_git.assert_called_once() + mk_build.assert_not_called() + + def test_rebuild_failure_propagates_the_exit_code(self): + with patch.object(make_server.build, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server.build.servers, "pid_for", + return_value=None), \ + patch.object(common, "git_head", side_effect=["a", "b"]), \ + patch.object(common, "git_update", return_value=0), \ + patch.object(make_server.build, "load_server_config", + return_value={"backend": "cuda"}), \ + patch.object(make_server.build, "build_audiocpp", + return_value=2): + rc = make_server.build.update() + self.assertEqual(rc, 2) + + def test_cancel_before_the_update_skips_everything_after_stopping(self): + cancel = threading.Event() + cancel.set() + with patch.object(make_server.build, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server.build.servers, "pid_for", + return_value=1234), \ + patch.object(make_server.build.servers, "stop") as mk_stop, \ + patch.object(common, "git_update") as mk_git: + rc = make_server.build.update(cancel=cancel) + self.assertEqual(rc, 130) + mk_stop.assert_called_once_with("audiocpp") + mk_git.assert_not_called() + + def test_cancel_after_the_checkout_skips_the_rebuild(self): + cancel = threading.Event() + cancel.set() + with patch.object(make_server.build, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server.build.servers, "pid_for", + return_value=None), \ + patch.object(common, "git_head", side_effect=["a", "b"]), \ + patch.object(common, "git_update", return_value=0), \ + patch.object(make_server.build, "build_audiocpp") as mk_build: + rc = make_server.build.update(cancel=cancel) + self.assertEqual(rc, 130) + mk_build.assert_not_called() + + class AudiocppDetectTests(unittest.TestCase): """backends.audiocpp.detect() status reporting.""" diff --git a/app/tests/test_backends_common.py b/app/tests/test_backends_common.py index 5bc967a..a94994b 100644 --- a/app/tests/test_backends_common.py +++ b/app/tests/test_backends_common.py @@ -73,6 +73,126 @@ class GitCloneTests(unittest.TestCase): self.assertEqual(run.call_args[1]["emit"], emit) +class GitUpdateTests(unittest.TestCase): + """git_update: fetch, then hard reset to origin's default branch.""" + + def _patched(self, fetch_rc=0, branch="main"): + """Patch run_console_subprocess (fetch/reset) and the branch probe.""" + run = mock.patch.object(common, "run_console_subprocess", + return_value=fetch_rc).start() + mock.patch.object(common, "_origin_default_branch", + return_value=branch).start() + return run + + def tearDown(self): + mock.patch.stopall() + + def test_fetch_then_hard_reset_to_origin_head(self): + run = self._patched() + self.assertEqual(common.git_update(common.Path("/co")), 0) + self.assertEqual( + run.call_args_list[0][0][0], + ["git", "-C", "/co", "fetch", "origin"]) + self.assertEqual( + run.call_args_list[1][0][0], + ["git", "-C", "/co", "reset", "--hard", "origin/main"]) + + def test_streaming_adds_progress_and_passes_emit(self): + emit = lambda line: None # noqa: E731 + run = self._patched() + self.assertEqual(common.git_update(common.Path("/co"), emit=emit), 0) + self.assertEqual( + run.call_args_list[0][0][0], + ["git", "-C", "/co", "fetch", "--progress", "origin"]) + self.assertEqual(run.call_args_list[0][1]["emit"], emit) + self.assertEqual(run.call_args_list[1][1]["emit"], emit) + + def test_fetch_failure_short_circuits_the_reset(self): + run = self._patched(fetch_rc=128) + self.assertEqual(common.git_update(common.Path("/co")), 128) + self.assertEqual(run.call_count, 1) + + def test_reset_uses_the_remote_default_branch(self): + run = self._patched(branch="trunk") + self.assertEqual(common.git_update(common.Path("/co")), 0) + self.assertEqual( + run.call_args_list[1][0][0], + ["git", "-C", "/co", "reset", "--hard", "origin/trunk"]) + + +class OriginDefaultBranchTests(unittest.TestCase): + def test_symbolic_ref_name_is_returned(self): + proc = mock.Mock(returncode=0, + stdout=b"refs/remotes/origin/master\n") + with mock.patch.object(common, "run_console_subprocess_quiet", + return_value=proc): + self.assertEqual(common._origin_default_branch( + common.Path("/co")), "master") + + def test_missing_ref_falls_back_to_main(self): + proc = mock.Mock(returncode=128, stdout=b"") + with mock.patch.object(common, "run_console_subprocess_quiet", + return_value=proc): + self.assertEqual(common._origin_default_branch( + common.Path("/co")), "main") + + def test_unstartable_probe_falls_back_to_main(self): + with mock.patch.object(common, "run_console_subprocess_quiet", + return_value=None): + self.assertEqual(common._origin_default_branch( + common.Path("/co")), "main") + + +class GitHeadTests(unittest.TestCase): + def test_head_sha_is_returned(self): + proc = mock.Mock(returncode=0, stdout=b"abc123\n") + with mock.patch.object(common, "run_console_subprocess_quiet", + return_value=proc) as run: + self.assertEqual(common.git_head(common.Path("/co")), "abc123") + self.assertEqual(run.call_args[0][0], + ["git", "-C", "/co", "rev-parse", "HEAD"]) + + def test_not_a_repo_yields_none(self): + proc = mock.Mock(returncode=128, stdout=b"") + with mock.patch.object(common, "run_console_subprocess_quiet", + return_value=proc): + self.assertIsNone(common.git_head(common.Path("/co"))) + + def test_unstartable_probe_yields_none(self): + with mock.patch.object(common, "run_console_subprocess_quiet", + return_value=None): + self.assertIsNone(common.git_head(common.Path("/co"))) + + +class GitCommitTimeTests(unittest.TestCase): + def test_committer_time_is_parsed(self): + proc = mock.Mock(returncode=0, stdout=b"1756300000\n") + with mock.patch.object(common, "run_console_subprocess_quiet", + return_value=proc) as run: + self.assertEqual(common.git_commit_time(common.Path("/co")), + 1756300000) + self.assertEqual(run.call_args[0][0], + ["git", "-C", "/co", "show", "-s", + "--format=%ct", "HEAD"]) + + def test_not_a_repo_yields_none(self): + proc = mock.Mock(returncode=128, stdout=b"") + with mock.patch.object(common, "run_console_subprocess_quiet", + return_value=proc): + self.assertIsNone(common.git_commit_time(common.Path("/co"))) + + def test_unparsable_output_yields_none(self): + proc = mock.Mock(returncode=0, stdout=b"not-a-number\n") + with mock.patch.object(common, "run_console_subprocess_quiet", + return_value=proc): + self.assertIsNone(common.git_commit_time(common.Path("/co"))) + + def test_unstartable_probe_yields_none(self): + with mock.patch.object(common, "run_console_subprocess_quiet", + return_value=None): + self.assertIsNone(common.git_commit_time(common.Path("/co"))) + + class ParseRequestOptionsTests(unittest.TestCase): """parse_request_options: the shared --option / TUI-field parser.""" diff --git a/app/tests/test_backends_envs.py b/app/tests/test_backends_envs.py index 8b17458..2f0f910 100644 --- a/app/tests/test_backends_envs.py +++ b/app/tests/test_backends_envs.py @@ -154,6 +154,34 @@ class PipInstallTests(unittest.TestCase): self.assertEqual(rc, 1) run.assert_not_called() + def test_upgrade_adds_the_u_flag(self): + calls = [] + + def fake_run(argv, **kwargs): + calls.append(list(argv)) + return 0 + + with patch.object(envs, "env_exists", return_value=True), \ + patch.object(envs.common, "run_console_subprocess", + side_effect=fake_run): + self.assertEqual(envs.pip_install(["qwen-tts"], upgrade=True), 0) + self.assertIn("-U", calls[0]) + # -U sits before the packages; nothing else about the argv changes. + self.assertLess(calls[0].index("-U"), calls[0].index("qwen-tts")) + + def test_no_upgrade_flag_by_default(self): + calls = [] + + def fake_run(argv, **kwargs): + calls.append(list(argv)) + return 0 + + with patch.object(envs, "env_exists", return_value=True), \ + patch.object(envs.common, "run_console_subprocess", + side_effect=fake_run): + self.assertEqual(envs.pip_install(["qwen-tts"]), 0) + self.assertNotIn("-U", calls[0]) + class PipUninstallTests(unittest.TestCase): def test_missing_env_is_success_without_running_pip(self): diff --git a/app/tests/test_backends_faster.py b/app/tests/test_backends_faster.py index 8f024bf..44f4907 100644 --- a/app/tests/test_backends_faster.py +++ b/app/tests/test_backends_faster.py @@ -470,3 +470,100 @@ class UninstallTests(unittest.TestCase): 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) 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 = {} |