fix: resolve the issue store from the project, not the plugin

`issue.store_root` and `_gitea.PAYLOAD_ROOT` were anchored on `__file__`, on
the reasoning that where an installation keeps its files is a fact about the
installation. That holds for an installation and not for a store.

Installed, the plugin therefore resolved every project's issues inside its own
directory — and a plugin cache is versioned, so the store moved on each
update:

    ~/.claude/plugins/cache/tea/tea/2.0.0/tmp/issues   5 files, 2 origin: local
    ~/.claude/plugins/cache/tea/tea/2.1.0/tmp/issues   12 files
    ~/.claude/plugins/cache/claude-skills/tea/2.2.0/   empty, the current one

Issues written from one project were invisible from the next, and an `origin:
local` file — which IS the issue, the only copy — was stranded a version bump
at a time. Two of them were.

The store is a fact about the project, exactly as the login pin is. So the
anchor is now an explicit marker an operator creates, `.tea/`, searched for up
from $CLAUDE_PROJECT_DIR and then cwd — the pin's order, so the two cannot
disagree about which project this is. Inferred markers were tried and are worse
than useless here: `.git` is in every clone including this plugin's own, and
the agents-sync hook writes an AGENTS.md next to every AGENTS.md, so the plugin
root always carried one and cwd never got a turn.

With no marker anywhere, `store_root()` is None and every entry point reports
which directories it searched. A store in a plausible-looking directory is the
failure this replaces, so nothing falls back to one.

- `.tea/` holds the store and the transport's scratchpad: `.tea/issues`,
  `.tea/payload`. One marker, one walk, one gitignore line.
- `issue_init.py` creates it, moves an old `tmp/issues` store in rather than
  copying, adds `.tea/` to `.gitignore`, and refuses to pick a winner when both
  sides hold the same file name.
- A linked worktree has no marker — it is gitignored — and reaches the main
  checkout's store by the hop the pin already took.
- `parents`, `gitdir_of` and `main_worktree` move from `pin.py` into the domain
  and `pin.py` imports them. The domain depends on nothing, so it is the layer
  all three callers can borrow from, and the walk stays written once: the
  guard, the transport and the store cannot disagree about a directory.

