11 Commits

Author SHA1 Message Date
naudachu 257c547e22 merge: merge checkbox state on pull instead of overwriting it 2026-08-10 15:52:50 +05:00
naudachu d4c43464e5 feat: merge checkbox state on pull instead of overwriting it
A tick was lost in both directions: pull wrote the server body as-is,
push sent the local body as-is, last writer won. Tick it in the web UI
and the first `push.py --update` dropped it; tick it locally and the
first pull dropped it.

The usual answer is drift tracking and a three-way merge, which this
repo rejected on purpose. It is not needed. A tick is monotone — an item
only travels `[ ]` -> `[x]` — so unioning the two sides is a set union,
not conflict resolution. One rule for one line type replaces the whole
mechanism, and the store stays "not a mirror".

`map.merge_checkbox_state` is pure and does the work; `from_api` takes
the local body as an optional argument; `pull.py` hands it the copy
already on disk. Checkbox parsing is imported from `skills/issue`
(`checkboxes` / `set_checkbox`), never redefined here — the domain layer
is untouched.

The same item text more than once is read as a set: one ticked local
item ticks every server line with that text. Pairing duplicates up by
order is the alternative, and it can still drop a tick — which is the
bug being fixed.

The price is documented, not hidden: unticking is not monotone, so a box
unticked in the web UI comes back on the next pull. Untick locally, then
push.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-10 15:52:11 +05:00
naudachu 47f53a7edc test: check the domain layer against sys.stdlib_module_names
The layering test pinned an exact import set, so any new stdlib import in
skills/issue tripped it — 'collections', added by issue_ac.py, did. Assert
the rule AGENTS.md actually states (stdlib only, never subprocess) instead
of a frozen snapshot of it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-10 15:43:24 +05:00
naudachu 484da64621 merge: tick in-body checkboxes from a domain script
# Conflicts:
#	AGENTS.md
2026-08-10 15:42:38 +05:00
naudachu 31f7c39155 merge: write issue dependencies to Gitea on push 2026-08-10 15:41:37 +05:00
naudachu f230f98f35 merge: reconcile the feature-container convention with the depends validator 2026-08-10 15:41:30 +05:00
naudachu b72f619fda merge: resolve the issue store path independently of the working directory 2026-08-10 15:41:19 +05:00
naudachu 62c8ff976d feat: tick in-body checkboxes from a domain script
A checkbox is the one part of a body that is state and not prose.
Everything else is written once; boxes get ticked as the work goes, and
until now the only ways to tick one were a human with an editor or a
model rewriting the whole body. The second is worse: the rewrite re-flows
lines and re-words sentences, so the issue's diff swells around a change
that means one character. Progress was invisible too — issue_index.py
builds INDEX.md from metadata and never looked inside a body, so "3 of 7
done" required opening the file.

All three pieces are domain: a checkbox is body syntax, which is part of
the answer to "what is an issue". The parser goes in issue.py so the sync
layer can reuse it instead of redefining the format on its own side.

issue.py gains checkboxes(text) -> [Checkbox(index, line, end_line,
checked, text, section)], plus set_checkbox(text, item, checked) and
checkbox_progress(text). All pure, no I/O, importable from another layer.
The scan covers the whole text, in any section: the type/feature template
keeps child issues as checkboxes under `## Issues`, so binding the parser
to `## Acceptance criteria` would silently lose half of them; the heading
is recorded, never required. Only a marker line opens an item, so a
wrapped continuation line belongs to the item above it rather than
counting as one of its own. A `- [ ]` inside a code fence is an example
of the markup and is skipped. Line numbers are relative to the text
given, which is what lets a caller work on a body or on a whole file.

issue_ac.py lists the items numbered, grouped by heading, and ticks one
by number or by substring. An ambiguous substring is an error that prints
the matches — a coin flip would tick the wrong box and look like it
worked. It patches the file rather than round-tripping through
Issue.to_text(), so exactly one character changes: metadata order,
wording, wrapping, trailing whitespace and CRLF endings all come back
byte for byte, proven by a diff in the tests.

INDEX.md gains a progress column: `3/7` for an issue with checkboxes,
blank for one without. Counted off the body at build time and stored in
no field — a second copy of the state would be wrong by the next edit.

issue_check.py is unchanged and stays that way on purpose: an unticked
box is work not done yet, not a malformed issue, and validate() carries a
comment saying so.

Delivering a tick to the tracker is out of scope — that is push.py
--update in /tea:sync.

format.md gets one clarifying bullet. It said acceptance criteria are
checkboxes but never said what a checkbox is, so the parser had to settle
questions the format left open: any section, wrapped items, fenced
examples. Those rules are now written down where the parser and the sync
layer can both point at them.

tests/ is new, and is the convention: plain stdlib unittest, no pytest
and no third-party deps, since the code under test may not have
dependencies either. Scripts are imported via sys.path.insert and every
fixture is built in a TemporaryDirectory, never in tmp/.

    python3 -m unittest discover -s tests -v     32 tests, OK

skills/issue/scripts/ still imports stdlib only, with no subprocess.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-10 15:41:15 +05:00
naudachu fb862554ed fix: reconcile the feature-container convention with the depends validator
`format.md` told a child issue to link back to its container through its own
`depends:`, while the validator wanted the container to list its children.
Satisfying both made a cycle, caught as an ERROR, so every `type/feature` with
a filled `## Issues` ended in either a warning or a hard failure — no third
option.

Variant B is chosen: the container depends on its children, and a child never
names its container. "The container is closed when its children are closed" IS
a dependency relation, so it belongs in the graph; "a child belongs to a
feature" is membership, and membership does not. The code already walked the
edge that way — `## Issues` is an edge source pointing container -> child — so
this rewrites the documentation to match instead of inverting the graph, and
the tree draws containers as roots for free.

- format.md: the `type/feature` template states the direction, shows the
  container's `depends:`, and says why the reverse cycles; the Dependencies
  section names `## Issues` as the second edge source.
- issue.py: `body_dep_ref_sections()` carries the section each reference came
  from, so the desync warning names `## Issues` on a container rather than a
  `## Depends on` that is not in the file. `body_dep_refs()` stays as a thin
  wrapper — `skills/sync/scripts/map.py` calls it and is untouched.
- tests/test_container_edges.py: the repo's first tests. Stdlib unittest,
  `python3 -m unittest discover -s tests`.

issue_check.py's cycle detector and issue_tree.py need no change: with the edge
pointing down there is no cycle to break and the container is already the root.

Refs claude-skills/tea#14

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-10 15:40:28 +05:00
naudachu 6d01ead245 fix: resolve the issue store path independently of the working directory
ISSUE_ROOT was the relative `tmp/issues`, so "the store" was whatever
directory the shell happened to be standing in. It is the --out default
in all eight scripts of both layers, which made one `cd` — and a `cd`
outlives the command that ran it — enough for readers to report an empty
store on a full one and for writers to quietly build a second store
beside the first. `issue_index.py` run from inside tmp/issues left
tmp/issues/tmp/issues/ behind and exited 0.

