diff options
Diffstat (limited to 'app/tests')
| -rw-r--r-- | app/tests/test_backends_common.py | 8 | ||||
| -rw-r--r-- | app/tests/test_hub.py | 159 | ||||
| -rw-r--r-- | app/tests/test_selfupdate.py | 241 |
3 files changed, 404 insertions, 4 deletions
diff --git a/app/tests/test_backends_common.py b/app/tests/test_backends_common.py index 68a5543..71f0c17 100644 --- a/app/tests/test_backends_common.py +++ b/app/tests/test_backends_common.py @@ -161,7 +161,7 @@ class GitUpdateTests(unittest.TestCase): """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", + mock.patch.object(common, "origin_default_branch", return_value=branch).start() return run @@ -207,20 +207,20 @@ class OriginDefaultBranchTests(unittest.TestCase): 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( + 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( + 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( + self.assertEqual(common.origin_default_branch( common.Path("/co")), "main") diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index 485a37f..228a103 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -394,6 +394,14 @@ class SubmenuStatusTableTests(unittest.TestCase): servers, showing running/stopped inline instead of the table. """ + def setUp(self): + # These tests concern the backend actions; the host's git-checkout + # state (which gates the "Update Generator" entry) must not leak in. + patcher = patch.object(hub.selfupdate, "is_git_checkout", + return_value=False) + patcher.start() + self.addCleanup(patcher.stop) + def _capture_menu(self, captured): def fake_menu(stdscr, title, options, **kwargs): captured["title"] = title @@ -4102,6 +4110,157 @@ class ConfigureBackendsDispatchTests(unittest.TestCase): # freshly detected status table. self.assertEqual(titles, ["Configure Backends", "Configure Backends"]) + def test_configure_menu_offers_update_generator_for_git_checkouts(self): + def options_for(git_checkout): + captured = {} + + def fake_menu(stdscr, title, options, **kwargs): + captured["options"] = options + return tui.Wizard.BACK + + info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 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.selfupdate, "is_git_checkout", + return_value=git_checkout): + hub._Hub(None).screen_configure() + return [opt[0] for opt in captured["options"]] + + self.assertIn("Update Generator", options_for(True)) + self.assertNotIn("Update Generator", options_for(False)) + + def test_selecting_update_generator_runs_the_action_and_reshows(self): + titles = [] + ran = [] + + def fake_menu(stdscr, title, options, **kwargs): + titles.append(title) + return "update_self" if len(titles) == 1 else tui.Wizard.BACK + + info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 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_self_action", + side_effect=lambda scr: ran.append(True)): + result = hub._Hub(None).screen_configure() + self.assertIs(result, tui.Wizard.BACK) + self.assertEqual(ran, [True]) + # An inline action: the same menu re-shows after the update. + self.assertEqual(titles, ["Configure Backends", "Configure Backends"]) + + def test_update_self_action_flashes_err_without_git_checkout(self): + patch_flash, flashes = self._capture_flashes() + with patch.object(hub.selfupdate, "is_git_checkout", + return_value=False), \ + patch.object(hub.taskview, "run_steps") as mk_run, \ + patch_flash: + hub._update_self_action(None) + mk_run.assert_not_called() + self.assertEqual(len(flashes), 1) + self.assertEqual(flashes[0][1], "err") + self.assertIn("not a git checkout", flashes[0][0]) + + def test_update_self_action_declined_confirm_aborts_before_running(self): + patch_flash, flashes = self._capture_flashes() + with patch.object(hub.selfupdate, "is_git_checkout", + return_value=True), \ + patch.object(hub.selfupdate, "current_commit", + return_value="aaa"), \ + patch.object(hub.selfupdate, "modified_tracked_files", + return_value=["app/converter/config.py"]), \ + patch.object(hub.tui, "confirm", + return_value=False) as mk_confirm, \ + patch.object(hub.taskview, "run_steps") as mk_run, \ + patch_flash: + hub._update_self_action(None) + mk_run.assert_not_called() + self.assertEqual(flashes, []) + # The confirm names the files that will be preserved. + body = mk_confirm.call_args[1]["body"] + self.assertIn("app/converter/config.py", body) + + def test_update_self_action_runs_step_and_flashes_update(self): + patch_flash, flashes = self._capture_flashes() + with patch.object(hub.selfupdate, "is_git_checkout", + return_value=True), \ + patch.object(hub.selfupdate, "current_commit", + side_effect=["aaa", "bbb"]), \ + patch.object(hub.selfupdate, "modified_tracked_files", + return_value=[]), \ + patch.object(hub.taskview, "run_steps", + return_value=0) as mk_run, \ + patch.object(hub.selfupdate, "update_generator", + return_value=0) as mk_update, \ + patch_flash: + hub._update_self_action(None) + mk_run.assert_called_once() + self.assertEqual(mk_run.call_args[0][0], None) + self.assertEqual(mk_run.call_args[0][1], "Update Generator") + steps = mk_run.call_args[0][2] + self.assertEqual([step.title for step in steps], + ["Fetch and reset the checkout"]) + + def emit(line): + pass + + steps[0].work(emit, "CANCEL") + mk_update.assert_called_once_with(emit=emit, cancel="CANCEL") + self.assertEqual(len(flashes), 1) + text, kind = flashes[0] + self.assertEqual(kind, "ok") + self.assertIn("aaa", text) + self.assertIn("bbb", text) + self.assertIn("Restart to apply", text) + + def test_update_self_action_flashes_already_up_to_date(self): + patch_flash, flashes = self._capture_flashes() + with patch.object(hub.selfupdate, "is_git_checkout", + return_value=True), \ + patch.object(hub.selfupdate, "current_commit", + side_effect=["aaa", "aaa"]), \ + patch.object(hub.selfupdate, "modified_tracked_files", + return_value=[]), \ + patch.object(hub.taskview, "run_steps", return_value=0), \ + patch_flash: + hub._update_self_action(None) + self.assertEqual(flashes, + [("The generator is already up to date (aaa).", + "ok")]) + + def test_update_self_action_flashes_warn_when_cancelled(self): + patch_flash, flashes = self._capture_flashes() + with patch.object(hub.selfupdate, "is_git_checkout", + return_value=True), \ + patch.object(hub.selfupdate, "current_commit", + side_effect=["aaa", "aaa"]), \ + patch.object(hub.selfupdate, "modified_tracked_files", + return_value=[]), \ + patch.object(hub.taskview, "run_steps", return_value=130), \ + patch_flash: + hub._update_self_action(None) + self.assertEqual(flashes[-1][1], "warn") + self.assertIn("cancelled", flashes[-1][0]) + + def test_update_self_action_flashes_err_when_failed(self): + patch_flash, flashes = self._capture_flashes() + with patch.object(hub.selfupdate, "is_git_checkout", + return_value=True), \ + patch.object(hub.selfupdate, "current_commit", + side_effect=["aaa", "aaa"]), \ + patch.object(hub.selfupdate, "modified_tracked_files", + return_value=[]), \ + patch.object(hub.taskview, "run_steps", return_value=1), \ + patch_flash: + hub._update_self_action(None) + self.assertEqual(flashes[-1][1], "err") + self.assertIn("did not complete", flashes[-1][0]) + def test_pick_backend_install_lists_uninstalled_only(self): captured = {} diff --git a/app/tests/test_selfupdate.py b/app/tests/test_selfupdate.py new file mode 100644 index 0000000..bf2f6b3 --- /dev/null +++ b/app/tests/test_selfupdate.py @@ -0,0 +1,241 @@ +"""Tests for the generator's self-update (selfupdate.py). + +The update is a fetch + conditional hard reset of the generator's own +checkout; its defining property is preservation: locally modified tracked +files (app/converter/config.py above all) are snapshotted before the reset +and written back afterwards, so an update never overwrites user config. +These tests simulate the reset's effect on the working tree via the +subprocess mocks' side effects and assert what survives. +""" + +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import selfupdate + + +def _proc(returncode=0, stdout=b""): + proc = mock.Mock() + proc.returncode = returncode + proc.stdout = stdout + return proc + + +class IsGitCheckoutTests(unittest.TestCase): + def test_git_directory_and_worktree_file_count(self): + with tempfile.TemporaryDirectory() as td: + root = Path(td) + self.assertFalse(selfupdate.is_git_checkout(root)) + (root / ".git").mkdir() + self.assertTrue(selfupdate.is_git_checkout(root)) + (root / ".git").rmdir() + (root / ".git").write_text("gitdir: /elsewhere\n", + encoding="utf-8") + self.assertTrue(selfupdate.is_git_checkout(root)) + + def test_defaults_to_the_install_root(self): + # The repo running the tests is a git checkout. + self.assertTrue(selfupdate.is_git_checkout()) + + +class CurrentCommitTests(unittest.TestCase): + def test_short_sha_is_parsed(self): + proc = _proc(0, b"abc1234\n") + with mock.patch.object(selfupdate.common, + "run_console_subprocess_quiet", + return_value=proc): + self.assertEqual(selfupdate.current_commit(Path("/co")), + "abc1234") + + def test_git_failure_returns_none(self): + proc = _proc(128, b"") + with mock.patch.object(selfupdate.common, + "run_console_subprocess_quiet", + return_value=proc): + self.assertIsNone(selfupdate.current_commit(Path("/co"))) + with mock.patch.object(selfupdate.common, + "run_console_subprocess_quiet", + return_value=None): + self.assertIsNone(selfupdate.current_commit(Path("/co"))) + + +class ModifiedTrackedFilesTests(unittest.TestCase): + def test_porcelain_lines_are_parsed(self): + # --untracked-files=no: git never reports "??" entries here. + proc = _proc(0, b" M app/converter/config.py\n" + b"M staged.py\n" + b" D deleted.py\n" + b"R old.py -> new.py\n") + with mock.patch.object(selfupdate.common, + "run_console_subprocess_quiet", + return_value=proc): + self.assertEqual( + selfupdate.modified_tracked_files(Path("/co")), + ["app/converter/config.py", "staged.py", "deleted.py", + "new.py"]) + + def test_git_failure_returns_empty(self): + proc = _proc(128, b"") + with mock.patch.object(selfupdate.common, + "run_console_subprocess_quiet", + return_value=proc): + self.assertEqual(selfupdate.modified_tracked_files(Path("/co")), + []) + + +class UpdateGeneratorTests(unittest.TestCase): + """update_generator against a fake checkout in a temp directory.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.root = Path(self._tmp.name) + self.config = self.root / "app" / "converter" / "config.py" + self.config.parent.mkdir(parents=True) + self.config.write_bytes(b"# user settings\n") + self.emitted = [] + + def tearDown(self): + mock.patch.stopall() + self._tmp.cleanup() + + def _patch_git(self, *, fetch_rc=0, reset_rc=0, head=b"aaa\n", + remote=b"bbb\n", status=b" M app/converter/config.py\n", + on_reset=None): + """Patch the subprocess layer; returns the run_console mock. + + run_console_subprocess serves fetch (first call) and reset (second); + ON_RESET, when given, runs before the reset's return code so a test + can simulate the working-tree damage a real reset does (rewriting or + deleting files). + """ + run = mock.patch.object( + selfupdate.common, "run_console_subprocess", + side_effect=self._run_side_effect(fetch_rc, reset_rc, + on_reset)).start() + + def quiet(argv, **kwargs): + if "status" in argv: + return _proc(0, status) + if "rev-parse" in argv: + ref = argv[argv.index("rev-parse") + 2] + return _proc(0, remote if ref.startswith("origin/") + else head) + return _proc(0, b"") + + mock.patch.object(selfupdate.common, + "run_console_subprocess_quiet", + side_effect=quiet).start() + mock.patch.object(selfupdate.common, "origin_default_branch", + return_value="main").start() + return run + + @staticmethod + def _run_side_effect(fetch_rc, reset_rc, on_reset): + def run(argv, **kwargs): + if "fetch" in argv: + return fetch_rc + if "reset" in argv: + if on_reset is not None: + on_reset() + return reset_rc + raise AssertionError(f"unexpected subprocess call: {argv}") + return run + + # -- up to date / fetch failure: the tree must never be touched ------ + + def test_already_up_to_date_skips_the_reset(self): + run = self._patch_git(head=b"bbb\n", remote=b"bbb\n") + rc = selfupdate.update_generator(root=self.root, + emit=self.emitted.append) + self.assertEqual(rc, 0) + # Only the fetch ran — a no-op update cannot discard anything. + self.assertEqual(run.call_count, 1) + self.assertEqual(self.config.read_bytes(), b"# user settings\n") + self.assertTrue(any("already up to date" in line + for line in self.emitted)) + + def test_fetch_failure_short_circuits(self): + run = self._patch_git(fetch_rc=128) + rc = selfupdate.update_generator(root=self.root) + self.assertEqual(rc, 128) + self.assertEqual(run.call_count, 1) + self.assertEqual(self.config.read_bytes(), b"# user settings\n") + + def test_console_mode_prints_instead_of_emitting(self): + with mock.patch("builtins.print") as mk_print: + self._patch_git() + selfupdate.update_generator(root=self.root) + self.assertTrue(mk_print.called) + + # -- the reset path --------------------------------------------------- + + def test_updates_and_preserves_the_modified_file(self): + run = self._patch_git(on_reset=lambda: self.config.write_bytes( + b"# upstream settings\n")) + rc = selfupdate.update_generator(root=self.root, + emit=self.emitted.append) + self.assertEqual(rc, 0) + # Fetch first, then the reset to origin's default branch. + self.assertEqual( + run.call_args_list[1][0][0], + ["git", "-C", str(self.root), "reset", "--hard", + "origin/main"]) + # The user's config survived the reset verbatim... + self.assertEqual(self.config.read_bytes(), b"# user settings\n") + # ...and the skip is reported. + self.assertTrue(any("Kept your local app/converter/config.py" + in line for line in self.emitted)) + + def test_streaming_adds_progress_and_passes_emit(self): + run = self._patch_git() + selfupdate.update_generator(root=self.root, emit=self.emitted.append) + fetch_argv = run.call_args_list[0][0][0] + self.assertEqual(fetch_argv[:4], + ["git", "-C", str(self.root), "fetch"]) + self.assertIn("--progress", fetch_argv) + self.assertEqual(run.call_args_list[0][1]["emit"], + self.emitted.append) + self.assertEqual(run.call_args_list[1][1]["emit"], + self.emitted.append) + + def test_reset_failure_still_restores_the_snapshot(self): + self._patch_git(reset_rc=1, on_reset=lambda: self.config.write_bytes( + b"# upstream settings\n")) + rc = selfupdate.update_generator(root=self.root) + self.assertEqual(rc, 1) + self.assertEqual(self.config.read_bytes(), b"# user settings\n") + + def test_staged_added_file_is_restored_after_the_reset_removes_it(self): + added = self.root / "new_config.py" + added.write_bytes(b"# user additions\n") + def damage(): + added.unlink() + self._patch_git(status=b"A new_config.py\n", on_reset=damage) + rc = selfupdate.update_generator(root=self.root) + self.assertEqual(rc, 0) + self.assertEqual(added.read_bytes(), b"# user additions\n") + + def test_deleted_tracked_file_is_not_resurrected(self): + # A deletion is not a modification with content: the reset's + # restore of the file stands (it shows up in the confirm instead). + self._patch_git(status=b" D gone.py\n") + rc = selfupdate.update_generator(root=self.root) + self.assertEqual(rc, 0) + self.assertFalse((self.root / "gone.py").exists()) + + def test_identical_upstream_version_is_not_reported_as_kept(self): + # Upstream's new file content equals what the user already has: + # the restore is a no-op and says nothing. + self._patch_git(on_reset=lambda: self.config.write_bytes( + b"# user settings\n")) + rc = selfupdate.update_generator(root=self.root, + emit=self.emitted.append) + self.assertEqual(rc, 0) + self.assertFalse(any("Kept your local" in line + for line in self.emitted)) + + +if __name__ == "__main__": + unittest.main() |