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 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()


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_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(argparse.Namespace(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(argparse.Namespace(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(argparse.Namespace(repo="cloud", into=dest))
        self.assertEqual(open(os.path.join(dest, "mine.txt")).read(),
                         "do not clobber")

    def test_get_unlinked_exits_like_add(self):
        with mock.patch.object(client, "load_config", lambda: {}):
            with self.assertRaises(SystemExit) as cm:
                client.cmd_get(argparse.Namespace(repo="cloud", into=None))
        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(argparse.Namespace(repo="blank", into=dest))
        meta = client.load_config()["folders"][os.path.abspath(dest)]
        self.assertEqual(meta["branch"], "main")

    def test_get_never_puts_token_in_remote_url(self):
        dest = os.path.join(self.tmp, "pulled3")
        client.cmd_get(argparse.Namespace(repo="cloud", into=dest))
        self.assertNotIn("sekrit", run_git(dest, "remote", "get-url", "granthi"))


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