The anchor is issue.py's own __file__, not cwd. A script's location is a
fact about the installation; cwd is a fact about the last `cd`, and the
scripts are invoked by path from wherever the agent happens to be. From
there `store_root()` walks up to the nearest repo marker — `.git`
(exists(), not isdir(): a worktree's .git is a file) or AGENTS.md for a
copy taken out of git — and joins tmp/issues. Markers rather than a
fixed number of `..` hops, because the layout is not a promise. cwd is
tried only if the scripts are not inside a repository at all.

The function lives in the domain layer and skills/sync imports it, so
both layers agree by construction — the direction the layering rule
allows. skills/issue stays stdlib-only.

An explicit --out still wins and is used exactly as typed: a relative
--out stays relative to cwd, because that is what the operator asked
for. No new environment surface.

Two consequences the issue also asked for:

- Missing is no longer reported as empty. `store_error()` returns one
  message for a path that is not there and another for a store with no
  issues in it.
- Nothing conjures a store as a side effect of a write. save() and
  issue_index.build() require it instead of os.makedirs'ing it; only
  issue_new.py and pull.py create one, and both say so on stderr.

Establishes tests/ — plain stdlib unittest, no pytest, no dependencies.
The store tests build a throwaway repo in a TemporaryDirectory (a .git
marker, a copy of both script layers, fixture issues) and run the real
scripts inside it as subprocesses from five different working
directories; tmp/issues/ is never touched. Against the pre-fix scripts
15 of the 21 fail, reproducing the report exactly — five stray stores,
including tmp/issues/tmp/issues.

    python3 -m unittest discover -s tests -v

Closes claude-skills/tea#15

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-10 15:40:07 +05:00
naudachu 0cf4baa429 fix: write issue dependencies to Gitea on push
The local `depends:` graph never reached the tracker. push.py sent dependent
issues in topological order but created no native links, so `native_deps` in
_gitea.py was a reader with no writer and the slugs in `## Depends on` stayed
dead prose for anyone reading the issue in Gitea.

Once an issue has its number, every `depends:` entry that also has one now
becomes a real link: POST /repos/{owner}/{repo}/issues/{index}/dependencies
with the blocker's IssueMeta. Topological order means the blocker is already
numbered, so no second pass is needed. Existing links are read back first, so
a repeat push is a no-op and never 409s; a link that fails anyway warns rather
than aborting a run that has already created issues. `--dry-run` prints the
links it would make and touches nothing.

The `## Depends on` prose is still passed through verbatim — the edge the
tracker acts on is the native link, not the text, which is exactly why the
text can be left alone. Removing a link that disappeared from `depends:` is
out of scope and now says so in push.py's docstring.

Establishes tests/: stdlib unittest, the transport stubbed at _gitea.api, no
network. Run with `python3 -m unittest discover -s tests`.

The POST body shape was confirmed against the instance's own swagger.v1.json
(Gitea 1.26.1), not assumed from upstream docs.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-10 15:39:52 +05:00
23 changed files with 2785 additions and 67 deletions
+30 -1
View File
@@ -47,9 +47,11 @@ the domain layer, it is in the wrong place.
- `skills/issue` — issues as units of work (`/tea:issue`), entirely offline - `skills/issue` — issues as units of work (`/tea:issue`), entirely offline
- `references/format.md` — canonical issue format; single source of truth - `references/format.md` — canonical issue format; single source of truth
- `scripts/issue.py` — domain module: slug identity, parse/render, validation, - `scripts/issue.py` — domain module: slug identity, parse/render, validation,
taxonomy, dependency graph taxonomy, dependency graph, body checkboxes
- `scripts/issue_new.py` — create a local issue from its type template - `scripts/issue_new.py` — create a local issue from its type template
- `scripts/issue_check.py` — validate against the format - `scripts/issue_check.py` — validate against the format
- `scripts/issue_ac.py` — list the body's checkboxes; tick one by number or
substring, changing exactly one character of the file
- `scripts/issue_tree.py` — draw the dependency graph - `scripts/issue_tree.py` — draw the dependency graph
- `scripts/issue_index.py` — rebuild `tmp/issues/INDEX.md` - `scripts/issue_index.py` — rebuild `tmp/issues/INDEX.md`
- `skills/sync` — move issues between the local store and Gitea (`/tea:sync`) - `skills/sync` — move issues between the local store and Gitea (`/tea:sync`)
@@ -68,6 +70,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 +96,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
+3 -1
View File
@@ -127,9 +127,11 @@ skills/
references/format.md canonical issue format (identity, types, templates) references/format.md canonical issue format (identity, types, templates)
scripts/ Python 3, stdlib only, no network: scripts/ Python 3, stdlib only, no network:
issue.py domain module: slug identity, parse/render, issue.py domain module: slug identity, parse/render,
validation, taxonomy, dependency graph validation, taxonomy, dependency graph,
body checkboxes
issue_new.py create a local issue from its type template issue_new.py create a local issue from its type template
issue_check.py validate against the format issue_check.py validate against the format
issue_ac.py list the body's checkboxes; tick one
issue_tree.py draw the dependency graph issue_tree.py draw the dependency graph
issue_index.py rebuild tmp/issues/INDEX.md issue_index.py rebuild tmp/issues/INDEX.md
sync/ /tea:sync — the bridge to Gitea sync/ /tea:sync — the bridge to Gitea
+7 -2
View File
@@ -30,7 +30,8 @@ to fill the gap yourself.
Load the skill, do not remember the flags: 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`, `remote.py`, `labels.py`
- `/tea:issue``issue_check.py`, `issue_tree.py`, `issue_index.py`, `issue_new.py` - `/tea:issue``issue_check.py`, `issue_tree.py`, `issue_index.py`,
`issue_new.py`, `issue_ac.py`
Invoke `Skill` with `tea:sync` or `tea:issue` at the start of the task, and use Invoke `Skill` with `tea:sync` or `tea:issue` at the start of the task, and use
the command table it gives you verbatim. The skill is the single source of the command table it gives you verbatim. The skill is the single source of
@@ -46,7 +47,11 @@ instead of trying it.
the `tea-guard` hook substitutes the pinned login. Never name a login. the `tea-guard` hook substitutes the pinned login. Never name a login.
2. **No writing to issue files.** You have no `Edit` and no `Write`. Scripts 2. **No writing to issue files.** You have no `Edit` and no `Write`. Scripts
write files; you do not. If a task needs a body edited or a metadata field write files; you do not. If a task needs a body edited or a metadata field
changed by hand, stop and say which file and which field. changed by hand, stop and say which file and which field. `issue_ac.py` is
the one script that touches a body, and it changes a single character: tick
only the items the caller named, by the number or the substring the caller
gave. Whether a criterion is actually met is a judgement about content, and
content is never yours.
3. **Push only what you were told to push.** `push.py` publishes to a tracker 3. **Push only what you were told to push.** `push.py` publishes to a tracker
other people read. Run it with the ids the caller named, or with the filter other people read. Run it with the ids the caller named, or with the filter
the caller named. Never widen the set, never run a bare `push.py` because it the caller named. Never widen the set, never run a bare `push.py` because it
+57 -3
View File
@@ -36,6 +36,7 @@ All offline, all in `<skill-base-dir>/scripts/`.
|---|---| |---|---|
| `issue_new.py --type T --title "…"` | create `tmp/issues/<slug>.md` from the type's template | | `issue_new.py --type T --title "…"` | create `tmp/issues/<slug>.md` from the type's template |
| `issue_check.py [id…]` | validate against the canonical format; exit 1 on errors | | `issue_check.py [id…]` | validate against the canonical format; exit 1 on errors |
| `issue_ac.py <id> [--check N\|TEXT]` | list the body's checkboxes; tick or untick one |
| `issue_tree.py [id…]` | draw the dependency graph from `depends:` | | `issue_tree.py [id…]` | draw the dependency graph from `depends:` |
| `issue_index.py` | rebuild `tmp/issues/INDEX.md` | | `issue_index.py` | rebuild `tmp/issues/INDEX.md` |
| `issue.py` | the domain module the others import — not a command | | `issue.py` | the domain module the others import — not a command |
@@ -47,6 +48,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`
@@ -92,13 +111,48 @@ decision — `/tea:sync` — and does not change the file's status here.
## Editing an issue ## Editing an issue
Edit the file. Change `state:` to close it, edit `labels:`, tick checkboxes in Edit the file. Change `state:` to close it, edit `labels:`, add ids to
`## Acceptance criteria`, add ids to `depends:`. Re-run `issue_check.py` `depends:`. Re-run `issue_check.py` afterwards, and `issue_index.py` to refresh
afterwards, and `issue_index.py` to refresh the table. the table. Checkboxes are the exception — use `issue_ac.py`, below.
If the issue is synced (`origin: gitea`), your edit is local until you run If the issue is synced (`origin: gitea`), your edit is local until you run
`push.py --update` from `/tea:sync`. Nothing tracks that drift automatically. `push.py --update` from `/tea:sync`. Nothing tracks that drift automatically.
## Ticking checkboxes
A checkbox is the one part of a body that is **state** and not prose, so it has
a command of its own. Never rewrite a body just to tick a box: the rewrite
re-flows lines and re-words sentences, and the issue's diff swells around a
change that means one character.
```bash
python3 <skill-base-dir>/scripts/issue_ac.py wire-sqlc-appclick
python3 <skill-base-dir>/scripts/issue_ac.py wire-sqlc-appclick --check 3
python3 <skill-base-dir>/scripts/issue_ac.py wire-sqlc-appclick --check "регресс"
python3 <skill-base-dir>/scripts/issue_ac.py wire-sqlc-appclick --uncheck 3
```
With no flag it prints the numbered list with each item's state, grouped by the
heading the item sits under. `--check` / `--uncheck` take that number or a
substring of the item's text (case-insensitive).
- **Every checkbox in the body counts, not just `## Acceptance criteria`.** A
`type/feature` keeps its children as checkboxes under `## Issues`, and they
are numbered in the same list. The script is named after the section most
boxes live in, nothing more.
- **A substring must match exactly one item.** Two matches is an error that
lists them; pick by number instead. It never guesses.
- **Exactly one character of the file changes.** Metadata, wording, wrapping
and trailing whitespace all come back byte for byte, so `git diff` and the
tracker's diff show the tick and nothing else.
- Examples inside a ``` fence are markup, not state — they are skipped.
- `INDEX.md` gains a `progress` column (`3/7`, blank when the issue has no
boxes), recomputed from the body on every build and stored in no field.
`issue_ac.py` rebuilds the index after a successful tick.
Getting the tick to the tracker is a separate step — `push.py --update` in
`/tea:sync`.
## Writing a proper description ## Writing a proper description
Issues get filed on the run — "comments aren't pulled", "the guard broke". Issues get filed on the run — "comments aren't pulled", "the guard broke".
+38 -4
View File
@@ -145,6 +145,11 @@ you — `issue_check.py` warns when the section names an id that `depends:` does
not list. Omit the section when there are no dependencies; never write an empty not list. Omit the section when there are no dependencies; never write an empty
one. one.
A `type/feature` container writes the same relation under `## Issues` instead
(see the template below). Same direction, same rule: every id named there also
belongs in that issue's `depends:`. The warning names whichever of the two
sections the reference actually came from.
Draw the graph with `issue_tree.py`. The reverse direction is a grep: Draw the graph with `issue_tree.py`. The reverse direction is a grep:
```bash ```bash
@@ -162,6 +167,13 @@ grep -ln 'depends:.*migrate-schema' tmp/issues/*.md
valid answer. valid answer.
- Acceptance criteria are `- [ ]` checkboxes; each item is an objectively - Acceptance criteria are `- [ ]` checkboxes; each item is an objectively
checkable condition, not an aspiration. checkable condition, not an aspiration.
- A checkbox is **item markup, not a property of one section**: `- [ ]`
unticked, `- [x]` ticked, and it means the same under `## Issues` as under
`## Acceptance criteria`. An item that wraps continues on an indented line
and is still one item. A `- [ ]` inside a ``` code fence is an example of the
markup, not state. Tick them with `issue_ac.py`, which reads the whole body
on exactly these rules and rewrites one character; progress (`3/7`) is
counted off the body and is never a metadata field.
- Code references use the `path/file.ext:line` form; related issues by id. - Code references use the `path/file.ext:line` form; related issues by id.
- Screenshots are allowed but their content must be duplicated as text — an - Screenshots are allowed but their content must be duplicated as text — an
LLM reading these files cannot see images. LLM reading these files cannot see images.
@@ -258,9 +270,31 @@ grep -ln 'depends:.*migrate-schema' tmp/issues/*.md
## Template: `type/feature` ## Template: `type/feature`
A container: one unit of business value delivered by several child issues. A container: one unit of business value delivered by several child issues.
Child issues carry their own `type/*` (`task`, `bug`, `test`, …) and link back Child issues carry their own `type/*` (`task`, `bug`, `test`, …) and know
via their `depends:`. Keep implementation detail in the children; the feature nothing about the container.
body stays at business level.
**The container depends on its children, never the reverse.** Every child id
goes in the container's own `depends:` and, as prose, in its `## Issues`
section; a child's `depends:` is for that child's real dependencies and must
not point back at the container. Keep implementation detail in the children;
the feature body stays at business level.
That direction is not a convention picked at random. "The container is closed
when its children are closed" *is* a dependency relation. "This child belongs
to that feature" is a membership relation, and membership has no place in a
dependency graph. Pointed the other way the two rules contradict each other:
the moment the container listed a child that already depended on it,
`issue_check.py` would report `ERROR cycle`. With the edge going down, the
graph reads as nesting — `issue_tree.py` draws the container as the root with
its children beneath it — and the check is green.
So the container's metadata block carries the children:
```markdown
depends: [wire-sqlc-appclick, add-pool-cfg]
```
and its body repeats them for a human:
```markdown ```markdown
## Summary ## Summary
@@ -274,7 +308,7 @@ body stays at business level.
## Issues ## Issues
- [ ] wire-sqlc-appclick — краткое описание части - [ ] wire-sqlc-appclick — краткое описание части
- [ ] - [ ] add-pool-cfg — краткое описание части
## Acceptance criteria ## Acceptance criteria
- [ ] все дочерние issues закрыты - [ ] все дочерние issues закрыты
+271 -15
View File
@@ -45,10 +45,71 @@ without a parser:
grep -l 'labels:.*type/bug' tmp/issues/*.md grep -l 'labels:.*type/bug' tmp/issues/*.md
grep -ln 'depends:.*migrate-schema' tmp/issues/*.md # who depends on it grep -ln 'depends:.*migrate-schema' tmp/issues/*.md # who depends on it
""" """
import collections
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.
@@ -80,13 +141,20 @@ EXCLUSIVE_NS = ("type/", "severity/")
REQUIRED_SECTIONS = ["## Summary", "## Spec"] REQUIRED_SECTIONS = ["## Summary", "## Spec"]
AC_SECTION = "## Acceptance criteria" AC_SECTION = "## Acceptance criteria"
DEPENDS_SECTION = "## Depends on" DEPENDS_SECTION = "## Depends on"
ISSUES_SECTION = "## Issues"
# Both sections name what an issue depends on, so both are edge sources and
# both point the same way. In a `type/feature` that reads container -> child:
# "the container is closed when its children are closed" IS a dependency.
# "a child belongs to a feature" is membership, and membership has no place in
# a dependency graph — which is why a child never names its container back.
DEP_SECTIONS = (DEPENDS_SECTION, ISSUES_SECTION)
# Per-type sections from the templates — absence is a warning, not a stop. # Per-type sections from the templates — absence is a warning, not a stop.
EXPECTED_SECTIONS = { EXPECTED_SECTIONS = {
"bug": ["## Steps to reproduce", "## Expected", "## Actual", "## Environment"], "bug": ["## Steps to reproduce", "## Expected", "## Actual", "## Environment"],
"task": ["## Motivation"], "task": ["## Motivation"],
"refactor": ["## Motivation", "## Invariants"], "refactor": ["## Motivation", "## Invariants"],
"test": ["## Motivation", "## Test cases"], "test": ["## Motivation", "## Test cases"],
"feature": ["## Motivation", "## Issues"], "feature": ["## Motivation", ISSUES_SECTION],
"draft": ["## Notes"], "draft": ["## Notes"],
} }
@@ -267,24 +335,158 @@ def section_body(body, header):
return "\n".join(out).strip() return "\n".join(out).strip()
def body_dep_refs(body): def body_dep_ref_sections(body):
"""Tokens referenced from `## Depends on` / `## Issues` only — never from """[(section, ref)] for every reference under one of DEP_SECTIONS — never
prose, or a graph walk would drag in half the backlog. Returns whatever was from prose, or a graph walk would drag in half the backlog. Refs are
written there (slugs, and `#N` on issues that came from a tracker).""" whatever was written there (slugs, and `#N` on issues that came from a
out, active = [], False tracker), deduplicated on first sight.
The section is carried out with the ref so a caller can name the one the
reader actually has in front of them: a container's children come from
`## Issues`, and pointing at `## Depends on` would name a section that is
not in the file."""
out, seen, section = [], set(), ""
for line in (body or "").splitlines(): for line in (body or "").splitlines():
if line.startswith("## "): if line.startswith("## "):
active = line.strip() in (DEPENDS_SECTION, "## Issues") head = line.strip()
section = head if head in DEP_SECTIONS else ""
continue continue
if not active: if not section:
continue continue
for tok in re.findall(r'#(\d+)|\b([a-z0-9]+(?:-[a-z0-9]+)+)\b', line): for tok in re.findall(r'#(\d+)|\b([a-z0-9]+(?:-[a-z0-9]+)+)\b', line):
ref = ("#" + tok[0]) if tok[0] else tok[1] ref = ("#" + tok[0]) if tok[0] else tok[1]
if ref not in out: if ref not in seen:
out.append(ref) seen.add(ref)
out.append((section, ref))
return out return out
def body_dep_refs(body):
"""Just the refs, in order of first appearance."""
return [ref for _, ref in body_dep_ref_sections(body)]
# --------------------------------------------------------------------------
# checkboxes
# --------------------------------------------------------------------------
# A checkbox is the one part of a body that is *state* and not prose, so the
# format gives it markup of its own (references/format.md:163-164). It is item
# markup, not a property of one section: `## Acceptance criteria` is the usual
# home, but a type/feature keeps its children as checkboxes under `## Issues`
# (format.md:275-277). The scan is therefore over the whole text and the
# heading is only recorded, never required.
CHECKBOX_RE = re.compile(
r'^(?P<indent>[ \t]*)(?P<marker>[-*+]|\d+[.)])[ \t]+'
r'\[(?P<box>[ xX])\](?=[ \t]|$)(?P<text>.*)$')
# Any list item — a sibling ends the item above it, checkbox or not.
LIST_ITEM_RE = re.compile(r'^[ \t]*([-*+]|\d+[.)])([ \t]|$)')
FENCE_RE = re.compile(r'^[ \t]{0,3}(`{3,}|~{3,})')
Checkbox = collections.namedtuple(
"Checkbox", "index line end_line checked text section")
def checkboxes(text):
"""Every checkbox item in `text`, in document order.
A pure function of the string it is given — no I/O, no store, no tracker.
Pass an issue body (`Issue.body`) to get body-relative line numbers, or a
whole file to get file-relative ones; nothing else changes.
Returns a list of `Checkbox` namedtuples:
index 1-based position in this list — what a user types to pick it
line 1-based line of the `- [ ]` marker, in the text given
end_line 1-based last line of the item, continuation lines included
checked True for `[x]` / `[X]`, False for `[ ]`
text the item's text; continuation lines joined with one space
section nearest preceding `## ` heading, "" above the first one
Rules:
- Only a line matching CHECKBOX_RE opens an item. A wrapped ("continuation")
line is part of the item above it, never an item of its own; the item
runs to the next blank line, heading, code fence, or list marker.
- Fenced code blocks are skipped whole: `- [ ]` inside a ``` fence is an
example of the markup, not a box anybody may tick.
- `-`, `*`, `+` and `1.` markers all count, at any indentation, so nested
lists are seen too.
"""
lines = (text or "").splitlines()
items, section, fence = [], "", ""
for n, line in enumerate(lines, 1):
m = FENCE_RE.match(line)
if m:
tok = m.group(1)
if not fence:
fence = tok
elif tok[0] == fence[0] and len(tok) >= len(fence):
fence = ""
continue
if fence:
continue
if line.startswith("## "):
section = line.strip()
continue
if line.startswith("# "):
section = ""
continue
m = CHECKBOX_RE.match(line)
if not m:
continue
end, parts = n, [m.group("text").strip()]
for k in range(n, len(lines)): # lines[k] is line number k + 1
nxt = lines[k]
if (not nxt.strip() or nxt.startswith("#")
or FENCE_RE.match(nxt) or LIST_ITEM_RE.match(nxt)):
break
end = k + 1
parts.append(nxt.strip())
items.append(Checkbox(len(items) + 1, n, end,
m.group("box") != " ",
" ".join(p for p in parts if p), section))
return items
def set_checkbox(text, item, checked=True):
"""Return `text` with one checkbox set to `checked`.
Pure, and deliberately surgical: exactly one character of the input
changes — the one between the brackets. Everything else, including
trailing whitespace and the item's own wording, comes back byte for byte.
That is the whole point of the function: ticking a box must not produce a
diff wider than the state that changed.
`item` is a `Checkbox` from `checkboxes(text)` — the same text, or the
line number will point at the wrong line — or a 1-based line number.
Already in the requested state is a no-op: `text` is returned unchanged,
and an existing `[X]` keeps its capital.
"""
line_no = item.line if isinstance(item, Checkbox) else int(item)
off = 0
for n, raw in enumerate(text.splitlines(True), 1):
if n == line_no:
m = CHECKBOX_RE.match(raw.rstrip("\r\n"))
if not m:
raise ValueError("line %d is not a checkbox item" % line_no)
if (m.group("box") != " ") == bool(checked):
return text
box = off + m.start("box")
return text[:box] + ("x" if checked else " ") + text[box + 1:]
off += len(raw)
raise ValueError("line %d is past the end of the text" % line_no)
def checkbox_progress(text):
"""(done, total) over every checkbox in `text`; (0, 0) when it has none.
Computed on the fly, on purpose. Progress is not a metadata field: it is
the body read back, and the body is the only place the state lives."""
items = checkboxes(text)
return sum(1 for c in items if c.checked), len(items)
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
# validation # validation
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
@@ -342,12 +544,17 @@ def validate(issue, known_ids=None):
warn.append("depends on %r, which is not in the store" % d) warn.append("depends on %r, which is not in the store" % d)
# `depends:` is the machine-readable graph; the body section is prose for # `depends:` is the machine-readable graph; the body section is prose for
# humans. They drift silently unless something says so. # humans. They drift silently unless something says so. Name the section
# the reference actually came from — for a container that is `## Issues`.
listed = set(issue.depends) listed = set(issue.depends)
for ref in body_dep_refs(issue.body): for section, ref in body_dep_ref_sections(issue.body):
if not ref.startswith("#") and ref not in listed: if not ref.startswith("#") and ref not in listed:
warn.append("%s mentions %r but `depends:` does not list it" warn.append("%s mentions %r but `depends:` does not list it"
% (DEPENDS_SECTION, ref)) % (section, ref))
# An unticked checkbox is never a finding — neither an error nor a
# warning. `- [ ]` is work not done yet, which is the normal state of a
# perfectly well-formed issue. Reading that state is issue_ac.py's job.
return err, warn return err, warn
@@ -356,6 +563,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 +633,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())
+134
View File
@@ -0,0 +1,134 @@
#!/usr/bin/env python3
"""
issue_ac.py — list and tick the checkboxes in an issue's body. Offline.
issue_ac.py wire-sqlc-appclick numbered list with state
issue_ac.py wire-sqlc-appclick --check 3 by number
issue_ac.py wire-sqlc-appclick --check регресс by substring
issue_ac.py wire-sqlc-appclick --uncheck 3
A checkbox is the one part of a body that is *state* and not prose. Everything
else is written once; boxes get ticked as the work goes, and until now the only
ways to tick one were a human with an editor or a model rewriting the whole
body — the second worse than the first, because the rewrite re-flows the text
and the issue's diff swells around a change of one character. This changes that
one character and nothing else.
Named after `## Acceptance criteria`, where most boxes live, but every checkbox
in the body is listed and tickable: a type/feature keeps its children under
`## Issues`, and binding this to one heading would silently lose half of them.
A substring picks an item only when it picks exactly one. Two matches is an
error listing both — a coin flip would tick the wrong box and look like it
worked.
Delivering the changed body to a tracker is not part of this: that is
`push.py --update` in /tea:sync.
"""
import argparse
import os
import re
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import issue # noqa: E402
import issue_index # noqa: E402
NUMBER = re.compile(r'^\d+$')
def box(c):
return "[x]" if c.checked else "[ ]"
def listing(items):
"""The numbered list, grouped by the heading each item sits under."""
out, section = [], None
for c in items:
if c.section != section:
section = c.section
out.append("")
out.append(section or "(above the first heading)")
out.append(" %2d %s %s" % (c.index, box(c), c.text))
return out
def select(items, needle):
"""Resolve a --check/--uncheck argument to exactly one item, or exit."""
needle = (needle or "").strip()
if not needle:
sys.exit("issue_ac.py: empty selector — give an item number or a substring")
if NUMBER.match(needle):
n = int(needle)
if not 1 <= n <= len(items):
sys.exit("issue_ac.py: no item %d — the issue has %d" % (n, len(items)))
return items[n - 1]
hits = [c for c in items if needle.lower() in c.text.lower()]
if not hits:
sys.exit("issue_ac.py: nothing matches %r" % needle)
if len(hits) > 1:
sys.exit("\n".join(
["issue_ac.py: %r matches %d items — narrow it down, or use a number:"
% (needle, len(hits))]
+ [" %2d %s %s" % (c.index, box(c), c.text) for c in hits]))
return hits[0]
def main(argv=None):
ap = argparse.ArgumentParser(
description="List and tick an issue's checkboxes (offline)")
ap.add_argument("id", help="issue id (the slug, without .md)")
g = ap.add_mutually_exclusive_group()
g.add_argument("--check", metavar="N|TEXT", help="tick one item: number or substring")
g.add_argument("--uncheck", metavar="N|TEXT", help="untick one item: number or substring")
ap.add_argument("--out", default=issue.ISSUE_ROOT, help="store root (default: tmp/issues)")
args = ap.parse_args(argv)
path = issue.path_of(args.out, args.id)
if not os.path.exists(path):
sys.exit("issue_ac.py: no issue %r in %s" % (args.id, args.out))
# newline="": no translation in either direction. Byte-for-byte means the
# line endings too — reading a CRLF file in text mode and writing it back
# would rewrite every line while claiming to have changed one character.
with open(path, newline="") as f:
text = f.read()
# The whole file, not just the body: line numbers then point at the file,
# and the metadata block is rewritten by nobody. Round-tripping through
# Issue.to_text() would re-render metadata and re-strip the body, which is
# exactly the byte-level churn this script exists to avoid.
items = issue.checkboxes(text)
needle = args.check if args.check is not None else args.uncheck
if not items:
if needle is not None:
sys.exit("issue_ac.py: %s has no checkboxes" % args.id)
print("%s — no checkboxes" % args.id)
return 0
if needle is None:
done = sum(1 for c in items if c.checked)
print("%s%d/%d %s" % (args.id, done, len(items), path))
print("\n".join(listing(items)))
return 0
checked = args.check is not None
item = select(items, needle)
new = issue.set_checkbox(text, item, checked)
verb = "checked" if checked else "unchecked"
if new == text:
print("unchanged %2d %s %s" % (item.index, box(item), item.text))
return 0
with open(path, "w", newline="") as f:
f.write(new)
issue_index.build(args.out)
done, total = issue.checkbox_progress(new)
print("%s %2d %s %s" % (verb, item.index, "[x]" if checked else "[ ]", item.text))
print("%s%d/%d %s:%d" % (args.id, done, total, path, item.line))
return 0
if __name__ == "__main__":
sys.exit(main())
+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
+36 -9
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
@@ -26,7 +30,20 @@ def cell(v):
return v.replace("|", "\\|") or "" return v.replace("|", "\\|") or ""
def progress(body):
"""`3/7` for a body with checkboxes, "" for one without.
Counted from the body every time the index is built and stored nowhere —
the boxes are the state, and a second copy of it in a metadata field would
be wrong by the next edit."""
done, total = issue.checkbox_progress(body)
return "%d/%d" % (done, total) if total else ""
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):
@@ -35,6 +52,7 @@ def build(root):
rows.append({ rows.append({
"id": i, "id": i,
"state": cell(iss.state), "state": cell(iss.state),
"progress": progress(iss.body),
"type": cell(iss.type), "type": cell(iss.type),
"labels": cell(rest), "labels": cell(rest),
"title": cell(iss.title), "title": cell(iss.title),
@@ -50,13 +68,16 @@ def build(root):
"Every issue this project knows about. `origin: local` means it " "Every issue this project knows about. `origin: local` means it "
"exists nowhere else — a complete state, not a pending one. Any " "exists nowhere else — a complete state, not a pending one. Any "
"other value names the tracker it also lives in; the handle is in " "other value names the tracker it also lives in; the handle is in "
"the file. Rebuild with `issue_index.py`.", ""] "the file. `progress` counts the body's checkboxes, ticked over "
"total, and is blank for an issue that has none — read off the "
"body at build time, stored nowhere. Rebuild with `issue_index.py`; "
"tick a box with `issue_ac.py`.", ""]
if rows: if rows:
out += ["| id | state | type | labels | title | milestone | depends | origin |", out += ["| id | state | progress | type | labels | title | milestone | depends | origin |",
"|---|---|---|---|---|---|---|---|"] "|---|---|---|---|---|---|---|---|---|"]
out += ["| [%s](%s.md) | %s | %s | %s | %s | %s | %s | %s |" % ( out += ["| [%s](%s.md) | %s | %s | %s | %s | %s | %s | %s | %s |" % (
r["id"], r["id"], r["state"], r["type"], r["labels"], r["title"], r["id"], r["id"], r["state"], r["progress"], r["type"], r["labels"],
r["milestone"], r["depends"], r["origin"]) for r in rows] r["title"], r["milestone"], r["depends"], r["origin"]) for r in rows]
else: else:
out.append("_empty_") out.append("_empty_")
@@ -71,7 +92,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 +99,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()
# 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) 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
+87 -8
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
@@ -83,7 +89,9 @@ not one per issue. Filters AND together; `--state` defaults to `open`;
`--limit` to 100. Keys and filters are mutually exclusive. `--limit` to 100. Keys and filters are mutually exclusive.
**A pull overwrites the local body.** It is a fetch, not a merge — unpushed **A pull overwrites the local body.** It is a fetch, not a merge — unpushed
local edits are lost. `--cached` skips issues already on disk. local edits are lost, with one exception: [checkbox
state](#checkboxes-are-the-one-exception). `--cached` skips issues already on
disk.
**Closed issues stay out of the store.** In filter mode they are enumerated **Closed issues stay out of the store.** In filter mode they are enumerated
but not written: `--state all` still shows the whole picture, only `--state but not written: `--state all` still shows the whole picture, only `--state
@@ -113,6 +121,39 @@ Two traps this handles for you:
After a pull, draw the graph with `/tea:issue`'s `issue_tree.py` — offline, no After a pull, draw the graph with `/tea:issue`'s `issue_tree.py` — offline, no
extra requests. extra requests.
### Checkboxes are the one exception
A checkbox is state, not prose, and it is the one thing a pull does **not**
overwrite. For a checkbox line whose **text** matches a line in the local copy,
`[x]` wins from whichever side has it — tick it in the web UI, tick it locally,
tick it in both, the tick survives.
| part of the body | what a pull does to it |
|---|---|
| prose, headings, everything not a checkbox | overwritten from the server, whole, as before |
| a checkbox whose text is in the local copy | `[x]` from **either** side wins |
| a checkbox whose text is not in the local copy | taken from the server as it stands, ticked or not |
| any issue the store has never seen | written exactly as the server sent it |
This is not drift tracking — [Drift](#drift) stands. A tick is **monotone**: an
item only travels `[ ]``[x]`, so joining the two sides is a set union, not a
conflict to resolve. No base version is kept and nothing is compared against
one; one rule for one line type replaces the whole mechanism.
**The price, and it is real: a box unticked in the web UI comes back on the next
pull.** Unticking is not monotone, so the union cannot see it. Untick locally,
then `push.py --update` — the body goes up whole and the server follows.
Matching is on the item's text after the domain parser has stripped it and
rejoined wrapped lines with single spaces, so rewrapping a long item keeps its
tick. Rewording one does not: different text is a different item. The same text
twice in a body is read as a set — one ticked local copy ticks every server line
with that text.
The parsing is `/tea:issue`'s (`issue.checkboxes` / `issue.set_checkbox`),
imported, never reimplemented here. The rule itself is
`map.merge_checkbox_state`: pure, and testable without a Gitea anywhere.
## Pushing ## Pushing
```bash ```bash
@@ -131,10 +172,40 @@ at most one `severity/*`, English title with no type prefix, `## Summary` /
`## Spec` / `## Acceptance criteria` present). `--force` posts anyway — say why `## Spec` / `## Acceptance criteria` present). `--force` posts anyway — say why
when you use it. when you use it.
Issues go up in topological order, dependencies first. A dependency that is ### Dependencies
still local-only is reported, not silently dropped: the body's `## Depends on`
prose is sent verbatim either way, but the `#N` cross-link will be missing Issues go up in topological order, dependencies first, and **the graph goes up
until that issue is pushed too. with them**. Once an issue has its number, every `depends:` entry that also has
one becomes a native Gitea link, so the tracker shows the blocking panel and
refuses to close a blocked issue before its blocker.
The two directions are symmetric, and they use the same endpoint:
| | direction | endpoint |
|---|---|---|
| `push.py` | `depends:` → native links | `POST …/issues/{n}/dependencies` |
| `pull.py --deps` | native links → `depends:` | `GET …/issues/{n}/dependencies` |
The POST body is Gitea's `IssueMeta``{"index", "owner", "repo"}` naming the
**blocker**, posted to the **blocked** issue's endpoint ("make the issue in the
url depend on the issue in the form"). `owner`/`repo` travel with it, so a
dependency in another repo links correctly.
- Topological order means the blocker already has its number — no second pass.
- A link the tracker already has is skipped: push GETs the existing ones first,
so a repeat push is a no-op and a 409 never happens. Should a link fail
anyway, it is a warning, not a dead run — the issues are already created.
- `--update` carries links that appeared in `depends:` after the first push.
- `--dry-run` prints every link it would make (`#?` for a number this run has
not handed out yet) and makes no request at all.
- **Removing a link is out of scope.** Push only adds. A dependency deleted
from `depends:` leaves its Gitea link standing; drop it in the web UI or with
`tea api -X DELETE …/issues/N/dependencies`.
A dependency that is still local-only is reported, not silently dropped: it has
no number, so it gets no link. The body's `## Depends on` prose is sent verbatim
either way — nothing is lost, but the tracker shows no edge until that issue is
pushed too.
Missing labels are created with the canonical color and, for `type/*` and Missing labels are created with the canonical color and, for `type/*` and
`severity/*`, `exclusive: true``tea labels create` cannot set that field `severity/*`, `exclusive: true``tea labels create` cannot set that field
@@ -173,19 +244,22 @@ never check out, create, or write anything.
| domain | Gitea | note | | domain | Gitea | note |
|---|---|---| |---|---|---|
| `id` (slug) | — | local only; the tracker never sees it | | `id` (slug) | — | local only; the tracker never sees it |
| title, body | `title`, `body` | verbatim, both directions | | title | `title` | verbatim, both directions |
| body | `body` | verbatim up; verbatim down except checkbox state, which is unioned |
| `state` | `state` | same vocabulary | | `state` | `state` | same vocabulary |
| `labels` | `labels[]` | names both ways; ids only on write | | `labels` | `labels[]` | names both ways; ids only on write |
| `assignees` | `assignees[]` | logins | | `assignees` | `assignees[]` | logins |
| `milestone` | `milestone.title` | resolved to an id on write | | `milestone` | `milestone.title` | resolved to an id on write |
| `depends` | — | slugs; seeded from `#N` on pull | | `depends` | native links | slugs here, `IssueMeta` there; push writes them, `pull --deps` reads them |
| — | `ref` | lands in `branch:`; sent only when non-empty | | — | `ref` | lands in `branch:`; sent only when non-empty |
| — | `number`, `html_url` | lands in `gitea:` / `url:` | | — | `number`, `html_url` | lands in `gitea:` / `url:` |
`depends:` is always slugs. The body's `## Depends on` section is human prose `depends:` is always slugs. The body's `## Depends on` section is human prose
and is passed through **unchanged** in both directions: a pull seeds `depends:` and is passed through **unchanged** in both directions: a pull seeds `depends:`
from the `#N` it finds there, a push never rewrites what the author wrote. A from the `#N` it finds there, a push never rewrites what the author wrote. A
translator that edits prose churns the body on every round trip. translator that edits prose churns the body on every round trip. The edge the
tracker acts on is the native link, not the text — which is exactly why the
text can be left alone.
Comments are **pull-only** in the store: `<id>.comments.md` is written by Comments are **pull-only** in the store: `<id>.comments.md` is written by
`pull.py` and `comment.py`, and editing it by hand changes nothing in Gitea. `pull.py` and `comment.py`, and editing it by hand changes nothing in Gitea.
@@ -197,6 +271,11 @@ nothing reconciles, nothing warns that a synced issue changed upstream.
`synced:` tells you how old your copy is; `remote-updated:` what the server `synced:` tells you how old your copy is; `remote-updated:` what the server
said at that moment. Re-pull when it matters. said at that moment. Re-pull when it matters.
Checkbox state is not an exception to this. The union a pull applies reads only
the two bodies in front of it — there is no base version, no history, and no
way for it to report that anything diverged. One rule for one line type,
precisely so the mechanism this section rules out is not needed.
## Rich payloads for everything else ## Rich payloads for everything else
Comments and issues are wrapped by the scripts above. For **other** entities Comments and issues are wrapped by the scripts above. For **other** entities
+40
View File
@@ -226,6 +226,46 @@ def native_deps(login, base, number):
return [i["number"] for i in got] if isinstance(got, list) else [] return [i["number"] for i in got] if isinstance(got, list) else []
def native_dep_pairs(login, base, number):
"""The same links as {(owner/repo, number)} — what a repeat push compares
against so it does not POST a link the tracker already has.
A bare number is ambiguous the moment a dependency lives in another repo,
and IssueMeta lets it, so the repo travels with it. The pair is a transport
fact; formatting it as `owner/repo#42` is map.py's job, not this module's."""
got = api(login, "%s/issues/%d/dependencies" % (base, number), allow_fail=True)
out = set()
for i in got if isinstance(got, list) else []:
repo = (i.get("repository") or {}).get("full_name") or ""
if "number" in i:
out.add((repo, int(i["number"])))
return out
def add_dependency(login, base, number, dep_repo, dep_number, out_root=None):
"""Make issue `number` depend on `dep_repo#dep_number`. True on success.
Confirmed against the instance's own swagger.v1.json (Gitea 1.26.1):
POST /repos/{owner}/{repo}/issues/{index}/dependencies
body: IssueMeta — {"index": <int>, "owner": "<owner>", "repo": "<name>"}
"Make the issue in the url depend on the issue in the form."
The URL names the blocked issue and the body the blocker, which is the same
direction native_deps reads back ("all issues that block this issue"). A
link that already exists answers 409, so a failure here is reported and not
fatal: one missing cross-link must not abort a push that has already
created issues. Callers pre-filter with native_dep_pairs."""
owner, _, name = (dep_repo or "").partition("/")
if not owner or not name:
return False
payload = {"index": int(dep_number), "owner": owner, "repo": name}
got = api(login, "%s/issues/%d/dependencies" % (base, number), "POST", payload,
payload_name="dep-%d-%d" % (number, dep_number),
out_root=out_root, allow_fail=True)
return got is not None
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
# labels # labels
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
+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)
+53 -4
View File
@@ -16,7 +16,10 @@ What crosses the boundary, and what does not:
domain Gitea note domain Gitea note
---------------------------------------------------------------------- ----------------------------------------------------------------------
id (slug) — local only; the tracker never sees it id (slug) — local only; the tracker never sees it
title, body title, body verbatim, both ways title title verbatim, both ways
body body verbatim up, verbatim down except
checkbox state — see
merge_checkbox_state
state state open/closed, same vocabulary state state open/closed, same vocabulary
labels labels[] names both ways; ids only on write labels labels[] names both ways; ids only on write
assignees assignees[] logins assignees assignees[] logins
@@ -104,13 +107,59 @@ def numbers_in_body(body):
return [int(r[1:]) for r in issue.body_dep_refs(body) if r.startswith("#")] return [int(r[1:]) for r in issue.body_dep_refs(body) if r.startswith("#")]
def from_api(payload, id, repo, id_for_number=None, extra_numbers=(), synced=None): def merge_checkbox_state(remote_body, local_body):
"""The remote body with every tick the local copy already had put back.
The one exception to "a pull overwrites the body", and it is deliberately
the narrowest one that works. A tick is **monotone** — an item only ever
travels `[ ]` -> `[x]` — so the two sides are joined by a set union, not
reconciled: no base version, no drift tracking, no conflict to resolve. The
set is a set of item TEXTS, and an item comes out ticked when either side
has it ticked. Everything else in the body is still the remote's word.
Matching is on `Checkbox.text`, which the domain parser has already
stripped and rejoined with single spaces, so rewrapping a long item does
not cost it its tick. It is otherwise literal: reword an item and it is a
different item — the tick stays with the wording it was put on.
**The same text more than once** is read as the rule says, as a set: one
ticked local item ticks every remote item with that text. The alternative —
pairing duplicates up by order — is the reading that can still drop a tick
(local `[ ]` then `[x]`, remote a single line: the ticked one pairs with
nothing), and dropping a tick is the bug this exists to fix. Two items
whose text is identical are the same item to whoever reads them.
Pure: no store, no tracker, no I/O. A `local_body` of None or "" — a first
pull, an empty store — returns the remote body untouched.
The price, accepted explicitly: UNticking is not monotone, so a box
unticked in the web UI comes back on the next pull. Untick locally, push.
"""
ticked = {c.text for c in issue.checkboxes(local_body) if c.checked}
if not ticked:
return remote_body
body = remote_body
# set_checkbox trades one character for one character, so line numbers read
# off `remote_body` stay valid against the partially rewritten `body`.
for c in issue.checkboxes(remote_body):
if not c.checked and c.text in ticked:
body = issue.set_checkbox(body, c.line, True)
return body
def from_api(payload, id, repo, id_for_number=None, extra_numbers=(), synced=None,
local_body=None):
"""Build a domain Issue from a Gitea issue payload. """Build a domain Issue from a Gitea issue payload.
id_for_number maps a Gitea number to a local slug — dependencies whose id_for_number maps a Gitea number to a local slug — dependencies whose
target has not been pulled yet are dropped from `depends:` (the body still target has not been pulled yet are dropped from `depends:` (the body still
names them, so nothing is lost) rather than invented.""" names them, so nothing is lost) rather than invented.
body = (payload.get("body") or "").strip()
`local_body` is the body of the copy already in the store, when there is
one. It contributes exactly one thing: its ticked checkboxes survive the
overwrite (merge_checkbox_state). Pass None and the remote body is taken
whole, which is what a first pull does."""
body = merge_checkbox_state((payload.get("body") or "").strip(), local_body)
id_for_number = id_for_number or {} id_for_number = id_for_number or {}
numbers = list(numbers_in_body(body)) numbers = list(numbers_in_body(body))
+19 -4
View File
@@ -43,7 +43,11 @@ Other flags:
--repo owner/repo default: auto-detect from the CWD git remote --repo owner/repo default: auto-detect from the CWD git remote
Pulling overwrites the local body: it is a fetch, not a merge. Local edits you Pulling overwrites the local body: it is a fetch, not a merge. Local edits you
have not pushed are lost. Draw the graph afterwards with the domain's own have not pushed are lost — with exactly one exception, checkbox state. A `[x]`
on either side wins for any item whose text matches, because a tick is monotone
and unioning the two sides is not conflict resolution (gmap.merge_checkbox_state
has the rule and its price). `--cached` skips an issue before any of that: it is
not read and not merged. Draw the graph afterwards with the domain's own
issue_tree.py — it needs no network. issue_tree.py — it needs no network.
Login: the operator's pin from .claude/settings.local.json (see /tea:auth). Login: the operator's pin from .claude/settings.local.json (see /tea:auth).
@@ -108,7 +112,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 +123,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 ------------------------------------------------------
@@ -181,13 +191,18 @@ def main():
store_ids.add(id) store_ids.add(id)
number_of_id[number] = id number_of_id[number] = id
if args.cached and stored: if args.cached and stored:
skipped.append(id) # untouched, and not one request spent on it skipped.append(id) # untouched, unread, and not one request spent
else: else:
extra = _gitea.native_deps(login, base, number) if args.deps else [] extra = _gitea.native_deps(login, base, number) if args.deps else []
# The copy already on disk, as it was when this run started. It
# contributes its ticked checkboxes and nothing else; None when
# the store has never seen this issue.
prev = issues.get(id)
iss, unresolved = gmap.from_api(payload, id, repo, iss, unresolved = gmap.from_api(payload, id, repo,
id_for_number=number_of_id, id_for_number=number_of_id,
extra_numbers=extra, extra_numbers=extra,
synced=_gitea.now_iso()) synced=_gitea.now_iso(),
local_body=prev.body if prev else None)
issue.save(root, iss) issue.save(root, iss)
sync_comments(login, base, root, id, number, payload.get("comments") or 0) sync_comments(login, base, root, id, number, payload.get("comments") or 0)
remote_map[gmap.remote_key(repo, number)] = id remote_map[gmap.remote_key(repo, number)] = id
+80 -6
View File
@@ -20,7 +20,25 @@ anyway; say why when you use it.
Dependencies are pushed in topological order so a parent is created after the Dependencies are pushed in topological order so a parent is created after the
issues it depends on. A dependency that is still local-only is reported, not issues it depends on. A dependency that is still local-only is reported, not
silently dropped — the body's `## Depends on` prose is sent verbatim either silently dropped — the body's `## Depends on` prose is sent verbatim either
way, so nothing is lost, but the `#N` cross-links will be missing. way, so nothing is lost, but the tracker shows no edge for it.
The graph goes up with them. Once an issue has its number, every `depends:`
entry that also has one becomes a **native Gitea link** — the same
`/dependencies` that `pull.py --deps` reads back, so the tracker shows the
blocking panel and refuses to close a blocked issue first. Topological order
means the blocker already has its number by then; no second pass is needed.
`--update` links whatever appeared in `depends:` since the last push. A link
the tracker already has is skipped, not re-POSTed. A dependency that stayed
local has no number and becomes no link — only the warning above.
REMOVING a link is OUT OF SCOPE. Push only ever adds: a dependency deleted
from `depends:` leaves its Gitea link standing, and nothing here will notice.
Unlink it in the web UI, or by hand with
`tea api -X DELETE --login "$GITEA_LOGIN" repos/OWNER/REPO/issues/N/dependencies`.
The `## Depends on` prose itself is never touched — slugs stay slugs and are
not rewritten to `#N`, so the body survives a pull -> push round trip byte for
byte. The link lives in Gitea's own graph, not in the text.
Missing labels are created with the canonical color and, for type/* and Missing labels are created with the canonical color and, for type/* and
severity/*, `exclusive: true` — `tea labels create` cannot set that field. severity/*, `exclusive: true` — `tea labels create` cannot set that field.
@@ -67,6 +85,27 @@ def select(issues, ids, update):
return chosen return chosen
def dep_state(iss, issues, pushing):
"""What each `depends:` entry is, as far as linking is concerned.
Yields (slug, remote_key, in_run) per dependency that exists in the store:
remote_key the dependency's `gitea:` value, or None while it is local
in_run this push is about to give it one
In the real run remote_key is all that matters — topological order means an
in-run blocker has already been stamped by the time its dependent is sent.
`--dry-run` has no numbers to stamp, so it leans on in_run to say which
links are coming and which cannot exist at all."""
out = []
for d in iss.depends:
dep = issues.get(d)
if dep is None:
continue # not in the store; validate() already warned
out.append((d, dep.extra.get("gitea") or None, d in pushing))
return out
def git_branch(): def git_branch():
"""The branch HEAD is on, or None. The only git call these scripts make — """The branch HEAD is on, or None. The only git call these scripts make —
read, never write. A detached HEAD prints `HEAD` and outside a repo git read, never write. A detached HEAD prints `HEAD` and outside a repo git
@@ -90,13 +129,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)
@@ -133,12 +174,27 @@ def main():
_gitea.warn("no current git branch (detached HEAD, or outside a git repo) " _gitea.warn("no current git branch (detached HEAD, or outside a git repo) "
"— no `ref` on: %s" % ", ".join(blank)) "— no `ref` on: %s" % ", ".join(blank))
pushing = set(order)
if args.dry_run: if args.dry_run:
links = 0
for id in order: for id in order:
iss = issues[id] iss = issues[id]
print("ok %s [type/%s] %s (%s)" print("ok %s [type/%s] %s (%s)"
% (id, iss.type or "?", iss.title, ", ".join(iss.labels) or "no labels")) % (id, iss.type or "?", iss.title, ", ".join(iss.labels) or "no labels"))
print("%d issue(s) would be %s" % (len(order), "updated" if args.update else "created")) # Not one request is made here: everything below is read off the
# store. `#?` is a number this run has not handed out yet.
for slug, key, in_run in dep_state(iss, issues, pushing):
if key:
print(" link -> %s (%s)" % (key, slug))
links += 1
elif in_run:
print(" link -> #? (%s, created by this run)" % slug)
links += 1
else:
print(" no link: %s is local-only" % slug)
print("%d issue(s) would be %s, %d dependency link(s) would be created"
% (len(order), "updated" if args.update else "created", links))
return return
login = _gitea.require_login() login = _gitea.require_login()
@@ -157,7 +213,7 @@ def main():
unsynced = [d for d in iss.depends unsynced = [d for d in iss.depends
if d in issues and not issues[d].extra.get("gitea") if d in issues and not issues[d].extra.get("gitea")
and d not in order] and d not in pushing]
if unsynced: if unsynced:
_gitea.warn("%s: depends on local-only issue(s) %s — no #N cross-link in Gitea" _gitea.warn("%s: depends on local-only issue(s) %s — no #N cross-link in Gitea"
% (id, ", ".join(unsynced))) % (id, ", ".join(unsynced)))
@@ -202,6 +258,24 @@ def main():
remote_map[gmap.remote_key(repo, number)] = id remote_map[gmap.remote_key(repo, number)] = id
print("%s %s #%d %s" % (verb, id, number, got.get("html_url", ""))) print("%s %s #%d %s" % (verb, id, number, got.get("html_url", "")))
# ---- the graph, as Gitea's own links ------------------------------
# Blockers came first in topological order, so each one that is going
# to have a number has one already — the store was stamped in place.
# The GET is the idempotence check: it costs one request per issue that
# has dependencies at all, and it is what makes a repeat push a no-op.
wanted_links = [(slug, gmap.parse_remote_key(key))
for slug, key, _ in dep_state(iss, issues, pushing) if key]
if wanted_links:
have = _gitea.native_dep_pairs(login, base, number)
for slug, (drepo, dnum) in wanted_links:
if not dnum or (drepo, dnum) in have:
continue
if _gitea.add_dependency(login, base, number, drepo, dnum, root):
print(" depends on %s#%d (%s)" % (drepo, dnum, slug))
else:
_gitea.warn("%s: could not link #%d -> %s#%d (%s) — link it by "
"hand or re-run push" % (id, number, drepo, dnum, slug))
_gitea.save_map(root, remote_map) _gitea.save_map(root, remote_map)
path, n = issue_index.build(root) path, n = issue_index.build(root)
print("index: %s%d issue(s)" % (path, n)) print("index: %s%d issue(s)" % (path, n))
+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()
+407
View File
@@ -0,0 +1,407 @@
#!/usr/bin/env python3
"""
Checkbox state survives a pull; everything else in the body does not.
Two levels, on purpose. `map.merge_checkbox_state` is pure, so most of the rule
is pinned down with plain strings and no store anywhere. The pull tests then
prove the rule is actually wired into the write path, with the transport
stubbed at the one seam `test_push_dependencies.py` uses — `_gitea.api`, the
single function that shells out to `tea`. Nothing here touches a network, and
no test may ever be made to.
`skills/*/scripts/` are not packages; they go on sys.path by hand.
"""
import contextlib
import io
import os
import re
import shutil
import sys
import tempfile
import unittest
import urllib.parse
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 issue # noqa: E402
import map as gmap # noqa: E402
import pull # noqa: E402
REPO = "claude-skills/tea"
BASE = "repos/%s" % REPO
def body(*criteria, **kw):
"""A body in the canonical shape, with the given `## Acceptance criteria`."""
summary = kw.get("summary", "Прозаическое описание.")
return ("## Summary\n%s\n\n## Spec\nnone\n\n## Acceptance criteria\n%s\n"
% (summary, "\n".join(criteria))).strip()
# --------------------------------------------------------------------------
# the rule itself — pure, no store, no tracker
# --------------------------------------------------------------------------
class MergeCheckboxStateTest(unittest.TestCase):
"""`[x]` wins from whichever side has it, for a matching item text."""
def test_a_local_tick_survives_the_overwrite(self):
got = gmap.merge_checkbox_state(body("- [ ] первое", "- [ ] второе"),
body("- [x] первое", "- [ ] второе"))
self.assertEqual(got, body("- [x] первое", "- [ ] второе"))
def test_a_remote_tick_is_kept(self):
got = gmap.merge_checkbox_state(body("- [x] первое", "- [ ] второе"),
body("- [ ] первое", "- [ ] второе"))
self.assertEqual(got, body("- [x] первое", "- [ ] второе"))
def test_both_sides_ticked_is_still_ticked(self):
one = body("- [x] первое")
self.assertEqual(gmap.merge_checkbox_state(one, one), one)
def test_the_union_is_taken_item_by_item(self):
got = gmap.merge_checkbox_state(
body("- [x] первое", "- [ ] второе", "- [ ] третье"),
body("- [ ] первое", "- [x] второе", "- [ ] третье"))
self.assertEqual(got, body("- [x] первое", "- [x] второе", "- [ ] третье"))
def test_an_item_the_local_copy_does_not_have_comes_from_the_server(self):
"""Including its state — both states, in both directions."""
got = gmap.merge_checkbox_state(
body("- [x] новое сверху", "- [ ] новое снизу"),
body("- [x] что-то совсем другое"))
self.assertEqual(got, body("- [x] новое сверху", "- [ ] новое снизу"))
def test_prose_is_not_merged(self):
got = gmap.merge_checkbox_state(
body("- [ ] пункт", summary="Новый текст с сервера."),
body("- [x] пункт", summary="Старый локальный текст."))
self.assertIn("Новый текст с сервера.", got)
self.assertNotIn("Старый локальный текст.", got)
self.assertIn("- [x] пункт", got)
def test_a_heading_the_local_copy_added_is_gone(self):
remote = body("- [x] пункт")
got = gmap.merge_checkbox_state(remote, remote + "\n\n## Notes\nмои заметки\n")
self.assertEqual(got, remote)
def test_no_local_copy_returns_the_server_body_untouched(self):
remote = body("- [ ] пункт")
for local in (None, "", " \n"):
self.assertIs(gmap.merge_checkbox_state(remote, local), remote,
"local_body=%r rewrote the body" % local)
def test_nothing_ticked_locally_returns_the_same_object(self):
remote = body("- [ ] пункт")
self.assertIs(gmap.merge_checkbox_state(remote, body("- [ ] пункт")), remote)
def test_no_matching_item_returns_the_same_object(self):
remote = body("- [ ] пункт")
self.assertIs(gmap.merge_checkbox_state(remote, body("- [x] другой")), remote)
def test_a_body_with_no_checkboxes_at_all(self):
remote = "## Summary\nодна проза\n"
self.assertIs(gmap.merge_checkbox_state(remote, "- [x] пункт"), remote)
self.assertEqual(gmap.merge_checkbox_state("- [ ] пункт", remote), "- [ ] пункт")
def test_exactly_one_character_changes(self):
"""Ticking a box must not produce a diff wider than the state."""
remote = body("- [ ] пункт", "- [ ] второй")
got = gmap.merge_checkbox_state(remote, body("- [x] пункт", "- [ ] второй"))
diff = [i for i, (a, b) in enumerate(zip(remote, got)) if a != b]
self.assertEqual(len(remote), len(got))
self.assertEqual(len(diff), 1)
self.assertEqual((remote[diff[0]], got[diff[0]]), (" ", "x"))
def test_a_rewrapped_item_keeps_its_tick(self):
"""`Checkbox.text` joins continuation lines with one space, which is
the whole reason matching survives a reflow."""
remote = body("- [ ] длинный пункт, который сервер\n"
" перенёс на две строки")
got = gmap.merge_checkbox_state(
remote, body("- [x] длинный пункт, который сервер перенёс на две строки"))
self.assertIn("- [x] длинный пункт", got)
def test_a_reworded_item_does_not_keep_its_tick(self):
"""Different text is a different item. The tick stays with the wording
it was put on — this is a match, not a guess."""
got = gmap.merge_checkbox_state(body("- [ ] пункт про sqlc"),
body("- [x] пункт про SQLC"))
self.assertEqual(got, body("- [ ] пункт про sqlc"))
def test_the_marker_style_does_not_have_to_match(self):
"""`-`, `*` and `1.` are all checkbox markers to the domain parser, so
the item is the same item however the two sides chose to render it."""
got = gmap.merge_checkbox_state(body("1. [ ] пункт"), body("* [x] пункт"))
self.assertEqual(got, body("1. [x] пункт"))
def test_a_moved_item_keeps_its_tick(self):
"""Matching is on text alone; the section is not part of the key. An
item promoted out of `## Acceptance criteria` is the same item."""
remote = "## Issues\n- [ ] пункт\n"
got = gmap.merge_checkbox_state(remote, body("- [x] пункт"))
self.assertEqual(got, "## Issues\n- [x] пункт\n")
def test_duplicate_text_is_read_as_a_set(self):
"""The documented reading: one ticked local item ticks every remote
line with that text. Pairing duplicates by order is the alternative,
and it is the one that can still drop a tick."""
got = gmap.merge_checkbox_state(body("- [ ] пункт", "- [ ] пункт"),
body("- [ ] пункт", "- [x] пункт"))
self.assertEqual(got, body("- [x] пункт", "- [x] пункт"))
def test_duplicate_text_never_loses_the_second_tick(self):
"""Two local lines, one remote: order-pairing would drop this tick."""
got = gmap.merge_checkbox_state(body("- [ ] пункт"),
body("- [ ] пункт", "- [x] пункт"))
self.assertEqual(got, body("- [x] пункт"))
def test_an_example_inside_a_code_fence_is_not_ticked(self):
"""The domain parser skips fences whole, and so does the merge: a
`- [ ]` in a fence is markup being shown, not a box anyone may tick."""
remote = "## Spec\n```md\n- [ ] пункт\n```\n\n## Acceptance criteria\n- [ ] пункт\n"
got = gmap.merge_checkbox_state(remote, body("- [x] пункт"))
self.assertEqual(got, remote.replace("## Acceptance criteria\n- [ ] пункт",
"## Acceptance criteria\n- [x] пункт"))
self.assertIn("```md\n- [ ] пункт\n```", got)
def test_an_existing_capital_X_is_left_alone(self):
remote = body("- [X] пункт")
self.assertIs(gmap.merge_checkbox_state(remote, body("- [x] пункт")), remote)
def test_a_capital_X_locally_still_counts_as_ticked(self):
got = gmap.merge_checkbox_state(body("- [ ] пункт"), body("- [X] пункт"))
self.assertEqual(got, body("- [x] пункт"))
def test_unticking_in_the_web_does_not_survive(self):
"""The accepted price, pinned so nobody 'fixes' it by accident:
unticking is not monotone, so a box unticked upstream comes back.
Untick locally and push."""
got = gmap.merge_checkbox_state(body("- [ ] пункт"), body("- [x] пункт"))
self.assertEqual(got, body("- [x] пункт"))
def test_parsing_is_the_domain_layer_s(self):
"""The acceptance criterion, asserted rather than eyeballed: the rule
calls into skills/issue and defines no checkbox syntax of its own."""
with mock.patch.object(issue, "checkboxes", wraps=issue.checkboxes) as cb, \
mock.patch.object(issue, "set_checkbox", wraps=issue.set_checkbox) as sc:
gmap.merge_checkbox_state(body("- [ ] пункт"), body("- [x] пункт"))
self.assertTrue(cb.called)
self.assertTrue(sc.called)
def test_no_checkbox_markup_is_spelled_out_in_the_sync_layer(self):
"""The same criterion from the other side: the bracket markup itself
appears nowhere under skills/sync/scripts. Knowing what `[ ]` looks
like is the domain's job, and there is only one copy of it."""
sync = os.path.join(_ROOT, "skills", "sync", "scripts")
for name in sorted(f for f in os.listdir(sync) if f.endswith(".py")):
with open(os.path.join(sync, name)) as f:
code = f.read().split('"""')[0::2] # docstrings dropped
for chunk in code:
for markup in ("[ xX]", "[xX]", "- [ ]", "- [x]"):
self.assertNotIn(markup, chunk,
"%s spells out %r" % (name, markup))
class FromApiTest(unittest.TestCase):
"""The seam between the rule and the translation."""
PAYLOAD = {"number": 42, "title": "T", "html_url": "u",
"body": body("- [ ] пункт")}
def test_local_body_is_optional_and_defaults_to_no_merge(self):
iss, _ = gmap.from_api(dict(self.PAYLOAD), "an-issue", REPO)
self.assertEqual(iss.body, body("- [ ] пункт"))
def test_local_body_contributes_its_ticks(self):
iss, _ = gmap.from_api(dict(self.PAYLOAD), "an-issue", REPO,
local_body=body("- [x] пункт"))
self.assertEqual(iss.body, body("- [x] пункт"))
def test_an_empty_remote_body_does_not_crash(self):
iss, _ = gmap.from_api({"number": 42, "title": "T", "body": None},
"an-issue", REPO, local_body=body("- [x] пункт"))
self.assertEqual(iss.body, "")
# --------------------------------------------------------------------------
# pull.py — the rule wired into the write path
# --------------------------------------------------------------------------
class FakeGitea(object):
"""A `tea api` that answers issues from memory and remembers the calls."""
def __init__(self):
self.calls = []
self.issues = {}
def add(self, number, title, text, **kw):
p = {"number": number, "title": title, "body": text, "state": "open",
"comments": 0, "html_url": "https://git.example/%s/issues/%d" % (REPO, number),
"labels": [{"name": "type/task"}], "assignees": [], "milestone": None,
"updated_at": "2026-08-10T00:00:00Z"}
p.update(kw)
self.issues[number] = p
return p
def api(self, login, endpoint, method="GET", payload=None, **kw):
self.calls.append((method, endpoint))
path, _, query = endpoint.partition("?")
params = dict(urllib.parse.parse_qsl(query))
m = re.match(r"^%s/issues/(\d+)$" % re.escape(BASE), path)
if m and method == "GET":
return self.issues.get(int(m.group(1)))
if path == "%s/issues" % BASE and method == "GET":
if int(params.get("page", 1)) > 1:
return []
return [self.issues[n] for n in sorted(self.issues)]
raise AssertionError("unstubbed call: %s %s" % (method, endpoint))
class PullTestCase(unittest.TestCase):
"""A temp store and a fake transport."""
def setUp(self):
self.root = tempfile.mkdtemp(prefix="tea-store-")
self.fake = FakeGitea()
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)
self.addCleanup(shutil.rmtree, self.root, True)
def write_local(self, id, text, number=42):
issue.save(self.root, issue.Issue(
id=id, title="An issue", body=text, labels=["type/task"],
origin="gitea", extra={"gitea": "%s#%d" % (REPO, number),
"url": "https://git.example/x", "synced": "old"}))
def run_pull(self, *argv):
out, err = io.StringIO(), io.StringIO()
args = ["pull.py", "--repo", REPO, "--out", self.root] + list(argv)
with mock.patch.object(sys, "argv", args), \
contextlib.redirect_stdout(out), \
contextlib.redirect_stderr(err):
pull.main()
return out.getvalue(), err.getvalue()
def stored_body(self, id):
return issue.load(self.root, id).body
def raw(self, id):
"""The file on disk, byte for byte — metadata included."""
with open(issue.path_of(self.root, id)) as f:
return f.read()
class PullMergesTicksTest(PullTestCase):
def test_a_tick_made_locally_survives_the_pull(self):
self.write_local("an-issue", body("- [x] первое", "- [ ] второе"))
self.fake.add(42, "An issue", body("- [ ] первое", "- [ ] второе"))
self.run_pull("42")
self.assertEqual(self.stored_body("an-issue"),
body("- [x] первое", "- [ ] второе"))
def test_a_tick_made_in_the_web_lands_locally(self):
self.write_local("an-issue", body("- [ ] первое", "- [ ] второе"))
self.fake.add(42, "An issue", body("- [x] первое", "- [ ] второе"))
self.run_pull("42")
self.assertEqual(self.stored_body("an-issue"),
body("- [x] первое", "- [ ] второе"))
def test_the_rest_of_the_body_is_still_overwritten(self):
self.write_local("an-issue", body("- [x] первое",
summary="Локальная правка прозы."))
self.fake.add(42, "An issue", body("- [ ] первое", "- [ ] новое с сервера",
summary="Серверная проза."))
self.run_pull("42")
got = self.stored_body("an-issue")
self.assertIn("Серверная проза.", got)
self.assertNotIn("Локальная правка прозы.", got)
self.assertIn("- [x] первое", got)
self.assertIn("- [ ] новое с сервера", got)
def test_an_item_the_server_added_arrives_ticked_if_the_server_ticked_it(self):
self.write_local("an-issue", body("- [ ] первое"))
self.fake.add(42, "An issue", body("- [ ] первое", "- [x] новое с сервера"))
self.run_pull("42")
self.assertEqual(self.stored_body("an-issue"),
body("- [ ] первое", "- [x] новое с сервера"))
def test_filter_mode_merges_too(self):
self.write_local("an-issue", body("- [x] первое"))
self.fake.add(42, "An issue", body("- [ ] первое"))
self.run_pull("--label", "type/task")
self.assertEqual(self.stored_body("an-issue"), body("- [x] первое"))
def test_a_retitled_issue_keeps_its_slug_and_its_ticks(self):
"""The merge hangs off the local id, which is resolved from the number
— a title change must not orphan the ticks."""
self.write_local("an-issue", body("- [x] первое"))
self.fake.add(42, "Completely different title", body("- [ ] первое"))
self.run_pull("42")
self.assertEqual(self.stored_body("an-issue"), body("- [x] первое"))
self.assertEqual(issue.load(self.root, "an-issue").title,
"Completely different title")
class PullIntoAnEmptyStoreTest(PullTestCase):
def test_no_local_file_writes_the_server_body_unchanged(self):
self.fake.add(42, "An issue", body("- [x] первое", "- [ ] второе"))
self.run_pull("42")
self.assertEqual(self.stored_body("an-issue"),
body("- [x] первое", "- [ ] второе"))
def test_a_store_that_does_not_exist_yet_is_created_and_not_merged(self):
shutil.rmtree(self.root)
self.fake.add(42, "An issue", body("- [ ] первое"))
_out, err = self.run_pull("42")
self.assertIn("created store", err)
self.assertEqual(self.stored_body("an-issue"), body("- [ ] первое"))
class CachedIsUnchangedTest(PullTestCase):
def test_a_skipped_issue_is_neither_read_nor_merged(self):
self.write_local("an-issue", body("- [x] первое"))
self.fake.add(42, "An issue", body("- [ ] первое", "- [ ] второе"))
before = self.raw("an-issue")
with mock.patch.object(gmap, "merge_checkbox_state") as merge:
out, _ = self.run_pull("42", "--cached")
merge.assert_not_called()
self.assertIn("(cached)", out)
self.assertEqual(self.raw("an-issue"), before)
def test_without_cached_the_same_issue_is_merged(self):
self.write_local("an-issue", body("- [x] первое"))
self.fake.add(42, "An issue", body("- [ ] первое", "- [ ] второе"))
self.run_pull("42")
self.assertEqual(self.stored_body("an-issue"),
body("- [x] первое", "- [ ] второе"))
class RoundTripTest(PullTestCase):
"""Pull twice with no change in between: the second is a no-op."""
def test_a_repeat_pull_does_not_churn_the_file(self):
self.write_local("an-issue", body("- [x] первое", "- [ ] второе"))
self.fake.add(42, "An issue", body("- [ ] первое", "- [ ] второе"))
self.run_pull("42")
first = self.raw("an-issue")
self.run_pull("42")
self.assertEqual(self.raw("an-issue"), first)
if __name__ == "__main__":
unittest.main()
+423
View File
@@ -0,0 +1,423 @@
#!/usr/bin/env python3
"""
Checkbox parsing, ticking, and the INDEX progress column.
Plain stdlib unittest — the scripts under test are stdlib-only by the layering
rule, and their tests have no business dragging in a dependency the code they
cover is forbidden to have. `skills/*/scripts/` are directories of scripts, not
packages, so they go on sys.path the same way the scripts do it to each other.
python3 -m unittest discover -s tests -v
Nothing here touches tmp/, the network, or the real store: every case builds
its own store in a TemporaryDirectory.
"""
import contextlib
import io
import os
import sys
import tempfile
import unittest
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(ROOT, "skills", "issue", "scripts"))
import issue # noqa: E402
import issue_ac # noqa: E402
import issue_check # noqa: E402
import issue_index # noqa: E402
# Boxes in two different sections, a wrapped item, a fenced example, and a
# plain list item that is not a checkbox at all. Line numbers are 1-based:
# the items sit on lines 8, 9, 12, 14 and 15.
BODY = """## Summary
Что-то про задачу.
## Spec
none
## Issues
- [x] wire-sqlc-appclick — первая часть
- [ ] add-pool-cfg — вторая часть
## Acceptance criteria
- [ ] в `issue.py` есть функция разбора чекбоксов тела:
возвращает пункты с номером строки, состоянием и текстом
- [X] чекбоксы ищутся по всему телу
- [ ] пример в блоке кода не считается пунктом:
```markdown
- [ ] это разметка из шаблона, а не галочка
- [x] и эта тоже
```
## Constraints
- не входит в объём: доставка тела в трекер
"""
NO_BOXES = """## Summary
Тело без единой галочки.
## Spec
none
## Notes
- обычный пункт списка
- ещё один
"""
# Metadata deliberately out of canonical order and missing optional keys, one
# item with trailing whitespace: a round-trip through Issue.to_text() would
# rewrite all of that, so this fixture catches a ticking path that re-renders
# the file instead of patching one character of it.
MESSY = """---
origin: local
labels: [type/task]
id: messy-issue
state: open
---
# Messy but valid
## Summary
Тело, которое нельзя перерисовывать.
## Spec
none
## Acceptance criteria
- [ ] первый пункт
- [ ] второй пункт
- [ ] третий пункт
"""
TASK_BODY = """## Summary
Что нужно сделать.
## Spec
none
## Motivation
Зачем это нужно.
## Acceptance criteria
- [ ] ничего ещё не сделано
- [ ] и это тоже не сделано
"""
def sole_difference(before, after):
"""The single character position at which the two strings differ.
Raises AssertionError when they differ in length or in more than one
place — the whole claim of `set_checkbox` is that this never happens."""
assert len(before) == len(after), (
"length changed: %d -> %d" % (len(before), len(after)))
diff = [i for i, (a, b) in enumerate(zip(before, after)) if a != b]
assert len(diff) == 1, "expected 1 differing character, got %d" % len(diff)
return diff[0]
@contextlib.contextmanager
def store(**files):
"""A throwaway issue store: {id: file text}."""
with tempfile.TemporaryDirectory() as root:
for id, text in files.items():
with open(os.path.join(root, "%s.md" % id), "w", newline="") as f:
f.write(text)
yield root
def run(fn, *argv):
"""Call a script entry point, returning (exit code or None, stdout)."""
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
rc = fn(list(argv))
return rc, buf.getvalue()
class TestParse(unittest.TestCase):
def test_finds_every_box_in_every_section(self):
items = issue.checkboxes(BODY)
self.assertEqual([c.index for c in items], [1, 2, 3, 4, 5])
self.assertEqual([c.line for c in items], [8, 9, 12, 14, 15])
self.assertEqual([c.checked for c in items],
[True, False, False, True, False])
self.assertEqual([c.section for c in items],
["## Issues", "## Issues"] + ["## Acceptance criteria"] * 3)
def test_boxes_outside_acceptance_criteria_are_items_too(self):
# A type/feature keeps its children under `## Issues`; binding the
# parser to one heading would lose them.
under_issues = [c for c in issue.checkboxes(BODY) if c.section == "## Issues"]
self.assertEqual(len(under_issues), 2)
self.assertTrue(under_issues[0].text.startswith("wire-sqlc-appclick"))
def test_continuation_line_is_part_of_the_item(self):
item = issue.checkboxes(BODY)[2]
self.assertEqual((item.line, item.end_line), (12, 13))
self.assertEqual(
item.text,
"в `issue.py` есть функция разбора чекбоксов тела: "
"возвращает пункты с номером строки, состоянием и текстом")
def test_fenced_example_is_not_an_item(self):
texts = [c.text for c in issue.checkboxes(BODY)]
self.assertNotIn("это разметка из шаблона, а не галочка", texts)
self.assertEqual(len(texts), 5)
def test_plain_list_item_is_not_a_checkbox(self):
self.assertNotIn("не входит в объём: доставка тела в трекер",
[c.text for c in issue.checkboxes(BODY)])
def test_markers_and_nesting(self):
text = ("* [ ] star\n"
"+ [x] plus\n"
"1. [ ] ordered\n"
"2) [x] ordered too\n"
" - [ ] nested\n"
"- [x]no space, not an item\n")
items = issue.checkboxes(text)
self.assertEqual([c.text for c in items],
["star", "plus", "ordered", "ordered too", "nested"])
self.assertEqual([c.checked for c in items],
[False, True, False, True, False])
def test_empty_text(self):
self.assertEqual(issue.checkboxes(""), [])
self.assertEqual(issue.checkboxes(None), [])
def test_line_numbers_are_relative_to_the_text_given(self):
# Same body, prefixed with a metadata block: the offsets move with it,
# which is what lets issue_ac.py work on a whole file.
head = "---\nid: x\nstate: open\n---\n# Title\n\n"
shift = head.count("\n")
self.assertEqual([c.line for c in issue.checkboxes(head + BODY)],
[c.line + shift for c in issue.checkboxes(BODY)])
def test_progress(self):
self.assertEqual(issue.checkbox_progress(BODY), (2, 5))
self.assertEqual(issue.checkbox_progress(NO_BOXES), (0, 0))
class TestToggle(unittest.TestCase):
def test_ticking_changes_exactly_one_character(self):
item = issue.checkboxes(BODY)[1] # line 9, unchecked
after = issue.set_checkbox(BODY, item, True)
at = sole_difference(BODY, after)
self.assertEqual(BODY[at], " ")
self.assertEqual(after[at], "x")
self.assertEqual(issue.checkbox_progress(after), (3, 5))
def test_unticking_changes_exactly_one_character(self):
item = issue.checkboxes(BODY)[0] # line 8, checked
after = issue.set_checkbox(BODY, item, False)
at = sole_difference(BODY, after)
self.assertEqual((BODY[at], after[at]), ("x", " "))
def test_every_item_toggles_in_isolation(self):
for item in issue.checkboxes(BODY):
after = issue.set_checkbox(BODY, item, not item.checked)
at = sole_difference(BODY, after)
self.assertEqual(after.splitlines()[item.line - 1].count("["), 1)
self.assertLess(at, len(BODY))
def test_no_op_when_already_in_that_state(self):
items = issue.checkboxes(BODY)
self.assertIs(issue.set_checkbox(BODY, items[0], True), BODY)
self.assertIs(issue.set_checkbox(BODY, items[1], False), BODY)
def test_capital_x_is_left_alone(self):
item = issue.checkboxes(BODY)[3] # `- [X]`
self.assertEqual(issue.set_checkbox(BODY, item, True), BODY)
def test_accepts_a_line_number(self):
after = issue.set_checkbox(BODY, 9, True)
self.assertEqual(after, issue.set_checkbox(BODY, issue.checkboxes(BODY)[1], True))
def test_refuses_a_line_that_is_not_a_checkbox(self):
with self.assertRaises(ValueError):
issue.set_checkbox(BODY, 1, True)
with self.assertRaises(ValueError):
issue.set_checkbox(BODY, 9999, True)
class TestScript(unittest.TestCase):
def test_lists_items_numbered_with_state(self):
with store(**{"messy-issue": MESSY}) as root:
rc, out = run(issue_ac.main, "messy-issue", "--out", root)
self.assertEqual(rc, 0)
self.assertIn("messy-issue — 0/3", out)
self.assertIn("## Acceptance criteria", out)
self.assertIn(" 1 [ ] первый пункт", out)
self.assertIn(" 3 [ ] третий пункт", out)
def test_check_by_number(self):
with store(**{"messy-issue": MESSY}) as root:
rc, out = run(issue_ac.main, "messy-issue", "--check", "2", "--out", root)
with open(os.path.join(root, "messy-issue.md")) as f:
after = f.read()
self.assertEqual(rc, 0)
self.assertIn("checked", out)
self.assertIn("1/3", out)
self.assertEqual(issue.checkbox_progress(after), (1, 3))
def test_check_by_substring(self):
with store(**{"messy-issue": MESSY}) as root:
run(issue_ac.main, "messy-issue", "--check", "ТРЕТИЙ", "--out", root)
with open(os.path.join(root, "messy-issue.md")) as f:
after = f.read()
self.assertTrue(issue.checkboxes(after)[2].checked)
self.assertEqual(issue.checkbox_progress(after), (1, 3))
def test_uncheck(self):
with store(**{"messy-issue": MESSY}) as root:
run(issue_ac.main, "messy-issue", "--check", "1", "--out", root)
rc, out = run(issue_ac.main, "messy-issue", "--uncheck", "1", "--out", root)
with open(os.path.join(root, "messy-issue.md")) as f:
after = f.read()
self.assertEqual(rc, 0)
self.assertIn("unchecked", out)
self.assertEqual(after, MESSY)
def test_toggling_through_the_script_changes_one_character_of_the_file(self):
with store(**{"messy-issue": MESSY}) as root:
path = os.path.join(root, "messy-issue.md")
with open(path) as f:
before = f.read()
run(issue_ac.main, "messy-issue", "--check", "второй", "--out", root)
with open(path) as f:
after = f.read()
at = sole_difference(before, after)
self.assertEqual((before[at], after[at]), (" ", "x"))
# The metadata block was neither reordered nor completed, and the
# trailing whitespace on the third item survived.
self.assertTrue(after.startswith("---\norigin: local\n"))
self.assertIn("- [ ] третий пункт \n", after)
def test_crlf_line_endings_survive(self):
crlf = MESSY.replace("\n", "\r\n")
with store(**{"messy-issue": crlf}) as root:
path = os.path.join(root, "messy-issue.md")
run(issue_ac.main, "messy-issue", "--check", "1", "--out", root)
with open(path, newline="") as f:
after = f.read()
at = sole_difference(crlf, after)
self.assertEqual((crlf[at], after[at]), (" ", "x"))
self.assertEqual(after.count("\r\n"), crlf.count("\r\n"))
def test_ambiguous_substring_is_an_error_listing_the_matches(self):
with store(**{"messy-issue": MESSY}) as root:
with self.assertRaises(SystemExit) as cm:
run(issue_ac.main, "messy-issue", "--check", "пункт", "--out", root)
with open(os.path.join(root, "messy-issue.md")) as f:
self.assertEqual(f.read(), MESSY) # nothing was picked
msg = str(cm.exception)
self.assertIn("matches 3 items", msg)
for want in ("1 [ ] первый пункт", "2 [ ] второй пункт", "3 [ ] третий пункт"):
self.assertIn(want, msg)
def test_substring_that_matches_nothing(self):
with store(**{"messy-issue": MESSY}) as root:
with self.assertRaises(SystemExit) as cm:
run(issue_ac.main, "messy-issue", "--check", "нетакого", "--out", root)
self.assertIn("nothing matches", str(cm.exception))
def test_number_out_of_range(self):
with store(**{"messy-issue": MESSY}) as root:
with self.assertRaises(SystemExit) as cm:
run(issue_ac.main, "messy-issue", "--check", "9", "--out", root)
self.assertIn("no item 9 — the issue has 3", str(cm.exception))
def test_issue_without_checkboxes(self):
with store(**{"plain": "---\nid: plain\n---\n# Plain\n\n" + NO_BOXES}) as root:
rc, out = run(issue_ac.main, "plain", "--out", root)
self.assertEqual((rc, out.strip()), (0, "plain — no checkboxes"))
with self.assertRaises(SystemExit) as cm:
run(issue_ac.main, "plain", "--check", "1", "--out", root)
self.assertIn("has no checkboxes", str(cm.exception))
def test_unknown_id(self):
with store() as root:
with self.assertRaises(SystemExit) as cm:
run(issue_ac.main, "nope", "--out", root)
self.assertIn("no issue 'nope'", str(cm.exception))
class TestIndexProgress(unittest.TestCase):
def files(self):
boxed = ("---\nid: boxed\nstate: open\nlabels: [type/task]\n"
"origin: local\n---\n# Boxed\n\n" + BODY)
plain = ("---\nid: plain\nstate: open\nlabels: [type/task]\n"
"origin: local\n---\n# Plain\n\n" + NO_BOXES)
return {"boxed": boxed, "plain": plain}
def index(self, root):
issue_index.build(root)
with open(os.path.join(root, "INDEX.md")) as f:
return f.read()
def row(self, text, id):
for line in text.splitlines():
if line.startswith("| [%s]" % id):
return [c.strip() for c in line.split("|")]
self.fail("no row for %r in INDEX.md" % id)
def test_column_exists_and_counts_the_body(self):
with store(**self.files()) as root:
text = self.index(root)
self.assertIn("| id | state | progress | type |", text)
self.assertEqual(self.row(text, "boxed")[3], "2/5")
def test_blank_for_an_issue_without_checkboxes(self):
with store(**self.files()) as root:
text = self.index(root)
self.assertEqual(self.row(text, "plain")[3], "")
def test_recomputed_on_the_fly_not_stored(self):
with store(**self.files()) as root:
self.assertEqual(self.row(self.index(root), "boxed")[3], "2/5")
run(issue_ac.main, "boxed", "--check", "add-pool-cfg", "--out", root)
self.assertEqual(self.row(self.index(root), "boxed")[3], "3/5")
# No metadata field anywhere holds it.
with open(os.path.join(root, "boxed.md")) as f:
head = f.read().split("---")[1]
self.assertNotIn("3/5", head)
self.assertNotIn("progress", head)
class TestCheckIgnoresUntickedBoxes(unittest.TestCase):
"""An unticked box is work not done yet, not a malformed issue."""
def test_validate_reports_nothing(self):
iss = issue.Issue(id="unticked-issue", title="Do the thing",
labels=["type/task"], body=TASK_BODY)
err, warn = issue.validate(iss, known_ids={"unticked-issue"})
self.assertEqual(err, [])
self.assertEqual(warn, [])
def test_issue_check_exits_clean(self):
text = ("---\nid: unticked-issue\nstate: open\nlabels: [type/task]\n"
"assignees: []\nmilestone: none\ndepends: []\norigin: local\n"
"---\n# Do the thing\n\n" + TASK_BODY)
argv = sys.argv
with store(**{"unticked-issue": text}) as root:
sys.argv = ["issue_check.py", "--out", root]
try:
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
rc = issue_check.main()
finally:
sys.argv = argv
out = buf.getvalue()
self.assertEqual(rc, 0, out)
self.assertIn("ok unticked-issue", out)
self.assertNotIn("ERROR", out)
self.assertNotIn("warn ", out)
if __name__ == "__main__":
unittest.main()
+234
View File
@@ -0,0 +1,234 @@
#!/usr/bin/env python3
"""
The container <-> child edge points ONE way: container -> child.
A `type/feature` is closed when its children are closed, and that is a
dependency relation, so the container lists its children in `depends:`. A child
belongs to a feature, which is a membership relation, and membership has no
place in a dependency graph — so a child never names its container back. These
tests pin that direction down in all three places it shows up: the validator,
the desync warning, and the drawn tree.
python3 -m unittest discover -s tests -v
Stdlib only, like the scripts under test. `skills/*/scripts/` are directories,
not packages, so they go on sys.path by hand.
"""
import contextlib
import io
import os
import sys
import tempfile
import unittest
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SCRIPTS = os.path.join(REPO, "skills", "issue", "scripts")
if SCRIPTS not in sys.path:
sys.path.insert(0, SCRIPTS)
import issue # noqa: E402
import issue_check # noqa: E402
import issue_new # noqa: E402
import issue_tree # noqa: E402
def run(module, argv):
"""Call a script's main() with argv, returning (exit_code, stdout).
stderr is swallowed: issue_new.py notes on it when a `--depends` id is not
in the store yet, which is fine and not what these tests are about."""
buf = io.StringIO()
old = sys.argv
sys.argv = [module.__name__ + ".py"] + argv
try:
with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(io.StringIO()):
code = module.main()
except SystemExit as e: # argparse / sys.exit("msg")
code = e.code if isinstance(e.code, int) else 1
finally:
sys.argv = old
return (code or 0), buf.getvalue()
def drawn(tree_output):
"""The rows inside the tree's code fence, header and blanks dropped."""
return [l for l in tree_output.splitlines() if l.rstrip().endswith(".md")]
class StoreCase(unittest.TestCase):
"""A scratch store per test. Never touches tmp/issues/."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.root = self._tmp.name
self.addCleanup(self._tmp.cleanup)
def new(self, type, id, title, depends=()):
argv = ["--type", type, "--id", id, "--title", title, "--out", self.root]
for d in depends:
argv += ["--depends", d]
code, _ = run(issue_new, argv)
self.assertEqual(code, 0, "issue_new.py failed for %s" % id)
def edit(self, id, old, new):
p = issue.path_of(self.root, id)
with open(p) as f:
text = f.read()
self.assertIn(old, text, "%s.md does not contain %r" % (id, old))
with open(p, "w") as f:
f.write(text.replace(old, new))
def fill_issues_section(self, id, *children):
"""Replace the type/feature template's `## Issues` placeholder."""
self.edit(id,
"- [ ] slug-дочернего-issue — краткое описание части\n- [ ] …\n",
"".join("- [ ] %s — часть\n" % c for c in children))
def container_and_child(self):
"""The canonical shape from references/format.md: the container names
the child in `depends:` AND in `## Issues`; the child names nobody."""
self.new("task", "child-y", "Child y")
self.new("feature", "feat-x", "Container x", depends=["child-y"])
self.fill_issues_section("feat-x", "child-y")
class CanonicalContainerIsClean(StoreCase):
"""A container from the template plus a child per format.md: green."""
def test_check_is_silent_and_exits_zero(self):
self.container_and_child()
code, out = run(issue_check, ["--out", self.root])
self.assertEqual(code, 0, out)
self.assertNotIn("ERROR", out)
self.assertNotIn("warn", out)
self.assertIn("ok feat-x", out)
self.assertIn("ok child-y", out)
def test_validate_reports_nothing_for_either_issue(self):
self.container_and_child()
issues = issue.load_all(self.root)
for id in ("feat-x", "child-y"):
err, warn = issue.validate(issues[id], known_ids=set(issues))
self.assertEqual((err, warn), ([], []), id)
def test_the_child_does_not_depend_on_its_container(self):
self.container_and_child()
issues = issue.load_all(self.root)
self.assertEqual(issues["feat-x"].depends, ["child-y"])
self.assertEqual(issues["child-y"].depends, [])
class OnlyOneDirectionIsLegal(StoreCase):
"""format.md and the validator agree on container -> child, and the
reverse edge is an error rather than a matter of taste."""
def test_the_reverse_edge_is_a_cycle(self):
self.container_and_child()
self.edit("child-y", "depends: []", "depends: [feat-x]")
code, out = run(issue_check, ["--out", self.root])
self.assertEqual(code, 1, out)
self.assertIn("ERROR cycle:", out)
self.assertIn("feat-x", out)
self.assertIn("child-y", out)
def test_a_child_pointing_at_its_container_alone_is_not_the_graph(self):
"""The shape format.md used to document: the child depends on the
container and the container's depends: is empty. It no longer matches
what the container's own `## Issues` says, so the check complains."""
self.new("feature", "feat-x", "Container x")
self.new("task", "child-y", "Child y", depends=["feat-x"])
self.fill_issues_section("feat-x", "child-y")
_, out = run(issue_check, ["--out", self.root])
self.assertIn("warn feat-x:", out)
def test_issues_section_is_an_edge_source_pointing_down(self):
body = "## Issues\n- [ ] child-y — часть\n"
self.assertEqual(issue.body_dep_refs(body), ["child-y"])
class WarningNamesItsOwnSection(StoreCase):
"""The desync warning quotes the section the reference came from, not
`## Depends on` unconditionally — a container has no such section."""
def test_container_warning_says_issues(self):
self.new("feature", "feat-x", "Container x")
self.new("task", "child-y", "Child y")
self.fill_issues_section("feat-x", "child-y") # but not depends:
issues = issue.load_all(self.root)
err, warn = issue.validate(issues["feat-x"], known_ids=set(issues))
self.assertEqual(err, [])
self.assertEqual(
warn, ["## Issues mentions 'child-y' but `depends:` does not list it"])
self.assertNotIn("## Depends on", "\n".join(warn))
body = issue.load(self.root, "feat-x").body
self.assertNotIn("## Depends on", body,
"the warning must not name a section that is not in the file")
def test_plain_issue_warning_still_says_depends_on(self):
self.new("task", "child-y", "Child y", depends=["migrate-schema"])
self.edit("child-y", "depends: [migrate-schema]", "depends: []")
issues = issue.load_all(self.root)
_, warn = issue.validate(issues["child-y"])
self.assertEqual(
warn,
["## Depends on mentions 'migrate-schema' but `depends:` does not list it"])
def test_each_reference_is_named_with_its_own_section(self):
body = ("## Depends on\n- migrate-schema\n\n"
"## Issues\n- [ ] child-y — часть\n")
self.assertEqual(
issue.body_dep_ref_sections(body),
[("## Depends on", "migrate-schema"), ("## Issues", "child-y")])
def test_body_dep_refs_still_returns_bare_strings(self):
"""skills/sync/scripts/map.py filters this list for `#N` refs."""
body = "## Depends on\n- migrate-schema\n- #42\n"
refs = issue.body_dep_refs(body)
self.assertEqual(refs, ["migrate-schema", "#42"])
self.assertTrue(all(isinstance(r, str) for r in refs))
def test_tracker_numbers_never_warn(self):
"""`#42` is a tracker handle, not a slug; `depends:` holds ids only."""
self.new("task", "child-y", "Child y")
self.edit("child-y", "## Motivation", "## Depends on\n- #42\n\n## Motivation")
issues = issue.load_all(self.root)
_, warn = issue.validate(issues["child-y"])
self.assertEqual(warn, [])
class TreePutsTheContainerOnTop(StoreCase):
def test_container_is_the_root_and_children_hang_below(self):
self.container_and_child()
code, out = run(issue_tree, ["--out", self.root])
self.assertEqual(code, 0, out)
rows = drawn(out)
self.assertEqual(len(rows), 2, out)
self.assertTrue(rows[0].startswith("feat-x "), out)
self.assertTrue(rows[1].startswith("└── child-y "), out)
self.assertEqual(out.count("child-y ["), 1, "child drawn more than once")
def test_the_container_is_the_only_root(self):
self.container_and_child()
_, out = run(issue_tree, ["--out", self.root])
self.assertIn("# Dependency tree — feat-x", out)
def test_two_children_hang_off_one_container(self):
self.new("task", "child-y", "Child y")
self.new("task", "child-z", "Child z")
self.new("feature", "feat-x", "Container x",
depends=["child-y", "child-z"])
self.fill_issues_section("feat-x", "child-y", "child-z")
code, out = run(issue_check, ["--out", self.root])
self.assertEqual(code, 0, out)
_, tree = run(issue_tree, ["--out", self.root])
self.assertEqual(tree.count("feat-x ["), 1,
"the container must not repeat once per child")
self.assertEqual([r.split(" ")[0] for r in drawn(tree)],
["feat-x", "├──", "└──"], tree)
self.assertIn("├── child-y ", tree)
self.assertIn("└── child-z ", tree)
if __name__ == "__main__":
unittest.main()
+403
View File
@@ -0,0 +1,403 @@
#!/usr/bin/env python3
"""
Native Gitea dependency links, written by push.py.
The transport is stubbed at exactly one seam — `_gitea.api`, the single
function that shells out to `tea` — so everything above it runs for real:
argument parsing, validation, topological order, the id map, map.py's payload
shapes and _gitea's own endpoint/body construction. Nothing here touches a
network, and no test may ever be made to.
`skills/*/scripts/` are not packages; they go on sys.path by hand.
"""
import contextlib
import io
import os
import shutil
import sys
import tempfile
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 issue # noqa: E402
import map as gmap # noqa: E402
import push # noqa: E402
REPO = "claude-skills/tea"
BASE = "repos/%s" % REPO
LABELS = {"type/task": 901, "type/bug": 902, "severity/medium": 903,
"comp/sync": 904}
LABEL_NAMES = {v: k for k, v in LABELS.items()}
BODY = """## Summary
Прозаическое описание.
## Spec
skills/issue/references/format.md
## Depends on
- first-thing — ставит фундамент, без него второй не собрать
## Acceptance criteria
- [ ] что-нибудь работает
"""
BODY_NO_DEPS = """## Summary
Прозаическое описание.
## Spec
skills/issue/references/format.md
## Acceptance criteria
- [ ] что-нибудь работает
"""
class FakeGitea(object):
"""A `tea api` that answers from memory and remembers what it was asked.
Dependency links are kept the way Gitea keeps them: per blocked issue, a
set of (repo, number) blockers. That is what makes the idempotence test
meaningful — the second push sees the link the first one made."""
def __init__(self, next_number=101):
self.calls = [] # (method, endpoint, payload)
self.next_number = next_number
self.deps = {} # number -> {(repo, number)}
self.titles = {} # number -> title
self.fail_dependency_post = False
# -- helpers -----------------------------------------------------------
@property
def writes(self):
"""Every non-GET call. `--dry-run` must produce an empty list."""
return [c for c in self.calls if c[0] != "GET"]
def dep_posts(self):
return [c for c in self.calls
if c[0] == "POST" and c[1].endswith("/dependencies")]
def issue_payload(self, number, labels=()):
return {"number": number,
"html_url": "https://git.example/%s/issues/%d" % (REPO, number),
"title": self.titles.get(number, ""),
"labels": [{"name": LABEL_NAMES[i]} for i in labels
if i in LABEL_NAMES],
"updated_at": "2026-08-10T00:00:00Z",
"repository": {"full_name": REPO}}
# -- the seam ----------------------------------------------------------
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 path == "%s/labels" % BASE and method == "GET":
# Every label the run could ask for, so nothing is ever created.
return [{"name": n, "id": i} for n, i in LABELS.items()]
if path == "%s/issues" % BASE and method == "POST":
number = self.next_number
self.next_number += 1
self.titles[number] = (payload or {}).get("title", "")
# Echo the labels back, or push re-applies them with a PUT.
return self.issue_payload(number, (payload or {}).get("labels") or [])
if path.endswith("/dependencies"):
number = int(path.split("/issues/")[1].split("/")[0])
if method == "GET":
return [dict(self.issue_payload(n), repository={"full_name": r})
for r, n in sorted(self.deps.get(number, set()))]
if method == "POST":
if self.fail_dependency_post:
return None
key = ("%s/%s" % (payload["owner"], payload["repo"]),
int(payload["index"]))
self.deps.setdefault(number, set()).add(key)
return self.issue_payload(number)
if "/issues/" in path and method == "PATCH":
number = int(path.rsplit("/", 1)[1])
self.titles[number] = (payload or {}).get("title", self.titles.get(number, ""))
return self.issue_payload(number, (payload or {}).get("labels") or [])
raise AssertionError("unstubbed call: %s %s" % (method, endpoint))
class PushTestCase(unittest.TestCase):
"""A temp store, a fake transport, and no git."""
def setUp(self):
self.root = tempfile.mkdtemp(prefix="tea-store-")
self.fake = FakeGitea()
patches = [
mock.patch.object(_gitea, "api", self.fake.api),
mock.patch.object(_gitea, "require_login", lambda: "test-login"),
# push reads the current branch from git; a temp store has none and
# the runner's branch would leak into the payload.
mock.patch.object(push, "git_branch", lambda: "test-branch"),
]
for p in patches:
p.start()
self.addCleanup(p.stop)
self.addCleanup(shutil.rmtree, self.root, True)
# -- fixtures ----------------------------------------------------------
def write_issue(self, id, title, body=BODY_NO_DEPS, depends=(), extra=None):
iss = issue.Issue(id=id, title=title, body=body, labels=["type/task"],
depends=list(depends), extra=dict(extra or {}))
issue.save(self.root, iss)
return iss
def two_issues(self):
"""first-thing, and second-thing which depends on it."""
self.write_issue("first-thing", "First thing")
self.write_issue("second-thing", "Second thing", body=BODY,
depends=["first-thing"])
def run_push(self, *argv):
out, err = io.StringIO(), io.StringIO()
args = ["push.py", "--repo", REPO, "--out", self.root] + list(argv)
with mock.patch.object(sys, "argv", args), \
contextlib.redirect_stdout(out), \
contextlib.redirect_stderr(err):
push.main()
return out.getvalue(), err.getvalue()
def number_of(self, id):
return gmap.number_of(issue.load(self.root, id))
# --------------------------------------------------------------------------
# _gitea: the POST body, and the pre-check that reads links back
# --------------------------------------------------------------------------
class AddDependencyTest(unittest.TestCase):
def test_post_body_is_issue_meta(self):
"""POST /issues/{index}/dependencies with IssueMeta for the BLOCKER.
Confirmed against the instance's swagger.v1.json (Gitea 1.26.1):
"Make the issue in the url depend on the issue in the form." """
calls = []
def fake_api(login, endpoint, method="GET", payload=None, **kw):
calls.append((method, endpoint, payload))
return {"number": 102}
with mock.patch.object(_gitea, "api", fake_api):
ok = _gitea.add_dependency("l", BASE, 102, REPO, 101, out_root=None)
self.assertTrue(ok)
method, endpoint, payload = calls[0]
self.assertEqual(method, "POST")
self.assertEqual(endpoint, "%s/issues/102/dependencies" % BASE)
self.assertEqual(payload, {"index": 101, "owner": "claude-skills",
"repo": "tea"})
def test_blocker_may_live_in_another_repo(self):
"""IssueMeta carries owner/repo precisely so it can."""
seen = {}
def fake_api(login, endpoint, method="GET", payload=None, **kw):
seen.update(payload or {})
return {"number": 1}
with mock.patch.object(_gitea, "api", fake_api):
_gitea.add_dependency("l", BASE, 102, "other-org/infra", 7)
self.assertEqual(seen, {"index": 7, "owner": "other-org", "repo": "infra"})
def test_failure_is_reported_not_raised(self):
"""409 (link already there) and friends come back as False."""
with mock.patch.object(_gitea, "api", lambda *a, **k: None):
self.assertFalse(_gitea.add_dependency("l", BASE, 102, REPO, 101))
def test_unparseable_repo_makes_no_request(self):
called = []
with mock.patch.object(_gitea, "api", lambda *a, **k: called.append(1)):
self.assertFalse(_gitea.add_dependency("l", BASE, 102, "tea", 101))
self.assertEqual(called, [])
def test_native_dep_pairs_reads_repo_and_number(self):
payload = [{"number": 101, "repository": {"full_name": REPO}},
{"number": 7, "repository": {"full_name": "other-org/infra"}}]
with mock.patch.object(_gitea, "api", lambda *a, **k: payload):
got = _gitea.native_dep_pairs("l", BASE, 102)
self.assertEqual(got, {(REPO, 101), ("other-org/infra", 7)})
def test_native_dep_pairs_empty_when_unsupported(self):
with mock.patch.object(_gitea, "api", lambda *a, **k: None):
self.assertEqual(_gitea.native_dep_pairs("l", BASE, 102), set())
# --------------------------------------------------------------------------
# push: the whole run
# --------------------------------------------------------------------------
class PushCreatesLinksTest(PushTestCase):
def test_link_created_after_both_have_numbers(self):
"""One run, topological order, one native link — no second pass."""
self.two_issues()
out, _ = self.run_push()
first, second = self.number_of("first-thing"), self.number_of("second-thing")
self.assertLess(first, second, "blocker must be created first")
self.assertEqual(self.fake.deps.get(second), {(REPO, first)})
self.assertIn("depends on %s#%d (first-thing)" % (REPO, first), out)
def test_link_direction_matches_what_pull_reads_back(self):
"""The link hangs off the BLOCKED issue, which is where native_deps
looks — push and `pull.py --deps` must agree or the round trip lies."""
self.two_issues()
self.run_push()
second = self.number_of("second-thing")
with mock.patch.object(_gitea, "api", self.fake.api):
self.assertEqual(_gitea.native_deps("l", BASE, second),
[self.number_of("first-thing")])
def test_issue_without_dependencies_makes_no_dependency_request(self):
"""Not even the idempotence GET — it is skipped when there is nothing
to link, so the common case costs no extra round trip."""
self.write_issue("lonely-thing", "Lonely thing")
self.run_push()
self.assertEqual([c for c in self.fake.calls if "dependencies" in c[1]], [])
class LocalOnlyDependencyTest(PushTestCase):
def test_local_dependency_is_warned_and_not_linked(self):
self.two_issues()
out, err = self.run_push("second-thing")
self.assertEqual(self.fake.dep_posts(), [])
self.assertIn("depends on local-only issue(s) first-thing", err)
self.assertNotIn("depends on ", out)
self.assertIsNone(self.number_of("first-thing"))
class IdempotenceTest(PushTestCase):
def test_repeat_push_does_not_duplicate_the_link(self):
self.two_issues()
self.run_push()
self.assertEqual(len(self.fake.dep_posts()), 1)
self.run_push("--update")
self.assertEqual(len(self.fake.dep_posts()), 1, "link re-POSTed")
self.assertEqual(self.fake.deps[self.number_of("second-thing")],
{(REPO, self.number_of("first-thing"))})
def test_a_failing_link_warns_and_the_run_finishes(self):
"""A 409 or any other refusal must not abort a push that has already
created issues."""
self.two_issues()
self.fake.fail_dependency_post = True
out, err = self.run_push()
self.assertIn("could not link", err)
self.assertIn("index:", out) # the run completed
self.assertIsNotNone(self.number_of("second-thing"))
class UpdateCarriesNewLinksTest(PushTestCase):
def test_dependency_added_after_the_first_push_is_linked_by_update(self):
self.write_issue("first-thing", "First thing")
self.write_issue("second-thing", "Second thing")
self.run_push()
self.assertEqual(self.fake.dep_posts(), [])
iss = issue.load(self.root, "second-thing")
iss.depends = ["first-thing"]
iss.body = BODY
issue.save(self.root, iss)
self.run_push("--update", "second-thing")
self.assertEqual(self.fake.deps[self.number_of("second-thing")],
{(REPO, self.number_of("first-thing"))})
class DryRunTest(PushTestCase):
def test_dry_run_names_the_links_and_writes_nothing(self):
self.two_issues()
out, _ = self.run_push("--dry-run")
self.assertEqual(self.fake.calls, [], "--dry-run made a request")
self.assertIn("link -> #? (first-thing, created by this run)", out)
self.assertIn("1 dependency link(s) would be created", out)
def test_dry_run_shows_a_known_number_when_the_blocker_is_pushed(self):
self.write_issue("first-thing", "First thing",
extra={"gitea": "%s#101" % REPO})
self.write_issue("second-thing", "Second thing", body=BODY,
depends=["first-thing"])
out, _ = self.run_push("--dry-run")
self.assertIn("link -> %s#101 (first-thing)" % REPO, out)
self.assertEqual(self.fake.writes, [])
def test_dry_run_says_a_local_dependency_gets_no_link(self):
self.two_issues()
out, _ = self.run_push("--dry-run", "second-thing")
self.assertIn("no link: first-thing is local-only", out)
self.assertIn("0 dependency link(s) would be created", out)
class BodyIsVerbatimTest(PushTestCase):
def test_depends_on_prose_is_not_rewritten_to_numbers(self):
"""map.py deliberately never edits the prose. Linking must not start."""
self.two_issues()
self.run_push()
created = [c for c in self.fake.calls
if c[0] == "POST" and c[1] == "%s/issues" % BASE]
sent = [c[2]["body"] for c in created]
second_body = [b for b in sent if "Depends on" in b][0]
self.assertIn("- first-thing — ставит фундамент", second_body)
self.assertNotIn("#101", second_body)
self.assertEqual(second_body, issue.load(self.root, "second-thing").body)
def test_body_survives_a_second_push_unchanged(self):
self.two_issues()
self.run_push()
before = issue.load(self.root, "second-thing").body
self.run_push("--update")
patched = [c for c in self.fake.calls if c[0] == "PATCH"]
self.assertIn(before, [c[2]["body"] for c in patched])
self.assertEqual(before, issue.load(self.root, "second-thing").body)
class DepStateTest(PushTestCase):
"""The classifier both the dry run and the real run read from."""
def test_classifies_linked_in_run_and_local(self):
issues = {
"pushed": issue.Issue(id="pushed", extra={"gitea": "%s#101" % REPO}),
"coming": issue.Issue(id="coming"),
"local": issue.Issue(id="local"),
}
iss = issue.Issue(id="dependent",
depends=["pushed", "coming", "local", "ghost"])
got = push.dep_state(iss, issues, {"coming", "dependent"})
self.assertEqual(got, [("pushed", "%s#101" % REPO, False),
("coming", None, True),
("local", None, False)])
if __name__ == "__main__":
unittest.main()
+437
View File
@@ -0,0 +1,437 @@
#!/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_ac", "issue_index"}
foreign = imported - local - sys.stdlib_module_names
self.assertEqual(foreign, set(),
"non-stdlib import in the domain layer: %s"
% ", ".join(sorted(foreign)))
self.assertNotIn("subprocess", imported)
if __name__ == "__main__":
unittest.main()