#!/usr/bin/env python3 """ Where the login pin is found, and that a git worktree is not a dead zone. python3 -m unittest discover -s tests -v Stdlib unittest, no third-party anything, and not one real network call: every run here is against a throwaway repository with a FAKE `tea` first on PATH. The bug: the pin was searched for by walking up from CWD only. A worktree is a *sibling* of the main checkout, and `.claude/settings.local.json` is untracked, so it lives in the main checkout and nowhere else — the whole sync layer died inside any worktree with "no login pinned", while `tea` in the same directory worked, because the tea-guard hook had a second, different copy of the search. So these tests hold two lines at once: the pin is reachable from a worktree, and the hook and the scripts get their answer from the same function. """ import json import os import shutil import stat import subprocess import sys import tempfile import unittest from unittest import mock REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) AUTH_SCRIPTS = os.path.join(REPO, "skills", "auth", "scripts") ISSUE_SCRIPTS = os.path.join(REPO, "skills", "issue", "scripts") SYNC_SCRIPTS = os.path.join(REPO, "skills", "sync", "scripts") HOOKS = os.path.join(REPO, "hooks") sys.path.insert(0, AUTH_SCRIPTS) import pin # noqa: E402 HAVE_GIT = shutil.which("git") is not None LOGIN = "fixture/user" ENV_KEY = pin.ENV_KEY # A `tea` that answers without a network: an empty list for every GET, a # created object for every write. It records its own argv, which is how a test # reads back the login the call actually ran under. FAKE_TEA = '''#!%s import json, os, sys argv = sys.argv[1:] with open(os.environ["TEA_CALL_LOG"], "a") as f: f.write("\\t".join(argv) + "\\n") sys.stdout.write(json.dumps({"id": 1, "number": 101, "name": "created", "html_url": "https://example.invalid/issues/101", "labels": []}) if "-X" in argv else "[]") ''' ISSUE = """\ --- id: pinned-work state: open labels: [type/task] assignees: [] milestone: none depends: [] origin: local --- # Pinned work ## Summary Issue фикстуры, живёт в сторе worktree. ## Spec none ## Motivation Нужен, чтобы push.py было что отправить. ## Acceptance criteria - [ ] проверяемое условие """ def write(path, text): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w") as f: f.write(text) class Worktree(object): """A repository with a pin, and a linked worktree beside it. Beside, not below: `main/` and `worktrees/feature/` are siblings, which is the entire shape of the bug. The pin is written after the clone is committed and is covered by .gitignore, so it exists in the main checkout only — exactly as `/tea:auth` leaves it.""" def __init__(self, pinned=LOGIN): self._tmp = tempfile.TemporaryDirectory(prefix="tea-pin-") # realpath: on macOS $TMPDIR is a symlink, and a child reporting its # own cwd would otherwise disagree with the path we handed it. self.root = os.path.realpath(self._tmp.name) self.main = os.path.join(self.root, "main") self.tree = os.path.join(self.root, "worktrees", "feature") self.calls = os.path.join(self.root, "calls.txt") skip = shutil.ignore_patterns("__pycache__") for layer in ("auth", "issue", "sync"): shutil.copytree(os.path.join(REPO, "skills", layer, "scripts"), os.path.join(self.main, "skills", layer, "scripts"), ignore=skip) shutil.copytree(HOOKS, os.path.join(self.main, "hooks"), ignore=skip) write(os.path.join(self.main, ".gitignore"), "tmp/\n.claude/\n") self.bin = os.path.join(self.root, "fakebin") os.makedirs(self.bin) tea = os.path.join(self.bin, "tea") write(tea, FAKE_TEA % sys.executable) os.chmod(tea, os.stat(tea).st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) self.git("init", cwd=self.main) self.git("add", "-A", cwd=self.main) self.git("commit", "-m", "fixture", cwd=self.main) self.git("worktree", "add", "-b", "feature", self.tree, cwd=self.main) if pinned: write(os.path.join(self.main, ".claude", "settings.local.json"), json.dumps({"env": {ENV_KEY: pinned}})) def cleanup(self): self._tmp.cleanup() def env(self): env = dict(os.environ) env.pop("PYTHONPATH", None) # no leakage from the harness into the child # The start of the search order, cleared: this fixture is about the # steps *after* it, and the developer's own project must not answer. env.pop(pin.PROJECT_DIR_ENV, None) env["PATH"] = self.bin + os.pathsep + env["PATH"] env["TEA_CALL_LOG"] = self.calls env["HOME"] = self.root # keep the developer's git config out env["GIT_CONFIG_NOSYSTEM"] = "1" env["GIT_CONFIG_GLOBAL"] = os.devnull return env def git(self, *args, **kw): cmd = ["git", "-c", "user.email=fixture@example.invalid", "-c", "user.name=fixture", "-c", "commit.gpgsign=false"] + list(args) p = subprocess.run(cmd, cwd=kw.pop("cwd", self.tree), env=self.env(), capture_output=True, text=True) if p.returncode != 0: raise AssertionError("%s failed:\n%s%s" % (" ".join(cmd), p.stdout, p.stderr)) return p.stdout.strip() def script(self, layer, name): """A script as the WORKTREE sees it — the copy the operator would run.""" return os.path.join(self.tree, "skills", layer, "scripts", name) def run(self, script, *args, **kw): p = subprocess.run([sys.executable, script] + list(args), cwd=kw.pop("cwd", self.tree), env=self.env(), capture_output=True, text=True) return p.returncode, p.stdout, p.stderr def tea_calls(self): if not os.path.isfile(self.calls): return [] with open(self.calls) as f: return [line.rstrip("\n").split("\t") for line in f if line.strip()] def logins_used(self): return [a[a.index("--login") + 1] for a in self.tea_calls() if "--login" in a] # -------------------------------------------------------------------------- # the search itself # -------------------------------------------------------------------------- class TestSearch(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory(prefix="tea-pin-unit-") self.root = os.path.realpath(self._tmp.name) self.addCleanup(self._tmp.cleanup) def path(self, *parts): return os.path.join(self.root, *parts) def pin_at(self, root, login=LOGIN): write(os.path.join(root, ".claude", "settings.local.json"), json.dumps({"env": {ENV_KEY: login}})) def test_the_parent_chain_is_searched(self): self.pin_at(self.root) os.makedirs(self.path("a", "b")) self.assertEqual(pin.search(self.path("a", "b"))[0], LOGIN) def test_no_pin_is_no_pin(self): os.makedirs(self.path("a")) self.assertEqual(pin.search(self.path("a")), (None, None)) def test_an_unreadable_pin_is_not_a_login(self): write(self.path(".claude", "settings.local.json"), "{ not json") self.assertEqual(pin.search(self.root), (None, None)) def test_an_empty_pin_is_not_a_login(self): write(self.path(".claude", "settings.local.json"), json.dumps({"env": {ENV_KEY: " "}})) self.assertEqual(pin.search(self.root), (None, None)) def test_a_git_file_pointing_at_a_worktree_reaches_the_main_checkout(self): """The hop, built by hand from the two files git writes — no git needed to state what the layout means.""" main, tree = self.path("main"), self.path("elsewhere", "feature") gitdir = os.path.join(main, ".git", "worktrees", "feature") os.makedirs(gitdir) os.makedirs(tree) write(os.path.join(gitdir, "commondir"), "../..\n") write(os.path.join(tree, ".git"), "gitdir: %s\n" % gitdir) self.pin_at(main) self.assertEqual(pin.main_worktree(tree), main) login, src = pin.search(tree) self.assertEqual(login, LOGIN) self.assertEqual(src, pin.settings_path(main)) def test_an_ordinary_clone_is_not_a_worktree(self): os.makedirs(self.path("clone", ".git")) self.assertIsNone(pin.main_worktree(self.path("clone"))) def test_a_submodule_pointer_is_not_a_worktree(self): """`.git` is a file there too, but it points into .git/modules/… and the tree it belongs to is already on the parent chain.""" sub = self.path("super", "lib") gitdir = self.path("super", ".git", "modules", "lib") os.makedirs(gitdir) os.makedirs(sub) write(os.path.join(sub, ".git"), "gitdir: %s\n" % gitdir) self.assertIsNone(pin.main_worktree(sub)) def test_the_chain_wins_over_the_hop(self): """The worktree branch may only find a pin the walk up would have missed entirely — it never overrides a nearer one.""" main, tree = self.path("main"), self.path("elsewhere", "feature") gitdir = os.path.join(main, ".git", "worktrees", "feature") os.makedirs(gitdir) os.makedirs(tree) write(os.path.join(gitdir, "commondir"), "../..\n") write(os.path.join(tree, ".git"), "gitdir: %s\n" % gitdir) self.pin_at(main, "main/login") self.pin_at(tree, "worktree/login") self.assertEqual(pin.search(tree)[0], "worktree/login") def test_start_dirs_are_ordered_and_deduplicated(self): with mock.patch.dict(os.environ, {pin.PROJECT_DIR_ENV: self.path("p")}): self.assertEqual(pin.start_dirs(self.path("h")), [self.path("p"), self.path("h"), os.path.abspath(os.getcwd())]) with mock.patch.dict(os.environ, {}, clear=True): self.assertEqual(pin.start_dirs(), [os.path.abspath(os.getcwd())]) # -------------------------------------------------------------------------- # a script run from a worktree # -------------------------------------------------------------------------- @unittest.skipUnless(HAVE_GIT, "git is not installed") class TestScriptsInAWorktree(unittest.TestCase): def setUp(self): self.wt = Worktree() self.addCleanup(self.wt.cleanup) def test_a_sync_script_run_from_the_worktree_finds_the_login(self): """The acceptance criterion, run for real: cwd inside the worktree, the pin in the main checkout, and the call goes out under it.""" rc, out, err = self.wt.run(self.wt.script("sync", "remote.py"), "--repo", "fixture/repo", "--state", "all") self.assertEqual(rc, 0, "remote.py failed:\n%s%s" % (out, err)) self.assertNotIn("no login pinned", err) self.assertEqual(self.wt.logins_used(), [LOGIN]) def test_it_does_not_pin_a_second_login_in_the_worktree(self): """Nothing here writes a settings file, and the worktree is the last place one should appear: it is deleted with the worktree.""" self.wt.run(self.wt.script("sync", "remote.py"), "--repo", "fixture/repo") self.assertFalse(os.path.exists(pin.settings_path(self.wt.tree)), "a second settings.local.json appeared in the worktree") def test_with_no_pin_anywhere_it_still_says_so(self): wt = Worktree(pinned=None) self.addCleanup(wt.cleanup) rc, out, err = wt.run(wt.script("sync", "remote.py"), "--repo", "fixture/repo") self.assertNotEqual(rc, 0) self.assertIn("no login pinned", err) self.assertEqual(wt.logins_used(), []) def test_the_scripts_own_directory_is_not_a_pin_source(self): """Run the worktree's script from a directory that is in no pinned tree. The script sits inside a repository that has a pin — and it must still refuse, because the pin belongs to the project being worked on, not to the installation.""" outside = os.path.join(self.wt.root, "outside") os.makedirs(outside) rc, out, err = self.wt.run(self.wt.script("sync", "remote.py"), "--repo", "fixture/repo", cwd=outside) self.assertNotEqual(rc, 0) self.assertIn("no login pinned", err) def test_push_from_a_worktree_sends_the_worktree_branch(self): """`branch:` -> Gitea `ref`. The workaround this fix removes — run the worktree's scripts with cwd in the main checkout — sent the main checkout's branch, which is the one field `branch:` exists for.""" write(os.path.join(self.wt.tree, "tmp", "issues", "pinned-work.md"), ISSUE) rc, out, err = self.wt.run(self.wt.script("sync", "push.py"), "pinned-work", "--repo", "fixture/repo") self.assertEqual(rc, 0, "push.py failed:\n%s%s" % (out, err)) self.assertIn("created pinned-work #101", out) with open(os.path.join(self.wt.tree, "tmp", "payload", "issue-pinned-work.json")) as f: payload = json.load(f) self.assertEqual(payload.get("ref"), "feature") self.assertEqual(self.wt.git("rev-parse", "--abbrev-ref", "HEAD"), "feature") self.assertNotEqual( self.wt.git("rev-parse", "--abbrev-ref", "HEAD", cwd=self.wt.main), "feature", "the fixture's two trees are on the same branch") # -------------------------------------------------------------------------- # one order, one copy of it # -------------------------------------------------------------------------- @unittest.skipUnless(HAVE_GIT, "git is not installed") class TestTheHookAndTheScriptsAgree(unittest.TestCase): def setUp(self): self.wt = Worktree() self.addCleanup(self.wt.cleanup) def guard(self, cwd): """The hook, as the harness calls it: payload on stdin, decision on stdout.""" payload = {"tool_input": {"command": 'tea api --login "$GITEA_LOGIN" repos/x/y'}, "cwd": cwd} p = subprocess.run([sys.executable, os.path.join(self.wt.tree, "hooks", "tea-guard.sh")], input=json.dumps(payload), cwd=cwd, env=self.wt.env(), capture_output=True, text=True) return p def test_the_hook_resolves_the_pin_from_the_worktree_too(self): p = self.guard(self.wt.tree) self.assertEqual(p.returncode, 0, p.stderr) got = json.loads(p.stdout)["hookSpecificOutput"]["updatedInput"]["command"] self.assertIn(LOGIN, got) self.assertNotIn("GITEA_LOGIN", got) def test_the_hook_and_a_script_answer_the_same_directory_alike(self): """The regression that started this: in one directory the hook resolved the login and every script said there was none.""" rc, out, err = self.wt.run(self.wt.script("sync", "remote.py"), "--repo", "fixture/repo") self.assertEqual(rc, 0, err) script_login = self.wt.logins_used()[0] hook_login = json.loads(self.guard(self.wt.tree).stdout)[ "hookSpecificOutput"]["updatedInput"]["command"].split("--login ")[1].split()[0] self.assertEqual(hook_login, script_login) def test_the_hook_still_blocks_when_nothing_is_pinned(self): wt = Worktree(pinned=None) self.addCleanup(wt.cleanup) payload = {"tool_input": {"command": 'tea api --login "$GITEA_LOGIN" repos/x/y'}, "cwd": wt.tree} p = subprocess.run([sys.executable, os.path.join(wt.tree, "hooks", "tea-guard.sh")], input=json.dumps(payload), cwd=wt.tree, env=wt.env(), capture_output=True, text=True) self.assertEqual(p.returncode, 2) self.assertIn("no login is pinned", p.stderr) class TestNobodyKeepsASecondCopy(unittest.TestCase): """Mechanical: the search order is written in pin.py, and the two callers spell neither the path nor the walk.""" CALLERS = (os.path.join(HOOKS, "tea-guard.sh"), os.path.join(SYNC_SCRIPTS, "_gitea.py")) def source(self, path): with open(path) as f: return f.read() def test_the_path_is_spelled_once(self): self.assertEqual(pin.SETTINGS_PARTS, (".claude", "settings.local.json")) for path in self.CALLERS: body = self.source(path) for literal in ('".claude"', "'.claude'"): self.assertNotIn(literal, body, "%s builds the settings path itself" % path) def test_both_callers_go_through_the_module(self): for path in self.CALLERS: self.assertIn("import pin", self.source(path), "%s does not resolve the pin through pin.py" % path) def test_the_domain_layer_never_learns_what_a_login_is(self): """The layer rule, unchanged by this: the identity module is imported by the bridge and by the hook, never by a domain.""" for layer in ("issue", "page"): d = os.path.join(REPO, "skills", layer, "scripts") for name in sorted(os.listdir(d)): if not name.endswith(".py"): continue body = self.source(os.path.join(d, name)) for banned in ("import pin", "GITEA_LOGIN", "settings.local.json"): self.assertNotIn(banned, body, "%s/%s: %s" % (layer, name, banned)) if __name__ == "__main__": unittest.main()