feat(client): snapshot backups, restore points, scoped bulk pull
Two modes per linked folder. 'mirror' keeps today's behaviour for a plain folder that add turned into a repo. 'snapshot' is new and is for a folder that already had a git history: nothing is ever committed on the user's behalf, and instead each pass builds a commit object from the working tree via a scratch index + commit-tree and pushes it to refs/granthi-backup/<device>/<ts>. HEAD, the index and every file stay exactly as the user left them, so uncommitted, unmerged, half-finished work leaves the machine with a timestamp to restore from. Verified on the beta forge (Gitea 1.27.2) that a custom ref namespace is accepted, readable via ls-remote, and absent from the branch list. Also: retention (all for 24h, hourly for 7d, daily beyond; unparseable timestamps kept), snapshots/restore commands, restore never writing over the working tree, get --all bounded by what the forge grants, list <pattern>, .gitignore seeding, an add size guard, and a persisted device_id. 141 tests (was 108).
This commit is contained in:
claude
+437
-9
@@ -33,6 +33,17 @@ def run_git(cwd, *args):
|
||||
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."""
|
||||
|
||||
@@ -341,7 +352,7 @@ class TestGet(GitScenarioBase):
|
||||
|
||||
def test_get_clones_registers_and_is_watchable(self):
|
||||
dest = os.path.join(self.tmp, "pulled")
|
||||
client.cmd_get(argparse.Namespace(repo="cloud", into=dest))
|
||||
client.cmd_get(get_ns(repo="cloud", into=dest))
|
||||
|
||||
# cloned content
|
||||
self.assertTrue(os.path.exists(os.path.join(dest, "hello.txt")))
|
||||
@@ -366,7 +377,7 @@ class TestGet(GitScenarioBase):
|
||||
|
||||
def test_get_accepts_owner_qualified_name(self):
|
||||
dest = os.path.join(self.tmp, "pulled2")
|
||||
client.cmd_get(argparse.Namespace(repo="alice/cloud", into=dest))
|
||||
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):
|
||||
@@ -374,14 +385,14 @@ class TestGet(GitScenarioBase):
|
||||
os.makedirs(dest)
|
||||
self.write(dest, "mine.txt", "do not clobber")
|
||||
with self.assertRaises(SystemExit):
|
||||
client.cmd_get(argparse.Namespace(repo="cloud", into=dest))
|
||||
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(argparse.Namespace(repo="cloud", into=None))
|
||||
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):
|
||||
@@ -389,7 +400,7 @@ class TestGet(GitScenarioBase):
|
||||
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(argparse.Namespace(repo="blank", into=dest))
|
||||
client.cmd_get(get_ns(repo="blank", into=dest))
|
||||
meta = client.load_config()["folders"][os.path.abspath(dest)]
|
||||
self.assertEqual(meta["branch"], "main")
|
||||
|
||||
@@ -399,17 +410,17 @@ class TestGet(GitScenarioBase):
|
||||
"", "alice/"]:
|
||||
with self.subTest(repo=bad):
|
||||
with self.assertRaises(SystemExit):
|
||||
client.cmd_get(argparse.Namespace(repo=bad, into=None))
|
||||
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(argparse.Namespace(repo="cloud", into=dest))
|
||||
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(argparse.Namespace(repo="cloud", into=dest))
|
||||
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"},
|
||||
@@ -426,9 +437,426 @@ class TestGet(GitScenarioBase):
|
||||
|
||||
def test_get_never_puts_token_in_remote_url(self):
|
||||
dest = os.path.join(self.tmp, "pulled3")
|
||||
client.cmd_get(argparse.Namespace(repo="cloud", into=dest))
|
||||
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])
|
||||
self.assertEqual(client.list_snapshots(self.local, "someone-else"), [])
|
||||
|
||||
|
||||
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), {})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user