fix: resolve the issue store path independently of the working directory
ISSUE_ROOT was the relative `tmp/issues`, so "the store" was whatever
directory the shell happened to be standing in. It is the --out default
in all eight scripts of both layers, which made one `cd` — and a `cd`
outlives the command that ran it — enough for readers to report an empty
store on a full one and for writers to quietly build a second store
beside the first. `issue_index.py` run from inside tmp/issues left
tmp/issues/tmp/issues/ behind and exited 0.
The anchor is issue.py's own __file__, not cwd. A script's location is a
fact about the installation; cwd is a fact about the last `cd`, and the
scripts are invoked by path from wherever the agent happens to be. From
there `store_root()` walks up to the nearest repo marker — `.git`
(exists(), not isdir(): a worktree's .git is a file) or AGENTS.md for a
copy taken out of git — and joins tmp/issues. Markers rather than a
fixed number of `..` hops, because the layout is not a promise. cwd is
tried only if the scripts are not inside a repository at all.
The function lives in the domain layer and skills/sync imports it, so
both layers agree by construction — the direction the layering rule
allows. skills/issue stays stdlib-only.
An explicit --out still wins and is used exactly as typed: a relative
--out stays relative to cwd, because that is what the operator asked
for. No new environment surface.
Two consequences the issue also asked for:
- Missing is no longer reported as empty. `store_error()` returns one
message for a path that is not there and another for a store with no
issues in it.
- Nothing conjures a store as a side effect of a write. save() and
issue_index.build() require it instead of os.makedirs'ing it; only
issue_new.py and pull.py create one, and both say so on stderr.
Establishes tests/ — plain stdlib unittest, no pytest, no dependencies.
The store tests build a throwaway repo in a TemporaryDirectory (a .git
marker, a copy of both script layers, fixture issues) and run the real
scripts inside it as subprocesses from five different working
directories; tmp/issues/ is never touched. Against the pre-fix scripts
15 of the 21 fail, reproducing the report exactly — five stray stores,
including tmp/issues/tmp/issues.
python3 -m unittest discover -s tests -v
Closes claude-skills/tea#15
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
naudachu
@@ -0,0 +1,435 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Where the issue store is, and that the answer does not depend on cwd.
|
||||
|
||||
python3 -m unittest discover -s tests -v
|
||||
|
||||
Stdlib unittest, no third-party anything — the same rule the scripts under test
|
||||
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
|
||||
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.
|
||||
"""
|
||||
import os
|
||||
import shutil
|
||||
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")
|
||||
SYNC_SCRIPTS = os.path.join(REPO, "skills", "sync", "scripts")
|
||||
|
||||
sys.path.insert(0, ISSUE_SCRIPTS)
|
||||
import issue # noqa: E402
|
||||
|
||||
|
||||
ALPHA = """\
|
||||
---
|
||||
id: alpha-issue
|
||||
state: open
|
||||
labels: [type/task]
|
||||
assignees: []
|
||||
milestone: none
|
||||
depends: []
|
||||
origin: local
|
||||
---
|
||||
# Alpha issue
|
||||
|
||||
## Summary
|
||||
Первый issue фикстуры.
|
||||
|
||||
## Spec
|
||||
none
|
||||
|
||||
## Motivation
|
||||
Нужен, чтобы в store что-то лежало.
|
||||
|
||||
## Acceptance criteria
|
||||
- [ ] проверяемое условие
|
||||
"""
|
||||
|
||||
BETA = """\
|
||||
---
|
||||
id: beta-issue
|
||||
state: open
|
||||
labels: [type/task]
|
||||
assignees: []
|
||||
milestone: none
|
||||
depends: [alpha-issue]
|
||||
origin: local
|
||||
---
|
||||
# Beta issue
|
||||
|
||||
## Summary
|
||||
Второй issue фикстуры, зависит от первого.
|
||||
|
||||
## Spec
|
||||
none
|
||||
|
||||
## Depends on
|
||||
- alpha-issue
|
||||
|
||||
## Motivation
|
||||
Нужен, чтобы у графа было ребро.
|
||||
|
||||
## Acceptance criteria
|
||||
- [ ] проверяемое условие
|
||||
"""
|
||||
|
||||
|
||||
def run(script, *args, **kw):
|
||||
"""Run one of the plugin's scripts and return (rc, stdout, stderr)."""
|
||||
cwd = kw.pop("cwd")
|
||||
env = dict(os.environ)
|
||||
env.pop("PYTHONPATH", None) # no leakage from the harness into the child
|
||||
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.
|
||||
|
||||
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)):
|
||||
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)
|
||||
os.makedirs(self.path("sub", "deeper"))
|
||||
|
||||
if with_store:
|
||||
os.makedirs(self.store)
|
||||
for text in issues:
|
||||
id = text.split("id: ", 1)[1].split("\n", 1)[0]
|
||||
with open(os.path.join(self.store, "%s.md" % id), "w") as f:
|
||||
f.write(text)
|
||||
|
||||
def cleanup(self):
|
||||
self._tmp.cleanup()
|
||||
|
||||
def path(self, *parts):
|
||||
return os.path.join(self.root, *parts)
|
||||
|
||||
@property
|
||||
def store(self):
|
||||
return self.path("tmp", "issues")
|
||||
|
||||
def script(self, layer, name):
|
||||
return self.path("skills", layer, "scripts", name)
|
||||
|
||||
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."""
|
||||
return [self.root, self.path("sub"), self.path("sub", "deeper"),
|
||||
self.path("skills", "issue", "scripts"), self.store]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# resolution, in isolation
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TestResolution(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.repo = FakeRepo()
|
||||
self.addCleanup(self.repo.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_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_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)
|
||||
|
||||
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_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"))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# the acceptance criterion: same answer from any subdirectory
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TestSameFromAnywhere(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.repo = FakeRepo()
|
||||
self.addCleanup(self.repo.cleanup)
|
||||
|
||||
def assertSameEverywhere(self, layer, name, *args):
|
||||
"""Run the script from the repo 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]))
|
||||
for d in dirs[1:]:
|
||||
self.assertEqual(run(self.repo.script(layer, name), *args, cwd=d), base,
|
||||
"%s disagrees when run from %s" % (name, d))
|
||||
return base
|
||||
|
||||
def test_issue_check(self):
|
||||
rc, out, _ = self.assertSameEverywhere("issue", "issue_check.py")
|
||||
self.assertIn("ok alpha-issue", out)
|
||||
self.assertIn("2 issue(s) checked, 0 with errors", out)
|
||||
|
||||
def test_issue_tree(self):
|
||||
_, out, _ = self.assertSameEverywhere("issue", "issue_tree.py")
|
||||
self.assertIn("beta-issue", out)
|
||||
self.assertIn("alpha-issue", out)
|
||||
|
||||
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)
|
||||
|
||||
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():
|
||||
for name in ("issue_index.py", "issue_check.py", "issue_tree.py"):
|
||||
run(self.repo.script("issue", name), cwd=d)
|
||||
|
||||
found = []
|
||||
for dirpath, dirnames, filenames in os.walk(self.repo.root):
|
||||
if "__pycache__" in dirnames:
|
||||
dirnames.remove("__pycache__")
|
||||
if "INDEX.md" in filenames:
|
||||
found.append(dirpath)
|
||||
self.assertEqual(found, [self.repo.store],
|
||||
"a second store appeared: %s" % found)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# missing is not empty
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TestMissingVersusEmpty(unittest.TestCase):
|
||||
|
||||
def test_missing_store_says_missing(self):
|
||||
repo = FakeRepo(with_store=False)
|
||||
self.addCleanup(repo.cleanup)
|
||||
for name in ("issue_check.py", "issue_tree.py", "issue_index.py"):
|
||||
rc, out, err = run(repo.script("issue", name), cwd=repo.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)
|
||||
for name in ("issue_check.py", "issue_tree.py"):
|
||||
rc, out, err = run(repo.script("issue", name), cwd=repo.root)
|
||||
msg = out + err
|
||||
self.assertNotEqual(rc, 0, name)
|
||||
self.assertIn("is empty", msg, name)
|
||||
self.assertNotIn("does not exist", msg, name)
|
||||
|
||||
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)
|
||||
self.assertEqual(rc, 0, err)
|
||||
self.assertIn("0 issue(s)", out)
|
||||
with open(os.path.join(repo.store, "INDEX.md")) as f:
|
||||
self.assertIn("_empty_", f.read())
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# nothing conjures a store
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TestNoSilentCreation(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.repo = FakeRepo(with_store=False)
|
||||
self.addCleanup(self.repo.cleanup)
|
||||
|
||||
def test_readers_and_the_indexer_create_nothing(self):
|
||||
for d in (self.repo.root, self.repo.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")),
|
||||
"the store was created by a read")
|
||||
self.assertFalse(os.path.exists(self.repo.path("sub", "tmp")),
|
||||
"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)
|
||||
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"),
|
||||
"--type", "task", "--title", "Bootstrap the store",
|
||||
cwd=self.repo.path("sub", "deeper"))
|
||||
self.assertEqual(rc, 0, err)
|
||||
self.assertIn("created store", err)
|
||||
self.assertIn(self.repo.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")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# an explicit --out is the operator's, not ours to rewrite
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TestExplicitOutWins(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.repo = FakeRepo()
|
||||
self.addCleanup(self.repo.cleanup)
|
||||
|
||||
def test_absolute_out_is_honored(self):
|
||||
other = self.repo.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)
|
||||
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
|
||||
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"))
|
||||
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)
|
||||
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"))
|
||||
self.assertEqual(rc, 0, err)
|
||||
self.assertIn("2 issue(s) checked", out)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# both layers, one root
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TestSyncLayerAgrees(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.repo = FakeRepo()
|
||||
self.addCleanup(self.repo.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")
|
||||
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)
|
||||
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():
|
||||
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)
|
||||
|
||||
def test_every_out_flag_defers_to_the_domain_layer(self):
|
||||
"""Both layers agree by construction, not by coincidence: no script
|
||||
spells the default out for itself."""
|
||||
for layer, names in (("issue", ("issue_new.py", "issue_check.py",
|
||||
"issue_tree.py", "issue_index.py")),
|
||||
("sync", ("pull.py", "push.py", "remote.py",
|
||||
"comment.py"))):
|
||||
for name in names:
|
||||
with open(os.path.join(REPO, "skills", layer, "scripts", name)) as f:
|
||||
src = f.read()
|
||||
self.assertIn('"--out", default=issue.ISSUE_ROOT', src,
|
||||
"%s/%s does not take its --out default from the "
|
||||
"domain layer" % (layer, name))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# the layering rule, mechanically
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TestLayering(unittest.TestCase):
|
||||
|
||||
def test_domain_layer_is_stdlib_only(self):
|
||||
"""skills/issue must keep working with skills/sync deleted — so no
|
||||
transport, and above all no subprocess, in the domain layer."""
|
||||
imported = set()
|
||||
for name in sorted(os.listdir(ISSUE_SCRIPTS)):
|
||||
if not name.endswith(".py"):
|
||||
continue
|
||||
with open(os.path.join(ISSUE_SCRIPTS, name)) as f:
|
||||
for line in f:
|
||||
if line.startswith(("import ", "from ")):
|
||||
imported.add(line.split()[1].split(".")[0])
|
||||
local = {"issue", "issue_index"}
|
||||
self.assertEqual(imported - local, {"argparse", "os", "re", "sys"},
|
||||
"non-stdlib or unexpected import in the domain layer")
|
||||
self.assertNotIn("subprocess", imported)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user