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 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 <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])
        # 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



class TestDefaultServer(unittest.TestCase):
    """A new computer must be able to sign in without joining a private
    network first -- that is the whole point of exposing the endpoint."""

    def test_default_server_is_public_https(self):
        self.assertTrue(client.DEFAULT_SERVER.startswith("https://"),
                        client.DEFAULT_SERVER)
        self.assertNotIn("100.111.", client.DEFAULT_SERVER)

    def test_env_override_wins_for_internal_machines(self):
        import importlib
        with mock.patch.dict(os.environ,
                             {"GRANTHI_LINK_SERVER": "http://10.0.0.5:3042"}):
            reloaded = importlib.reload(client)
            self.assertEqual(reloaded.DEFAULT_SERVER, "http://10.0.0.5:3042")
        importlib.reload(client)  # restore for the rest of the suite


class TestWorkspaceBootstrap(GitScenarioBase):
    """A new computer set up from a manifest -- pulling only what the forge
    grants, and never installing an application behind someone's back."""

    def setUp(self):
        super().setUp()
        self.forge = os.path.join(self.tmp, "forge")
        for full in ("alice/notes", "alice/reports"):
            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, "f.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.ws = os.path.join(self.tmp, "ws")
        os.makedirs(self.ws)
        self.repos = [{"name": "notes", "full_name": "alice/notes"},
                      {"name": "reports", "full_name": "alice/reports"}]

    def manifest(self, obj):
        with open(os.path.join(self.ws, "workspace.json"), "w") as f:
            json.dump(obj, f)

    def ns(self, **kw):
        kw.setdefault("folder", self.ws)
        kw.setdefault("into", os.path.join(self.tmp, "out"))
        kw.setdefault("dry_run", False)
        return argparse.Namespace(**kw)

    def test_pulls_the_repos_the_manifest_names(self):
        self.manifest({"name": "acme",
                       "repos": ["notes", {"name": "reports", "mode": "mirror"}]})
        with mock.patch.object(client, "list_repos", lambda c: (self.repos, False)), \
             mock.patch("sys.stdout", new_callable=io.StringIO):
            client.cmd_bootstrap(self.ns())
        out = os.path.join(self.tmp, "out")
        self.assertTrue(os.path.exists(os.path.join(out, "notes", "f.txt")))
        self.assertTrue(os.path.exists(os.path.join(out, "reports", "f.txt")))
        modes = {m["name"]: m["mode"]
                 for m in client.load_config()["folders"].values()}
        self.assertEqual(modes["reports"], "mirror")   # manifest override
        self.assertEqual(modes["notes"], "snapshot")   # safe default

    def test_a_repo_the_account_cannot_see_is_reported_not_attempted(self):
        """The forge decides. A manifest naming someone else's repo is a
        permissions answer, not an error to route around."""
        self.manifest({"repos": ["notes", "someone-elses-secrets"]})
        with mock.patch.object(client, "list_repos", lambda c: (self.repos, False)), \
             mock.patch("sys.stdout", new_callable=io.StringIO) as out:
            client.cmd_bootstrap(self.ns())
        self.assertIn("NOT GRANTED", out.getvalue())
        self.assertIn("1 not granted", out.getvalue())
        self.assertFalse(os.path.exists(
            os.path.join(self.tmp, "out", "someone-elses-secrets")))

    def test_apps_are_described_never_installed(self):
        self.manifest({"repos": [], "apps": [
            {"id": "dash", "repo": "nirpa/hermes-agent", "model": "aum/70b",
             "setup": "docs/SETUP.md", "needs_keys": ["anthropic"]}]})
        with mock.patch.object(client, "list_repos", lambda c: (self.repos, False)), \
             mock.patch("sys.stdout", new_callable=io.StringIO) as out:
            client.cmd_bootstrap(self.ns())
        text = out.getvalue()
        self.assertIn("NOT installed by granthi-sync", text)
        self.assertIn("nirpa/hermes-agent", text)
        self.assertIn("aum/70b", text)
        self.assertIn("shre-cred request", text)   # how to supply the key
        # nothing was cloned or written for the app
        self.assertFalse(os.path.exists(os.path.join(self.tmp, "out", "dash")))

    def test_dry_run_changes_nothing(self):
        self.manifest({"repos": ["notes"]})
        with mock.patch.object(client, "list_repos", lambda c: (self.repos, False)), \
             mock.patch("sys.stdout", new_callable=io.StringIO) as out:
            client.cmd_bootstrap(self.ns(dry_run=True))
        self.assertIn("WOULD PULL", out.getvalue())
        self.assertFalse(os.path.exists(os.path.join(self.tmp, "out", "notes")))
        self.assertEqual(client.load_config()["folders"], {})

    def test_rerun_is_idempotent(self):
        self.manifest({"repos": ["notes"]})
        with mock.patch.object(client, "list_repos", lambda c: (self.repos, False)), \
             mock.patch("sys.stdout", new_callable=io.StringIO):
            client.cmd_bootstrap(self.ns())
        with mock.patch.object(client, "list_repos", lambda c: (self.repos, False)), \
             mock.patch("sys.stdout", new_callable=io.StringIO) as out:
            client.cmd_bootstrap(self.ns())
        self.assertIn("already here", out.getvalue())

    def test_a_broken_manifest_says_what_is_wrong(self):
        with open(os.path.join(self.ws, "workspace.json"), "w") as f:
            f.write("{not json")
        with self.assertRaises(SystemExit) as e:
            client.cmd_bootstrap(self.ns())
        self.assertIn("not valid JSON", str(e.exception))
        os.remove(os.path.join(self.ws, "workspace.json"))
        with self.assertRaises(SystemExit) as e:
            client.cmd_bootstrap(self.ns())
        self.assertIn("no workspace.json", str(e.exception))