The suite stopped copying the script layers into its fixtures. That is what hid
this: with the scripts inside the fixture, the installation and the project
were the same directory. They are now deliberately far apart, and a regression
test asserts the plugin tree gains no files when commands run against a project
somewhere else.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
naudachu
2026-08-11 13:38:39 +05:00
parent 27e4b6b1da
commit fb5445915f
30 changed files with 1193 additions and 430 deletions
+203
View File
@@ -0,0 +1,203 @@
#!/usr/bin/env python3
"""
`issue_init.py` — the statement that makes a directory a project.
python3 -m unittest discover -s tests -v
The marker is the anchor every other script resolves from, so the command that
creates it carries the whole contract: it is idempotent, it never picks a
winner between two versions of one issue, and it migrates the old `tmp/` layout
by MOVING — a store left behind at the old path is a store somebody edits by
accident.
Like the rest of the suite, these run the real script against throwaway
directories: the script stays where it is installed, the project is somewhere
else entirely.
"""
import os
import subprocess
import sys
import tempfile
import unittest
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
ISSUE_SCRIPTS = os.path.join(REPO, "skills", "issue", "scripts")
INIT = os.path.join(ISSUE_SCRIPTS, "issue_init.py")
sys.path.insert(0, ISSUE_SCRIPTS)
import issue # noqa: E402
def run(*args, **kw):
env = dict(os.environ)
env.pop("PYTHONPATH", None)
env.pop("CLAUDE_PROJECT_DIR", None)
p = subprocess.run([sys.executable, INIT] + list(args),
cwd=kw.pop("cwd"), env=env, capture_output=True, text=True)
return p.returncode, p.stdout, p.stderr
def write(path, text):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f:
f.write(text)
class Dir(object):
def __init__(self):
self._tmp = tempfile.TemporaryDirectory()
self.root = os.path.realpath(self._tmp.name)
def cleanup(self):
self._tmp.cleanup()
def path(self, *parts):
return os.path.join(self.root, *parts)
class TestInit(unittest.TestCase):
def setUp(self):
self.d = Dir()
self.addCleanup(self.d.cleanup)
def test_it_creates_the_marker_and_both_directories(self):
rc, out, err = run(cwd=self.d.root)
self.assertEqual(rc, 0, err)
self.assertTrue(os.path.isdir(self.d.path(issue.MARKER, "issues")))
self.assertTrue(os.path.isdir(self.d.path(issue.MARKER, "payload")))
self.assertEqual(issue.project_root(self.d.root), self.d.root)
def test_the_store_resolves_from_a_subdirectory_afterwards(self):
run(cwd=self.d.root)
deep = self.d.path("a", "b", "c")
os.makedirs(deep)
self.assertEqual(issue.store_root(deep),
self.d.path(*issue.STORE_PARTS))
def test_running_it_twice_changes_nothing(self):
run(cwd=self.d.root)
before = sorted(os.walk(self.d.root))
rc, out, err = run(cwd=self.d.root)
self.assertEqual(rc, 0, err)
self.assertIn("already initialized", out)
self.assertEqual(sorted(os.walk(self.d.root)), before)
def test_a_dry_run_touches_nothing(self):
rc, out, err = run("--dry-run", cwd=self.d.root)
self.assertEqual(rc, 0, err)
self.assertIn("would:", out)
self.assertFalse(os.path.exists(self.d.path(issue.MARKER)))
def test_at_initializes_somewhere_else(self):
other = Dir()
self.addCleanup(other.cleanup)
rc, out, err = run("--at", other.root, cwd=self.d.root)
self.assertEqual(rc, 0, err)
self.assertTrue(os.path.isdir(other.path(issue.MARKER)))
self.assertFalse(os.path.exists(self.d.path(issue.MARKER)))
class TestGitignore(unittest.TestCase):
def setUp(self):
self.d = Dir()
self.addCleanup(self.d.cleanup)
def test_the_marker_is_added(self):
run(cwd=self.d.root)
with open(self.d.path(".gitignore")) as f:
self.assertIn(issue.MARKER + "/", f.read().split())
def test_an_existing_gitignore_keeps_its_contents(self):
write(self.d.path(".gitignore"), "node_modules/\n*.log\n")
run(cwd=self.d.root)
with open(self.d.path(".gitignore")) as f:
lines = f.read().split()
self.assertIn("node_modules/", lines)
self.assertIn("*.log", lines)
self.assertIn(issue.MARKER + "/", lines)
def test_it_is_not_added_twice(self):
write(self.d.path(".gitignore"), issue.MARKER + "\n")
run(cwd=self.d.root)
with open(self.d.path(".gitignore")) as f:
body = f.read()
self.assertEqual(body.count(issue.MARKER), 1, body)
class TestMigration(unittest.TestCase):
"""The old layout moves in. Moves, not copies: two stores is the state this
whole change exists to prevent."""
def setUp(self):
self.d = Dir()
self.addCleanup(self.d.cleanup)
def test_an_old_store_is_moved_in(self):
write(self.d.path("tmp", "issues", "old-work.md"), "id: old-work\n")
write(self.d.path("tmp", "payload", "request.json"), "{}")
rc, out, err = run(cwd=self.d.root)
self.assertEqual(rc, 0, err)
self.assertTrue(os.path.isfile(
self.d.path(issue.MARKER, "issues", "old-work.md")))
self.assertTrue(os.path.isfile(
self.d.path(issue.MARKER, "payload", "request.json")))
self.assertFalse(os.path.exists(self.d.path("tmp", "issues")),
"the old store was left behind for somebody to edit")
self.assertIn("moved 1 file(s)", out)
def test_a_clash_stops_everything_and_moves_nothing(self):
write(self.d.path("tmp", "issues", "same.md"), "old version\n")
write(self.d.path(issue.MARKER, "issues", "same.md"), "new version\n")
rc, out, err = run(cwd=self.d.root)
self.assertNotEqual(rc, 0)
self.assertIn("same.md", out + err)
with open(self.d.path("tmp", "issues", "same.md")) as f:
self.assertEqual(f.read(), "old version\n")
with open(self.d.path(issue.MARKER, "issues", "same.md")) as f:
self.assertEqual(f.read(), "new version\n")
def test_a_dry_run_reports_the_move_without_making_it(self):
write(self.d.path("tmp", "issues", "old-work.md"), "id: old-work\n")
rc, out, err = run("--dry-run", cwd=self.d.root)
self.assertEqual(rc, 0, err)
self.assertIn("moved 1 file(s)", out)
self.assertTrue(os.path.isfile(self.d.path("tmp", "issues", "old-work.md")))
self.assertFalse(os.path.exists(self.d.path(issue.MARKER)))
def test_no_old_layout_is_not_an_error(self):
rc, out, err = run(cwd=self.d.root)
self.assertEqual(rc, 0, err)
self.assertNotIn("moved", out)
class TestNesting(unittest.TestCase):
def setUp(self):
self.d = Dir()
self.addCleanup(self.d.cleanup)
run(cwd=self.d.root)
def test_initializing_inside_a_project_warns(self):
inner = self.d.path("packages", "api")
os.makedirs(inner)
rc, out, err = run(cwd=inner)
self.assertEqual(rc, 0, err)
self.assertIn("already sits inside the project", err)
self.assertIn(self.d.root, err)
def test_the_warning_does_not_stop_it(self):
"""A monorepo package that genuinely wants its own issues is allowed to
say so. The warning is that the nearer marker wins from then on, which
is a consequence worth reading, not an error."""
inner = self.d.path("packages", "api")
os.makedirs(inner)
run(cwd=inner)
self.assertEqual(issue.project_root(inner), inner)
self.assertEqual(issue.project_root(self.d.root), self.d.root)
if __name__ == "__main__":
unittest.main()
+30 -4
View File
@@ -108,7 +108,12 @@ class Worktree(object):
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")
write(os.path.join(self.main, ".gitignore"), ".tea/\n.claude/\n")
# The project marker, in the MAIN checkout only — it is gitignored, so
# a linked worktree never has one, exactly like the pin. One project,
# one store, reached from the worktree by the same hop.
os.makedirs(os.path.join(self.main, ".tea", "issues"))
self.bin = os.path.join(self.root, "fakebin")
os.makedirs(self.bin)
@@ -305,17 +310,38 @@ class TestScriptsInAWorktree(unittest.TestCase):
self.assertNotEqual(rc, 0)
self.assertIn("no login pinned", err)
def test_the_store_resolves_to_the_main_checkout_from_a_worktree(self):
"""The marker is gitignored, so a linked worktree never has one. It is
the same project on another branch and it gets the same store — by the
same hop the pin takes. Initializing in the worktree instead would give
one project two stores, in a directory deleted with the branch."""
code = ("import sys; sys.path.insert(0, %r)\n"
"import issue\nprint(issue.project_root() or '')\n"
"print(issue.store_root() or '')\n"
% os.path.join(self.wt.main, "skills", "issue", "scripts"))
env = self.wt.env()
env.pop("CLAUDE_PROJECT_DIR", None)
p = subprocess.run([sys.executable, "-c", code], cwd=self.wt.tree,
env=env, capture_output=True, text=True)
self.assertEqual(p.returncode, 0, p.stderr)
root, store = p.stdout.strip().splitlines()
self.assertEqual(root, self.wt.main)
self.assertEqual(store, os.path.join(self.wt.main, ".tea", "issues"))
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)
checkout's branch, which is the one field `branch:` exists for.
The store is the main checkout's, reached by the hop; the branch is the
worktree's, read from cwd. Two questions, two answers, one command."""
write(os.path.join(self.wt.main, ".tea", "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",
with open(os.path.join(self.wt.main, ".tea", "payload",
"issue-pinned-work.json")) as f:
payload = json.load(f)
self.assertEqual(payload.get("ref"), "feature")
+57 -35
View File
@@ -49,7 +49,11 @@ sys.stdout.write(json.dumps({"id": 1, "name": "created"})
class FakeRepo(object):
"""A self-contained repository with no store and no tmp/ at all."""
"""An initialized project with no store and nothing under `.tea/` yet.
The scripts are NOT copied in: they stay at their real installed path, so
what these tests exercise is a plugin operating on somebody else's project
— which is every use of it but this repository's own."""
def __init__(self):
self._tmp = tempfile.TemporaryDirectory()
@@ -57,12 +61,7 @@ class FakeRepo(object):
# own cwd would otherwise disagree with the path we handed it.
self.root = os.path.realpath(self._tmp.name)
os.makedirs(os.path.join(self.root, ".git")) # the repo marker
skip = shutil.ignore_patterns("__pycache__")
shutil.copytree(ISSUE_SCRIPTS, self.path("skills", "issue", "scripts"), ignore=skip)
shutil.copytree(SYNC_SCRIPTS, self.path("skills", "sync", "scripts"), ignore=skip)
# the transport resolves the login pin through skills/auth/scripts
shutil.copytree(AUTH_SCRIPTS, self.path("skills", "auth", "scripts"), ignore=skip)
os.makedirs(os.path.join(self.root, issue.MARKER)) # the project marker
os.makedirs(self.path("sub", "deeper"))
# the login pin the transport insists on, local to this fixture
@@ -85,18 +84,22 @@ class FakeRepo(object):
@property
def store(self):
return self.path("tmp", "issues")
return self.path(*issue.STORE_PARTS)
@property
def payloads(self):
return self.path("tmp", "payload")
return self.path(*_gitea.PAYLOAD_PARTS)
def script(self, layer, name):
return self.path("skills", layer, "scripts", name)
"""The real script, several directories away from this fixture."""
return os.path.join(REPO, "skills", layer, "scripts", name)
def run(self, script, *args, **kw):
env = dict(os.environ)
env.pop("PYTHONPATH", None) # no leakage from the harness into the child
# the first anchor of both walks: left in place, every fixture would
# resolve to this repository instead of itself
env.pop("CLAUDE_PROJECT_DIR", None)
env["PATH"] = self.bin + os.pathsep + env["PATH"]
env["TEA_CALL_LOG"] = self.root
p = subprocess.run([sys.executable, script] + list(args),
@@ -118,26 +121,43 @@ class FakeRepo(object):
class TestPayloadRoot(unittest.TestCase):
def test_root_is_absolute_and_repo_anchored(self):
self.assertTrue(os.path.isabs(_gitea.PAYLOAD_ROOT), _gitea.PAYLOAD_ROOT)
self.assertEqual(_gitea.PAYLOAD_ROOT, os.path.join(REPO, "tmp", "payload"))
def setUp(self):
self.repo = FakeRepo()
self.addCleanup(self.repo.cleanup)
def test_root_is_absolute_and_project_anchored(self):
root = _gitea.payload_root(self.repo.path("sub", "deeper"))
self.assertTrue(os.path.isabs(root), root)
self.assertEqual(root, self.repo.payloads)
def test_it_is_not_the_issue_store_and_not_inside_one(self):
"""The acceptance criterion, as a path fact: a request body is not
store content, so it may not live in a store or under one."""
self.assertNotEqual(_gitea.PAYLOAD_ROOT, issue.ISSUE_ROOT)
self.assertFalse(_gitea.PAYLOAD_ROOT.startswith(issue.ISSUE_ROOT + os.sep))
self.assertFalse(issue.ISSUE_ROOT.startswith(_gitea.PAYLOAD_ROOT + os.sep))
start = self.repo.path("sub", "deeper")
payload = _gitea.payload_root(start)
store = issue.store_root(start)
self.assertNotEqual(payload, store)
self.assertFalse(payload.startswith(store + os.sep))
self.assertFalse(store.startswith(payload + os.sep))
def test_the_name_says_what_it_holds(self):
"""Named so the distinction is visible: a top-level directory called
`payload`, not a dotdir hiding among an issue's files."""
self.assertEqual(os.path.basename(_gitea.PAYLOAD_ROOT), "payload")
self.assertFalse(os.path.basename(_gitea.PAYLOAD_ROOT).startswith("."))
"""Named so the distinction is visible: `payload`, a sibling of the
store under the marker, not a dotdir hiding among an issue's files."""
payload = _gitea.payload_root(self.repo.root)
self.assertEqual(os.path.basename(payload), "payload")
self.assertEqual(os.path.dirname(payload),
self.repo.path(issue.MARKER))
def test_no_project_means_no_payload_root(self):
"""Same answer as the store gives: nothing, rather than a directory
picked because it was the only one at hand."""
plain = tempfile.TemporaryDirectory()
self.addCleanup(plain.cleanup)
self.assertIsNone(_gitea.payload_root(os.path.realpath(plain.name)))
def test_gitignore_covers_it(self):
"""The rule is `tmp/` is ignored, not which file says so: this plugin
lives under `plugins/` in a marketplace repo, and git reads every
"""The rule is that the marker is ignored, not which file says so: this
plugin lives under `plugins/` in a marketplace repo, and git reads every
.gitignore on the way up. So walk up the same way git does."""
ignored = set()
d = REPO
@@ -150,15 +170,16 @@ class TestPayloadRoot(unittest.TestCase):
if parent == d or os.path.isdir(os.path.join(d, ".git")):
break
d = parent
self.assertEqual(_gitea.PAYLOAD_PARTS[0], "tmp")
self.assertIn("tmp/", ignored,
self.assertEqual(_gitea.PAYLOAD_PARTS[0], issue.MARKER)
self.assertIn(issue.MARKER + "/", ignored,
"the payload directory is not covered by .gitignore")
def test_resolution_is_anchored_on_the_module_not_on_cwd(self):
repo = FakeRepo()
self.addCleanup(repo.cleanup)
self.assertEqual(_gitea.payload_root(repo.path("sub", "deeper")),
repo.payloads)
def test_resolution_follows_the_project_not_the_module(self):
"""The bug, as a path fact: the scripts live somewhere else entirely,
and the answer is still this project's directory."""
self.assertEqual(_gitea.payload_root(self.repo.path("sub", "deeper")),
self.repo.payloads)
self.assertFalse(self.repo.payloads.startswith(REPO + os.sep))
# --------------------------------------------------------------------------
@@ -211,17 +232,18 @@ class TestLabelsTouchesNoStore(unittest.TestCase):
def test_a_dry_run_writes_nothing_at_all(self):
out, _ = self.bootstrap("--dry-run")
self.assertIn("nothing was written", out)
self.assertFalse(os.path.exists(self.repo.path("tmp")),
"a dry run left something behind in tmp/")
self.assertEqual(os.listdir(self.repo.path(issue.MARKER)), [],
"a dry run left something behind under the marker")
def test_the_directory_does_not_follow_cwd(self):
"""Run from a subdirectory: still one payload root, at the repo root.
A cwd-relative directory is how the store ended up with a second copy
of itself, and this one is resolved the same way to avoid the same
"""Run from a subdirectory: still one payload root, at the project
root. A cwd-relative directory is how the store ended up with a second
copy of itself, and this one is resolved the same way to avoid the same
class of bug."""
self.bootstrap(cwd=self.repo.path("sub", "deeper"))
self.assertTrue(os.path.isdir(self.repo.payloads))
self.assertFalse(os.path.exists(self.repo.path("sub", "deeper", "tmp")))
self.assertFalse(os.path.exists(
self.repo.path("sub", "deeper", issue.MARKER)))
self.assertFalse(os.path.exists(self.repo.store))
+297 -117
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
Where the issue store is, and that the answer does not depend on cwd.
Where the issue store is: the project the operator marked, never the plugin.
python3 -m unittest discover -s tests -v
@@ -8,11 +8,22 @@ Stdlib unittest, no third-party anything — the same rule the scripts under tes
live by. `skills/*/scripts/` are not packages, so the domain module is imported
by path.
Most of these tests do not touch this repository at all. They build a throwaway
repo in a temp directory — a `.git` marker, a copy of both script layers, a
These tests build a throwaway project in a temp directory — a `.tea/` marker, a
store with two issues — and run the real scripts inside it as subprocesses with
different working directories. That is the only honest way to test a cwd bug:
importing the module would resolve the store once, against the wrong tree.
different working directories.
**The scripts are deliberately NOT copied into the fixture.** They stay where
they really live, several directories away from the project under test, because
that separation IS the thing being tested: a plugin is installed in one place
and used on projects in another, and the store belongs to the project. The
suite used to copy both script layers in, which made the two locations the same
directory and hid the bug completely — issues written from a project landed in
`~/.claude/plugins/cache/tea/tea/<version>/tmp/issues` and vanished on the next
version bump.
`CLAUDE_PROJECT_DIR` is stripped from the child environment except where a test
is about it: it is the first anchor, so leaving the harness's own value in place
would point every fixture at this repository.
"""
import os
import shutil
@@ -24,7 +35,6 @@ import unittest
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
ISSUE_SCRIPTS = os.path.join(REPO, "skills", "issue", "scripts")
SYNC_SCRIPTS = os.path.join(REPO, "skills", "sync", "scripts")
AUTH_SCRIPTS = os.path.join(REPO, "skills", "auth", "scripts")
sys.path.insert(0, ISSUE_SCRIPTS)
import issue # noqa: E402
@@ -85,34 +95,38 @@ none
def run(script, *args, **kw):
"""Run one of the plugin's scripts and return (rc, stdout, stderr)."""
"""Run one of the plugin's real scripts and return (rc, stdout, stderr).
`project_dir` sets CLAUDE_PROJECT_DIR for the child; by default the variable
is removed, so cwd alone decides which project answers."""
cwd = kw.pop("cwd")
project_dir = kw.pop("project_dir", None)
env = dict(os.environ)
env.pop("PYTHONPATH", None) # no leakage from the harness into the child
env.pop("CLAUDE_PROJECT_DIR", None)
if project_dir:
env["CLAUDE_PROJECT_DIR"] = project_dir
p = subprocess.run([sys.executable, script] + list(args), cwd=cwd, env=env,
capture_output=True, text=True)
return p.returncode, p.stdout, p.stderr
class FakeRepo(object):
"""A self-contained repository in a temp directory.
def script(layer, name):
"""A script at its real installed path — never a copy inside a fixture."""
return os.path.join(REPO, "skills", layer, "scripts", name)
Both script layers are copied in, so `__file__`-anchored resolution lands
inside the fixture and never on the developer's real store.
"""
def __init__(self, with_store=True, issues=(ALPHA, BETA)):
class FakeProject(object):
"""An initialized project in a temp directory, far from the scripts."""
def __init__(self, with_store=True, marker=True, issues=(ALPHA, BETA)):
self._tmp = tempfile.TemporaryDirectory()
# realpath: on macOS $TMPDIR is a symlink, and a child process reporting
# its own cwd would otherwise disagree with the path we handed it.
self.root = os.path.realpath(self._tmp.name)
os.makedirs(os.path.join(self.root, ".git")) # the repo marker
skip = shutil.ignore_patterns("__pycache__")
shutil.copytree(ISSUE_SCRIPTS, self.path("skills", "issue", "scripts"), ignore=skip)
shutil.copytree(SYNC_SCRIPTS, self.path("skills", "sync", "scripts"), ignore=skip)
# the transport resolves the login pin through skills/auth/scripts
shutil.copytree(AUTH_SCRIPTS, self.path("skills", "auth", "scripts"), ignore=skip)
if marker:
os.makedirs(os.path.join(self.root, issue.MARKER))
os.makedirs(self.path("sub", "deeper"))
if with_store:
@@ -130,17 +144,14 @@ class FakeRepo(object):
@property
def store(self):
return self.path("tmp", "issues")
def script(self, layer, name):
return self.path("skills", layer, "scripts", name)
return self.path(*issue.STORE_PARTS)
def everywhere(self):
"""Working directories that must all produce the same answer: the repo
root, a plain subdirectory, a deeper one, the script directory itself,
and — the case from the bug report — inside the store."""
"""Working directories that must all produce the same answer: the
project root, a plain subdirectory, a deeper one, and — the case from
the original bug report — inside the store itself."""
return [self.root, self.path("sub"), self.path("sub", "deeper"),
self.path("skills", "issue", "scripts"), self.store]
self.store]
# --------------------------------------------------------------------------
@@ -150,38 +161,99 @@ class FakeRepo(object):
class TestResolution(unittest.TestCase):
def setUp(self):
self.repo = FakeRepo()
self.addCleanup(self.repo.cleanup)
self.project = FakeProject()
self.addCleanup(self.project.cleanup)
def test_repo_root_found_from_any_depth(self):
for start in self.repo.everywhere():
self.assertEqual(issue.repo_root(start), self.repo.root, start)
def test_the_marker_is_found_from_any_depth(self):
for start in self.project.everywhere():
self.assertEqual(issue.project_root(start), self.project.root, start)
def test_agents_md_works_as_a_marker(self):
"""A checkout without .git — the plugin copied out of git — still
resolves, because AGENTS.md marks the root too."""
shutil.rmtree(self.repo.path(".git"))
open(self.repo.path("AGENTS.md"), "w").close()
self.assertEqual(issue.repo_root(self.repo.path("sub", "deeper")),
self.repo.root)
def test_git_alone_is_not_a_marker(self):
"""The whole point of an explicit marker. `.git` is in every clone,
including this plugin's own — inferring the root from one is how the
plugin came to answer with itself. An uninitialized repository is not a
project this tool knows about, and it says so instead of guessing."""
plain = FakeProject(marker=False, with_store=False)
self.addCleanup(plain.cleanup)
os.makedirs(plain.path(".git"))
open(plain.path("AGENTS.md"), "w").close()
self.assertIsNone(issue.project_root(plain.path("sub", "deeper")))
self.assertIsNone(issue.store_root(plain.path("sub", "deeper")))
def test_nearest_marker_wins(self):
"""A repo inside a repo (a worktree, a vendored copy) resolves to the
inner one, not the outer."""
inner = self.repo.path("sub", "inner")
os.makedirs(os.path.join(inner, ".git"))
self.assertEqual(issue.repo_root(inner), inner)
self.assertEqual(issue.repo_root(self.repo.root), self.repo.root)
"""A project inside a project (a vendored copy, a nested checkout)
resolves to the inner one, not the outer."""
inner = self.project.path("sub", "inner")
os.makedirs(os.path.join(inner, issue.MARKER))
self.assertEqual(issue.project_root(inner), inner)
self.assertEqual(issue.project_root(self.project.root), self.project.root)
def test_store_root_is_repo_root_plus_tmp_issues(self):
self.assertEqual(issue.store_root(self.repo.path("sub", "deeper")),
self.repo.store)
def test_store_root_is_project_root_plus_marker(self):
self.assertEqual(issue.store_root(self.project.path("sub", "deeper")),
self.project.store)
self.assertTrue(os.path.isabs(issue.store_root(self.project.root)))
def test_default_root_is_absolute(self):
"""The whole point: a default that cannot mean two directories."""
self.assertTrue(os.path.isabs(issue.ISSUE_ROOT), issue.ISSUE_ROOT)
self.assertEqual(issue.ISSUE_ROOT,
os.path.join(REPO, "tmp", "issues"))
def test_no_marker_anywhere_resolves_to_nothing(self):
"""Not a default, not cwd, not the script's own directory: None. A
wrong directory that looks like it worked is the failure this
replaces."""
plain = FakeProject(marker=False, with_store=False)
self.addCleanup(plain.cleanup)
self.assertIsNone(issue.store_root(plain.path("sub", "deeper")))
def test_the_error_names_the_directories_it_searched(self):
msg = issue.no_project_error("/nowhere-at-all")
self.assertIn(issue.MARKER, msg)
self.assertIn("/nowhere-at-all", msg)
self.assertIn("issue_init.py", msg)
# --------------------------------------------------------------------------
# the regression: an installed plugin never answers with itself
# --------------------------------------------------------------------------
class TestTheStoreIsNeverThePlugin(unittest.TestCase):
"""The bug this contract exists for.
Anchored on `__file__`, every one of these commands resolved the store
inside the plugin — a versioned cache directory — so work written from a
project was invisible from it and disappeared on the next plugin update."""
def setUp(self):
self.project = FakeProject()
self.addCleanup(self.project.cleanup)
self.before = self._plugin_tree()
def _plugin_tree(self):
out = set()
for dirpath, dirnames, filenames in os.walk(REPO):
dirnames[:] = [d for d in dirnames if d != "__pycache__"]
for f in filenames:
out.add(os.path.join(dirpath, f))
return out
def test_scripts_run_from_a_project_write_only_into_that_project(self):
for d in self.project.everywhere():
for name in ("issue_index.py", "issue_check.py", "issue_tree.py"):
run(script("issue", name), cwd=d)
rc, out, err = run(script("issue", "issue_new.py"),
"--type", "task", "--title", "Written from a project",
cwd=self.project.path("sub", "deeper"))
self.assertEqual(rc, 0, err)
self.assertTrue(os.path.isfile(
os.path.join(self.project.store, "written-from-a-project.md")))
new = self._plugin_tree() - self.before
self.assertEqual(new, set(),
"these commands wrote into the plugin: %s"
% ", ".join(sorted(new)))
def test_the_plugins_own_marker_does_not_leak_into_a_project(self):
"""Should this repository ever be initialized for its own issues, that
marker must not become the answer for a project that has one."""
for d in self.project.everywhere():
self.assertEqual(issue.project_root(d), self.project.root, d)
# --------------------------------------------------------------------------
@@ -191,17 +263,18 @@ class TestResolution(unittest.TestCase):
class TestSameFromAnywhere(unittest.TestCase):
def setUp(self):
self.repo = FakeRepo()
self.addCleanup(self.repo.cleanup)
self.project = FakeProject()
self.addCleanup(self.project.cleanup)
def assertSameEverywhere(self, layer, name, *args):
"""Run the script from the repo root and from every other directory;
"""Run the script from the project root and from every other directory;
every result must be byte-identical to the one from the root."""
dirs = self.repo.everywhere()
base = run(self.repo.script(layer, name), *args, cwd=dirs[0])
self.assertEqual(base[0], 0, "%s failed at the repo root:\n%s" % (name, base[2]))
dirs = self.project.everywhere()
base = run(script(layer, name), *args, cwd=dirs[0])
self.assertEqual(base[0], 0, "%s failed at the project root:\n%s"
% (name, base[2]))
for d in dirs[1:]:
self.assertEqual(run(self.repo.script(layer, name), *args, cwd=d), base,
self.assertEqual(run(script(layer, name), *args, cwd=d), base,
"%s disagrees when run from %s" % (name, d))
return base
@@ -218,24 +291,111 @@ class TestSameFromAnywhere(unittest.TestCase):
def test_issue_index(self):
_, out, _ = self.assertSameEverywhere("issue", "issue_index.py")
self.assertIn("2 issue(s)", out)
self.assertIn(os.path.join(self.repo.store, "INDEX.md"), out)
self.assertIn(os.path.join(self.project.store, "INDEX.md"), out)
def test_no_second_store_is_ever_created(self):
"""The bug's worst symptom: `issue_index.py` run from inside the store
used to leave tmp/issues/tmp/issues/ behind, silently."""
for d in self.repo.everywhere():
"""The old bug's worst symptom: `issue_index.py` run from inside the
store used to leave tmp/issues/tmp/issues/ behind, silently."""
for d in self.project.everywhere():
for name in ("issue_index.py", "issue_check.py", "issue_tree.py"):
run(self.repo.script("issue", name), cwd=d)
run(script("issue", name), cwd=d)
found = []
for dirpath, dirnames, filenames in os.walk(self.repo.root):
for dirpath, dirnames, filenames in os.walk(self.project.root):
if "__pycache__" in dirnames:
dirnames.remove("__pycache__")
if "INDEX.md" in filenames:
found.append(dirpath)
self.assertEqual(found, [self.repo.store],
self.assertEqual(found, [self.project.store],
"a second store appeared: %s" % found)
def test_a_cd_into_another_project_answers_with_that_project(self):
"""Walking up is not cwd-independence for its own sake: two projects
are two stores, and the one you are standing in is the one you meant."""
other = FakeProject(issues=(ALPHA,))
self.addCleanup(other.cleanup)
_, mine, _ = run(script("issue", "issue_check.py"), cwd=self.project.root)
_, theirs, _ = run(script("issue", "issue_check.py"), cwd=other.root)
self.assertIn("2 issue(s) checked", mine)
self.assertIn("1 issue(s) checked", theirs)
# --------------------------------------------------------------------------
# which anchor wins
# --------------------------------------------------------------------------
class TestAnchorOrder(unittest.TestCase):
def setUp(self):
self.project = FakeProject()
self.other = FakeProject(issues=(ALPHA,))
self.addCleanup(self.project.cleanup)
self.addCleanup(self.other.cleanup)
def test_claude_project_dir_is_asked_before_cwd(self):
"""The editor's project is the project, even when a command happens to
run from somewhere else — the same order the login pin uses, so the two
cannot disagree about which project this is."""
_, out, err = run(script("issue", "issue_check.py"),
cwd=self.other.root, project_dir=self.project.root)
self.assertIn("2 issue(s) checked", out, err)
def test_an_unmarked_claude_project_dir_falls_through_to_cwd(self):
"""First hit wins, not first anchor tried: a project dir with no marker
above it is no answer at all, and cwd still gets its turn."""
plain = FakeProject(marker=False, with_store=False)
self.addCleanup(plain.cleanup)
_, out, err = run(script("issue", "issue_check.py"),
cwd=self.other.root, project_dir=plain.root)
self.assertIn("1 issue(s) checked", out, err)
# --------------------------------------------------------------------------
# no project at all
# --------------------------------------------------------------------------
class TestNoProject(unittest.TestCase):
def setUp(self):
self.plain = FakeProject(marker=False, with_store=False)
self.addCleanup(self.plain.cleanup)
def test_readers_report_it_and_name_where_they_looked(self):
for name in ("issue_check.py", "issue_tree.py", "issue_index.py"):
rc, out, err = run(script("issue", name), cwd=self.plain.root)
msg = out + err
self.assertNotEqual(rc, 0, "%s should fail with no project" % name)
self.assertIn("no %s/ found" % issue.MARKER, msg, name)
self.assertIn(self.plain.root, msg, name)
def test_no_domain_script_ever_shows_a_traceback(self):
""""No project" is an ordinary answer, not a crash. A TypeError on a
None path is how an unresolved root announced itself while this was
being written — every entry point is swept, so a new one cannot
quietly reintroduce it."""
args = {"issue_ac.py": ["alpha-issue"],
"issue_evict.py": ["--dry-run"],
"issue_new.py": ["--type", "task", "--title", "Nowhere"]}
entries = [n for n in sorted(os.listdir(ISSUE_SCRIPTS))
if n.endswith(".py") and n not in ("issue.py", "issue_init.py")]
self.assertTrue(entries)
for name in entries:
rc, out, err = run(script("issue", name), *args.get(name, []),
cwd=self.plain.path("sub", "deeper"))
self.assertNotIn("Traceback", err, "%s crashed:\n%s" % (name, err))
self.assertNotEqual(rc, 0, name)
self.assertIn("no %s/ found" % issue.MARKER, out + err, name)
def test_a_writer_refuses_to_invent_a_project(self):
rc, out, err = run(script("issue", "issue_new.py"),
"--type", "task", "--title", "Nowhere to put this",
cwd=self.plain.path("sub", "deeper"))
self.assertNotEqual(rc, 0)
self.assertIn("no %s/ found" % issue.MARKER, out + err)
self.assertFalse(os.path.exists(self.plain.path(issue.MARKER)))
self.assertFalse(os.path.exists(self.plain.path("sub", "deeper",
issue.MARKER)))
# --------------------------------------------------------------------------
# missing is not empty
@@ -244,20 +404,20 @@ class TestSameFromAnywhere(unittest.TestCase):
class TestMissingVersusEmpty(unittest.TestCase):
def test_missing_store_says_missing(self):
repo = FakeRepo(with_store=False)
self.addCleanup(repo.cleanup)
project = FakeProject(with_store=False)
self.addCleanup(project.cleanup)
for name in ("issue_check.py", "issue_tree.py", "issue_index.py"):
rc, out, err = run(repo.script("issue", name), cwd=repo.root)
rc, out, err = run(script("issue", name), cwd=project.root)
msg = out + err
self.assertNotEqual(rc, 0, "%s should fail on a missing store" % name)
self.assertIn("does not exist", msg, name)
self.assertNotIn("is empty", msg, name)
def test_empty_store_says_empty(self):
repo = FakeRepo(issues=())
self.addCleanup(repo.cleanup)
project = FakeProject(issues=())
self.addCleanup(project.cleanup)
for name in ("issue_check.py", "issue_tree.py"):
rc, out, err = run(repo.script("issue", name), cwd=repo.root)
rc, out, err = run(script("issue", name), cwd=project.root)
msg = out + err
self.assertNotEqual(rc, 0, name)
self.assertIn("is empty", msg, name)
@@ -266,12 +426,12 @@ class TestMissingVersusEmpty(unittest.TestCase):
def test_index_of_an_empty_store_is_legitimate(self):
"""An existing store with nothing in it gets an index saying so. Only a
missing directory is an error."""
repo = FakeRepo(issues=())
self.addCleanup(repo.cleanup)
rc, out, err = run(repo.script("issue", "issue_index.py"), cwd=repo.root)
project = FakeProject(issues=())
self.addCleanup(project.cleanup)
rc, out, err = run(script("issue", "issue_index.py"), cwd=project.root)
self.assertEqual(rc, 0, err)
self.assertIn("0 issue(s)", out)
with open(os.path.join(repo.store, "INDEX.md")) as f:
with open(os.path.join(project.store, "INDEX.md")) as f:
self.assertIn("_empty_", f.read())
@@ -282,39 +442,41 @@ class TestMissingVersusEmpty(unittest.TestCase):
class TestNoSilentCreation(unittest.TestCase):
def setUp(self):
self.repo = FakeRepo(with_store=False)
self.addCleanup(self.repo.cleanup)
self.project = FakeProject(with_store=False)
self.addCleanup(self.project.cleanup)
def test_readers_and_the_indexer_create_nothing(self):
for d in (self.repo.root, self.repo.path("sub")):
for d in (self.project.root, self.project.path("sub")):
for name in ("issue_check.py", "issue_tree.py", "issue_index.py"):
run(self.repo.script("issue", name), cwd=d)
self.assertFalse(os.path.exists(self.repo.path("tmp")),
run(script("issue", name), cwd=d)
self.assertFalse(os.path.exists(self.project.store),
"the store was created by a read")
self.assertFalse(os.path.exists(self.repo.path("sub", "tmp")),
self.assertFalse(os.path.exists(self.project.path("sub", issue.MARKER)),
"a store was created relative to cwd")
def test_explicit_out_pointing_nowhere_is_an_error_not_a_mkdir(self):
target = self.repo.path("sub", "nowhere")
rc, out, err = run(self.repo.script("issue", "issue_index.py"),
"--out", target, cwd=self.repo.root)
target = self.project.path("sub", "nowhere")
rc, out, err = run(script("issue", "issue_index.py"),
"--out", target, cwd=self.project.root)
self.assertNotEqual(rc, 0)
self.assertIn("does not exist", out + err)
self.assertFalse(os.path.exists(target))
def test_issue_new_creates_the_store_and_says_so(self):
"""Creating the first issue in a fresh checkout must still work — but
out loud, and at the repo root, not below whatever cwd happens to be."""
rc, out, err = run(self.repo.script("issue", "issue_new.py"),
"""Creating the first issue in a fresh project must still work — but
out loud, and at the project root, not below whatever cwd happens to
be."""
rc, out, err = run(script("issue", "issue_new.py"),
"--type", "task", "--title", "Bootstrap the store",
cwd=self.repo.path("sub", "deeper"))
cwd=self.project.path("sub", "deeper"))
self.assertEqual(rc, 0, err)
self.assertIn("created store", err)
self.assertIn(self.repo.store, err)
self.assertIn(self.project.store, err)
self.assertTrue(os.path.isfile(
os.path.join(self.repo.store, "bootstrap-the-store.md")))
self.assertFalse(os.path.exists(self.repo.path("sub", "deeper", "tmp")),
"a store was created relative to cwd")
os.path.join(self.project.store, "bootstrap-the-store.md")))
self.assertFalse(
os.path.exists(self.project.path("sub", "deeper", issue.MARKER)),
"a store was created relative to cwd")
# --------------------------------------------------------------------------
@@ -324,39 +486,38 @@ class TestNoSilentCreation(unittest.TestCase):
class TestExplicitOutWins(unittest.TestCase):
def setUp(self):
self.repo = FakeRepo()
self.addCleanup(self.repo.cleanup)
self.project = FakeProject()
self.addCleanup(self.project.cleanup)
def test_absolute_out_is_honored(self):
other = self.repo.path("sub", "other-store")
other = self.project.path("sub", "other-store")
os.makedirs(other)
shutil.copy(os.path.join(self.repo.store, "alpha-issue.md"), other)
rc, out, err = run(self.repo.script("issue", "issue_check.py"),
"--out", other, cwd=self.repo.root)
shutil.copy(os.path.join(self.project.store, "alpha-issue.md"), other)
rc, out, err = run(script("issue", "issue_check.py"),
"--out", other, cwd=self.project.root)
self.assertEqual(rc, 0, err)
self.assertIn("1 issue(s) checked", out)
def test_relative_out_stays_relative_to_cwd(self):
"""`--out tmp/issues` typed from a subdirectory means that
subdirectory's tmp/issues — which is not there. Auto-resolution must
"""`--out .tea/issues` typed from a subdirectory means that
subdirectory's `.tea/issues` — which is not there. Auto-resolution must
not step in and "fix" what the operator typed."""
rc, out, err = run(self.repo.script("issue", "issue_check.py"),
"--out", os.path.join("tmp", "issues"),
cwd=self.repo.path("sub"))
rel = os.path.join(*issue.STORE_PARTS)
rc, out, err = run(script("issue", "issue_check.py"),
"--out", rel, cwd=self.project.path("sub"))
self.assertNotEqual(rc, 0)
self.assertIn("does not exist", out + err)
# the same relative path from the root does resolve, by cwd alone
rc, out, err = run(self.repo.script("issue", "issue_check.py"),
"--out", os.path.join("tmp", "issues"),
cwd=self.repo.root)
rc, out, err = run(script("issue", "issue_check.py"),
"--out", rel, cwd=self.project.root)
self.assertEqual(rc, 0, err)
self.assertIn("2 issue(s) checked", out)
def test_relative_out_can_climb(self):
rc, out, err = run(self.repo.script("issue", "issue_check.py"),
"--out", os.path.join("..", "tmp", "issues"),
cwd=self.repo.path("sub"))
rc, out, err = run(script("issue", "issue_check.py"),
"--out", os.path.join("..", *issue.STORE_PARTS),
cwd=self.project.path("sub"))
self.assertEqual(rc, 0, err)
self.assertIn("2 issue(s) checked", out)
@@ -368,33 +529,52 @@ class TestExplicitOutWins(unittest.TestCase):
class TestSyncLayerAgrees(unittest.TestCase):
def setUp(self):
self.repo = FakeRepo()
self.addCleanup(self.repo.cleanup)
self.project = FakeProject()
self.addCleanup(self.project.cleanup)
def _probe(self, layer, cwd):
"""Ask one layer, from `cwd`, which module defines the store and where
it lands. The sync scripts put the issue scripts on sys.path themselves
— `import map` is how they do it — so each layer is asked its own way.
"""
scripts = self.repo.path("skills", layer, "scripts")
scripts = SYNC_SCRIPTS if layer == "sync" else ISSUE_SCRIPTS
entry = "import map, issue" if layer == "sync" else "import issue"
code = ("import sys; sys.path.insert(0, %r)\n%s\n"
"print(issue.__file__)\nprint(issue.ISSUE_ROOT)\n") % (scripts, entry)
env = dict(os.environ)
env.pop("PYTHONPATH", None)
env.pop("CLAUDE_PROJECT_DIR", None)
p = subprocess.run([sys.executable, "-c", code], cwd=cwd, env=env,
capture_output=True, text=True)
self.assertEqual(p.returncode, 0, p.stderr)
return p.stdout.strip().splitlines()
def test_both_layers_resolve_the_same_store_from_anywhere(self):
for d in self.repo.everywhere():
for d in self.project.everywhere():
mod_i, root_i = self._probe("issue", d)
mod_s, root_s = self._probe("sync", d)
# sync does not redefine the store; it imports the domain module
self.assertEqual(os.path.realpath(mod_i), os.path.realpath(mod_s), d)
self.assertEqual(root_i, self.repo.store, d)
self.assertEqual(root_s, self.repo.store, d)
self.assertEqual(root_i, self.project.store, d)
self.assertEqual(root_s, self.project.store, d)
def test_the_payload_root_is_a_sibling_of_the_store(self):
"""One marker, one walk: the transport's scratchpad and the domain's
store cannot end up in different projects, and the scratchpad is never
inside the store."""
code = ("import sys; sys.path.insert(0, %r)\n"
"import _gitea\nprint(_gitea.PAYLOAD_ROOT)\n") % SYNC_SCRIPTS
env = dict(os.environ)
env.pop("PYTHONPATH", None)
env.pop("CLAUDE_PROJECT_DIR", None)
for d in self.project.everywhere():
p = subprocess.run([sys.executable, "-c", code], cwd=d, env=env,
capture_output=True, text=True)
self.assertEqual(p.returncode, 0, p.stderr)
payload = p.stdout.strip()
self.assertEqual(payload,
self.project.path(issue.MARKER, "payload"), d)
self.assertFalse(payload.startswith(self.project.store + os.sep), d)
def test_every_out_flag_defers_to_the_domain_layer(self):
"""Both layers agree by construction, not by coincidence: no script