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


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