class TestResolveGranted(unittest.TestCase):
    """`get <name>` has to find the repo the forge actually grants.

    Found by running the real flow against production: of 172 repos granted
    to this estate's own admin, 157 are owned by an ORG, so a bare name failed
    for 91% of what the user could see -- and the error read "Repository not
    found", which sounds like a permissions problem rather than a naming one.
    """

    def setUp(self):
        self.cfg = {"login": "alice", "gitea_base": "http://forge.example",
                    "token": "t"}
        self.repos = [{"full_name": "Nirlabinc/Ai-Assistant"},
                      {"full_name": "alice/notes"},
                      {"full_name": "Shreai/notes"}]

    def test_a_bare_name_finds_an_org_owned_repo(self):
        self.assertEqual(
            client.resolve_granted(self.cfg, "Ai-Assistant", self.repos),
            "Nirlabinc/Ai-Assistant")

    def test_an_owner_qualified_name_is_taken_as_given(self):
        self.assertEqual(
            client.resolve_granted(self.cfg, "Nirlabinc/Ai-Assistant",
                                   self.repos),
            "Nirlabinc/Ai-Assistant")

    def test_an_ambiguous_bare_name_is_refused_with_the_candidates(self):
        """Two teams with a repo called `notes` is not a guess worth making
        on someone's behalf."""
        with self.assertRaises(SystemExit) as e:
            client.resolve_granted(self.cfg, "notes", self.repos)
        msg = str(e.exception)
        self.assertIn("ambiguous", msg)
        self.assertIn("Shreai/notes", msg)
        self.assertIn("alice/notes", msg)

    def test_an_unknown_name_falls_back_to_your_own_namespace(self):
        self.assertEqual(
            client.resolve_granted(self.cfg, "brand-new", self.repos),
            "alice/brand-new")

    def test_an_unreachable_listing_does_not_block_a_clone(self):
        """A network blip must not stop someone cloning a repo they named."""
        def boom(cfg):
            raise SystemExit("listing repos failed (HTTP 502)")
        with mock.patch.object(client, "list_repos", boom), \
             mock.patch("sys.stdout", new_callable=io.StringIO):
            self.assertEqual(client.resolve_granted(self.cfg, "notes"),
                             "alice/notes")
            self.assertEqual(
                client.resolve_granted(self.cfg, "Nirlabinc/Ai-Assistant"),
                "Nirlabinc/Ai-Assistant")


class TestDeviceFlowOutputIsVisible(unittest.TestCase):
    """The code has a five-minute life. If it is sitting in a buffer, the user
    never sees it and it expires -- which is what happened on 2026-08-23 while
    driving a first real sign-in through a wrapper."""

    def test_the_url_and_code_are_flushed_immediately(self):
        seen = []
        real_print = print

        def spy(*a, **kw):
            seen.append(kw.get("flush", False))
            return real_print(*a, **{k: v for k, v in kw.items() if k != "flush"})

        resp = {"verification_uri_complete": "https://id.example/device?user_code=AB-CD",
                "user_code": "AB-CD", "device_code": "dc", "interval": 0,
                "expires_in": 0}
        with mock.patch.object(client, "http_json", lambda *a, **k: (200, resp)), \
             mock.patch("builtins.print", spy), \
             mock.patch.object(client.time, "sleep", lambda *_: None):
            with self.assertRaises(SystemExit):   # expires_in 0 -> times out
                client.device_flow()
        self.assertTrue(seen, "device_flow printed nothing")
        self.assertTrue(all(seen[:2]),
                        "the URL and code must be printed with flush=True")


