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