merge: resolve the issue store path independently of the working directory

This commit is contained in:
naudachu
2026-08-10 15:41:19 +05:00
13 changed files with 651 additions and 19 deletions
+27
View File
@@ -68,6 +68,25 @@ the domain layer, it is in the wrong place.
- `hooks/` — PreToolUse hooks: `tea-guard` blocks or rewrites `tea` invocations - `hooks/` — PreToolUse hooks: `tea-guard` blocks or rewrites `tea` invocations
that don't use the pinned login; `agents-sync` keeps every directory canonical that don't use the pinned login; `agents-sync` keeps every directory canonical
(`AGENTS.md` real file, `CLAUDE.md` symlink to it) (`AGENTS.md` real file, `CLAUDE.md` symlink to it)
- `tests/` — stdlib `unittest`, no third-party anything
## Tests
```bash
python3 -m unittest discover -s tests -v
```
Plain `unittest`; no pytest, no dependencies — the scripts under test are
stdlib-only and the tests hold the same line. `skills/*/scripts/` are not
packages, so a test that needs the domain module imports it with
`sys.path.insert`.
**A test never touches `tmp/issues/`.** Anything that needs a store builds a
throwaway repository in a `tempfile.TemporaryDirectory()` — a `.git` marker, a
copy of the script layers, fixture issues — and runs the real scripts inside it
as subprocesses. That is the only way to test behavior that depends on where a
script is run from, and it keeps the developer's own store out of the blast
radius.
## Local issue store ## Local issue store
@@ -75,6 +94,14 @@ the domain layer, it is in the wrong place.
markdown file per issue, named by its slug, with one metadata field per line so markdown file per issue, named by its slug, with one metadata field per line so
plain grep works without a parser. plain grep works without a parser.
- **The path is `<repo root>/tmp/issues`, resolved from `issue.py`'s own
location, not from cwd.** `issue.store_root()` walks up from `__file__` to the
nearest `.git` or `AGENTS.md` — so every script in both layers sees one store
whatever directory it is run from. An explicit `--out` overrides it and is
used exactly as typed; a relative `--out` stays relative to cwd.
- Nothing creates the store as a side effect of a write. Readers distinguish
"does not exist" from "is empty"; only `issue_new.py` and `pull.py` create it,
and they say so on stderr.
- Identity is the slug (`wire-sqlc-appclick.md`), never a tracker number. - Identity is the slug (`wire-sqlc-appclick.md`), never a tracker number.
Numbers live in the `gitea:` field. Numbers live in the `gitea:` field.
- `origin: local` is a durable state. An issue that never leaves this machine is - `origin: local` is a durable state. An issue that never leaves this machine is
+18
View File
@@ -47,6 +47,24 @@ tmp/issues/wire-sqlc.comments.md comment thread (written by /tea:sync only)
tmp/issues/tree-<id>.md saved graph (issue_tree.py --write) tmp/issues/tree-<id>.md saved graph (issue_tree.py --write)
``` ```
## Where the store is
`<repo root>/tmp/issues`**not** `tmp/issues` relative to wherever you are
standing. The scripts resolve it by walking up from their own file to the
nearest `.git` or `AGENTS.md`, so they all see one store no matter which
directory you run them from, and a `cd` earlier in the session changes nothing.
`--out` overrides that and is taken **literally**: an absolute path is used as
given, a relative one stays relative to the current directory. Nothing rewrites
what you typed.
Two things follow, and both are deliberate:
- A store that is not there reports `does not exist`; a store with no issues in
it reports `is empty`. They are different problems.
- No script conjures a store as a side effect of writing. Only `issue_new.py`
creates one — the first issue in a fresh checkout — and it says so on stderr.
## Reading: grep, don't parse ## Reading: grep, don't parse
Metadata is one field per line with inline lists precisely so plain `grep` Metadata is one field per line with inline lists precisely so plain `grep`
+111 -2
View File
@@ -48,7 +48,67 @@ without a parser:
import os import os
import re import re
ISSUE_ROOT = os.path.join("tmp", "issues") # --------------------------------------------------------------------------
# where the store lives
# --------------------------------------------------------------------------
# `<repo root>/tmp/issues`, absolute, resolved once at import.
#
# It used to be the relative `tmp/issues`, which made "the store" whatever
# directory the shell happened to be standing in. One `cd` — and a `cd` outlives
# the command that ran it — was enough for readers to report an empty store on a
# full one and for writers to quietly build a second store beside the first.
#
# The anchor is THIS FILE, not the working directory. A script's own location is
# a fact about the installation; cwd is a fact about the last `cd`. Walking up
# from __file__ therefore hands every script in both layers the same answer no
# matter where it is invoked from — including from inside tmp/issues itself.
#
# An explicit --out still wins over all of this, and is used exactly as typed: a
# relative --out stays relative to cwd, because that is what the operator asked
# for. There is no environment override; the store is where the repo is.
STORE_PARTS = ("tmp", "issues")
# `.git` is a directory in a normal clone and a FILE in a worktree — hence
# exists(), not isdir(). AGENTS.md is the fallback for a plugin copied out of
# git; the agents-sync hook only ever puts one at a repository root.
REPO_MARKERS = (".git", "AGENTS.md")
_HERE = os.path.dirname(os.path.abspath(__file__))
def repo_root(start):
"""Nearest ancestor of `start` (inclusive) carrying a repo marker, or None.
Markers, not a fixed number of `..` hops: how deep this file sits below the
root is an implementation detail of the repo layout, and the layout is not
a promise."""
d = os.path.abspath(start)
while True:
if any(os.path.exists(os.path.join(d, m)) for m in REPO_MARKERS):
return d
parent = os.path.dirname(d)
if parent == d:
return None
d = parent
def store_root(start=None):
"""Absolute path of the issue store.
`start` overrides the anchor and exists so the resolution can be exercised
against a scratch tree. When these scripts are not inside a repository at
all, cwd gets a turn; failing that the historical cwd-relative location
stands, made absolute so an error message can name the directory it really
looked in."""
for anchor in ([start] if start is not None else [_HERE, os.getcwd()]):
root = repo_root(anchor)
if root:
return os.path.join(root, *STORE_PARTS)
return os.path.abspath(os.path.join(*STORE_PARTS))
ISSUE_ROOT = store_root()
# Domain-owned metadata, in render order. Foreign keys render after these, # Domain-owned metadata, in render order. Foreign keys render after these,
# sorted, so the sync layer can add fields without touching this list. # sorted, so the sync layer can add fields without touching this list.
@@ -356,6 +416,55 @@ def validate(issue, known_ids=None):
# store # store
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
class StoreMissing(Exception):
"""The store directory is not there.
Deliberately a different answer from "the store is empty". One is a path
that does not exist, the other is a repository with no issues filed yet, and
conflating the two is exactly what made a missed directory look like an
empty backlog."""
def __init__(self, root):
self.root = root
Exception.__init__(self, "store %s does not exist" % root)
def store_exists(root):
return os.path.isdir(root)
def require_store(root):
"""Assert the store is there before reading or writing it."""
if not os.path.isdir(root):
raise StoreMissing(root)
return root
def create_store(root):
"""Create the store; True when it actually made the directory.
Only the commands that legitimately bootstrap a store call this — issue_new
and pull — and both announce it. Nothing creates a store as a side effect of
a write any more: a missing directory is something to report, not something
to conjure."""
if os.path.isdir(root):
return False
os.makedirs(root)
return True
def store_error(root):
"""Why `root` cannot be read as a store, or None when it holds issues.
The two messages are distinct on purpose — see StoreMissing."""
if not os.path.isdir(root):
return ("store %s does not exist — nothing was created; pass --out to "
"point elsewhere" % root)
if not all_ids(root):
return "store %s exists but is empty" % root
return None
def path_of(root, id): def path_of(root, id):
return os.path.join(root, "%s.md" % id) return os.path.join(root, "%s.md" % id)
@@ -377,7 +486,7 @@ def load_all(root):
def save(root, issue): def save(root, issue):
os.makedirs(root, exist_ok=True) require_store(root)
p = path_of(root, issue.id) p = path_of(root, issue.id)
with open(p, "w") as f: with open(p, "w") as f:
f.write(issue.to_text()) f.write(issue.to_text())
+6 -3
View File
@@ -26,16 +26,19 @@ def main():
ap.add_argument("ids", nargs="*", help="ids to check (default: all)") ap.add_argument("ids", nargs="*", help="ids to check (default: all)")
ap.add_argument("--quiet", action="store_true", help="exit code only") ap.add_argument("--quiet", action="store_true", help="exit code only")
ap.add_argument("--strict", action="store_true", help="treat warnings as errors") ap.add_argument("--strict", action="store_true", help="treat warnings as errors")
ap.add_argument("--out", default=issue.ISSUE_ROOT, help="store root (default: tmp/issues)") ap.add_argument("--out", default=issue.ISSUE_ROOT,
help="store root (default: <repo>/tmp/issues)")
args = ap.parse_args() args = ap.parse_args()
problem = issue.store_error(args.out)
if problem:
sys.exit("issue_check.py: %s" % problem)
issues = issue.load_all(args.out) issues = issue.load_all(args.out)
ids = args.ids or sorted(issues) ids = args.ids or sorted(issues)
for i in ids: for i in ids:
if i not in issues: if i not in issues:
sys.exit("issue_check.py: no issue %r in %s" % (i, args.out)) sys.exit("issue_check.py: no issue %r in %s" % (i, args.out))
if not ids:
sys.exit("issue_check.py: store %s is empty" % args.out)
known = set(issues) known = set(issues)
bad = 0 bad = 0
+17 -4
View File
@@ -7,8 +7,12 @@ the index acknowledges that a tracker exists: `local` means the issue has never
left this machine, `gitea` means the sync layer has pushed or pulled it. Both left this machine, `gitea` means the sync layer has pushed or pulled it. Both
are ordinary issues here. are ordinary issues here.
The store is <repo root>/tmp/issues unless --out says otherwise; an existing
store with nothing in it gets an "_empty_" table, a store that is not there is
an error rather than a directory to create.
Usage: Usage:
issue_index.py [--out tmp/issues] issue_index.py [--out DIR]
""" """
import argparse import argparse
import os import os
@@ -27,6 +31,9 @@ def cell(v):
def build(root): def build(root):
# An index of a store that is not there is not an empty index, it is a bad
# path. Raising beats writing INDEX.md into a directory nobody asked for.
issue.require_store(root)
issues = issue.load_all(root) issues = issue.load_all(root)
rows = [] rows = []
for i in sorted(issues): for i in sorted(issues):
@@ -71,7 +78,6 @@ def build(root):
out.append("") out.append("")
path = os.path.join(root, "INDEX.md") path = os.path.join(root, "INDEX.md")
os.makedirs(root, exist_ok=True)
with open(path, "w") as f: with open(path, "w") as f:
f.write("\n".join(out)) f.write("\n".join(out))
return path, len(rows) return path, len(rows)
@@ -79,9 +85,16 @@ def build(root):
def main(): def main():
ap = argparse.ArgumentParser(description="Rebuild the local issue index (offline)") ap = argparse.ArgumentParser(description="Rebuild the local issue index (offline)")
ap.add_argument("--out", default=issue.ISSUE_ROOT, help="store root (default: tmp/issues)") ap.add_argument("--out", default=issue.ISSUE_ROOT,
help="store root (default: <repo>/tmp/issues)")
args = ap.parse_args() args = ap.parse_args()
path, n = build(args.out) # An existing store with nothing in it is a legitimate thing to index — it
# gets an "_empty_" table. A store that is not there is not.
try:
path, n = build(args.out)
except issue.StoreMissing as e:
sys.exit("issue_index.py: %s — nothing was created; create an issue with "
"issue_new.py, or pass --out" % e)
print("%s%d issue(s)" % (path, n)) print("%s%d issue(s)" % (path, n))
+7 -1
View File
@@ -151,7 +151,8 @@ def main():
ap.add_argument("--assignee", action="append", default=[], help="assignee; repeat") ap.add_argument("--assignee", action="append", default=[], help="assignee; repeat")
ap.add_argument("--depends", action="append", default=[], ap.add_argument("--depends", action="append", default=[],
help="id this issue depends on; repeat") help="id this issue depends on; repeat")
ap.add_argument("--out", default=issue.ISSUE_ROOT, help="store root (default: tmp/issues)") ap.add_argument("--out", default=issue.ISSUE_ROOT,
help="store root (default: <repo>/tmp/issues)")
args = ap.parse_args() args = ap.parse_args()
labels = ["type/%s" % args.type] labels = ["type/%s" % args.type]
@@ -177,6 +178,11 @@ def main():
labels=labels, assignees=args.assignee, milestone=args.milestone, labels=labels, assignees=args.assignee, milestone=args.milestone,
depends=args.depends) depends=args.depends)
# The first issue in a fresh checkout has to create the store, but it says
# so — and it says where, because the path is absolute.
if issue.create_store(args.out):
sys.stderr.write("created store %s\n" % os.path.abspath(args.out))
path = issue.save(args.out, iss) path = issue.save(args.out, iss)
issue_index.build(args.out) issue_index.build(args.out)
print("%s [type/%s] %s" % (path, args.type, args.title)) print("%s [type/%s] %s" % (path, args.type, args.title))
+6 -3
View File
@@ -66,12 +66,15 @@ def main():
ap.add_argument("--depth", type=int, default=6, help="max depth (default: 6)") ap.add_argument("--depth", type=int, default=6, help="max depth (default: 6)")
ap.add_argument("--write", action="store_true", ap.add_argument("--write", action="store_true",
help="also write tmp/issues/tree-<slug>.md") help="also write tmp/issues/tree-<slug>.md")
ap.add_argument("--out", default=issue.ISSUE_ROOT, help="store root (default: tmp/issues)") ap.add_argument("--out", default=issue.ISSUE_ROOT,
help="store root (default: <repo>/tmp/issues)")
args = ap.parse_args() args = ap.parse_args()
problem = issue.store_error(args.out)
if problem:
sys.exit("issue_tree.py: %s" % problem)
issues = issue.load_all(args.out) issues = issue.load_all(args.out)
if not issues:
sys.exit("issue_tree.py: store %s is empty" % args.out)
edges = issue.graph(issues) edges = issue.graph(issues)
roots = args.ids roots = args.ids
+6
View File
@@ -48,6 +48,12 @@ Key forms for `<key>`: `42`, `#42`, `owner/repo#42`, or a full issue URL. Repo
defaults to the current directory's git remote; add `--repo owner/repo` outside defaults to the current directory's git remote; add `--repo owner/repo` outside
one. one.
`--out` defaults to `issue.ISSUE_ROOT` on every one of them — the domain layer's
`<repo root>/tmp/issues`, resolved from the scripts' own location rather than
cwd. Both layers therefore address the same store by construction, from any
directory. Pass `--out` to override; a relative one stays relative to cwd. Only
`pull.py` will create a missing store, and it says so on stderr.
## Identity mapping ## Identity mapping
The local id is a slug; Gitea's is a number. The pair is recorded in the issue The local id is a slug; Gitea's is a number. The pair is recorded in the issue
+4 -1
View File
@@ -43,10 +43,13 @@ def main():
ap.add_argument("--edit", type=int, metavar="COMMENT_ID", ap.add_argument("--edit", type=int, metavar="COMMENT_ID",
help="PATCH an existing comment instead of posting a new one") help="PATCH an existing comment instead of posting a new one")
ap.add_argument("--repo", help="owner/repo (default: auto-detect from CWD git remote)") ap.add_argument("--repo", help="owner/repo (default: auto-detect from CWD git remote)")
ap.add_argument("--out", default=issue.ISSUE_ROOT, help="store root (default: tmp/issues)") ap.add_argument("--out", default=issue.ISSUE_ROOT,
help="store root (default: <repo>/tmp/issues)")
args = ap.parse_args() args = ap.parse_args()
root = args.out root = args.out
if not issue.store_exists(root):
_gitea.die("store %s does not exist — nothing was created" % root)
if not os.path.isfile(issue.path_of(root, args.id)): if not os.path.isfile(issue.path_of(root, args.id)):
_gitea.die("no issue %r in %s" % (args.id, root)) _gitea.die("no issue %r in %s" % (args.id, root))
iss = issue.load(root, args.id) iss = issue.load(root, args.id)
+7 -1
View File
@@ -108,7 +108,8 @@ def main():
ap.add_argument("--cached", action="store_true", ap.add_argument("--cached", action="store_true",
help="skip issues already on disk instead of refetching") help="skip issues already on disk instead of refetching")
ap.add_argument("--repo", help="owner/repo (default: auto-detect from CWD git remote)") ap.add_argument("--repo", help="owner/repo (default: auto-detect from CWD git remote)")
ap.add_argument("--out", default=issue.ISSUE_ROOT, help="store root (default: tmp/issues)") ap.add_argument("--out", default=issue.ISSUE_ROOT,
help="store root (default: <repo>/tmp/issues)")
args = ap.parse_args() args = ap.parse_args()
filtered = bool(args.milestone or args.label or args.query) filtered = bool(args.milestone or args.label or args.query)
@@ -118,6 +119,11 @@ def main():
_gitea.die("nothing to pull: pass issue keys, or --milestone / --label / -q") _gitea.die("nothing to pull: pass issue keys, or --milestone / --label / -q")
root = args.out root = args.out
# A first pull into a fresh checkout has to create the store; it says so,
# and the path is absolute, so it cannot be a stray cwd.
if issue.create_store(root):
sys.stderr.write("created store %s\n" % os.path.abspath(root))
login = _gitea.require_login() login = _gitea.require_login()
# ---- which repo ------------------------------------------------------ # ---- which repo ------------------------------------------------------
+5 -3
View File
@@ -90,13 +90,15 @@ def main():
ap.add_argument("--dry-run", action="store_true", help="validate only, no network") ap.add_argument("--dry-run", action="store_true", help="validate only, no network")
ap.add_argument("--force", action="store_true", help="push despite format violations") ap.add_argument("--force", action="store_true", help="push despite format violations")
ap.add_argument("--repo", help="owner/repo (default: auto-detect from CWD git remote)") ap.add_argument("--repo", help="owner/repo (default: auto-detect from CWD git remote)")
ap.add_argument("--out", default=issue.ISSUE_ROOT, help="store root (default: tmp/issues)") ap.add_argument("--out", default=issue.ISSUE_ROOT,
help="store root (default: <repo>/tmp/issues)")
args = ap.parse_args() args = ap.parse_args()
root = args.out root = args.out
problem = issue.store_error(root)
if problem:
_gitea.die("%s — create an issue with issue_new.py first" % problem)
issues = issue.load_all(root) issues = issue.load_all(root)
if not issues:
_gitea.die("store %s is empty — create an issue with issue_new.py first" % root)
chosen = select(issues, args.ids, args.update) chosen = select(issues, args.ids, args.update)
+2 -1
View File
@@ -39,7 +39,8 @@ def main():
ap.add_argument("--milestone", help="milestone id or title") ap.add_argument("--milestone", help="milestone id or title")
ap.add_argument("--limit", type=int, default=30) ap.add_argument("--limit", type=int, default=30)
ap.add_argument("--repo", help="owner/repo (default: auto-detect from CWD git remote)") ap.add_argument("--repo", help="owner/repo (default: auto-detect from CWD git remote)")
ap.add_argument("--out", default=issue.ISSUE_ROOT, help="store root (default: tmp/issues)") ap.add_argument("--out", default=issue.ISSUE_ROOT,
help="store root (default: <repo>/tmp/issues)")
args = ap.parse_args() args = ap.parse_args()
login = _gitea.require_login() login = _gitea.require_login()
+435
View File
@@ -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()