claude-skills/marketplace
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

merge: close issues through a script

Browse Source
This commit was merged in pull request #25.
This commit is contained in:
claude
2026-08-10 13:29:36 +00:00
parent 74a0e3b173 f5977fa4fc
commit 17567cd6a2
6 changed files with 960 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS.md
+1
View File
@@ -74,6 +74,7 @@ If a tracker concept (issue number, login, HTTP call, label color, `sub_url`,
- `scripts/_gitea.py` — transport: `tea api`, pagination, filters, label ids,
the remote-id map, `tmp/payload/`; the login comes from `auth/pin.py`
- `scripts/pull.py`, `push.py`, `remote.py`, `comment.py`
- `scripts/close.py` — the state field, both ways; explicit ids only
- `scripts/labels.py` — put the canonical `type/*` and `severity/*` set into a
repository; reads the domain taxonomy, never the store
- `skills/page` — a discussion's artifacts as a page tree (`/tea:page`),
agents/tea-runner.md
+12 -5
View File
@@ -29,7 +29,8 @@ to fill the gap yourself.
Load the skill, do not remember the flags:
- `/tea:sync` — `pull.py`, `push.py`, `comment.py`, `remote.py`, `labels.py`
- `/tea:sync` — `pull.py`, `push.py`, `comment.py`, `close.py`, `remote.py`,
`labels.py`
- `/tea:issue` — `issue_check.py`, `issue_tree.py`, `issue_index.py`,
`issue_new.py`, `issue_ac.py`
- `/tea:wiki` — `wiki_ls.py`, `wiki_pull.py`, `wiki_push.py`
@@ -65,10 +66,15 @@ instead of trying it.
`wiki_push.py` needs `-m`; use the caller's words, never your own summary.
Report the number and URL `push.py` printed; that is now the only address
the issue has.
4. **Do not close, delete, or retitle anything** on either side. On the wiki
that means no `--retitle`: renaming a published page abandons the old one.
The one deletion you may cause is push's own, on the issue you were told to
push.
4. **Close only the ids the caller named.** Closing is a script now
(`close.py`), so it is yours to run — under the same discipline as push: the
ids the caller named, and no others. Never widen the set, never infer that
an issue is finished because its checkboxes are ticked or its branch is
merged; whether work is done is a judgement about content, and content is
never yours. `--reopen` is the same rule backwards. **Deleting and
retitling stay forbidden** on both sides — on the wiki that means no
`--retitle`, since renaming a published page abandons the old one. The one
deletion you may cause is push's own, on the issue you were told to push.
5. **One retry, maximum.** A command that fails twice is a finding. Do not
permute flags looking for one that works.
6. **No payload dumps.** Never run `tea issues -o json`, never `cat` a pulled
@@ -122,3 +128,4 @@ Report these and halt; none of them is yours to resolve.
| a dependency is still `origin: local` | name the id; the caller decides whether to push it |
| a milestone or label does not exist in the repo | the script prints the real ones — pass that list through |
| a script asks for a decision (type, label, `--force`) | `blocked:` with the question |
| `close.py` is refused by Gitea because the issue is still blocked | the tracker's own line, and the blocker's number; the caller decides |
skills/issue/SKILL.md
+3 -1
View File
@@ -117,7 +117,9 @@ the table. Checkboxes are the exception — use `issue_ac.py`, below.
If the issue is synced (`origin: gitea`), the file is a working copy: your edit
is local until you run `push.py --update` from `/tea:sync`, and that push
**deletes the file** once Gitea has it. Nothing tracks drift, and with one copy
**deletes the file** once Gitea has it. Closing one of those is `close.py` from
`/tea:sync` — it moves the state on both sides in a single run; editing
`state:` here alone would only ever tell this machine. Nothing tracks drift, and with one copy
at a time there is little to track — a file that is still here has not been
pushed. Get it back with `pull.py <n>`; the slug does not change.
skills/sync/SKILL.md
+50 -1
View File
@@ -1,6 +1,6 @@
---
name: sync
description: Move issues between the local store and Gitea — pull issues into tmp/issues/, push local issues up, post comments. Load when the user asks to fetch/read a Gitea issue, publish an issue, list what exists in the tracker, or comment on one. Working with an issue's content (writing, grepping, validating, dependency graph) is /tea:issue and needs no network.
description: Move issues between the local store and Gitea — pull issues into tmp/issues/, push local issues up, post comments, close and reopen them. Load when the user asks to fetch/read a Gitea issue, publish an issue, list what exists in the tracker, comment on one, or close/reopen one. Working with an issue's content (writing, grepping, validating, dependency graph) is /tea:issue and needs no network.
---
# /tea:sync — the bridge between the local store and Gitea
@@ -46,6 +46,7 @@ there is nothing to pin a second time. No pin anywhere → exit with a pointer t
| `pull.py <key…>` or `pull.py --milestone M \| --label L \| -q TEXT [--limit N]` | Gitea → `tmp/issues/<id>.md`, plus `<id>.comments.md` when the thread is not empty; follows dependencies by default (`--no-deps` to stop); `--limit` caps what is **stored** (default 100) |
| `push.py [id…] [--update] [--dry-run]` | local → Gitea; validates first, **deletes the local file on success** and prints where it lives now |
| `comment.py <id> --file F \| --body TEXT [--edit N]` | post or edit a comment, then refetch the thread |
| `close.py <id…> [--reopen] [--dry-run]` | set `state` in Gitea and in the local copy with it; explicit ids only, no bulk filter |
| `labels.py [--dry-run] [--fix]` | bootstrap the canonical `type/*` + `severity/*` set in a repo; exact names left alone, lookalikes reported, drift fixed only with `--fix` |
| `map.py`, `_gitea.py` | the two layers the commands import — not commands |
@@ -365,6 +366,54 @@ where the pin resolves from: the old workaround for the pin — run the scripts
with cwd in the main checkout — sent the main checkout's branch as `ref`, which
is the one thing `branch:` exists to record.
## Closing and reopening
```bash
python3 <skill-base-dir>/scripts/close.py wire-sqlc-appclick # by slug
python3 <skill-base-dir>/scripts/close.py 42 '#43' # by number
python3 <skill-base-dir>/scripts/close.py --reopen 42
python3 <skill-base-dir>/scripts/close.py --dry-run 42 43 # no request at all
```
`close.py` is the only supported way to move `state:`. Never hand-roll
`tea api -X PATCH -d '{"state":"closed"}' repos/OWNER/REPO/issues/N`: it spells
out the owner, the repo and the request body — the three things this layer
exists to hide — and it needs a `Bash(tea api *)` permission that also covers
`-X DELETE` on the repository.
**State only.** The payload is `{"state": …}` and nothing else — no title, no
body, no labels, no milestone. Closing is not an edit; editing is `pull.py` →
change → `push.py --update`.
**Explicit ids only.** There is no `--milestone` and no `--label`: which issues
are finished is a judgement about content, and this script only carries one
out, one named id at a time. Deleting an issue is out of scope too — Gitea can,
and it is not an operation of this workflow.
What may be named, and what happens to the local copy:
| named | resolved through | local file |
|---|---|---|
| a slug with a file on disk | its `gitea:` field | `state:` rewritten, `synced:` refreshed |
| a slug whose file push dropped | `.remote.json` | none to write — say so and move on |
| `42`, `#42`, `owner/repo#42`, a URL | the key itself; the ledger supplies the slug | rewritten when a file of that slug is there |
| a slug with `origin: local` | — | **refused**: it is not in the tracker, and the error names the id |
The local file is written only after the tracker has confirmed *this* write: an
object carrying the very number that was PATCHed, in the state that was asked
for. A non-2xx, a `tea` that would not run, an answer for another issue, a 200
that still says `open` — the run stops and the file is byte for byte what it
was. `--dry-run` prints the same lines and makes no request at all, so it needs
no pinned login.
Gitea refuses to close an issue that its own dependency graph still blocks. The
refusal arrives as a non-2xx with the tracker's own words: close the blockers
first, or unlink them in the web UI.
The index is rebuilt when at least one local file changed, so `INDEX.md` never
outlives the state it reports. Nothing is deleted here — unlike a push, a close
leaves the working copy where it is.
## What crosses the boundary, and what does not
| domain | Gitea | note |
skills/sync/scripts/close.py
+253
View File
@@ -0,0 +1,253 @@
#!/usr/bin/env python3
"""
close.py — change an issue's state in Gitea, and in the local copy with it.
The one regular tracker operation that used to have no script: closing. Without
it the only way to move `state:` was a raw `tea api -X PATCH -d '{"state":
"closed"}' repos/OWNER/REPO/issues/N`, which spells out the owner, the repo and
the request body — the three things `_gitea.py` exists to hide — and which needs
`Bash(tea api *)`, a permission that also covers `-X DELETE` on the repository.
close.py wire-sqlc-appclick one issue, by slug
close.py wire-sqlc-appclick 42 #43 several, by slug or number
close.py --reopen 42 the same thing backwards
close.py --dry-run 42 43 what would happen, no request at all
STATE ONLY. This script sends `{"state": …}` and nothing else: no title, no
body, no labels, no milestone. Editing an issue is `pull.py` -> edit ->
`push.py --update`; closing it is not an edit.
**What may be named.** A local slug, or a Gitea key (`42`, `#42`,
`owner/repo#42`, an issue URL) — the same forms `pull.py` takes. Both are
needed, and for the same reason: a push deletes the local file, so most issues
in the tracker have no slug on disk to name them by. A slug is resolved through
the file's `gitea:` field when the file is there, and through the ledger
(`.remote.json`) when push has already dropped it.
**An `origin: local` issue cannot be closed.** It is not in the tracker, so
there is nothing to close there, and the run stops naming the id rather than
quietly editing one field of a local file. Delete it, or push it first.
**Explicit ids only.** No `--milestone`, no `--label`, no "close everything
that looks done". Which issues are finished is a judgement about content; this
script only carries it out, one named id at a time. Nothing here deletes an
issue either — Gitea can, and it is not an operation of this workflow.
The local file is written only after the tracker has confirmed the write:
1. `tea` ran and exited 0 (a non-2xx exits the run inside `_gitea.api`), and
2. the answer is an object carrying the very number that was PATCHed, and
3. its `state` is the state we asked for.
Anything else and the file is left exactly as it was — see `confirmed`. An
issue whose local copy is gone (pushed and dropped) is closed in Gitea and
nothing is written; the state comes down with the next `pull.py`.
Gitea refuses to close an issue that its own dependency graph still blocks. That
refusal arrives as a non-2xx and stops the run with the tracker's own words:
close the blockers first, or unlink them in the web UI.
Login: the operator's pin from .claude/settings.local.json (see /tea:auth).
"""
import argparse
import os
import re
import sys
_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path[:0] = [_HERE, os.path.normpath(os.path.join(_HERE, "..", "..", "issue", "scripts"))]
import _gitea # noqa: E402
import issue # noqa: E402
import issue_index # noqa: E402
import map as gmap # noqa: E402
# What `_gitea.parse_key` accepts, asked as a question instead of an assertion:
# parse_key exits on anything it cannot read, and here "not a key" is the normal
# case — it means the argument is a slug. A slug never contains `#`, `/` or `:`,
# so the two vocabularies cannot collide.
KEY_RE = re.compile(r'^(#?\d+|[\w.-]+/[\w.-]+#\d+|https?://\S+)$')
def looks_like_key(arg):
return bool(KEY_RE.match((arg or "").strip()))
def ledger_pairs(remote_map, repo=None):
"""[(repo, number, slug)] from `.remote.json`, filtered to `repo`.
A `--repo` that was not given means "whatever the ledger holds": resolving
the repo's real name costs a request, and a dry run is required to make
none. The ambiguity that opens — one number under two repos — is caught at
lookup time rather than papered over."""
out = []
for key, slug in sorted(remote_map.items()):
r, n = gmap.parse_remote_key(key)
if n:
if repo is None or r == repo:
out.append((r, n, slug))
return out
def one(candidates, what, arg):
"""The single `(repo, value)` in `candidates`, None when empty, or exit.
Two answers mean the ledger knows this number (or this slug) under more than
one repository, and only `--repo` can settle that."""
got = sorted(set(candidates))
if len(got) > 1:
_gitea.die("%r matches %s under more than one repo (%s) — pass "
"--repo owner/repo" % (arg, what, ", ".join(r for r, _v in got)))
return got[0] if got else None
def resolve(arg, issues, pairs):
"""(id, number, repo) for one argument. Either of `id` and `repo` is None
when nothing this machine holds names it.
Order, and it is the order of what is most authoritative about this machine:
a file on disk, then the ledger, then nothing. A key skips straight to the
ledger — its number is already the tracker's answer, and the slug is only
wanted so the local copy, if there is one, can be kept honest.
`repo` travels out with the number because a key may name one
(`owner/repo#42`) and a `gitea:` field always does. Sending a foreign key to
whatever repo the CWD happens to be in would close somebody else's issue of
the same number, so the caller reconciles them before anything goes out."""
if looks_like_key(arg):
number, repo = _gitea.parse_key(arg)
hit = one([(r, s) for r, n, s in pairs
if n == number and (repo is None or r == repo)], "a slug", arg)
return (hit[1] if hit else None), number, repo or (hit[0] if hit else None)
iss = issues.get(arg)
if iss is not None:
repo, number = gmap.parse_remote_key(iss.extra.get("gitea", ""))
if not number:
_gitea.die("%s is not in the tracker (origin: %s, no gitea: field) — "
"there is no state there to change; push.py %s first"
% (arg, iss.origin, arg))
return arg, number, repo
hit = one([(r, n) for r, n, s in pairs if s == arg], "a number", arg)
if hit:
return arg, hit[1], hit[0] # pushed, and its file went with the push
_gitea.die("no issue %r in the store or the ledger — pass a Gitea number "
"(42, #42, owner/repo#42, a URL) to close one this machine has "
"never seen" % arg)
def confirmed(got, number, state):
"""True when the tracker's answer confirms THIS write, and nothing else.
The gate in front of the local write, and deliberately boring: an answer
counts only when it is an object carrying the very number that was PATCHed
(`bool` rejected explicitly — `True` is an `int`) and the state that was
asked for. A non-2xx and a `tea` that would not run never reach here at all;
`_gitea.api` exits on both, so the file survives those by never being
written."""
if not isinstance(got, dict):
return False
n = got.get("number")
if isinstance(n, bool) or not isinstance(n, int) or n != number:
return False
return got.get("state") == state
def apply_state(root, iss, state, got):
"""Write the confirmed state onto the local file; return its path.
`state:` is the domain's own field, so it is set on the issue and written
out by the domain's own writer. The sync-owned freshness fields travel with
it: the answer that authorized this write is also the newest thing the
tracker has said about the issue, so `synced:` and `remote-updated:` are
stamped from it rather than left describing an older read."""
iss.state = state
iss.extra["synced"] = _gitea.now_iso()
if got.get("updated_at"):
iss.extra["remote-updated"] = got["updated_at"]
return issue.save(root, iss)
def main():
ap = argparse.ArgumentParser(description="Close (or reopen) issues in Gitea")
ap.add_argument("ids", nargs="+",
help="local ids, or Gitea keys: 42, #42, owner/repo#42, URL")
ap.add_argument("--reopen", action="store_true",
help="set the state back to open instead of closed")
ap.add_argument("--dry-run", action="store_true",
help="print what would change; makes no request at all")
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: <repo>/tmp/issues)")
args = ap.parse_args()
root = args.out
state = "open" if args.reopen else "closed"
verb = "reopen" if args.reopen else "close"
past = "reopened" if args.reopen else "closed"
# A store that is not there is not an error here: a number needs no local
# file, and closing an issue whose copy was dropped by push is the normal
# case. `load_all` reads an absent directory as an empty one.
issues = issue.load_all(root)
pairs = ledger_pairs(_gitea.load_map(root), args.repo)
# Every argument is resolved before anything is sent, so a typo in the third
# id does not leave the first two closed.
targets = []
for arg in args.ids:
got = resolve(arg, issues, pairs)
if got not in targets:
targets.append(got)
# One run, one repo. An explicit --repo is the operator's word and wins;
# without one, the repo comes from what the ids themselves said, and two
# answers are a question rather than a guess — `repo_base` would otherwise
# let `tea` fill the blank from the CWD and close the wrong #42.
named = {r for _i, _n, r in targets if r}
if not args.repo and len(named) > 1:
_gitea.die("all ids must belong to one repo, got: %s" % ", ".join(sorted(named)))
repo_arg = args.repo or (sorted(named)[0] if named else None)
if args.dry_run:
for id, number, _repo in targets:
iss = issues.get(id)
where = ("%s (state: %s)" % (issue.path_of(root, id), iss.state)
if iss is not None else "no local copy")
print("would %s %s #%d — %s" % (verb, id or "?", number, where))
print("%d issue(s) would be %s; no request was made"
% (len(targets), past))
return
login = _gitea.require_login()
base = _gitea.repo_base(repo_arg)
touched = 0
for id, number, _repo in targets:
got = _gitea.api(login, "%s/issues/%d" % (base, number), "PATCH",
{"state": state}, payload_name="state-%d" % number)
# The gate. Above it nothing local has been written; below it the file
# is about to say something the tracker had better agree with.
if not confirmed(got, number, state):
_gitea.die("#%d: %s failed — the tracker's answer does not confirm the "
"write (%.200r). Nothing local was changed."
% (number, verb, got))
print("%s %s #%d %s" % (past, id or "?", number,
got.get("html_url", "")))
iss = issues.get(id)
if iss is None:
print(" no local copy — pull.py %d to get one" % number)
continue
print(" state: %s %s" % (state, apply_state(root, iss, state, got)))
touched += 1
if touched:
path, n = issue_index.build(root)
print("index: %s — %d issue(s)" % (path, n))
if __name__ == "__main__":
main()
tests/test_close.py
+641
View File
@@ -0,0 +1,641 @@
#!/usr/bin/env python3
"""
close.py — the state changes in Gitea, and the local file follows it or nothing
happens at all.
Two halves, and the second is the one that matters:
1. **It closes.** A slug, a number, several of either in one run, and
`--reopen` going the other way. What goes out is a PATCH carrying `state`
and nothing else; what comes back is written into `state:` on the local
file, and the index is rebuilt so the store's own table agrees.
2. **It changes nothing local unless the tracker confirmed it.** A `tea` that
exited non-zero, an answer with no number, an answer for another issue, an
answer that still says `open`, an `origin: local` issue, a `--dry-run`: in
every one of those the file on disk is byte for byte what it was. A bug here
makes the store lie about the tracker, so each path is asserted on its own.
The transport is stubbed at `_gitea.api`, as `test_drop_after_push.py` does,
with the same deliberate exception: the non-2xx test stubs `_gitea.subprocess`
and lets the real `_gitea.api` run, so "tea exited 1" is proved end to end.
Nothing here touches a network, and nothing here touches the developer's store:
every test builds its own in a `tempfile.TemporaryDirectory()`.
"""
import contextlib
import io
import json
import os
import sys
import tempfile
import types
import unittest
from unittest import mock
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
for _p in (os.path.join(_ROOT, "skills", "sync", "scripts"),
os.path.join(_ROOT, "skills", "issue", "scripts")):
if _p not in sys.path:
sys.path.insert(0, _p)
import _gitea # noqa: E402
import close # noqa: E402
import issue # noqa: E402
import map as gmap # noqa: E402
# Captured before any test patches it — the non-2xx test needs the real thing.
REAL_API = _gitea.api
REPO = "claude-skills/tea"
BASE = "repos/%s" % REPO
BODY = """## Summary
Прозаическое описание задачи.
## Spec
skills/issue/references/format.md
## Acceptance criteria
- [x] что-нибудь работает
"""
class FakeTracker(object):
"""`tea api` answered from memory, for state writes only.
It keeps a `state` per number and flips it on a PATCH, which is the whole
contract close.py has with the far side."""
def __init__(self):
self.calls = []
self.states = {} # number -> "open" / "closed"
self.raise_on_write = None # an exception instance to raise
self.answer_override = None # what a write answers instead
def payload_of(self, number):
return {"number": number, "state": self.states[number],
"title": "A thing", "updated_at": "2026-08-11T00:00:00Z",
"html_url": "https://git.example/%s/issues/%d" % (REPO, number)}
def writes(self):
return [c for c in self.calls if c[0] != "GET"]
def api(self, login, endpoint, method="GET", payload=None,
payload_name=None, out_root=None, allow_fail=False):
self.calls.append((method, endpoint, payload))
path = endpoint.split("?")[0]
if "/issues/" in path and method == "PATCH":
number = int(path.rsplit("/", 1)[1])
if self.raise_on_write is not None:
raise self.raise_on_write
self.states.setdefault(number, "open")
if "state" in (payload or {}):
self.states[number] = payload["state"]
if self.answer_override is not None:
return self.answer_override
return self.payload_of(number)
if "/issues/" in path and method == "GET":
n = int(path.rsplit("/", 1)[1])
return self.payload_of(n) if n in self.states else None
raise AssertionError("unstubbed call: %s %s" % (method, endpoint))
class StoreTestCase(unittest.TestCase):
"""A temp store and a fake tracker."""
def setUp(self):
tmp = tempfile.TemporaryDirectory(prefix="tea-close-")
self.addCleanup(tmp.cleanup)
self.root = tmp.name
self.fake = FakeTracker()
for p in (mock.patch.object(_gitea, "api", self.fake.api),
mock.patch.object(_gitea, "require_login", lambda: "test-login")):
p.start()
self.addCleanup(p.stop)
# -- fixtures ----------------------------------------------------------
def synced(self, id="a-thing", number=101, state="open"):
"""An issue that is in the tracker and on disk, the way a pull leaves
it: `origin: gitea`, a `gitea:` field, and a ledger entry."""
key = gmap.remote_key(REPO, number)
iss = issue.Issue(id=id, title="A thing", body=BODY, state=state,
labels=["type/task"], origin=gmap.ORIGIN,
extra={"gitea": key, "url": "https://git.example/x",
"synced": "2026-08-10T00:00:00Z"})
issue.save(self.root, iss)
m = _gitea.load_map(self.root)
m[key] = id
_gitea.save_map(self.root, m)
self.fake.states[number] = state
return iss
def local_only(self, id="local-thing"):
"""An issue that has never left this machine."""
iss = issue.Issue(id=id, title="Local thing", body=BODY,
labels=["type/task"])
issue.save(self.root, iss)
return iss
def dropped(self, id="gone-thing", number=205, state="open"):
"""Pushed, and its file went with the push: ledger only."""
m = _gitea.load_map(self.root)
m[gmap.remote_key(REPO, number)] = id
_gitea.save_map(self.root, m)
self.fake.states[number] = state
return number
# -- runner ------------------------------------------------------------
def run_close(self, *argv):
self.out, self.err = io.StringIO(), io.StringIO()
args = ["close.py", "--repo", REPO, "--out", self.root] + list(argv)
with mock.patch.object(sys, "argv", args), \
contextlib.redirect_stdout(self.out), \
contextlib.redirect_stderr(self.err):
close.main()
return self.out.getvalue(), self.err.getvalue()
# -- assertions --------------------------------------------------------
def state_on_disk(self, id):
return issue.load(self.root, id).state
def raw(self, id):
with open(issue.path_of(self.root, id)) as f:
return f.read()
def assertUnchanged(self, id, before, why=""):
self.assertEqual(self.raw(id), before,
"%s.md was rewritten%s" % (id, why and " — " + why))
# --------------------------------------------------------------------------
# it closes
# --------------------------------------------------------------------------
class ClosesTest(StoreTestCase):
def test_a_slug_closes_the_issue_it_names(self):
self.synced("a-thing", 101)
out, _ = self.run_close("a-thing")
self.assertEqual(self.fake.states[101], "closed")
self.assertIn("closed a-thing #101", out)
def test_the_local_state_follows(self):
self.synced("a-thing", 101)
self.run_close("a-thing")
self.assertEqual(self.state_on_disk("a-thing"), "closed")
def test_only_the_state_is_sent(self):
"""Closing is not an edit: no title, no body, no labels ride along."""
self.synced("a-thing", 101)
self.run_close("a-thing")
writes = self.fake.writes()
self.assertEqual(len(writes), 1)
method, endpoint, payload = writes[0]
self.assertEqual((method, endpoint), ("PATCH", "%s/issues/101" % BASE))
self.assertEqual(payload, {"state": "closed"})
def test_a_number_closes_it_too(self):
"""The normal case for a pushed issue — the file is long gone."""
self.synced("a-thing", 101)
self.run_close("101")
self.assertEqual(self.fake.states[101], "closed")
self.assertEqual(self.state_on_disk("a-thing"), "closed")
def test_every_key_form_is_accepted(self):
forms = {110: "110", 111: "#111", 112: "%s#112" % REPO,
113: "https://git.example/%s/issues/113" % REPO}
for n in forms:
self.fake.states[n] = "open"
for n, arg in forms.items():
with self.subTest(arg=arg):
self.run_close(arg)
self.assertEqual(self.fake.states[n], "closed")
def test_several_ids_in_one_run(self):
self.synced("a-thing", 101)
self.synced("b-thing", 102)
self.run_close("a-thing", "102")
self.assertEqual(self.fake.states, {101: "closed", 102: "closed"})
self.assertEqual(self.state_on_disk("a-thing"), "closed")
self.assertEqual(self.state_on_disk("b-thing"), "closed")
def test_the_same_issue_named_twice_is_written_once(self):
self.synced("a-thing", 101)
self.run_close("a-thing", "#101")
self.assertEqual(len(self.fake.writes()), 1)
def test_the_index_is_rebuilt(self):
self.synced("a-thing", 101)
out, _ = self.run_close("a-thing")
self.assertIn("index:", out)
with open(os.path.join(self.root, "INDEX.md")) as f:
self.assertIn("closed", f.read())
def test_the_body_survives_untouched(self):
"""One metadata field changes; the prose and the ticks do not."""
self.synced("a-thing", 101)
before = issue.load(self.root, "a-thing").body
self.run_close("a-thing")
self.assertEqual(issue.load(self.root, "a-thing").body, before)
def test_synced_is_refreshed(self):
self.synced("a-thing", 101)
self.run_close("a-thing")
iss = issue.load(self.root, "a-thing")
self.assertNotEqual(iss.extra.get("synced"), "2026-08-10T00:00:00Z")
self.assertEqual(iss.extra.get("remote-updated"), "2026-08-11T00:00:00Z")
def test_an_issue_whose_file_was_dropped_still_closes(self):
"""No local copy at all: the ledger names it, the tracker takes it, and
nothing is written locally."""
self.dropped("gone-thing", 205)
out, _ = self.run_close("gone-thing")
self.assertEqual(self.fake.states[205], "closed")
self.assertIn("no local copy", out)
self.assertNotIn("index:", out)
def test_a_number_nobody_here_knows_closes_without_a_slug(self):
self.fake.states[777] = "open"
out, _ = self.run_close("777")
self.assertEqual(self.fake.states[777], "closed")
self.assertIn("#777", out)
class ReopensTest(StoreTestCase):
def test_reopen_sends_open(self):
self.synced("a-thing", 101, state="closed")
out, _ = self.run_close("--reopen", "a-thing")
self.assertEqual(self.fake.writes()[0][2], {"state": "open"})
self.assertIn("reopened a-thing #101", out)
def test_reopen_writes_the_local_state_back(self):
self.synced("a-thing", 101, state="closed")
self.run_close("--reopen", "a-thing")
self.assertEqual(self.state_on_disk("a-thing"), "open")
def test_close_then_reopen_is_a_round_trip(self):
self.synced("a-thing", 101)
self.run_close("a-thing")
self.run_close("--reopen", "a-thing")
self.assertEqual(self.fake.states[101], "open")
self.assertEqual(self.state_on_disk("a-thing"), "open")
# --------------------------------------------------------------------------
# it refuses
# --------------------------------------------------------------------------
class LocalOnlyTest(StoreTestCase):
"""An `origin: local` issue is not in the tracker, so it cannot be closed
there — and the local field is not quietly edited instead."""
def test_it_exits(self):
self.local_only("local-thing")
with self.assertRaises(SystemExit):
self.run_close("local-thing")
def test_the_error_names_the_id_and_says_it_is_not_in_the_tracker(self):
self.local_only("local-thing")
with self.assertRaises(SystemExit):
self.run_close("local-thing")
err = self.err.getvalue()
self.assertIn("local-thing", err)
self.assertIn("not in the tracker", err)
def test_nothing_is_sent(self):
self.local_only("local-thing")
with self.assertRaises(SystemExit):
self.run_close("local-thing")
self.assertEqual(self.fake.calls, [])
def test_the_file_is_untouched(self):
self.local_only("local-thing")
before = self.raw("local-thing")
with self.assertRaises(SystemExit):
self.run_close("local-thing")
self.assertUnchanged("local-thing", before)
def test_a_bad_id_stops_the_whole_run_before_anything_is_sent(self):
"""Resolution happens up front, so a typo in the second id does not
leave the first one closed."""
self.synced("a-thing", 101)
with self.assertRaises(SystemExit):
self.run_close("a-thing", "local-thing")
self.assertEqual(self.fake.states[101], "open")
self.assertEqual(self.fake.calls, [])
def test_an_unknown_slug_exits(self):
with self.assertRaises(SystemExit):
self.run_close("no-such-thing")
self.assertIn("no-such-thing", self.err.getvalue())
class DryRunTest(StoreTestCase):
def test_not_one_request_is_made(self):
self.synced("a-thing", 101)
self.run_close("--dry-run", "a-thing")
self.assertEqual(self.fake.calls, [])
def test_the_file_is_untouched(self):
self.synced("a-thing", 101)
before = self.raw("a-thing")
self.run_close("--dry-run", "a-thing")
self.assertUnchanged("a-thing", before, "--dry-run must write nothing")
def test_it_says_what_would_be_closed(self):
self.synced("a-thing", 101)
self.synced("b-thing", 102)
out, _ = self.run_close("--dry-run", "a-thing", "102")
self.assertIn("would close a-thing #101", out)
self.assertIn("would close b-thing #102", out)
self.assertIn("2 issue(s) would be closed", out)
def test_it_says_reopen_under_reopen(self):
self.synced("a-thing", 101, state="closed")
out, _ = self.run_close("--dry-run", "--reopen", "a-thing")
self.assertIn("would reopen a-thing #101", out)
self.assertIn("would be reopened", out)
def test_it_needs_no_login(self):
"""A dry run must work before /tea:auth has ever been run."""
self.synced("a-thing", 101)
with mock.patch.object(_gitea, "require_login",
lambda: self.fail("dry run asked for a login")):
self.run_close("--dry-run", "a-thing")
def test_a_local_only_issue_is_still_refused(self):
self.local_only("local-thing")
with self.assertRaises(SystemExit):
self.run_close("--dry-run", "local-thing")
# --------------------------------------------------------------------------
# the tracker said no
# --------------------------------------------------------------------------
class TrackerFailureTest(StoreTestCase):
"""The criterion that matters most: a write that was not confirmed leaves
the local file exactly as it was."""
def test_a_non_2xx_answer_leaves_the_file(self):
"""The real `_gitea.api` against a `tea` that exits 1 — the path a 422
or a 500 actually takes, and it ends in `die()`."""
self.synced("a-thing", 101)
before = self.raw("a-thing")
def fake_run(cmd, capture_output=False, text=False):
return types.SimpleNamespace(
returncode=1, stdout="",
stderr="422 Unprocessable Entity: issue is blocked")
with mock.patch.object(_gitea, "api", REAL_API), \
mock.patch.object(_gitea, "subprocess",
types.SimpleNamespace(run=fake_run)), \
self.assertRaises(SystemExit):
self.run_close("a-thing")
self.assertUnchanged("a-thing", before, "tea exited non-zero")
self.assertEqual(self.state_on_disk("a-thing"), "open")
def test_a_transport_exception_leaves_the_file(self):
self.synced("a-thing", 101)
before = self.raw("a-thing")
self.fake.raise_on_write = OSError("tea: command not found")
with self.assertRaises(OSError):
self.run_close("a-thing")
self.assertUnchanged("a-thing", before, "the transport raised")
def test_an_answer_without_a_number_leaves_the_file(self):
self.synced("a-thing", 101)
before = self.raw("a-thing")
self.fake.answer_override = {"ok": True, "state": "closed"}
with self.assertRaises(SystemExit):
self.run_close("a-thing")
self.assertUnchanged("a-thing", before)
def test_an_answer_for_another_issue_leaves_the_file(self):
self.synced("a-thing", 101)
before = self.raw("a-thing")
self.fake.answer_override = {"number": 999, "state": "closed"}
with self.assertRaises(SystemExit):
self.run_close("a-thing")
self.assertUnchanged("a-thing", before)
def test_an_answer_that_did_not_change_the_state_leaves_the_file(self):
"""A 200 that still says `open` is not a close."""
self.synced("a-thing", 101)
before = self.raw("a-thing")
self.fake.answer_override = {"number": 101, "state": "open"}
with self.assertRaises(SystemExit):
self.run_close("a-thing")
self.assertUnchanged("a-thing", before)
def test_an_empty_answer_leaves_the_file(self):
self.synced("a-thing", 101)
before = self.raw("a-thing")
self.fake.answer_override = None
real_api = self.fake.api
self.fake.api = lambda *a, **kw: (real_api(*a, **kw), None)[1]
with mock.patch.object(_gitea, "api", self.fake.api), \
self.assertRaises(SystemExit):
self.run_close("a-thing")
self.assertUnchanged("a-thing", before)
def test_the_error_says_nothing_local_changed(self):
self.synced("a-thing", 101)
self.fake.answer_override = {"ok": True}
with self.assertRaises(SystemExit):
self.run_close("a-thing")
self.assertIn("Nothing local was changed", self.err.getvalue())
def test_a_failure_partway_through_keeps_the_rest(self):
"""Two issues, the second one is not confirmed. The first is
legitimately closed; the second's file still says open."""
self.synced("aaa-thing", 101)
self.synced("zzz-thing", 102)
before = self.raw("zzz-thing")
real = self.fake.api
seen = []
def once(login, endpoint, method="GET", payload=None, **kw):
got = real(login, endpoint, method, payload, **kw)
if method != "GET":
seen.append(endpoint)
return {"nope": True} if len(seen) > 1 else got
with mock.patch.object(_gitea, "api", once), \
self.assertRaises(SystemExit):
self.run_close("aaa-thing", "zzz-thing")
self.assertEqual(self.state_on_disk("aaa-thing"), "closed")
self.assertUnchanged("zzz-thing", before, "its write was not confirmed")
# --------------------------------------------------------------------------
# the pure parts
# --------------------------------------------------------------------------
class ConfirmedTest(unittest.TestCase):
"""The gate itself. Everything below it rewrites a file."""
def test_a_matching_close_is_confirmed(self):
self.assertTrue(close.confirmed({"number": 42, "state": "closed"}, 42, "closed"))
def test_a_mismatched_number_is_not(self):
self.assertFalse(close.confirmed({"number": 43, "state": "closed"}, 42, "closed"))
def test_the_wrong_state_is_not(self):
self.assertFalse(close.confirmed({"number": 42, "state": "open"}, 42, "closed"))
def test_a_missing_state_is_not(self):
self.assertFalse(close.confirmed({"number": 42}, 42, "closed"))
def test_none_and_lists_are_not(self):
self.assertFalse(close.confirmed(None, 42, "closed"))
self.assertFalse(close.confirmed([{"number": 42, "state": "closed"}], 42, "closed"))
def test_true_is_not_a_number(self):
self.assertFalse(close.confirmed({"number": True, "state": "closed"}, 1, "closed"))
def test_a_string_number_is_not(self):
self.assertFalse(close.confirmed({"number": "42", "state": "closed"}, 42, "closed"))
class KeyFormTest(unittest.TestCase):
"""A slug and a key are two vocabularies that must not collide."""
def test_keys_are_keys(self):
for k in ("42", "#42", "owner/repo#42",
"https://git.example/owner/repo/issues/42"):
self.assertTrue(close.looks_like_key(k), k)
def test_slugs_are_not_keys(self):
for s in ("a-thing", "wire-sqlc-appclick", "close-issues-through-a-script"):
self.assertFalse(close.looks_like_key(s), s)
class LedgerPairsTest(unittest.TestCase):
def setUp(self):
self.m = {"%s#7" % REPO: "a-thing", "other/repo#7": "b-thing",
"not-a-key": "c-thing"}
def test_it_filters_by_repo(self):
self.assertEqual(close.ledger_pairs(self.m, REPO), [(REPO, 7, "a-thing")])
def test_without_a_repo_it_keeps_everything_parseable(self):
got = close.ledger_pairs(self.m)
self.assertEqual(sorted(s for _r, _n, s in got), ["a-thing", "b-thing"])
def test_an_ambiguous_number_exits(self):
pairs = close.ledger_pairs(self.m)
with self.assertRaises(SystemExit):
with contextlib.redirect_stderr(io.StringIO()):
close.resolve("7", {}, pairs)
class AmbiguityTest(StoreTestCase):
"""Two repos, one number, no --repo: settle it rather than guess."""
def test_the_error_points_at_repo(self):
_gitea.save_map(self.root, {"%s#7" % REPO: "a-thing",
"other/repo#7": "b-thing"})
err = io.StringIO()
args = ["close.py", "--out", self.root, "7"]
with mock.patch.object(sys, "argv", args), \
contextlib.redirect_stdout(io.StringIO()), \
contextlib.redirect_stderr(err), \
self.assertRaises(SystemExit):
close.main()
self.assertIn("--repo", err.getvalue())
class RepoOfTheKeyTest(StoreTestCase):
"""A key that names its own repo is sent there, not to whatever repo the
CWD happens to be — otherwise `#42` closes somebody else's issue."""
def run_bare(self, *argv):
"""No `--repo`, so the ids have to say where they live."""
self.out, self.err = io.StringIO(), io.StringIO()
args = ["close.py", "--out", self.root] + list(argv)
with mock.patch.object(sys, "argv", args), \
contextlib.redirect_stdout(self.out), \
contextlib.redirect_stderr(self.err):
close.main()
return self.out.getvalue(), self.err.getvalue()
def test_a_foreign_key_goes_to_its_own_repo(self):
self.run_bare("other/repo#42")
self.assertEqual(self.fake.writes()[0][1], "repos/other/repo/issues/42")
def test_a_slug_goes_to_the_repo_its_gitea_field_names(self):
self.synced("a-thing", 101)
self.run_bare("a-thing")
self.assertEqual(self.fake.writes()[0][1], "%s/issues/101" % BASE)
def test_two_repos_in_one_run_is_a_question_not_a_guess(self):
self.synced("a-thing", 101)
with self.assertRaises(SystemExit):
self.run_bare("a-thing", "other/repo#42")
self.assertIn("one repo", self.err.getvalue())
self.assertEqual(self.fake.calls, [])
def test_an_explicit_repo_settles_it(self):
self.synced("a-thing", 101)
self.run_close("a-thing", "other/repo#42")
self.assertEqual({c[1] for c in self.fake.writes()},
{"%s/issues/101" % BASE, "%s/issues/42" % BASE})
class NoStoreTest(StoreTestCase):
"""A number needs no local file, and a store that is not there is not an
error — closing an issue whose copy push dropped is the normal case."""
def test_a_number_closes_with_no_store_at_all(self):
missing = os.path.join(self.root, "nowhere")
self.fake.states[303] = "open"
args = ["close.py", "--repo", REPO, "--out", missing, "303"]
with mock.patch.object(sys, "argv", args), \
contextlib.redirect_stdout(io.StringIO()), \
contextlib.redirect_stderr(io.StringIO()):
close.main()
self.assertEqual(self.fake.states[303], "closed")
self.assertFalse(os.path.isdir(missing), "no store was conjured")
class PayloadFileTest(StoreTestCase):
"""The request body goes to the transport's own scratchpad.
Not to a directory this script picks: `close.py` names the payload and
nothing else, the way every other caller does. Where PAYLOAD_ROOT lands is
_gitea's business, and test_payload_root.py is where that is tested."""
def test_the_payload_lands_in_the_transports_scratchpad(self):
self.synced("a-thing", 101)
payloads = os.path.join(self.root, "payload")
with mock.patch.object(_gitea, "PAYLOAD_ROOT", payloads), \
mock.patch.object(_gitea, "api", REAL_API), \
mock.patch.object(
_gitea, "subprocess",
types.SimpleNamespace(run=lambda cmd, **kw: types.SimpleNamespace(
returncode=0, stderr="",
stdout=json.dumps({"number": 101, "state": "closed"})))):
self.run_close("a-thing")
p = os.path.join(payloads, "state-101.json")
self.assertTrue(os.path.isfile(p))
with open(p) as f:
self.assertEqual(json.load(f), {"state": "closed"})
if __name__ == "__main__":
unittest.main()