granthi-sync/tests/test_client.py

"""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 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 <base>/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])
        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), {})


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)


if __name__ == "__main__":
    unittest.main()