"""Unit tests for the granthi-sync client: autocommit / ff / diverged logic, config handling, device-flow polling (mocked HTTP). Stdlib unittest only.""" import argparse import io import json import os import shlex import shutil import subprocess import sys import tempfile import time import unittest from unittest import mock sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "client")) # Point client config at a temp home BEFORE import side effects. _TMP_HOME = tempfile.mkdtemp(prefix="granthi-test-home-") os.environ["GRANTHI_SYNC_HOME"] = _TMP_HOME import granthi_sync_client as client # noqa: E402 GIT_ENV = { "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t", "HOME": _TMP_HOME, "PATH": os.environ["PATH"], } def run_git(cwd, *args): return subprocess.run(["git", "-C", cwd] + list(args), check=True, capture_output=True, text=True, env=GIT_ENV).stdout.strip() def get_ns(**kw): """Namespace for cmd_get with the parser's defaults filled in, so a test exercises the same shape argparse hands the command.""" kw.setdefault("all", False) kw.setdefault("mode", None) kw.setdefault("into", None) kw.setdefault("repo", None) return argparse.Namespace(**kw) class GitScenarioBase(unittest.TestCase): """bare 'cloud' repo + two working clones to simulate device vs remote.""" def setUp(self): self.tmp = tempfile.mkdtemp(prefix="granthi-test-") self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) self.bare = os.path.join(self.tmp, "cloud.git") subprocess.run(["git", "init", "--bare", "-b", "main", self.bare], check=True, capture_output=True, env=GIT_ENV) self.local = os.path.join(self.tmp, "local") os.makedirs(self.local) client.ensure_repo(self.local) run_git(self.local, "remote", "add", "granthi", self.bare) # git() in the client inherits our env via subprocess default; set # identity locally in the repo so commits work. run_git(self.local, "config", "user.name", "t") run_git(self.local, "config", "user.email", "t@t") def write(self, repo, name, content): with open(os.path.join(repo, name), "w") as f: f.write(content) def other_clone(self): other = os.path.join(self.tmp, "other") subprocess.run(["git", "clone", self.bare, other], check=True, capture_output=True, env=GIT_ENV) run_git(other, "config", "user.name", "o") run_git(other, "config", "user.email", "o@o") return other class TestAutocommit(GitScenarioBase): def test_autocommit_commits_changes(self): self.write(self.local, "a.txt", "one") self.assertTrue(client.autocommit(self.local)) msg = run_git(self.local, "log", "-1", "--format=%s") self.assertTrue(msg.startswith("sync: "), msg) def test_autocommit_noop_when_clean(self): self.write(self.local, "a.txt", "one") client.autocommit(self.local) self.assertFalse(client.autocommit(self.local)) class TestSyncFolder(GitScenarioBase): def test_initial_push(self): self.write(self.local, "a.txt", "one") outcome, _ = client.sync_folder(self.local) self.assertEqual(outcome, "pushed") self.assertIn("a.txt", run_git(self.local, "ls-tree", "--name-only", "granthi/main")) def test_ff_pull_when_remote_ahead(self): self.write(self.local, "a.txt", "one") client.sync_folder(self.local) other = self.other_clone() self.write(other, "b.txt", "from-other") run_git(other, "add", "-A") run_git(other, "commit", "-m", "remote change") run_git(other, "push", "origin", "main") outcome, detail = client.sync_folder(self.local) self.assertEqual((outcome, detail), ("synced", "ff-pulled")) self.assertTrue(os.path.exists(os.path.join(self.local, "b.txt"))) def test_diverged_is_skipped_never_forced(self): self.write(self.local, "a.txt", "one") client.sync_folder(self.local) other = self.other_clone() self.write(other, "b.txt", "remote side") run_git(other, "add", "-A") run_git(other, "commit", "-m", "remote change") run_git(other, "push", "origin", "main") remote_sha = run_git(other, "rev-parse", "HEAD") self.write(self.local, "a.txt", "local side") # divergence outcome, _ = client.sync_folder(self.local) self.assertEqual(outcome, "diverged") # remote must be untouched (not forced, not merged) bare_sha = run_git(self.bare, "rev-parse", "main") self.assertEqual(bare_sha, remote_sha) def test_clean_when_in_sync(self): self.write(self.local, "a.txt", "one") client.sync_folder(self.local) outcome, _ = client.sync_folder(self.local) self.assertEqual(outcome, "clean") class TestConfig(unittest.TestCase): def test_save_creates_0600(self): client.save_config({"login": "x", "folders": {}}) st = os.stat(client.CONFIG_PATH) self.assertEqual(st.st_mode & 0o777, 0o600) self.assertEqual(client.load_config()["login"], "x") def test_save_is_0600_even_with_permissive_umask(self): """Finding 4: the token file must be born 0600 (O_CREAT mode), not chmod'ed after write -- a wide-open umask must not widen it.""" old = os.umask(0o000) try: client.save_config({"token": "sekrit", "folders": {}}) finally: os.umask(old) st = os.stat(client.CONFIG_PATH) self.assertEqual(st.st_mode & 0o777, 0o600) def test_save_never_calls_chmod(self): """The 0600 mode must come from creation, not a later chmod (which would leave a window where the file is world-readable).""" with mock.patch.object(client.os, "chmod", side_effect=AssertionError( "chmod used; file must be created 0600")): client.save_config({"token": "sekrit", "folders": {}}) st = os.stat(client.CONFIG_PATH) self.assertEqual(st.st_mode & 0o777, 0o600) def test_load_missing_returns_empty(self): with mock.patch.object(client, "CONFIG_PATH", "/nonexistent/nope.json"): self.assertEqual(client.load_config(), {}) class TestCredentialHelperQuoting(unittest.TestCase): """Finding 5: helper command paths must be shlex-quoted.""" def test_paths_with_spaces_are_quoted(self): with mock.patch.object(client.sys, "executable", "/opt/py dir/bin/python3"), \ mock.patch.object(client, "__file__", "/home/a user/granthi sync/client.py"): val = client.credential_helper_value() self.assertTrue(val.startswith("!")) self.assertIn("'/opt/py dir/bin/python3'", val) self.assertIn("'/home/a user/granthi sync/client.py'", val) # shell round-trip yields exactly [python, script, subcommand] parts = shlex.split(val[1:]) self.assertEqual(parts, ["/opt/py dir/bin/python3", "/home/a user/granthi sync/client.py", "git-credential"]) def test_metacharacters_do_not_inject(self): evil = "/tmp/x; rm -rf ~; echo/client.py" with mock.patch.object(client, "__file__", evil): val = client.credential_helper_value() parts = shlex.split(val[1:]) self.assertEqual(parts[1], os.path.abspath(evil)) self.assertEqual(len(parts), 3) def test_plain_paths_still_work(self): val = client.credential_helper_value() parts = shlex.split(val[1:]) self.assertEqual(parts[0], sys.executable) self.assertEqual(parts[2], "git-credential") class TestDeviceFlow(unittest.TestCase): def test_device_flow_polls_until_token(self): calls = [] def fake_http(method, url, headers=None, body=None, form=None, timeout=30): calls.append(url) if url.endswith("/device_authorization"): return 200, {"device_code": "dc", "user_code": "AB-CD", "verification_uri": "https://id/device", "verification_uri_complete": "https://id/device?u=AB-CD", "interval": 0, "expires_in": 300} if len([c for c in calls if c.endswith("/token")]) < 3: return 400, {"error": "authorization_pending"} return 200, {"access_token": "ZTOK"} with mock.patch.object(client, "http_json", fake_http), \ mock.patch.object(client.time, "sleep"): tok = client.device_flow() self.assertEqual(tok, "ZTOK") self.assertEqual(len([c for c in calls if c.endswith("/token")]), 3) def test_device_flow_slow_down_backs_off(self): state = {"n": 0} def fake_http(method, url, headers=None, body=None, form=None, timeout=30): if url.endswith("/device_authorization"): return 200, {"device_code": "dc", "user_code": "AB", "verification_uri": "u", "interval": 1, "expires_in": 300} state["n"] += 1 if state["n"] == 1: return 400, {"error": "slow_down"} return 200, {"access_token": "T"} sleeps = [] with mock.patch.object(client, "http_json", fake_http), \ mock.patch.object(client.time, "sleep", sleeps.append): self.assertEqual(client.device_flow(), "T") self.assertIn(6, sleeps) # 1 + 5 backoff after slow_down class TestCredentialHelper(GitScenarioBase): def test_helper_emits_creds_for_matching_host(self): client.save_config({"gitea_base": "http://100.111.127.127:3041", "login": "alice", "token": "sekrit", "folders": {}}) stdin = "protocol=http\nhost=100.111.127.127:3041\n\n" out = subprocess.run( [sys.executable, client.__file__, "git-credential", "get"], input=stdin, capture_output=True, text=True, env={**GIT_ENV, "GRANTHI_SYNC_HOME": _TMP_HOME}) self.assertIn("username=alice", out.stdout) self.assertIn("password=sekrit", out.stdout) def test_helper_silent_for_other_host(self): client.save_config({"gitea_base": "http://100.111.127.127:3041", "login": "alice", "token": "sekrit", "folders": {}}) out = subprocess.run( [sys.executable, client.__file__, "git-credential", "get"], input="protocol=https\nhost=github.com\n\n", capture_output=True, text=True, env={**GIT_ENV, "GRANTHI_SYNC_HOME": _TMP_HOME}) self.assertNotIn("password=", out.stdout) class TestListRepos(unittest.TestCase): """Pagination must be followed, and a bounded page must never be presented as the complete set.""" def setUp(self): self.cfg = {"gitea_base": "http://100.111.127.127:3041", "login": "alice", "token": "sekrit", "folders": {}} @staticmethod def _page(n, count): return [{"name": f"r{n}-{i}", "full_name": f"alice/r{n}-{i}", "private": True, "updated_at": "2026-08-20T00:00:00Z"} for i in range(count)] def test_follows_pagination_until_short_page(self): calls = [] def fake_http(method, url, headers=None, **kw): calls.append((method, url, headers)) page = int(url.split("page=")[1].split("&")[0]) # two full pages, then a short one ends the walk return 200, self._page(page, 50 if page <= 2 else 7) with mock.patch.object(client, "http_json", fake_http): repos, truncated = client.list_repos(self.cfg) self.assertEqual(len(repos), 107) self.assertFalse(truncated) self.assertEqual(len(calls), 3) self.assertEqual(calls[0][2]["Authorization"], "token sekrit") self.assertIn("/api/v1/user/repos", calls[0][1]) def test_truncation_is_reported_not_hidden(self): with mock.patch.object(client, "http_json", lambda *a, **k: (200, self._page(1, 50))): repos, truncated = client.list_repos(self.cfg) self.assertTrue(truncated) self.assertEqual(len(repos), client.FORGE_MAX_PAGES * 50) def test_exact_multiple_of_page_size_is_not_truncated(self): """Every page full up to the cap does not imply more exist -- a total that is an exact multiple ends on a full page. The sentinel fetch past the cap is what tells the two apart.""" def fake_http(method, url, headers=None, **kw): page = int(url.split("page=")[1].split("&")[0]) return 200, self._page(page, 0 if page > client.FORGE_MAX_PAGES else 50) with mock.patch.object(client, "http_json", fake_http): repos, truncated = client.list_repos(self.cfg) self.assertFalse(truncated) self.assertEqual(len(repos), client.FORGE_MAX_PAGES * 50) def test_http_error_is_fatal_not_silent_empty(self): with mock.patch.object(client, "http_json", lambda *a, **k: (401, {"error": "bad token"})): with self.assertRaises(SystemExit): client.list_repos(self.cfg) def test_list_unlinked_exits_like_add(self): with mock.patch.object(client, "load_config", lambda: {}): with self.assertRaises(SystemExit) as cm: client.cmd_list(argparse.Namespace()) self.assertIn("not linked", str(cm.exception)) class TestGet(GitScenarioBase): """`get` is only useful if `watch` subsequently picks the folder up.""" def setUp(self): super().setUp() # a bare repo standing in for the forge, at /alice/cloud.git self.forge = os.path.join(self.tmp, "forge") self.remote_path = os.path.join(self.forge, "alice", "cloud.git") os.makedirs(os.path.dirname(self.remote_path)) subprocess.run(["git", "init", "--bare", "-b", "main", self.remote_path], check=True, capture_output=True, env=GIT_ENV) seed = os.path.join(self.tmp, "seed") subprocess.run(["git", "clone", self.remote_path, seed], check=True, capture_output=True, env=GIT_ENV) run_git(seed, "config", "user.name", "s") run_git(seed, "config", "user.email", "s@s") self.write(seed, "hello.txt", "from the forge") run_git(seed, "add", "-A") run_git(seed, "commit", "-m", "seed") run_git(seed, "push", "origin", "main") client.save_config({"gitea_base": self.forge, "login": "alice", "token": "sekrit", "folders": {}}) def test_get_clones_registers_and_is_watchable(self): dest = os.path.join(self.tmp, "pulled") client.cmd_get(get_ns(repo="cloud", into=dest)) # cloned content self.assertTrue(os.path.exists(os.path.join(dest, "hello.txt"))) # remote is 'granthi', which is the name sync_folder/watch use self.assertEqual(run_git(dest, "remote"), "granthi") # credential helper persisted into the new repo self.assertIn("git-credential", run_git(dest, "config", "credential.helper")) # registered with the same shape `add` writes meta = client.load_config()["folders"][os.path.abspath(dest)] self.assertEqual(meta["name"], "cloud") self.assertEqual(meta["branch"], "main") self.assertFalse(meta["diverged"]) self.assertIn("last_sync", meta) # the real proof: a watch pass sees it and reports it in-sync # rather than skipping it as unknown. run_git(dest, "config", "user.name", "t") run_git(dest, "config", "user.email", "t@t") outcome, _ = client.sync_folder(dest, branch=meta["branch"]) self.assertEqual(outcome, "clean") def test_get_accepts_owner_qualified_name(self): dest = os.path.join(self.tmp, "pulled2") client.cmd_get(get_ns(repo="alice/cloud", into=dest)) self.assertTrue(os.path.exists(os.path.join(dest, "hello.txt"))) def test_get_refuses_non_empty_destination(self): dest = os.path.join(self.tmp, "occupied") os.makedirs(dest) self.write(dest, "mine.txt", "do not clobber") with self.assertRaises(SystemExit): client.cmd_get(get_ns(repo="cloud", into=dest)) self.assertEqual(open(os.path.join(dest, "mine.txt")).read(), "do not clobber") def test_get_unlinked_exits_like_add(self): with mock.patch.object(client, "load_config", lambda: {}): with self.assertRaises(SystemExit) as cm: client.cmd_get(get_ns(repo="cloud", into=None)) self.assertIn("not linked", str(cm.exception)) def test_get_empty_repo_falls_back_to_main(self): empty = os.path.join(self.forge, "alice", "blank.git") subprocess.run(["git", "init", "--bare", "-b", "main", empty], check=True, capture_output=True, env=GIT_ENV) dest = os.path.join(self.tmp, "blank") client.cmd_get(get_ns(repo="blank", into=dest)) meta = client.load_config()["folders"][os.path.abspath(dest)] self.assertEqual(meta["branch"], "main") def test_get_rejects_hostile_repo_arguments(self): for bad in ["../../etc/passwd", "alice/cloud?x=1", "alice/cloud#frag", "a/b/c", "..", "-flag", "alice/../bob", "cloud%2f..", "", "alice/"]: with self.subTest(repo=bad): with self.assertRaises(SystemExit): client.cmd_get(get_ns(repo=bad, into=None)) def test_get_records_full_name_so_list_matches_the_right_owner(self): dest = os.path.join(self.tmp, "pulled4") client.cmd_get(get_ns(repo="cloud", into=dest)) meta = client.load_config()["folders"][os.path.abspath(dest)] self.assertEqual(meta["full_name"], "alice/cloud") def test_list_does_not_mark_a_same_named_other_owner_repo_as_local(self): dest = os.path.join(self.tmp, "pulled5") client.cmd_get(get_ns(repo="cloud", into=dest)) cfg = client.load_config() repos = [{"name": "cloud", "full_name": "alice/cloud", "private": True, "updated_at": "2026-08-20T00:00:00Z"}, {"name": "cloud", "full_name": "bob/cloud", "private": False, "updated_at": "2026-08-20T00:00:00Z"}] with mock.patch.object(client, "list_repos", lambda c: (repos, False)), \ mock.patch("sys.stdout", new_callable=io.StringIO) as out: client.cmd_list(argparse.Namespace()) lines = {l.split()[0]: l for l in out.getvalue().splitlines() if l.strip()} self.assertIn(os.path.abspath(dest), lines["alice/cloud"]) self.assertNotIn(os.path.abspath(dest), lines["bob/cloud"]) def test_get_never_puts_token_in_remote_url(self): dest = os.path.join(self.tmp, "pulled3") client.cmd_get(get_ns(repo="cloud", into=dest)) self.assertNotIn("sekrit", run_git(dest, "remote", "get-url", "granthi")) class TestSnapshots(GitScenarioBase): """The whole point of snapshot mode: work that was never committed still leaves the machine, and the user's own history is not touched.""" DEV = "dev0123456789" def test_snapshot_captures_uncommitted_work_without_moving_head(self): self.write(self.local, "a.txt", "committed") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "real commit") head_before = run_git(self.local, "rev-parse", "HEAD") self.write(self.local, "a.txt", "UNCOMMITTED EDIT") self.write(self.local, "new.txt", "never staged") status_before = run_git(self.local, "status", "--porcelain") # (run_git strips, so the leading space of ' M' is gone here) self.assertEqual(status_before, "M a.txt\n?? new.txt") commit, tree = client.build_snapshot(self.local) self.assertEqual(run_git(self.local, "rev-parse", "HEAD"), head_before) # the index is untouched: a.txt is still merely modified, not staged, # and new.txt is still untracked. A snapshot that quietly staged the # user's files would corrupt whatever they were in the middle of. self.assertEqual(run_git(self.local, "status", "--porcelain"), status_before) # working tree still holds exactly what the user left there with open(os.path.join(self.local, "a.txt")) as f: self.assertEqual(f.read(), "UNCOMMITTED EDIT") # ...and the snapshot commit holds it too blob = run_git(self.local, "show", f"{commit}:a.txt") self.assertEqual(blob, "UNCOMMITTED EDIT") self.assertIn("new.txt", run_git(self.local, "ls-tree", "--name-only", tree)) self.assertEqual(run_git(self.local, "log", "-1", "--format=%P", commit), head_before) def test_snapshot_is_none_when_nothing_is_uncommitted(self): self.write(self.local, "a.txt", "one") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "c") self.assertIsNone(client.build_snapshot(self.local)) def test_push_snapshot_lands_in_the_backup_namespace(self): self.write(self.local, "a.txt", "one") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "c") run_git(self.local, "push", "-u", "granthi", "main") self.write(self.local, "a.txt", "work in progress") ref = client.push_snapshot(self.local, self.DEV) self.assertTrue(ref.startswith(f"refs/granthi-backup/{self.DEV}/"), ref) refs = run_git(self.bare, "for-each-ref", "--format=%(refname)") self.assertIn(ref, refs.splitlines()) # it is NOT a branch: the user's branch list stays theirs self.assertNotIn("refs/heads/granthi-backup", refs) def test_push_snapshot_skips_an_unchanged_tree(self): self.write(self.local, "a.txt", "one") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "c") run_git(self.local, "push", "-u", "granthi", "main") self.write(self.local, "a.txt", "work in progress") self.assertIsNotNone(client.push_snapshot(self.local, self.DEV)) self.assertIsNone(client.push_snapshot(self.local, self.DEV)) def test_snapshot_mode_never_commits_for_the_user(self): self.write(self.local, "a.txt", "one") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "mine") run_git(self.local, "push", "-u", "granthi", "main") head_before = run_git(self.local, "rev-parse", "HEAD") self.write(self.local, "a.txt", "dirty") outcome, detail = client.sync_folder(self.local, mode="snapshot", dev=self.DEV) self.assertEqual(run_git(self.local, "rev-parse", "HEAD"), head_before) self.assertIn("backed up", detail) self.assertEqual(outcome, "clean") self.assertTrue(run_git(self.local, "status", "--porcelain")) def test_diverged_folder_is_still_backed_up(self): """Divergence is when work is most at risk -- the least acceptable moment to skip the backup.""" self.write(self.local, "a.txt", "one") client.sync_folder(self.local) # mirror push to establish the branch other = self.other_clone() self.write(other, "b.txt", "remote side") run_git(other, "add", "-A") run_git(other, "commit", "-m", "remote") run_git(other, "push", "origin", "main") remote_sha = run_git(other, "rev-parse", "HEAD") self.write(self.local, "a.txt", "local side") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "local") self.write(self.local, "c.txt", "and uncommitted too") outcome, detail = client.sync_folder(self.local, mode="snapshot", dev=self.DEV) self.assertEqual(outcome, "diverged") self.assertIn("backed up", detail) self.assertEqual(run_git(self.bare, "rev-parse", "main"), remote_sha) snaps = client.list_snapshots(self.local, self.DEV) self.assertEqual(len(snaps), 1) self.assertIn("c.txt", run_git(self.local, "ls-tree", "--name-only", snaps[0]["sha"])) def test_dirty_tree_blocks_the_pull_but_not_the_backup(self): self.write(self.local, "a.txt", "one") client.sync_folder(self.local) other = self.other_clone() self.write(other, "b.txt", "remote side") run_git(other, "add", "-A") run_git(other, "commit", "-m", "remote") run_git(other, "push", "origin", "main") self.write(self.local, "wip.txt", "half-finished") head_before = run_git(self.local, "rev-parse", "HEAD") outcome, detail = client.sync_folder(self.local, mode="snapshot", dev=self.DEV) self.assertEqual(run_git(self.local, "rev-parse", "HEAD"), head_before) self.assertFalse(os.path.exists(os.path.join(self.local, "b.txt"))) self.assertIn("not pulling", detail) self.assertIn("backed up", detail) def test_snapshots_are_listed_from_the_remote(self): self.write(self.local, "a.txt", "one") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "c") run_git(self.local, "push", "-u", "granthi", "main") self.write(self.local, "a.txt", "wip") ref = client.push_snapshot(self.local, self.DEV) snaps = client.list_snapshots(self.local, self.DEV) self.assertEqual([s["ref"] for s in snaps], [ref]) # writing is device-scoped: the ref carries THIS device's id, so two # machines cannot overwrite each other self.assertEqual(snaps[0]["device"], self.DEV) self.assertEqual(client.list_snapshots(self.local, "someone-else"), []) def test_a_new_machine_can_see_the_dead_machine_s_backups(self): """The case the whole feature exists for. Reads must NOT be scoped to this device's id -- a replacement laptop has a new id, and scoping would show an empty list while the backups sit on the forge.""" self.write(self.local, "a.txt", "committed") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "c") run_git(self.local, "push", "-u", "granthi", "main") self.write(self.local, "a.txt", "PRECIOUS UNCOMMITTED WORK") ref = client.push_snapshot(self.local, "laptop-that-died") # a fresh clone standing in for the replacement machine newbox = os.path.join(self.tmp, "newbox") subprocess.run(["git", "clone", "-q", "--origin", "granthi", self.bare, newbox], check=True, capture_output=True, env=GIT_ENV) client.save_config({"gitea_base": self.tmp, "login": "alice", "token": "t", "device_id": "brand-new-laptop", "folders": {newbox: {"name": "cloud", "full_name": "alice/cloud", "branch": "main", "mode": "snapshot"}}}) seen = client.list_snapshots(newbox) self.assertEqual([s["ref"] for s in seen], [ref]) self.assertEqual(seen[0]["device"], "laptop-that-died") # and it can actually restore it dest = os.path.join(self.tmp, "recovered") client.cmd_restore(argparse.Namespace(folder=newbox, at=seen[0]["ts"], into=dest)) with open(os.path.join(dest, "a.txt")) as f: self.assertEqual(f.read(), "PRECIOUS UNCOMMITTED WORK") class TestRetention(unittest.TestCase): """Retention is what keeps 30-second backups from being a disk leak -- and what must never quietly eat the one restore point someone needs.""" NOW = client.datetime(2026, 8, 23, 12, 0, 0, tzinfo=client.timezone.utc) def snap(self, when): ts = when.strftime(client.SNAPSHOT_TS_FMT) return {"ts": ts, "ref": f"refs/granthi-backup/d/{ts}", "sha": "x"} def test_everything_recent_is_kept(self): snaps = [self.snap(self.NOW - client.timedelta(minutes=m)) for m in range(0, 24 * 60, 30)] self.assertEqual(client.snapshots_to_prune(snaps, now=self.NOW), []) def test_older_than_a_day_thins_to_hourly(self): base = self.NOW - client.timedelta(days=2) snaps = [self.snap(base + client.timedelta(minutes=m)) for m in (0, 10, 20, 60, 70)] pruned = client.snapshots_to_prune(snaps, now=self.NOW) self.assertEqual(len(pruned), 3) # 5 in 2 hourly buckets -> keep 2 def test_older_than_a_week_thins_to_daily(self): base = self.NOW - client.timedelta(days=30) snaps = [self.snap(base + client.timedelta(hours=h)) for h in (0, 1, 2, 25)] pruned = client.snapshots_to_prune(snaps, now=self.NOW) self.assertEqual(len(pruned), 2) # 2 days -> keep 1 each def test_unparseable_timestamps_are_kept_not_deleted(self): snaps = [{"ts": "not-a-timestamp", "ref": "refs/granthi-backup/d/not-a-timestamp", "sha": "x"}] self.assertEqual(client.snapshots_to_prune(snaps, now=self.NOW), []) class TestPrune(GitScenarioBase): DEV = "devprune" def test_prune_deletes_stale_refs_on_the_remote(self): self.write(self.local, "a.txt", "one") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "c") run_git(self.local, "push", "-u", "granthi", "main") sha = run_git(self.local, "rev-parse", "HEAD") old = "20260101T000000Z" older = "20260101T001000Z" for ts in (old, older): run_git(self.local, "push", "granthi", f"{sha}:refs/granthi-backup/{self.DEV}/{ts}") self.assertEqual(len(client.list_snapshots(self.local, self.DEV)), 2) gone = client.prune_snapshots(self.local, self.DEV) self.assertEqual(gone, 1) # same hour, older one dropped left = client.list_snapshots(self.local, self.DEV) self.assertEqual([s["ts"] for s in left], [older]) class TestAddGuards(GitScenarioBase): def test_gitignore_is_seeded_only_when_absent(self): self.assertTrue(client.seed_gitignore(self.local)) with open(os.path.join(self.local, ".gitignore")) as f: body = f.read() self.assertIn(".env", body) with open(os.path.join(self.local, ".gitignore"), "w") as f: f.write("mine-only\n") self.assertFalse(client.seed_gitignore(self.local)) with open(os.path.join(self.local, ".gitignore")) as f: self.assertEqual(f.read(), "mine-only\n") def test_seeded_gitignore_keeps_secrets_out_of_snapshots(self): client.seed_gitignore(self.local) self.write(self.local, ".env", "SECRET=hunter2") self.write(self.local, "ok.txt", "fine") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "c") self.write(self.local, "ok.txt", "changed") commit, tree = client.build_snapshot(self.local) names = run_git(self.local, "ls-tree", "-r", "--name-only", tree) self.assertIn("ok.txt", names) self.assertNotIn(".env", names.splitlines()) def test_measure_folder_stops_counting_past_the_cap(self): for i in range(12): self.write(self.local, f"f{i}.txt", "x" * 10) files, size = client.measure_folder(self.local, max_files=5) self.assertEqual(files, 6) # bounded: stopped one past the cap self.assertLess(size, 12 * 10) def test_measure_folder_ignores_dot_git(self): files, _ = client.measure_folder(self.local) self.assertEqual(files, 0) def test_detect_mode(self): plain = os.path.join(self.tmp, "plain") os.makedirs(plain) self.assertEqual(client.detect_mode(plain, had_git=False), "mirror") self.write(self.local, "a.txt", "one") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "real history") self.assertEqual(client.detect_mode(self.local, had_git=True), "snapshot") empty = os.path.join(self.tmp, "empty-repo") os.makedirs(empty) client.ensure_repo(empty) self.assertEqual(client.detect_mode(empty, had_git=True), "mirror") class TestMatchRepo(unittest.TestCase): def repo(self, full): return {"full_name": full, "name": full.split("/")[-1]} def test_substring_is_case_insensitive_and_matches_bare_name(self): self.assertTrue(client.match_repo(self.repo("alice/Notes"), "notes")) self.assertTrue(client.match_repo(self.repo("alice/notes"), "ALICE")) self.assertFalse(client.match_repo(self.repo("alice/notes"), "ledger")) def test_glob_syntax_switches_to_glob(self): self.assertTrue(client.match_repo(self.repo("alice/work-2026"), "work-*")) self.assertFalse(client.match_repo(self.repo("alice/homework"), "work-*")) def test_empty_pattern_matches_everything(self): self.assertTrue(client.match_repo(self.repo("alice/x"), None)) class TestRestore(GitScenarioBase): DEV = "devrestore" def setUp(self): super().setUp() client.save_config({"gitea_base": self.tmp, "login": "alice", "token": "sekrit", "device_id": self.DEV, "folders": {self.local: { "name": "cloud", "full_name": "alice/cloud", "branch": "main", "mode": "snapshot"}}}) def test_restore_writes_a_new_folder_and_leaves_the_working_tree_alone(self): self.write(self.local, "a.txt", "original") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "c") run_git(self.local, "push", "-u", "granthi", "main") self.write(self.local, "a.txt", "the version I want back") ref = client.push_snapshot(self.local, self.DEV) ts = ref.rsplit("/", 1)[-1] self.write(self.local, "a.txt", "what I have now") dest = os.path.join(self.tmp, "restored") client.cmd_restore(argparse.Namespace(folder=self.local, at=ts, into=dest)) with open(os.path.join(dest, "a.txt")) as f: self.assertEqual(f.read(), "the version I want back") with open(os.path.join(self.local, "a.txt")) as f: self.assertEqual(f.read(), "what I have now") def test_restore_refuses_a_non_empty_destination(self): self.write(self.local, "a.txt", "one") run_git(self.local, "add", "-A") run_git(self.local, "commit", "-m", "c") run_git(self.local, "push", "-u", "granthi", "main") self.write(self.local, "a.txt", "two") ref = client.push_snapshot(self.local, self.DEV) busy = os.path.join(self.tmp, "busy") os.makedirs(busy) with open(os.path.join(busy, "keepme"), "w") as f: f.write("do not clobber") with self.assertRaises(SystemExit): client.cmd_restore(argparse.Namespace( folder=self.local, at=ref.rsplit("/", 1)[-1], into=busy)) self.assertTrue(os.path.exists(os.path.join(busy, "keepme"))) def test_unknown_restore_point_is_an_error_not_an_empty_folder(self): with self.assertRaises(SystemExit): client.cmd_restore(argparse.Namespace( folder=self.local, at="20990101T000000Z", into=None)) def test_restore_refuses_a_folder_that_is_not_linked(self): with self.assertRaises(SystemExit): client.cmd_restore(argparse.Namespace( folder=os.path.join(self.tmp, "nowhere"), at="x", into=None)) class TestDeviceId(unittest.TestCase): def test_device_id_is_stable_and_persisted(self): cfg = {} first = client.device_id(cfg) self.assertEqual(client.device_id(cfg), first) self.assertEqual(cfg["device_id"], first) def test_two_installs_get_different_ids(self): self.assertNotEqual(client.device_id({}), client.device_id({})) class TestGetAll(GitScenarioBase): """--all must pull exactly what the forge grants, and one bad repo must not abandon the rest.""" def setUp(self): super().setUp() self.forge = os.path.join(self.tmp, "forge") for full in ("alice/one", "alice/two"): path = os.path.join(self.forge, full + ".git") os.makedirs(os.path.dirname(path), exist_ok=True) subprocess.run(["git", "init", "--bare", "-b", "main", path], check=True, capture_output=True, env=GIT_ENV) seed = os.path.join(self.tmp, "seed-" + full.replace("/", "-")) subprocess.run(["git", "clone", path, seed], check=True, capture_output=True, env=GIT_ENV) run_git(seed, "config", "user.name", "s") run_git(seed, "config", "user.email", "s@s") self.write(seed, "f.txt", full) run_git(seed, "add", "-A") run_git(seed, "commit", "-m", "seed") run_git(seed, "push", "origin", "main") client.save_config({"gitea_base": self.forge, "login": "alice", "token": "sekrit", "folders": {}}) self.repos = [{"name": "one", "full_name": "alice/one"}, {"name": "two", "full_name": "alice/two"}] def test_all_clones_every_granted_repo(self): into = os.path.join(self.tmp, "workspace") os.makedirs(into) with mock.patch.object(client, "list_repos", lambda c: (self.repos, False)): rc = client.cmd_get(get_ns(all=True, into=into)) self.assertEqual(rc, 0) for name in ("one", "two"): self.assertTrue(os.path.exists(os.path.join(into, name, "f.txt"))) self.assertEqual(len(client.load_config()["folders"]), 2) def test_all_skips_what_is_already_here(self): into = os.path.join(self.tmp, "workspace2") os.makedirs(into) with mock.patch.object(client, "list_repos", lambda c: (self.repos, False)): client.cmd_get(get_ns(all=True, into=into)) with mock.patch("sys.stdout", new_callable=io.StringIO) as out: client.cmd_get(get_ns(all=True, into=into)) self.assertIn("already present", out.getvalue()) self.assertEqual(len(client.load_config()["folders"]), 2) def test_all_defaults_cloned_repos_to_snapshot_mode(self): into = os.path.join(self.tmp, "workspace3") os.makedirs(into) with mock.patch.object(client, "list_repos", lambda c: (self.repos, False)): client.cmd_get(get_ns(all=True, into=into)) modes = {m["mode"] for m in client.load_config()["folders"].values()} self.assertEqual(modes, {"snapshot"}) def test_all_says_so_loudly_when_the_listing_was_truncated(self): into = os.path.join(self.tmp, "workspace4") os.makedirs(into) with mock.patch.object(client, "list_repos", lambda c: (self.repos, True)), \ mock.patch("sys.stdout", new_callable=io.StringIO) as out: client.cmd_get(get_ns(all=True, into=into)) self.assertIn("NOT every repo", out.getvalue()) class TestMarkerRoundTrip(GitScenarioBase): """A plain folder synced on machine A must behave the same on machine B: the intent travels in the repo, not in one machine's config.""" def test_marker_written_by_add_makes_get_choose_mirror(self): client.write_marker(self.local, "mirror") self.assertEqual(client.read_marker(self.local)["mode"], "mirror") def test_missing_or_corrupt_marker_falls_back_to_the_safe_mode(self): self.assertEqual(client.read_marker(self.local), {}) with open(os.path.join(self.local, client.MARKER_FILE), "w") as f: f.write("{not json") self.assertEqual(client.read_marker(self.local), {}) class TestCredentialHelperIsolation(GitScenarioBase): """A repo-local helper is not enough on a normal machine: git consults system + global helpers too, and they both shadow us and copy the token into plaintext. Found by live QA against the beta forge, not by a unit test -- so it gets one now.""" def test_install_leaves_exactly_one_helper(self): run_git(self.local, "config", "--add", "credential.helper", "store") client.install_credential_helper(self.local) # --get-all merges system + global + local, so entries inherited from # the machine still appear. What matters is that the last two are the # reset and ours: git reads an empty value as "forget every helper # inherited so far", so nothing before it can answer. helpers = run_git(self.local, "config", "--get-all", "credential.helper").splitlines() self.assertEqual(helpers[-2], "", helpers) self.assertIn("git-credential", helpers[-1]) # the repo-level 'store' this test added is gone, not merely outvoted self.assertNotIn("store", helpers) def test_inherited_helper_cannot_answer_for_the_forge(self): """The end-to-end property: with a poisoned outer helper configured, the credential git actually resolves is ours.""" fake = os.path.join(self.tmp, "poison.sh") with open(fake, "w") as f: f.write("#!/bin/sh\n" "echo username=wrong-user\necho password=stale-token\n") os.chmod(fake, 0o755) run_git(self.local, "config", "--add", "credential.helper", f"!{shlex.quote(fake)}") client.save_config({"gitea_base": "http://forge.example:3041", "login": "alice", "token": "the-right-token"}) client.install_credential_helper(self.local) out = subprocess.run( ["git", "-C", self.local, "credential", "fill"], input="protocol=http\nhost=forge.example:3041\n\n", capture_output=True, text=True, env=dict( GIT_ENV, GRANTHI_SYNC_HOME=os.environ["GRANTHI_SYNC_HOME"])) self.assertIn("password=the-right-token", out.stdout) self.assertNotIn("stale-token", out.stdout) class TestMirrorRestore(GitScenarioBase): """Mirror is the default for a plain folder, so its restore path is the one most people will use -- and the timestamp `snapshots` prints has to be a timestamp `restore` accepts, or the user just loops.""" def setUp(self): super().setUp() client.save_config({"gitea_base": self.tmp, "login": "alice", "token": "t", "device_id": "d1", "folders": {self.local: { "name": "cloud", "full_name": "alice/cloud", "branch": "main", "mode": "mirror"}}}) def test_restore_accepts_the_timestamp_snapshots_printed(self): self.write(self.local, "a.txt", "the version I want back") client.sync_folder(self.local, mode="mirror") time.sleep(1.1) # %cI has one-second resolution self.write(self.local, "a.txt", "later junk") client.sync_folder(self.local, mode="mirror") with mock.patch("sys.stdout", new_callable=io.StringIO) as out: client.cmd_snapshots(argparse.Namespace(folder=self.local, limit=20)) listed = [l.split() for l in out.getvalue().splitlines() if l.startswith(" ")] wanted_ts = listed[-1][0] # first column of the oldest entry dest = os.path.join(self.tmp, "mirror-restore") client.cmd_restore(argparse.Namespace(folder=self.local, at=wanted_ts, into=dest)) with open(os.path.join(dest, "a.txt")) as f: self.assertEqual(f.read(), "the version I want back") def test_two_commits_in_the_same_second_are_refused_not_guessed(self): """%cI has one-second resolution. Picking one silently would restore something the user did not choose.""" self.write(self.local, "a.txt", "first") client.sync_folder(self.local, mode="mirror") self.write(self.local, "a.txt", "second") client.sync_folder(self.local, mode="mirror") stamps = run_git(self.local, "log", "--format=%cI").splitlines() if len(set(stamps)) != 1: self.skipTest("commits did not land in the same second") with self.assertRaises(SystemExit) as caught: client.cmd_restore(argparse.Namespace(folder=self.local, at=stamps[0], into=None)) self.assertIn("matches 2 restore points", str(caught.exception)) def test_restore_still_accepts_a_sha(self): self.write(self.local, "a.txt", "one") client.sync_folder(self.local, mode="mirror") sha = run_git(self.local, "rev-parse", "HEAD") dest = os.path.join(self.tmp, "by-sha") client.cmd_restore(argparse.Namespace(folder=self.local, at=sha, into=dest)) self.assertTrue(os.path.exists(os.path.join(dest, "a.txt"))) def test_default_destination_is_a_usable_path(self): self.write(self.local, "a.txt", "one") client.sync_folder(self.local, mode="mirror") with mock.patch("sys.stdout", new_callable=io.StringIO) as out: client.cmd_snapshots(argparse.Namespace(folder=self.local, limit=5)) ts = [l.split() for l in out.getvalue().splitlines() if l.startswith(" ")][0][0] client.cmd_restore(argparse.Namespace(folder=self.local, at=ts, into=None)) made = [d for d in os.listdir(self.tmp) if d.startswith("local-restore-")] self.assertEqual(len(made), 1, made) self.assertNotIn(":", made[0]) class TestMeasureHonoursGitignore(GitScenarioBase): """The guard tells people to add a .gitignore. That advice has to work.""" def test_ignored_files_are_not_counted(self): os.makedirs(os.path.join(self.local, "bulkdata")) for i in range(30): self.write(self.local, f"bulkdata/x{i}.bin", "y" * 100) self.write(self.local, "real.txt", "mine") before, _ = client.measure_folder(self.local) self.write(self.local, ".gitignore", "bulkdata/\n") after, _ = client.measure_folder(self.local) self.assertGreater(before, 30) self.assertEqual(after, 2) # real.txt + .gitignore def test_measure_leaves_no_git_dir_behind(self): plain = os.path.join(self.tmp, "untouched") os.makedirs(plain) self.write(plain, "a.txt", "x") client.measure_folder(plain) self.assertEqual(os.listdir(plain), ["a.txt"]) class TestGetAllRobustness(GitScenarioBase): def setUp(self): super().setUp() self.forge = os.path.join(self.tmp, "forge") for full in ("alice/notes", "bob/notes"): path = os.path.join(self.forge, full + ".git") os.makedirs(os.path.dirname(path), exist_ok=True) subprocess.run(["git", "init", "-q", "--bare", "-b", "main", path], check=True, capture_output=True, env=GIT_ENV) seed = os.path.join(self.tmp, "seed-" + full.replace("/", "-")) subprocess.run(["git", "clone", "-q", path, seed], check=True, capture_output=True, env=GIT_ENV) run_git(seed, "config", "user.name", "s") run_git(seed, "config", "user.email", "s@s") self.write(seed, "who.txt", full) run_git(seed, "add", "-A") run_git(seed, "commit", "-m", "seed") run_git(seed, "push", "-q", "origin", "main") client.save_config({"gitea_base": self.forge, "login": "alice", "token": "t", "folders": {}}) self.repos = [{"name": "notes", "full_name": "alice/notes"}, {"name": "notes", "full_name": "bob/notes"}] def test_same_named_repos_from_two_owners_both_land(self): into = os.path.join(self.tmp, "ws") os.makedirs(into) with mock.patch.object(client, "list_repos", lambda c: (self.repos, False)), \ mock.patch("sys.stdout", new_callable=io.StringIO) as out: client.cmd_get(get_ns(all=True, into=into)) self.assertIn("name clash", out.getvalue()) with open(os.path.join(into, "notes", "who.txt")) as f: self.assertEqual(f.read(), "alice/notes") with open(os.path.join(into, "bob-notes", "who.txt")) as f: self.assertEqual(f.read(), "bob/notes") self.assertEqual(len(client.load_config()["folders"]), 2) def test_a_git_failure_on_one_repo_does_not_abandon_the_rest(self): into = os.path.join(self.tmp, "ws2") os.makedirs(into) real_clone = client.clone_one def flaky(cfg, full_name, dest, mode=None): if full_name == "alice/notes": raise RuntimeError("git clone failed: pretend network blip") return real_clone(cfg, full_name, dest, mode) with mock.patch.object(client, "list_repos", lambda c: (self.repos, False)), \ mock.patch.object(client, "clone_one", flaky), \ mock.patch("sys.stdout", new_callable=io.StringIO) as out: rc = client.cmd_get(get_ns(all=True, into=into)) self.assertEqual(rc, 1) # reported, not hidden self.assertIn("FAILED alice/notes", out.getvalue()) self.assertTrue(os.path.exists(os.path.join(into, "notes", "who.txt"))) class TestPruneClockIsPersisted(GitScenarioBase): def test_watch_once_does_not_prune_every_run(self): client.save_config({"gitea_base": self.tmp, "login": "alice", "token": "t", "device_id": "d1", "folders": {}}) with mock.patch.object(client, "prune_snapshots") as pruner: client.watch_pass(now=1_000_000.0) self.assertEqual(client.load_config()["last_prune"], 1_000_000.0) client.watch_pass(now=1_000_060.0) # a minute later: not due client.watch_pass(now=1_003_700.0) # an hour later: due again self.assertEqual(client.load_config()["last_prune"], 1_003_700.0) self.assertEqual(pruner.call_count, 0) # no snapshot folders linked if __name__ == "__main__": unittest.main()