diff options
Diffstat (limited to 'app/tests/test_backends.py')
| -rw-r--r-- | app/tests/test_backends.py | 55 |
1 files changed, 55 insertions, 0 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) |