class TestCodexReviewFindings(GitScenarioBase):
    """Regressions for the three issues codex found in today's merged work."""

    def test_staged_only_work_survives_a_snapshot(self):
        """[P2] Stage a hunk, edit further, lose the laptop: the staged version
        must still be recoverable, not just the later worktree one."""
        self.write(self.local, "a.txt", "committed")
        run_git(self.local, "add", "-A")
        run_git(self.local, "commit", "-m", "base")
        self.write(self.local, "a.txt", "THE CAREFULLY STAGED VERSION")
        run_git(self.local, "add", "a.txt")          # staged
        self.write(self.local, "a.txt", "later scratch edit")   # worktree moved on

        commit, tree = client.build_snapshot(self.local)

        # the worktree state is the snapshot's own tree
        self.assertEqual(run_git(self.local, "show", f"{commit}:a.txt"),
                         "later scratch edit")
        # ...and the staged state is reachable through the extra parent
        parents = run_git(self.local, "log", "-1", "--format=%P", commit).split()
        staged = [p for p in parents
                  if run_git(self.local, "show", f"{p}:a.txt")
                  == "THE CAREFULLY STAGED VERSION"]
        self.assertTrue(staged, f"staged version unreachable from {parents}")

    def test_a_snapshot_does_not_disturb_the_index(self):
        self.write(self.local, "a.txt", "one")
        run_git(self.local, "add", "-A")
        run_git(self.local, "commit", "-m", "base")
        self.write(self.local, "a.txt", "staged")
        run_git(self.local, "add", "a.txt")
        before = run_git(self.local, "status", "--porcelain")
        client.build_snapshot(self.local)
        self.assertEqual(run_git(self.local, "status", "--porcelain"), before)

    def test_index_only_work_survives_a_pushed_snapshot(self):
        """A staged version remains backed up when worktree bytes equal HEAD."""
        self.write(self.local, "a.txt", "committed")
        run_git(self.local, "add", "-A")
        run_git(self.local, "commit", "-m", "base")
        run_git(self.local, "push", "-u", "granthi", "main")
        head_before = run_git(self.local, "rev-parse", "HEAD")

        self.write(self.local, "a.txt", "INDEX-ONLY STAGED VERSION")
        run_git(self.local, "add", "a.txt")
        self.write(self.local, "a.txt", "committed")
        status_before = run_git(self.local, "status", "--porcelain")
        index_before = run_git(self.local, "write-tree")
        with open(os.path.join(self.local, "a.txt")) as f:
            file_before = f.read()
        self.assertEqual(status_before, "MM a.txt")

        ref = client.push_snapshot(self.local, "dev-index-only")

        self.assertIsNotNone(ref)
        snapshot = run_git(self.bare, "rev-parse", ref)
        parents = run_git(self.bare, "log", "-1", "--format=%P", snapshot).split()
        self.assertTrue(
            any(run_git(self.bare, "show", f"{parent}:a.txt")
                == "INDEX-ONLY STAGED VERSION" for parent in parents),
            f"staged version unreachable from {parents}")
        self.assertEqual(run_git(self.local, "rev-parse", "HEAD"), head_before)
        self.assertEqual(run_git(self.local, "write-tree"), index_before)
        self.assertEqual(run_git(self.local, "status", "--porcelain"), status_before)
        with open(os.path.join(self.local, "a.txt")) as f:
            self.assertEqual(f.read(), file_before)

        self.write(self.local, "a.txt", "A NEW STAGED VERSION")
        run_git(self.local, "add", "a.txt")
        self.write(self.local, "a.txt", "committed")
        second_ref = f"refs/granthi-backup/dev-index-only/second"
        with mock.patch.object(client, "snapshot_ref",
                               return_value=(second_ref, "second")):
            self.assertEqual(
                client.push_snapshot(self.local, "dev-index-only"), second_ref)
        second = run_git(self.bare, "rev-parse", second_ref)
        second_parents = run_git(
            self.bare, "log", "-1", "--format=%P", second).split()
        self.assertTrue(
            any(run_git(self.bare, "show", f"{parent}:a.txt")
                == "A NEW STAGED VERSION" for parent in second_parents),
            f"updated staged version unreachable from {second_parents}")

    def test_a_truncated_listing_refuses_instead_of_guessing(self):
        """[P3] A name that is merely beyond the page cap must not resolve to a
        different repo that happens to exist under your own account."""
        cfg = {"login": "alice", "gitea_base": "http://forge.example", "token": "t"}
        with self.assertRaises(SystemExit) as e:
            client.resolve_granted(cfg, "notes", repos=[], truncated=True)
        self.assertIn("truncated", str(e.exception))
        # a complete listing still falls back, because absence is then real
        self.assertEqual(
            client.resolve_granted(cfg, "notes", repos=[], truncated=False),
            "alice/notes")

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