naudachu
83f73c5cea
tea and tdl were two repositories, each carrying its own
.claude-plugin/marketplace.json — two marketplaces to register for what
is one collection. Fold them into one.
The repo root is now the marketplace and nothing else: a single
.claude-plugin/marketplace.json whose entries point at ./plugins/tea and
./plugins/tdl. A plugin's root is its own directory under plugins/, so
${CLAUDE_PLUGIN_ROOT} still resolves inside it and every path a plugin
uses stays relative to itself — the hooks and the test roots needed no
adjustment beyond the move.
tea's files move with git mv, so its history and blame follow. tdl
arrives as a plain copy; its history stays in claude-skills/threedotslab.
test_payload_root asserted `tmp/` was ignored by REPO/.gitignore. The
rule is that tmp/ is ignored, not which file says so, and git reads every
.gitignore on the way up — so the test now walks up to the repo root the
same way git does.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
419 lines
18 KiB
Python
419 lines
18 KiB
Python
#!/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")
|
|
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()
|