naudachu
6d01ead245
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>
436 lines
17 KiB
Python
436 lines
17 KiB
Python
#!/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()
|