Merge origin/main into feat/evict-closed-issues
Three doc conflicts, all unions: the script lists in AGENTS.md and the runner gain both close.py and evict.py, and the sync skill keeps both the closing and the evicting sections. Rule 4 of the runner is rewritten once to carry both halves — closing is now a script it may run on named ids, retitling and remote deletion stay forbidden, and the two allowed local deletions (push's own, and eviction) are listed together.
This commit is contained in:
naudachu
@@ -255,6 +255,11 @@ class FakeGitea(object):
|
||||
path, _, query = endpoint.partition("?")
|
||||
params = dict(urllib.parse.parse_qsl(query))
|
||||
|
||||
# Every pull asks for an issue's native links now (dependencies are the
|
||||
# default). Nothing here has any; the answer just has to exist.
|
||||
if path.endswith("/dependencies"):
|
||||
return []
|
||||
|
||||
m = re.match(r"^%s/issues/(\d+)$" % re.escape(BASE), path)
|
||||
if m and method == "GET":
|
||||
return self.issues.get(int(m.group(1)))
|
||||
|
||||
@@ -0,0 +1,641 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
close.py — the state changes in Gitea, and the local file follows it or nothing
|
||||
happens at all.
|
||||
|
||||
Two halves, and the second is the one that matters:
|
||||
|
||||
1. **It closes.** A slug, a number, several of either in one run, and
|
||||
`--reopen` going the other way. What goes out is a PATCH carrying `state`
|
||||
and nothing else; what comes back is written into `state:` on the local
|
||||
file, and the index is rebuilt so the store's own table agrees.
|
||||
|
||||
2. **It changes nothing local unless the tracker confirmed it.** A `tea` that
|
||||
exited non-zero, an answer with no number, an answer for another issue, an
|
||||
answer that still says `open`, an `origin: local` issue, a `--dry-run`: in
|
||||
every one of those the file on disk is byte for byte what it was. A bug here
|
||||
makes the store lie about the tracker, so each path is asserted on its own.
|
||||
|
||||
The transport is stubbed at `_gitea.api`, as `test_drop_after_push.py` does,
|
||||
with the same deliberate exception: the non-2xx test stubs `_gitea.subprocess`
|
||||
and lets the real `_gitea.api` run, so "tea exited 1" is proved end to end.
|
||||
|
||||
Nothing here touches a network, and nothing here touches the developer's store:
|
||||
every test builds its own in a `tempfile.TemporaryDirectory()`.
|
||||
"""
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
for _p in (os.path.join(_ROOT, "skills", "sync", "scripts"),
|
||||
os.path.join(_ROOT, "skills", "issue", "scripts")):
|
||||
if _p not in sys.path:
|
||||
sys.path.insert(0, _p)
|
||||
|
||||
import _gitea # noqa: E402
|
||||
import close # noqa: E402
|
||||
import issue # noqa: E402
|
||||
import map as gmap # noqa: E402
|
||||
|
||||
# Captured before any test patches it — the non-2xx test needs the real thing.
|
||||
REAL_API = _gitea.api
|
||||
|
||||
REPO = "claude-skills/tea"
|
||||
BASE = "repos/%s" % REPO
|
||||
|
||||
BODY = """## Summary
|
||||
Прозаическое описание задачи.
|
||||
|
||||
## Spec
|
||||
skills/issue/references/format.md
|
||||
|
||||
## Acceptance criteria
|
||||
- [x] что-нибудь работает
|
||||
"""
|
||||
|
||||
|
||||
class FakeTracker(object):
|
||||
"""`tea api` answered from memory, for state writes only.
|
||||
|
||||
It keeps a `state` per number and flips it on a PATCH, which is the whole
|
||||
contract close.py has with the far side."""
|
||||
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
self.states = {} # number -> "open" / "closed"
|
||||
self.raise_on_write = None # an exception instance to raise
|
||||
self.answer_override = None # what a write answers instead
|
||||
|
||||
def payload_of(self, number):
|
||||
return {"number": number, "state": self.states[number],
|
||||
"title": "A thing", "updated_at": "2026-08-11T00:00:00Z",
|
||||
"html_url": "https://git.example/%s/issues/%d" % (REPO, number)}
|
||||
|
||||
def writes(self):
|
||||
return [c for c in self.calls if c[0] != "GET"]
|
||||
|
||||
def api(self, login, endpoint, method="GET", payload=None,
|
||||
payload_name=None, out_root=None, allow_fail=False):
|
||||
self.calls.append((method, endpoint, payload))
|
||||
path = endpoint.split("?")[0]
|
||||
|
||||
if "/issues/" in path and method == "PATCH":
|
||||
number = int(path.rsplit("/", 1)[1])
|
||||
if self.raise_on_write is not None:
|
||||
raise self.raise_on_write
|
||||
self.states.setdefault(number, "open")
|
||||
if "state" in (payload or {}):
|
||||
self.states[number] = payload["state"]
|
||||
if self.answer_override is not None:
|
||||
return self.answer_override
|
||||
return self.payload_of(number)
|
||||
|
||||
if "/issues/" in path and method == "GET":
|
||||
n = int(path.rsplit("/", 1)[1])
|
||||
return self.payload_of(n) if n in self.states else None
|
||||
|
||||
raise AssertionError("unstubbed call: %s %s" % (method, endpoint))
|
||||
|
||||
|
||||
class StoreTestCase(unittest.TestCase):
|
||||
"""A temp store and a fake tracker."""
|
||||
|
||||
def setUp(self):
|
||||
tmp = tempfile.TemporaryDirectory(prefix="tea-close-")
|
||||
self.addCleanup(tmp.cleanup)
|
||||
self.root = tmp.name
|
||||
self.fake = FakeTracker()
|
||||
for p in (mock.patch.object(_gitea, "api", self.fake.api),
|
||||
mock.patch.object(_gitea, "require_login", lambda: "test-login")):
|
||||
p.start()
|
||||
self.addCleanup(p.stop)
|
||||
|
||||
# -- fixtures ----------------------------------------------------------
|
||||
|
||||
def synced(self, id="a-thing", number=101, state="open"):
|
||||
"""An issue that is in the tracker and on disk, the way a pull leaves
|
||||
it: `origin: gitea`, a `gitea:` field, and a ledger entry."""
|
||||
key = gmap.remote_key(REPO, number)
|
||||
iss = issue.Issue(id=id, title="A thing", body=BODY, state=state,
|
||||
labels=["type/task"], origin=gmap.ORIGIN,
|
||||
extra={"gitea": key, "url": "https://git.example/x",
|
||||
"synced": "2026-08-10T00:00:00Z"})
|
||||
issue.save(self.root, iss)
|
||||
m = _gitea.load_map(self.root)
|
||||
m[key] = id
|
||||
_gitea.save_map(self.root, m)
|
||||
self.fake.states[number] = state
|
||||
return iss
|
||||
|
||||
def local_only(self, id="local-thing"):
|
||||
"""An issue that has never left this machine."""
|
||||
iss = issue.Issue(id=id, title="Local thing", body=BODY,
|
||||
labels=["type/task"])
|
||||
issue.save(self.root, iss)
|
||||
return iss
|
||||
|
||||
def dropped(self, id="gone-thing", number=205, state="open"):
|
||||
"""Pushed, and its file went with the push: ledger only."""
|
||||
m = _gitea.load_map(self.root)
|
||||
m[gmap.remote_key(REPO, number)] = id
|
||||
_gitea.save_map(self.root, m)
|
||||
self.fake.states[number] = state
|
||||
return number
|
||||
|
||||
# -- runner ------------------------------------------------------------
|
||||
|
||||
def run_close(self, *argv):
|
||||
self.out, self.err = io.StringIO(), io.StringIO()
|
||||
args = ["close.py", "--repo", REPO, "--out", self.root] + list(argv)
|
||||
with mock.patch.object(sys, "argv", args), \
|
||||
contextlib.redirect_stdout(self.out), \
|
||||
contextlib.redirect_stderr(self.err):
|
||||
close.main()
|
||||
return self.out.getvalue(), self.err.getvalue()
|
||||
|
||||
# -- assertions --------------------------------------------------------
|
||||
|
||||
def state_on_disk(self, id):
|
||||
return issue.load(self.root, id).state
|
||||
|
||||
def raw(self, id):
|
||||
with open(issue.path_of(self.root, id)) as f:
|
||||
return f.read()
|
||||
|
||||
def assertUnchanged(self, id, before, why=""):
|
||||
self.assertEqual(self.raw(id), before,
|
||||
"%s.md was rewritten%s" % (id, why and " — " + why))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# it closes
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class ClosesTest(StoreTestCase):
|
||||
|
||||
def test_a_slug_closes_the_issue_it_names(self):
|
||||
self.synced("a-thing", 101)
|
||||
out, _ = self.run_close("a-thing")
|
||||
self.assertEqual(self.fake.states[101], "closed")
|
||||
self.assertIn("closed a-thing #101", out)
|
||||
|
||||
def test_the_local_state_follows(self):
|
||||
self.synced("a-thing", 101)
|
||||
self.run_close("a-thing")
|
||||
self.assertEqual(self.state_on_disk("a-thing"), "closed")
|
||||
|
||||
def test_only_the_state_is_sent(self):
|
||||
"""Closing is not an edit: no title, no body, no labels ride along."""
|
||||
self.synced("a-thing", 101)
|
||||
self.run_close("a-thing")
|
||||
writes = self.fake.writes()
|
||||
self.assertEqual(len(writes), 1)
|
||||
method, endpoint, payload = writes[0]
|
||||
self.assertEqual((method, endpoint), ("PATCH", "%s/issues/101" % BASE))
|
||||
self.assertEqual(payload, {"state": "closed"})
|
||||
|
||||
def test_a_number_closes_it_too(self):
|
||||
"""The normal case for a pushed issue — the file is long gone."""
|
||||
self.synced("a-thing", 101)
|
||||
self.run_close("101")
|
||||
self.assertEqual(self.fake.states[101], "closed")
|
||||
self.assertEqual(self.state_on_disk("a-thing"), "closed")
|
||||
|
||||
def test_every_key_form_is_accepted(self):
|
||||
forms = {110: "110", 111: "#111", 112: "%s#112" % REPO,
|
||||
113: "https://git.example/%s/issues/113" % REPO}
|
||||
for n in forms:
|
||||
self.fake.states[n] = "open"
|
||||
for n, arg in forms.items():
|
||||
with self.subTest(arg=arg):
|
||||
self.run_close(arg)
|
||||
self.assertEqual(self.fake.states[n], "closed")
|
||||
|
||||
def test_several_ids_in_one_run(self):
|
||||
self.synced("a-thing", 101)
|
||||
self.synced("b-thing", 102)
|
||||
self.run_close("a-thing", "102")
|
||||
self.assertEqual(self.fake.states, {101: "closed", 102: "closed"})
|
||||
self.assertEqual(self.state_on_disk("a-thing"), "closed")
|
||||
self.assertEqual(self.state_on_disk("b-thing"), "closed")
|
||||
|
||||
def test_the_same_issue_named_twice_is_written_once(self):
|
||||
self.synced("a-thing", 101)
|
||||
self.run_close("a-thing", "#101")
|
||||
self.assertEqual(len(self.fake.writes()), 1)
|
||||
|
||||
def test_the_index_is_rebuilt(self):
|
||||
self.synced("a-thing", 101)
|
||||
out, _ = self.run_close("a-thing")
|
||||
self.assertIn("index:", out)
|
||||
with open(os.path.join(self.root, "INDEX.md")) as f:
|
||||
self.assertIn("closed", f.read())
|
||||
|
||||
def test_the_body_survives_untouched(self):
|
||||
"""One metadata field changes; the prose and the ticks do not."""
|
||||
self.synced("a-thing", 101)
|
||||
before = issue.load(self.root, "a-thing").body
|
||||
self.run_close("a-thing")
|
||||
self.assertEqual(issue.load(self.root, "a-thing").body, before)
|
||||
|
||||
def test_synced_is_refreshed(self):
|
||||
self.synced("a-thing", 101)
|
||||
self.run_close("a-thing")
|
||||
iss = issue.load(self.root, "a-thing")
|
||||
self.assertNotEqual(iss.extra.get("synced"), "2026-08-10T00:00:00Z")
|
||||
self.assertEqual(iss.extra.get("remote-updated"), "2026-08-11T00:00:00Z")
|
||||
|
||||
def test_an_issue_whose_file_was_dropped_still_closes(self):
|
||||
"""No local copy at all: the ledger names it, the tracker takes it, and
|
||||
nothing is written locally."""
|
||||
self.dropped("gone-thing", 205)
|
||||
out, _ = self.run_close("gone-thing")
|
||||
self.assertEqual(self.fake.states[205], "closed")
|
||||
self.assertIn("no local copy", out)
|
||||
self.assertNotIn("index:", out)
|
||||
|
||||
def test_a_number_nobody_here_knows_closes_without_a_slug(self):
|
||||
self.fake.states[777] = "open"
|
||||
out, _ = self.run_close("777")
|
||||
self.assertEqual(self.fake.states[777], "closed")
|
||||
self.assertIn("#777", out)
|
||||
|
||||
|
||||
class ReopensTest(StoreTestCase):
|
||||
|
||||
def test_reopen_sends_open(self):
|
||||
self.synced("a-thing", 101, state="closed")
|
||||
out, _ = self.run_close("--reopen", "a-thing")
|
||||
self.assertEqual(self.fake.writes()[0][2], {"state": "open"})
|
||||
self.assertIn("reopened a-thing #101", out)
|
||||
|
||||
def test_reopen_writes_the_local_state_back(self):
|
||||
self.synced("a-thing", 101, state="closed")
|
||||
self.run_close("--reopen", "a-thing")
|
||||
self.assertEqual(self.state_on_disk("a-thing"), "open")
|
||||
|
||||
def test_close_then_reopen_is_a_round_trip(self):
|
||||
self.synced("a-thing", 101)
|
||||
self.run_close("a-thing")
|
||||
self.run_close("--reopen", "a-thing")
|
||||
self.assertEqual(self.fake.states[101], "open")
|
||||
self.assertEqual(self.state_on_disk("a-thing"), "open")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# it refuses
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class LocalOnlyTest(StoreTestCase):
|
||||
"""An `origin: local` issue is not in the tracker, so it cannot be closed
|
||||
there — and the local field is not quietly edited instead."""
|
||||
|
||||
def test_it_exits(self):
|
||||
self.local_only("local-thing")
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_close("local-thing")
|
||||
|
||||
def test_the_error_names_the_id_and_says_it_is_not_in_the_tracker(self):
|
||||
self.local_only("local-thing")
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_close("local-thing")
|
||||
err = self.err.getvalue()
|
||||
self.assertIn("local-thing", err)
|
||||
self.assertIn("not in the tracker", err)
|
||||
|
||||
def test_nothing_is_sent(self):
|
||||
self.local_only("local-thing")
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_close("local-thing")
|
||||
self.assertEqual(self.fake.calls, [])
|
||||
|
||||
def test_the_file_is_untouched(self):
|
||||
self.local_only("local-thing")
|
||||
before = self.raw("local-thing")
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_close("local-thing")
|
||||
self.assertUnchanged("local-thing", before)
|
||||
|
||||
def test_a_bad_id_stops_the_whole_run_before_anything_is_sent(self):
|
||||
"""Resolution happens up front, so a typo in the second id does not
|
||||
leave the first one closed."""
|
||||
self.synced("a-thing", 101)
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_close("a-thing", "local-thing")
|
||||
self.assertEqual(self.fake.states[101], "open")
|
||||
self.assertEqual(self.fake.calls, [])
|
||||
|
||||
def test_an_unknown_slug_exits(self):
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_close("no-such-thing")
|
||||
self.assertIn("no-such-thing", self.err.getvalue())
|
||||
|
||||
|
||||
class DryRunTest(StoreTestCase):
|
||||
|
||||
def test_not_one_request_is_made(self):
|
||||
self.synced("a-thing", 101)
|
||||
self.run_close("--dry-run", "a-thing")
|
||||
self.assertEqual(self.fake.calls, [])
|
||||
|
||||
def test_the_file_is_untouched(self):
|
||||
self.synced("a-thing", 101)
|
||||
before = self.raw("a-thing")
|
||||
self.run_close("--dry-run", "a-thing")
|
||||
self.assertUnchanged("a-thing", before, "--dry-run must write nothing")
|
||||
|
||||
def test_it_says_what_would_be_closed(self):
|
||||
self.synced("a-thing", 101)
|
||||
self.synced("b-thing", 102)
|
||||
out, _ = self.run_close("--dry-run", "a-thing", "102")
|
||||
self.assertIn("would close a-thing #101", out)
|
||||
self.assertIn("would close b-thing #102", out)
|
||||
self.assertIn("2 issue(s) would be closed", out)
|
||||
|
||||
def test_it_says_reopen_under_reopen(self):
|
||||
self.synced("a-thing", 101, state="closed")
|
||||
out, _ = self.run_close("--dry-run", "--reopen", "a-thing")
|
||||
self.assertIn("would reopen a-thing #101", out)
|
||||
self.assertIn("would be reopened", out)
|
||||
|
||||
def test_it_needs_no_login(self):
|
||||
"""A dry run must work before /tea:auth has ever been run."""
|
||||
self.synced("a-thing", 101)
|
||||
with mock.patch.object(_gitea, "require_login",
|
||||
lambda: self.fail("dry run asked for a login")):
|
||||
self.run_close("--dry-run", "a-thing")
|
||||
|
||||
def test_a_local_only_issue_is_still_refused(self):
|
||||
self.local_only("local-thing")
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_close("--dry-run", "local-thing")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# the tracker said no
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TrackerFailureTest(StoreTestCase):
|
||||
"""The criterion that matters most: a write that was not confirmed leaves
|
||||
the local file exactly as it was."""
|
||||
|
||||
def test_a_non_2xx_answer_leaves_the_file(self):
|
||||
"""The real `_gitea.api` against a `tea` that exits 1 — the path a 422
|
||||
or a 500 actually takes, and it ends in `die()`."""
|
||||
self.synced("a-thing", 101)
|
||||
before = self.raw("a-thing")
|
||||
|
||||
def fake_run(cmd, capture_output=False, text=False):
|
||||
return types.SimpleNamespace(
|
||||
returncode=1, stdout="",
|
||||
stderr="422 Unprocessable Entity: issue is blocked")
|
||||
|
||||
with mock.patch.object(_gitea, "api", REAL_API), \
|
||||
mock.patch.object(_gitea, "subprocess",
|
||||
types.SimpleNamespace(run=fake_run)), \
|
||||
self.assertRaises(SystemExit):
|
||||
self.run_close("a-thing")
|
||||
|
||||
self.assertUnchanged("a-thing", before, "tea exited non-zero")
|
||||
self.assertEqual(self.state_on_disk("a-thing"), "open")
|
||||
|
||||
def test_a_transport_exception_leaves_the_file(self):
|
||||
self.synced("a-thing", 101)
|
||||
before = self.raw("a-thing")
|
||||
self.fake.raise_on_write = OSError("tea: command not found")
|
||||
with self.assertRaises(OSError):
|
||||
self.run_close("a-thing")
|
||||
self.assertUnchanged("a-thing", before, "the transport raised")
|
||||
|
||||
def test_an_answer_without_a_number_leaves_the_file(self):
|
||||
self.synced("a-thing", 101)
|
||||
before = self.raw("a-thing")
|
||||
self.fake.answer_override = {"ok": True, "state": "closed"}
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_close("a-thing")
|
||||
self.assertUnchanged("a-thing", before)
|
||||
|
||||
def test_an_answer_for_another_issue_leaves_the_file(self):
|
||||
self.synced("a-thing", 101)
|
||||
before = self.raw("a-thing")
|
||||
self.fake.answer_override = {"number": 999, "state": "closed"}
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_close("a-thing")
|
||||
self.assertUnchanged("a-thing", before)
|
||||
|
||||
def test_an_answer_that_did_not_change_the_state_leaves_the_file(self):
|
||||
"""A 200 that still says `open` is not a close."""
|
||||
self.synced("a-thing", 101)
|
||||
before = self.raw("a-thing")
|
||||
self.fake.answer_override = {"number": 101, "state": "open"}
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_close("a-thing")
|
||||
self.assertUnchanged("a-thing", before)
|
||||
|
||||
def test_an_empty_answer_leaves_the_file(self):
|
||||
self.synced("a-thing", 101)
|
||||
before = self.raw("a-thing")
|
||||
self.fake.answer_override = None
|
||||
real_api = self.fake.api
|
||||
self.fake.api = lambda *a, **kw: (real_api(*a, **kw), None)[1]
|
||||
with mock.patch.object(_gitea, "api", self.fake.api), \
|
||||
self.assertRaises(SystemExit):
|
||||
self.run_close("a-thing")
|
||||
self.assertUnchanged("a-thing", before)
|
||||
|
||||
def test_the_error_says_nothing_local_changed(self):
|
||||
self.synced("a-thing", 101)
|
||||
self.fake.answer_override = {"ok": True}
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_close("a-thing")
|
||||
self.assertIn("Nothing local was changed", self.err.getvalue())
|
||||
|
||||
def test_a_failure_partway_through_keeps_the_rest(self):
|
||||
"""Two issues, the second one is not confirmed. The first is
|
||||
legitimately closed; the second's file still says open."""
|
||||
self.synced("aaa-thing", 101)
|
||||
self.synced("zzz-thing", 102)
|
||||
before = self.raw("zzz-thing")
|
||||
|
||||
real = self.fake.api
|
||||
seen = []
|
||||
|
||||
def once(login, endpoint, method="GET", payload=None, **kw):
|
||||
got = real(login, endpoint, method, payload, **kw)
|
||||
if method != "GET":
|
||||
seen.append(endpoint)
|
||||
return {"nope": True} if len(seen) > 1 else got
|
||||
|
||||
with mock.patch.object(_gitea, "api", once), \
|
||||
self.assertRaises(SystemExit):
|
||||
self.run_close("aaa-thing", "zzz-thing")
|
||||
|
||||
self.assertEqual(self.state_on_disk("aaa-thing"), "closed")
|
||||
self.assertUnchanged("zzz-thing", before, "its write was not confirmed")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# the pure parts
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class ConfirmedTest(unittest.TestCase):
|
||||
"""The gate itself. Everything below it rewrites a file."""
|
||||
|
||||
def test_a_matching_close_is_confirmed(self):
|
||||
self.assertTrue(close.confirmed({"number": 42, "state": "closed"}, 42, "closed"))
|
||||
|
||||
def test_a_mismatched_number_is_not(self):
|
||||
self.assertFalse(close.confirmed({"number": 43, "state": "closed"}, 42, "closed"))
|
||||
|
||||
def test_the_wrong_state_is_not(self):
|
||||
self.assertFalse(close.confirmed({"number": 42, "state": "open"}, 42, "closed"))
|
||||
|
||||
def test_a_missing_state_is_not(self):
|
||||
self.assertFalse(close.confirmed({"number": 42}, 42, "closed"))
|
||||
|
||||
def test_none_and_lists_are_not(self):
|
||||
self.assertFalse(close.confirmed(None, 42, "closed"))
|
||||
self.assertFalse(close.confirmed([{"number": 42, "state": "closed"}], 42, "closed"))
|
||||
|
||||
def test_true_is_not_a_number(self):
|
||||
self.assertFalse(close.confirmed({"number": True, "state": "closed"}, 1, "closed"))
|
||||
|
||||
def test_a_string_number_is_not(self):
|
||||
self.assertFalse(close.confirmed({"number": "42", "state": "closed"}, 42, "closed"))
|
||||
|
||||
|
||||
class KeyFormTest(unittest.TestCase):
|
||||
"""A slug and a key are two vocabularies that must not collide."""
|
||||
|
||||
def test_keys_are_keys(self):
|
||||
for k in ("42", "#42", "owner/repo#42",
|
||||
"https://git.example/owner/repo/issues/42"):
|
||||
self.assertTrue(close.looks_like_key(k), k)
|
||||
|
||||
def test_slugs_are_not_keys(self):
|
||||
for s in ("a-thing", "wire-sqlc-appclick", "close-issues-through-a-script"):
|
||||
self.assertFalse(close.looks_like_key(s), s)
|
||||
|
||||
|
||||
class LedgerPairsTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.m = {"%s#7" % REPO: "a-thing", "other/repo#7": "b-thing",
|
||||
"not-a-key": "c-thing"}
|
||||
|
||||
def test_it_filters_by_repo(self):
|
||||
self.assertEqual(close.ledger_pairs(self.m, REPO), [(REPO, 7, "a-thing")])
|
||||
|
||||
def test_without_a_repo_it_keeps_everything_parseable(self):
|
||||
got = close.ledger_pairs(self.m)
|
||||
self.assertEqual(sorted(s for _r, _n, s in got), ["a-thing", "b-thing"])
|
||||
|
||||
def test_an_ambiguous_number_exits(self):
|
||||
pairs = close.ledger_pairs(self.m)
|
||||
with self.assertRaises(SystemExit):
|
||||
with contextlib.redirect_stderr(io.StringIO()):
|
||||
close.resolve("7", {}, pairs)
|
||||
|
||||
|
||||
class AmbiguityTest(StoreTestCase):
|
||||
"""Two repos, one number, no --repo: settle it rather than guess."""
|
||||
|
||||
def test_the_error_points_at_repo(self):
|
||||
_gitea.save_map(self.root, {"%s#7" % REPO: "a-thing",
|
||||
"other/repo#7": "b-thing"})
|
||||
err = io.StringIO()
|
||||
args = ["close.py", "--out", self.root, "7"]
|
||||
with mock.patch.object(sys, "argv", args), \
|
||||
contextlib.redirect_stdout(io.StringIO()), \
|
||||
contextlib.redirect_stderr(err), \
|
||||
self.assertRaises(SystemExit):
|
||||
close.main()
|
||||
self.assertIn("--repo", err.getvalue())
|
||||
|
||||
|
||||
class RepoOfTheKeyTest(StoreTestCase):
|
||||
"""A key that names its own repo is sent there, not to whatever repo the
|
||||
CWD happens to be — otherwise `#42` closes somebody else's issue."""
|
||||
|
||||
def run_bare(self, *argv):
|
||||
"""No `--repo`, so the ids have to say where they live."""
|
||||
self.out, self.err = io.StringIO(), io.StringIO()
|
||||
args = ["close.py", "--out", self.root] + list(argv)
|
||||
with mock.patch.object(sys, "argv", args), \
|
||||
contextlib.redirect_stdout(self.out), \
|
||||
contextlib.redirect_stderr(self.err):
|
||||
close.main()
|
||||
return self.out.getvalue(), self.err.getvalue()
|
||||
|
||||
def test_a_foreign_key_goes_to_its_own_repo(self):
|
||||
self.run_bare("other/repo#42")
|
||||
self.assertEqual(self.fake.writes()[0][1], "repos/other/repo/issues/42")
|
||||
|
||||
def test_a_slug_goes_to_the_repo_its_gitea_field_names(self):
|
||||
self.synced("a-thing", 101)
|
||||
self.run_bare("a-thing")
|
||||
self.assertEqual(self.fake.writes()[0][1], "%s/issues/101" % BASE)
|
||||
|
||||
def test_two_repos_in_one_run_is_a_question_not_a_guess(self):
|
||||
self.synced("a-thing", 101)
|
||||
with self.assertRaises(SystemExit):
|
||||
self.run_bare("a-thing", "other/repo#42")
|
||||
self.assertIn("one repo", self.err.getvalue())
|
||||
self.assertEqual(self.fake.calls, [])
|
||||
|
||||
def test_an_explicit_repo_settles_it(self):
|
||||
self.synced("a-thing", 101)
|
||||
self.run_close("a-thing", "other/repo#42")
|
||||
self.assertEqual({c[1] for c in self.fake.writes()},
|
||||
{"%s/issues/101" % BASE, "%s/issues/42" % BASE})
|
||||
|
||||
|
||||
class NoStoreTest(StoreTestCase):
|
||||
"""A number needs no local file, and a store that is not there is not an
|
||||
error — closing an issue whose copy push dropped is the normal case."""
|
||||
|
||||
def test_a_number_closes_with_no_store_at_all(self):
|
||||
missing = os.path.join(self.root, "nowhere")
|
||||
self.fake.states[303] = "open"
|
||||
args = ["close.py", "--repo", REPO, "--out", missing, "303"]
|
||||
with mock.patch.object(sys, "argv", args), \
|
||||
contextlib.redirect_stdout(io.StringIO()), \
|
||||
contextlib.redirect_stderr(io.StringIO()):
|
||||
close.main()
|
||||
self.assertEqual(self.fake.states[303], "closed")
|
||||
self.assertFalse(os.path.isdir(missing), "no store was conjured")
|
||||
|
||||
|
||||
class PayloadFileTest(StoreTestCase):
|
||||
"""The request body goes to the transport's own scratchpad.
|
||||
|
||||
Not to a directory this script picks: `close.py` names the payload and
|
||||
nothing else, the way every other caller does. Where PAYLOAD_ROOT lands is
|
||||
_gitea's business, and test_payload_root.py is where that is tested."""
|
||||
|
||||
def test_the_payload_lands_in_the_transports_scratchpad(self):
|
||||
self.synced("a-thing", 101)
|
||||
payloads = os.path.join(self.root, "payload")
|
||||
with mock.patch.object(_gitea, "PAYLOAD_ROOT", payloads), \
|
||||
mock.patch.object(_gitea, "api", REAL_API), \
|
||||
mock.patch.object(
|
||||
_gitea, "subprocess",
|
||||
types.SimpleNamespace(run=lambda cmd, **kw: types.SimpleNamespace(
|
||||
returncode=0, stderr="",
|
||||
stdout=json.dumps({"number": 101, "state": "closed"})))):
|
||||
self.run_close("a-thing")
|
||||
p = os.path.join(payloads, "state-101.json")
|
||||
self.assertTrue(os.path.isfile(p))
|
||||
with open(p) as f:
|
||||
self.assertEqual(json.load(f), {"state": "closed"})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -124,7 +124,7 @@ class FakeTracker(object):
|
||||
# -- the seam ----------------------------------------------------------
|
||||
|
||||
def api(self, login, endpoint, method="GET", payload=None,
|
||||
payload_name=None, out_root=None, allow_fail=False):
|
||||
payload_name=None, allow_fail=False):
|
||||
self.calls.append((method, endpoint, payload))
|
||||
path = endpoint.split("?")[0]
|
||||
|
||||
@@ -194,7 +194,13 @@ class StoreTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.root = tempfile.mkdtemp(prefix="tea-drop-")
|
||||
self.fake = FakeTracker()
|
||||
# PAYLOAD_ROOT is the repo's own tmp/payload, and a test that stubs the
|
||||
# transport one layer down (see the non-2xx case) reaches the real
|
||||
# write. Point it at the fixture: a test writes in its temp directory
|
||||
# and nowhere else.
|
||||
for p in (mock.patch.object(_gitea, "api", self.fake.api),
|
||||
mock.patch.object(_gitea, "PAYLOAD_ROOT",
|
||||
os.path.join(self.root, "payload")),
|
||||
mock.patch.object(_gitea, "require_login", lambda: "test-login"),
|
||||
mock.patch.object(push, "git_branch", lambda: "test-branch")):
|
||||
p.start()
|
||||
|
||||
@@ -0,0 +1,418 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Where the login pin is found, and that a git worktree is not a dead zone.
|
||||
|
||||
python3 -m unittest discover -s tests -v
|
||||
|
||||
Stdlib unittest, no third-party anything, and not one real network call: every
|
||||
run here is against a throwaway repository with a FAKE `tea` first on PATH.
|
||||
|
||||
The bug: the pin was searched for by walking up from CWD only. A worktree is a
|
||||
*sibling* of the main checkout, and `.claude/settings.local.json` is untracked,
|
||||
so it lives in the main checkout and nowhere else — the whole sync layer died
|
||||
inside any worktree with "no login pinned", while `tea` in the same directory
|
||||
worked, because the tea-guard hook had a second, different copy of the search.
|
||||
|
||||
So these tests hold two lines at once: the pin is reachable from a worktree,
|
||||
and the hook and the scripts get their answer from the same function.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
AUTH_SCRIPTS = os.path.join(REPO, "skills", "auth", "scripts")
|
||||
SYNC_SCRIPTS = os.path.join(REPO, "skills", "sync", "scripts")
|
||||
HOOKS = os.path.join(REPO, "hooks")
|
||||
|
||||
sys.path.insert(0, AUTH_SCRIPTS)
|
||||
import pin # noqa: E402
|
||||
|
||||
HAVE_GIT = shutil.which("git") is not None
|
||||
|
||||
LOGIN = "fixture/user"
|
||||
ENV_KEY = pin.ENV_KEY
|
||||
|
||||
# A `tea` that answers without a network: an empty list for every GET, a
|
||||
# created object for every write. It records its own argv, which is how a test
|
||||
# reads back the login the call actually ran under.
|
||||
FAKE_TEA = '''#!%s
|
||||
import json, os, sys
|
||||
argv = sys.argv[1:]
|
||||
with open(os.environ["TEA_CALL_LOG"], "a") as f:
|
||||
f.write("\\t".join(argv) + "\\n")
|
||||
sys.stdout.write(json.dumps({"id": 1, "number": 101, "name": "created",
|
||||
"html_url": "https://example.invalid/issues/101",
|
||||
"labels": []})
|
||||
if "-X" in argv else "[]")
|
||||
'''
|
||||
|
||||
ISSUE = """\
|
||||
---
|
||||
id: pinned-work
|
||||
state: open
|
||||
labels: [type/task]
|
||||
assignees: []
|
||||
milestone: none
|
||||
depends: []
|
||||
origin: local
|
||||
---
|
||||
# Pinned work
|
||||
|
||||
## Summary
|
||||
Issue фикстуры, живёт в сторе worktree.
|
||||
|
||||
## Spec
|
||||
none
|
||||
|
||||
## Motivation
|
||||
Нужен, чтобы push.py было что отправить.
|
||||
|
||||
## Acceptance criteria
|
||||
- [ ] проверяемое условие
|
||||
"""
|
||||
|
||||
|
||||
def write(path, text):
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
with open(path, "w") as f:
|
||||
f.write(text)
|
||||
|
||||
|
||||
class Worktree(object):
|
||||
"""A repository with a pin, and a linked worktree beside it.
|
||||
|
||||
Beside, not below: `main/` and `worktrees/feature/` are siblings, which is
|
||||
the entire shape of the bug. The pin is written after the clone is
|
||||
committed and is covered by .gitignore, so it exists in the main checkout
|
||||
only — exactly as `/tea:auth` leaves it."""
|
||||
|
||||
def __init__(self, pinned=LOGIN):
|
||||
self._tmp = tempfile.TemporaryDirectory(prefix="tea-pin-")
|
||||
# realpath: on macOS $TMPDIR is a symlink, and a child reporting its
|
||||
# own cwd would otherwise disagree with the path we handed it.
|
||||
self.root = os.path.realpath(self._tmp.name)
|
||||
self.main = os.path.join(self.root, "main")
|
||||
self.tree = os.path.join(self.root, "worktrees", "feature")
|
||||
self.calls = os.path.join(self.root, "calls.txt")
|
||||
|
||||
skip = shutil.ignore_patterns("__pycache__")
|
||||
for layer in ("auth", "issue", "sync"):
|
||||
shutil.copytree(os.path.join(REPO, "skills", layer, "scripts"),
|
||||
os.path.join(self.main, "skills", layer, "scripts"),
|
||||
ignore=skip)
|
||||
shutil.copytree(HOOKS, os.path.join(self.main, "hooks"), ignore=skip)
|
||||
write(os.path.join(self.main, ".gitignore"), "tmp/\n.claude/\n")
|
||||
|
||||
self.bin = os.path.join(self.root, "fakebin")
|
||||
os.makedirs(self.bin)
|
||||
tea = os.path.join(self.bin, "tea")
|
||||
write(tea, FAKE_TEA % sys.executable)
|
||||
os.chmod(tea, os.stat(tea).st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
|
||||
|
||||
self.git("init", cwd=self.main)
|
||||
self.git("add", "-A", cwd=self.main)
|
||||
self.git("commit", "-m", "fixture", cwd=self.main)
|
||||
self.git("worktree", "add", "-b", "feature", self.tree, cwd=self.main)
|
||||
|
||||
if pinned:
|
||||
write(os.path.join(self.main, ".claude", "settings.local.json"),
|
||||
json.dumps({"env": {ENV_KEY: pinned}}))
|
||||
|
||||
def cleanup(self):
|
||||
self._tmp.cleanup()
|
||||
|
||||
def env(self):
|
||||
env = dict(os.environ)
|
||||
env.pop("PYTHONPATH", None) # no leakage from the harness into the child
|
||||
# The start of the search order, cleared: this fixture is about the
|
||||
# steps *after* it, and the developer's own project must not answer.
|
||||
env.pop(pin.PROJECT_DIR_ENV, None)
|
||||
env["PATH"] = self.bin + os.pathsep + env["PATH"]
|
||||
env["TEA_CALL_LOG"] = self.calls
|
||||
env["HOME"] = self.root # keep the developer's git config out
|
||||
env["GIT_CONFIG_NOSYSTEM"] = "1"
|
||||
env["GIT_CONFIG_GLOBAL"] = os.devnull
|
||||
return env
|
||||
|
||||
def git(self, *args, **kw):
|
||||
cmd = ["git", "-c", "user.email=fixture@example.invalid",
|
||||
"-c", "user.name=fixture", "-c", "commit.gpgsign=false"] + list(args)
|
||||
p = subprocess.run(cmd, cwd=kw.pop("cwd", self.tree), env=self.env(),
|
||||
capture_output=True, text=True)
|
||||
if p.returncode != 0:
|
||||
raise AssertionError("%s failed:\n%s%s" % (" ".join(cmd), p.stdout, p.stderr))
|
||||
return p.stdout.strip()
|
||||
|
||||
def script(self, layer, name):
|
||||
"""A script as the WORKTREE sees it — the copy the operator would run."""
|
||||
return os.path.join(self.tree, "skills", layer, "scripts", name)
|
||||
|
||||
def run(self, script, *args, **kw):
|
||||
p = subprocess.run([sys.executable, script] + list(args),
|
||||
cwd=kw.pop("cwd", self.tree), env=self.env(),
|
||||
capture_output=True, text=True)
|
||||
return p.returncode, p.stdout, p.stderr
|
||||
|
||||
def tea_calls(self):
|
||||
if not os.path.isfile(self.calls):
|
||||
return []
|
||||
with open(self.calls) as f:
|
||||
return [line.rstrip("\n").split("\t") for line in f if line.strip()]
|
||||
|
||||
def logins_used(self):
|
||||
return [a[a.index("--login") + 1] for a in self.tea_calls() if "--login" in a]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# the search itself
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TestSearch(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory(prefix="tea-pin-unit-")
|
||||
self.root = os.path.realpath(self._tmp.name)
|
||||
self.addCleanup(self._tmp.cleanup)
|
||||
|
||||
def path(self, *parts):
|
||||
return os.path.join(self.root, *parts)
|
||||
|
||||
def pin_at(self, root, login=LOGIN):
|
||||
write(os.path.join(root, ".claude", "settings.local.json"),
|
||||
json.dumps({"env": {ENV_KEY: login}}))
|
||||
|
||||
def test_the_parent_chain_is_searched(self):
|
||||
self.pin_at(self.root)
|
||||
os.makedirs(self.path("a", "b"))
|
||||
self.assertEqual(pin.search(self.path("a", "b"))[0], LOGIN)
|
||||
|
||||
def test_no_pin_is_no_pin(self):
|
||||
os.makedirs(self.path("a"))
|
||||
self.assertEqual(pin.search(self.path("a")), (None, None))
|
||||
|
||||
def test_an_unreadable_pin_is_not_a_login(self):
|
||||
write(self.path(".claude", "settings.local.json"), "{ not json")
|
||||
self.assertEqual(pin.search(self.root), (None, None))
|
||||
|
||||
def test_an_empty_pin_is_not_a_login(self):
|
||||
write(self.path(".claude", "settings.local.json"),
|
||||
json.dumps({"env": {ENV_KEY: " "}}))
|
||||
self.assertEqual(pin.search(self.root), (None, None))
|
||||
|
||||
def test_a_git_file_pointing_at_a_worktree_reaches_the_main_checkout(self):
|
||||
"""The hop, built by hand from the two files git writes — no git
|
||||
needed to state what the layout means."""
|
||||
main, tree = self.path("main"), self.path("elsewhere", "feature")
|
||||
gitdir = os.path.join(main, ".git", "worktrees", "feature")
|
||||
os.makedirs(gitdir)
|
||||
os.makedirs(tree)
|
||||
write(os.path.join(gitdir, "commondir"), "../..\n")
|
||||
write(os.path.join(tree, ".git"), "gitdir: %s\n" % gitdir)
|
||||
self.pin_at(main)
|
||||
|
||||
self.assertEqual(pin.main_worktree(tree), main)
|
||||
login, src = pin.search(tree)
|
||||
self.assertEqual(login, LOGIN)
|
||||
self.assertEqual(src, pin.settings_path(main))
|
||||
|
||||
def test_an_ordinary_clone_is_not_a_worktree(self):
|
||||
os.makedirs(self.path("clone", ".git"))
|
||||
self.assertIsNone(pin.main_worktree(self.path("clone")))
|
||||
|
||||
def test_a_submodule_pointer_is_not_a_worktree(self):
|
||||
"""`.git` is a file there too, but it points into .git/modules/… and
|
||||
the tree it belongs to is already on the parent chain."""
|
||||
sub = self.path("super", "lib")
|
||||
gitdir = self.path("super", ".git", "modules", "lib")
|
||||
os.makedirs(gitdir)
|
||||
os.makedirs(sub)
|
||||
write(os.path.join(sub, ".git"), "gitdir: %s\n" % gitdir)
|
||||
self.assertIsNone(pin.main_worktree(sub))
|
||||
|
||||
def test_the_chain_wins_over_the_hop(self):
|
||||
"""The worktree branch may only find a pin the walk up would have
|
||||
missed entirely — it never overrides a nearer one."""
|
||||
main, tree = self.path("main"), self.path("elsewhere", "feature")
|
||||
gitdir = os.path.join(main, ".git", "worktrees", "feature")
|
||||
os.makedirs(gitdir)
|
||||
os.makedirs(tree)
|
||||
write(os.path.join(gitdir, "commondir"), "../..\n")
|
||||
write(os.path.join(tree, ".git"), "gitdir: %s\n" % gitdir)
|
||||
self.pin_at(main, "main/login")
|
||||
self.pin_at(tree, "worktree/login")
|
||||
self.assertEqual(pin.search(tree)[0], "worktree/login")
|
||||
|
||||
def test_start_dirs_are_ordered_and_deduplicated(self):
|
||||
with mock.patch.dict(os.environ, {pin.PROJECT_DIR_ENV: self.path("p")}):
|
||||
self.assertEqual(pin.start_dirs(self.path("h")),
|
||||
[self.path("p"), self.path("h"),
|
||||
os.path.abspath(os.getcwd())])
|
||||
with mock.patch.dict(os.environ, {}, clear=True):
|
||||
self.assertEqual(pin.start_dirs(), [os.path.abspath(os.getcwd())])
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# a script run from a worktree
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
@unittest.skipUnless(HAVE_GIT, "git is not installed")
|
||||
class TestScriptsInAWorktree(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.wt = Worktree()
|
||||
self.addCleanup(self.wt.cleanup)
|
||||
|
||||
def test_a_sync_script_run_from_the_worktree_finds_the_login(self):
|
||||
"""The acceptance criterion, run for real: cwd inside the worktree,
|
||||
the pin in the main checkout, and the call goes out under it."""
|
||||
rc, out, err = self.wt.run(self.wt.script("sync", "remote.py"),
|
||||
"--repo", "fixture/repo", "--state", "all")
|
||||
self.assertEqual(rc, 0, "remote.py failed:\n%s%s" % (out, err))
|
||||
self.assertNotIn("no login pinned", err)
|
||||
self.assertEqual(self.wt.logins_used(), [LOGIN])
|
||||
|
||||
def test_it_does_not_pin_a_second_login_in_the_worktree(self):
|
||||
"""Nothing here writes a settings file, and the worktree is the last
|
||||
place one should appear: it is deleted with the worktree."""
|
||||
self.wt.run(self.wt.script("sync", "remote.py"), "--repo", "fixture/repo")
|
||||
self.assertFalse(os.path.exists(pin.settings_path(self.wt.tree)),
|
||||
"a second settings.local.json appeared in the worktree")
|
||||
|
||||
def test_with_no_pin_anywhere_it_still_says_so(self):
|
||||
wt = Worktree(pinned=None)
|
||||
self.addCleanup(wt.cleanup)
|
||||
rc, out, err = wt.run(wt.script("sync", "remote.py"), "--repo", "fixture/repo")
|
||||
self.assertNotEqual(rc, 0)
|
||||
self.assertIn("no login pinned", err)
|
||||
self.assertEqual(wt.logins_used(), [])
|
||||
|
||||
def test_the_scripts_own_directory_is_not_a_pin_source(self):
|
||||
"""Run the worktree's script from a directory that is in no pinned
|
||||
tree. The script sits inside a repository that has a pin — and it must
|
||||
still refuse, because the pin belongs to the project being worked on,
|
||||
not to the installation."""
|
||||
outside = os.path.join(self.wt.root, "outside")
|
||||
os.makedirs(outside)
|
||||
rc, out, err = self.wt.run(self.wt.script("sync", "remote.py"),
|
||||
"--repo", "fixture/repo", cwd=outside)
|
||||
self.assertNotEqual(rc, 0)
|
||||
self.assertIn("no login pinned", err)
|
||||
|
||||
def test_push_from_a_worktree_sends_the_worktree_branch(self):
|
||||
"""`branch:` -> Gitea `ref`. The workaround this fix removes — run the
|
||||
worktree's scripts with cwd in the main checkout — sent the main
|
||||
checkout's branch, which is the one field `branch:` exists for."""
|
||||
write(os.path.join(self.wt.tree, "tmp", "issues", "pinned-work.md"), ISSUE)
|
||||
rc, out, err = self.wt.run(self.wt.script("sync", "push.py"),
|
||||
"pinned-work", "--repo", "fixture/repo")
|
||||
self.assertEqual(rc, 0, "push.py failed:\n%s%s" % (out, err))
|
||||
self.assertIn("created pinned-work #101", out)
|
||||
|
||||
with open(os.path.join(self.wt.tree, "tmp", "payload",
|
||||
"issue-pinned-work.json")) as f:
|
||||
payload = json.load(f)
|
||||
self.assertEqual(payload.get("ref"), "feature")
|
||||
self.assertEqual(self.wt.git("rev-parse", "--abbrev-ref", "HEAD"), "feature")
|
||||
self.assertNotEqual(
|
||||
self.wt.git("rev-parse", "--abbrev-ref", "HEAD", cwd=self.wt.main),
|
||||
"feature", "the fixture's two trees are on the same branch")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# one order, one copy of it
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
@unittest.skipUnless(HAVE_GIT, "git is not installed")
|
||||
class TestTheHookAndTheScriptsAgree(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.wt = Worktree()
|
||||
self.addCleanup(self.wt.cleanup)
|
||||
|
||||
def guard(self, cwd):
|
||||
"""The hook, as the harness calls it: payload on stdin, decision on
|
||||
stdout."""
|
||||
payload = {"tool_input": {"command": 'tea api --login "$GITEA_LOGIN" repos/x/y'},
|
||||
"cwd": cwd}
|
||||
p = subprocess.run([sys.executable, os.path.join(self.wt.tree, "hooks",
|
||||
"tea-guard.sh")],
|
||||
input=json.dumps(payload), cwd=cwd, env=self.wt.env(),
|
||||
capture_output=True, text=True)
|
||||
return p
|
||||
|
||||
def test_the_hook_resolves_the_pin_from_the_worktree_too(self):
|
||||
p = self.guard(self.wt.tree)
|
||||
self.assertEqual(p.returncode, 0, p.stderr)
|
||||
got = json.loads(p.stdout)["hookSpecificOutput"]["updatedInput"]["command"]
|
||||
self.assertIn(LOGIN, got)
|
||||
self.assertNotIn("GITEA_LOGIN", got)
|
||||
|
||||
def test_the_hook_and_a_script_answer_the_same_directory_alike(self):
|
||||
"""The regression that started this: in one directory the hook
|
||||
resolved the login and every script said there was none."""
|
||||
rc, out, err = self.wt.run(self.wt.script("sync", "remote.py"),
|
||||
"--repo", "fixture/repo")
|
||||
self.assertEqual(rc, 0, err)
|
||||
script_login = self.wt.logins_used()[0]
|
||||
hook_login = json.loads(self.guard(self.wt.tree).stdout)[
|
||||
"hookSpecificOutput"]["updatedInput"]["command"].split("--login ")[1].split()[0]
|
||||
self.assertEqual(hook_login, script_login)
|
||||
|
||||
def test_the_hook_still_blocks_when_nothing_is_pinned(self):
|
||||
wt = Worktree(pinned=None)
|
||||
self.addCleanup(wt.cleanup)
|
||||
payload = {"tool_input": {"command": 'tea api --login "$GITEA_LOGIN" repos/x/y'},
|
||||
"cwd": wt.tree}
|
||||
p = subprocess.run([sys.executable, os.path.join(wt.tree, "hooks", "tea-guard.sh")],
|
||||
input=json.dumps(payload), cwd=wt.tree, env=wt.env(),
|
||||
capture_output=True, text=True)
|
||||
self.assertEqual(p.returncode, 2)
|
||||
self.assertIn("no login is pinned", p.stderr)
|
||||
|
||||
|
||||
class TestNobodyKeepsASecondCopy(unittest.TestCase):
|
||||
"""Mechanical: the search order is written in pin.py, and the two callers
|
||||
spell neither the path nor the walk."""
|
||||
|
||||
CALLERS = (os.path.join(HOOKS, "tea-guard.sh"),
|
||||
os.path.join(SYNC_SCRIPTS, "_gitea.py"))
|
||||
|
||||
def source(self, path):
|
||||
with open(path) as f:
|
||||
return f.read()
|
||||
|
||||
def test_the_path_is_spelled_once(self):
|
||||
self.assertEqual(pin.SETTINGS_PARTS, (".claude", "settings.local.json"))
|
||||
for path in self.CALLERS:
|
||||
body = self.source(path)
|
||||
for literal in ('".claude"', "'.claude'"):
|
||||
self.assertNotIn(literal, body,
|
||||
"%s builds the settings path itself" % path)
|
||||
|
||||
def test_both_callers_go_through_the_module(self):
|
||||
for path in self.CALLERS:
|
||||
self.assertIn("import pin", self.source(path),
|
||||
"%s does not resolve the pin through pin.py" % path)
|
||||
|
||||
def test_the_domain_layer_never_learns_what_a_login_is(self):
|
||||
"""The layer rule, unchanged by this: the identity module is imported
|
||||
by the bridge and by the hook, never by a domain."""
|
||||
for layer in ("issue", "page"):
|
||||
d = os.path.join(REPO, "skills", layer, "scripts")
|
||||
for name in sorted(os.listdir(d)):
|
||||
if not name.endswith(".py"):
|
||||
continue
|
||||
body = self.source(os.path.join(d, name))
|
||||
for banned in ("import pin", "GITEA_LOGIN", "settings.local.json"):
|
||||
self.assertNotIn(banned, body, "%s/%s: %s" % (layer, name, banned))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -329,13 +329,12 @@ class TestImportScript(unittest.TestCase):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.root = self.tmp.name
|
||||
os.makedirs(os.path.join(self.root, ".git"))
|
||||
for layer in ("page", "wiki"):
|
||||
# auth is in the list because the transport resolves the login pin
|
||||
# through skills/auth/scripts/pin.py — one search order, one module.
|
||||
for layer in ("page", "wiki", "sync", "auth"):
|
||||
shutil.copytree(os.path.join(REPO, "skills", layer, "scripts"),
|
||||
os.path.join(self.root, "skills", layer, "scripts"),
|
||||
ignore=shutil.ignore_patterns("__pycache__"))
|
||||
shutil.copytree(SYNC_SCRIPTS,
|
||||
os.path.join(self.root, "skills", "sync", "scripts"),
|
||||
ignore=shutil.ignore_patterns("__pycache__"))
|
||||
self.src = build_artifacts(os.path.join(self.root, "artifacts"))
|
||||
self.scripts = os.path.join(self.root, "skills", "page", "scripts")
|
||||
self.space = os.path.join(self.root, "tmp", "wiki", "s")
|
||||
|
||||
@@ -0,0 +1,250 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Where request bodies land, and that writing one never conjures a store.
|
||||
|
||||
python3 -m unittest discover -s tests -v
|
||||
|
||||
Stdlib unittest, no third-party anything. The bug these tests pin down:
|
||||
`labels.py --bootstrap` on a fresh checkout left `tmp/issues/.payload/` behind,
|
||||
because the only place `_gitea.api` had to put a request file was whatever root
|
||||
the caller handed it — and the label bootstrap, which touches no issue at all,
|
||||
handed it the issue store. A store materialized as a side effect of an
|
||||
operation that has nothing to do with issues.
|
||||
|
||||
Every run here is against a throwaway repository with a FAKE `tea` first on
|
||||
PATH, so nothing reaches the network and the developer's own store is never in
|
||||
the blast radius.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import stat
|
||||
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")
|
||||
WIKI_SCRIPTS = os.path.join(REPO, "skills", "wiki", "scripts")
|
||||
AUTH_SCRIPTS = os.path.join(REPO, "skills", "auth", "scripts")
|
||||
|
||||
sys.path.insert(0, SYNC_SCRIPTS)
|
||||
sys.path.insert(0, ISSUE_SCRIPTS)
|
||||
import _gitea # noqa: E402
|
||||
import issue # noqa: E402
|
||||
|
||||
|
||||
# A `tea` that answers without a network: an empty list for every GET (so the
|
||||
# repository looks like it has no labels yet) and a created object for every
|
||||
# write. It also records its own argv, which is how a test can tell that the
|
||||
# payload file the script wrote is the one the call actually referenced.
|
||||
FAKE_TEA = '''#!%s
|
||||
import json, os, sys
|
||||
with open(os.path.join(os.environ["TEA_CALL_LOG"], "calls.txt"), "a") as f:
|
||||
f.write("\\t".join(sys.argv[1:]) + "\\n")
|
||||
sys.stdout.write(json.dumps({"id": 1, "name": "created", "sub_url": "Page"})
|
||||
if "-X" in sys.argv else "[]")
|
||||
'''
|
||||
|
||||
|
||||
class FakeRepo(object):
|
||||
"""A self-contained repository with no store and no tmp/ at all."""
|
||||
|
||||
def __init__(self):
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
# realpath: on macOS $TMPDIR is a symlink, and a child 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)
|
||||
# the transport resolves the login pin through skills/auth/scripts
|
||||
shutil.copytree(AUTH_SCRIPTS, self.path("skills", "auth", "scripts"), ignore=skip)
|
||||
os.makedirs(self.path("sub", "deeper"))
|
||||
|
||||
# the login pin the transport insists on, local to this fixture
|
||||
os.makedirs(self.path(".claude"))
|
||||
with open(self.path(".claude", "settings.local.json"), "w") as f:
|
||||
json.dump({"env": {"GITEA_LOGIN": "fixture/user"}}, f)
|
||||
|
||||
self.bin = self.path("fakebin")
|
||||
os.makedirs(self.bin)
|
||||
tea = os.path.join(self.bin, "tea")
|
||||
with open(tea, "w") as f:
|
||||
f.write(FAKE_TEA % sys.executable)
|
||||
os.chmod(tea, os.stat(tea).st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
|
||||
|
||||
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")
|
||||
|
||||
@property
|
||||
def payloads(self):
|
||||
return self.path("tmp", "payload")
|
||||
|
||||
def script(self, layer, name):
|
||||
return self.path("skills", layer, "scripts", name)
|
||||
|
||||
def run(self, script, *args, **kw):
|
||||
env = dict(os.environ)
|
||||
env.pop("PYTHONPATH", None) # no leakage from the harness into the child
|
||||
env["PATH"] = self.bin + os.pathsep + env["PATH"]
|
||||
env["TEA_CALL_LOG"] = self.root
|
||||
p = subprocess.run([sys.executable, script] + list(args),
|
||||
cwd=kw.pop("cwd", self.root), env=env,
|
||||
capture_output=True, text=True)
|
||||
return p.returncode, p.stdout, p.stderr
|
||||
|
||||
def calls(self):
|
||||
p = os.path.join(self.root, "calls.txt")
|
||||
if not os.path.isfile(p):
|
||||
return []
|
||||
with open(p) as f:
|
||||
return [line.rstrip("\n").split("\t") for line in f if line.strip()]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# resolution
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TestPayloadRoot(unittest.TestCase):
|
||||
|
||||
def test_root_is_absolute_and_repo_anchored(self):
|
||||
self.assertTrue(os.path.isabs(_gitea.PAYLOAD_ROOT), _gitea.PAYLOAD_ROOT)
|
||||
self.assertEqual(_gitea.PAYLOAD_ROOT, os.path.join(REPO, "tmp", "payload"))
|
||||
|
||||
def test_it_is_not_the_issue_store_and_not_inside_one(self):
|
||||
"""The acceptance criterion, as a path fact: a request body is not
|
||||
store content, so it may not live in a store or under one."""
|
||||
self.assertNotEqual(_gitea.PAYLOAD_ROOT, issue.ISSUE_ROOT)
|
||||
self.assertFalse(_gitea.PAYLOAD_ROOT.startswith(issue.ISSUE_ROOT + os.sep))
|
||||
self.assertFalse(issue.ISSUE_ROOT.startswith(_gitea.PAYLOAD_ROOT + os.sep))
|
||||
|
||||
def test_the_name_says_what_it_holds(self):
|
||||
"""Named so the distinction is visible: a top-level directory called
|
||||
`payload`, not a dotdir hiding among an issue's files."""
|
||||
self.assertEqual(os.path.basename(_gitea.PAYLOAD_ROOT), "payload")
|
||||
self.assertFalse(os.path.basename(_gitea.PAYLOAD_ROOT).startswith("."))
|
||||
|
||||
def test_gitignore_covers_it(self):
|
||||
with open(os.path.join(REPO, ".gitignore")) as f:
|
||||
ignored = {line.strip() for line in f}
|
||||
self.assertEqual(_gitea.PAYLOAD_PARTS[0], "tmp")
|
||||
self.assertIn("tmp/", ignored,
|
||||
"the payload directory is not covered by .gitignore")
|
||||
|
||||
def test_resolution_is_anchored_on_the_module_not_on_cwd(self):
|
||||
repo = FakeRepo()
|
||||
self.addCleanup(repo.cleanup)
|
||||
self.assertEqual(_gitea.payload_root(repo.path("sub", "deeper")),
|
||||
repo.payloads)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# the bug: a label bootstrap that materialized the store
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TestLabelsTouchesNoStore(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.repo = FakeRepo()
|
||||
self.addCleanup(self.repo.cleanup)
|
||||
|
||||
def bootstrap(self, *args, **kw):
|
||||
rc, out, err = self.repo.run(self.repo.script("sync", "labels.py"),
|
||||
"--repo", "fixture/repo", *args, **kw)
|
||||
self.assertEqual(rc, 0, "labels.py failed:\n%s%s" % (out, err))
|
||||
return out, err
|
||||
|
||||
def test_bootstrap_creates_no_store(self):
|
||||
"""The reproduction from the report, run for real: no tmp/issues, and
|
||||
no complaint about one either."""
|
||||
out, _ = self.bootstrap()
|
||||
self.assertIn("created", out)
|
||||
self.assertFalse(os.path.exists(self.repo.store),
|
||||
"labels.py created the issue store")
|
||||
|
||||
def test_bootstrap_writes_its_payloads_to_the_payload_root(self):
|
||||
self.bootstrap()
|
||||
self.assertTrue(os.path.isdir(self.repo.payloads),
|
||||
"no payload directory: %s" % self.repo.payloads)
|
||||
written = os.listdir(self.repo.payloads)
|
||||
self.assertIn("label-type-bug.json", written)
|
||||
for name in written:
|
||||
self.assertTrue(name.startswith("label-"), name)
|
||||
|
||||
# and the file named on the command line is the one that was written
|
||||
sent = [a[a.index("-d") + 1][1:] for a in self.repo.calls() if "-d" in a]
|
||||
self.assertTrue(sent)
|
||||
for path in sent:
|
||||
self.assertEqual(os.path.dirname(path), self.repo.payloads)
|
||||
self.assertTrue(os.path.isfile(path), path)
|
||||
|
||||
def test_the_payload_is_the_request_body(self):
|
||||
self.bootstrap()
|
||||
with open(os.path.join(self.repo.payloads, "label-type-bug.json")) as f:
|
||||
body = json.load(f)
|
||||
self.assertEqual(body.get("name"), "type/bug")
|
||||
self.assertTrue(body.get("color"))
|
||||
|
||||
def test_a_dry_run_writes_nothing_at_all(self):
|
||||
out, _ = self.bootstrap("--dry-run")
|
||||
self.assertIn("nothing was written", out)
|
||||
self.assertFalse(os.path.exists(self.repo.path("tmp")),
|
||||
"a dry run left something behind in tmp/")
|
||||
|
||||
def test_the_directory_does_not_follow_cwd(self):
|
||||
"""Run from a subdirectory: still one payload root, at the repo root.
|
||||
A cwd-relative directory is how the store ended up with a second copy
|
||||
of itself, and this one is resolved the same way to avoid the same
|
||||
class of bug."""
|
||||
self.bootstrap(cwd=self.repo.path("sub", "deeper"))
|
||||
self.assertTrue(os.path.isdir(self.repo.payloads))
|
||||
self.assertFalse(os.path.exists(self.repo.path("sub", "deeper", "tmp")))
|
||||
self.assertFalse(os.path.exists(self.repo.store))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# one place, every caller
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TestOnePlaceForEveryCaller(unittest.TestCase):
|
||||
|
||||
def hits(self, needle, skip_transport=False):
|
||||
"""Every `layer/script.py:line` mentioning `needle`."""
|
||||
out = []
|
||||
for d in (SYNC_SCRIPTS, WIKI_SCRIPTS):
|
||||
layer = os.path.basename(os.path.dirname(d))
|
||||
for name in sorted(os.listdir(d)):
|
||||
if not name.endswith(".py") or (skip_transport and name == "_gitea.py"):
|
||||
continue
|
||||
with open(os.path.join(d, name)) as f:
|
||||
for n, line in enumerate(f, 1):
|
||||
if needle in line:
|
||||
out.append("%s/%s:%d" % (layer, name, n))
|
||||
return out
|
||||
|
||||
def test_no_caller_chooses_where_its_payload_goes(self):
|
||||
"""Whatever the answer is, it has to be the same for all of them —
|
||||
payload files scattered across two stores and a wiki space is the
|
||||
state this replaced."""
|
||||
self.assertEqual(self.hits("out_root"), [],
|
||||
"a caller still picks a payload directory of its own")
|
||||
|
||||
def test_only_the_transport_names_the_directory(self):
|
||||
self.assertEqual(self.hits("PAYLOAD", skip_transport=True), [],
|
||||
"the payload directory is named outside the transport")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,354 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
A pull returns the unit of work: the issue AND what blocks it.
|
||||
|
||||
`--deps` used to be opt-in, so `pull.py 42` wrote a file with an empty
|
||||
`depends:` and `issue_tree.py` drew it as a root with no blockers. The edge was
|
||||
not lost — it lives in Gitea's native dependency graph — but it was not asked
|
||||
for, and the body cannot supply it: `map.from_api` writes slugs into the
|
||||
`## Depends on` prose and never `#N`. Following the graph is now the default.
|
||||
|
||||
What is asserted here:
|
||||
|
||||
1. **The default fills the graph.** A bare `pull.py <n>` fills `depends:` and
|
||||
pulls the blocker too, down to `--depth`.
|
||||
2. **`--no-deps` is the way out, and it is free.** No `depends:`, no recursion,
|
||||
and not one request beyond the issue itself.
|
||||
3. **`--deps` still works and means nothing.** Calls written against the old
|
||||
default keep running and get what they always got.
|
||||
4. **The cost is one request per stored issue.** The native links are fetched
|
||||
once and used twice — for `depends:` and for the walk. Never twice.
|
||||
5. **Filter mode follows blockers out of the selection, deliberately.** A
|
||||
blocker no filter selected still lands in the store and does not spend
|
||||
`--limit`; a closed one is dropped like any other closed issue, and so is
|
||||
the edge to it. An issue the filter dropped costs no link request at all.
|
||||
|
||||
The transport is stubbed at `_gitea.api`, as the other suites do it, and the
|
||||
stub records every call so "how many requests" is an observation. No network,
|
||||
and no test touches the developer's store: each builds its own in a
|
||||
`tempfile.TemporaryDirectory()`.
|
||||
"""
|
||||
import contextlib
|
||||
import io
|
||||
import os
|
||||
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 pull # noqa: E402
|
||||
|
||||
REPO = "claude-skills/tea"
|
||||
BASE = "repos/%s" % REPO
|
||||
|
||||
BODY = """## Summary
|
||||
Прозаическое описание задачи.
|
||||
|
||||
## Spec
|
||||
skills/issue/references/format.md
|
||||
|
||||
## Acceptance criteria
|
||||
- [ ] что-нибудь работает
|
||||
"""
|
||||
|
||||
|
||||
def payload(number, title, state="open"):
|
||||
return {"number": number, "title": title, "body": BODY, "state": state,
|
||||
"comments": 0, "labels": [{"name": "type/task"}], "assignees": [],
|
||||
"milestone": None, "ref": "main", "updated_at": "2026-08-10T00:00:00Z",
|
||||
"html_url": "https://git.example/%s/issues/%d" % (REPO, number),
|
||||
"repository": {"full_name": REPO}}
|
||||
|
||||
|
||||
class FakeTracker(object):
|
||||
"""`tea api` answered from memory, with a native dependency graph.
|
||||
|
||||
`listed` is what the list endpoint serves — the filter's selection. `extra`
|
||||
exists and is fetchable by number but is in no selection, which is how a
|
||||
blocker outside the filter is modelled. `deps` maps a blocked issue's number
|
||||
to the numbers that block it, the direction `GET …/dependencies` reads.
|
||||
"""
|
||||
|
||||
def __init__(self, listed=(), extra=(), deps=None):
|
||||
self.listed = list(listed)
|
||||
self.issues = {p["number"]: p for p in list(listed) + list(extra)}
|
||||
self.deps = {int(k): list(v) for k, v in (deps or {}).items()}
|
||||
self.calls = [] # (method, path), in request order
|
||||
|
||||
# -- what the tests read off it ----------------------------------------
|
||||
|
||||
def paths(self, suffix):
|
||||
return [p for m, p in self.calls if p.endswith(suffix)]
|
||||
|
||||
def issue_gets(self):
|
||||
"""`GET …/issues/<n>` — one issue fetched by number."""
|
||||
return [p for m, p in self.calls
|
||||
if m == "GET" and p.startswith("%s/issues/" % BASE)
|
||||
and p.rsplit("/", 1)[1].isdigit()]
|
||||
|
||||
# -- the seam ----------------------------------------------------------
|
||||
|
||||
def api(self, login, endpoint, method="GET", payload=None,
|
||||
payload_name=None, out_root=None, allow_fail=False):
|
||||
path, _, qs = endpoint.partition("?")
|
||||
q = urllib.parse.parse_qs(qs)
|
||||
self.calls.append((method, path))
|
||||
|
||||
if path == "%s/issues" % BASE and method == "GET":
|
||||
page, per = int(q["page"][0]), int(q["limit"][0])
|
||||
return self.listed[(page - 1) * per:(page - 1) * per + per]
|
||||
|
||||
if path.endswith("/comments"):
|
||||
return []
|
||||
|
||||
if path.endswith("/dependencies") and method == "GET":
|
||||
n = int(path.split("/issues/")[1].split("/")[0])
|
||||
return [self.issues[b] for b in self.deps.get(n, []) if b in self.issues]
|
||||
|
||||
if path.startswith("%s/issues/" % BASE) and method == "GET":
|
||||
return self.issues.get(int(path.rsplit("/", 1)[1]))
|
||||
|
||||
raise AssertionError("unstubbed call: %s %s" % (method, endpoint))
|
||||
|
||||
|
||||
class PullDepsTestCase(unittest.TestCase):
|
||||
"""A temp store, a fake tracker, no git and no network."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory(prefix="tea-deps-")
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.root = os.path.join(self.tmp.name, "tmp", "issues")
|
||||
os.makedirs(self.root)
|
||||
p = mock.patch.object(_gitea, "require_login", lambda: "test-login")
|
||||
p.start()
|
||||
self.addCleanup(p.stop)
|
||||
|
||||
def serve(self, listed=(), extra=(), deps=None):
|
||||
self.fake = FakeTracker(listed, extra, deps)
|
||||
p = mock.patch.object(_gitea, "api", self.fake.api)
|
||||
p.start()
|
||||
self.addCleanup(p.stop)
|
||||
return self.fake
|
||||
|
||||
def blocked_pair(self):
|
||||
"""#10 "Second thing" is blocked by #7 "First thing"."""
|
||||
return self.serve(listed=[payload(10, "Second thing"),
|
||||
payload(7, "First thing")],
|
||||
deps={10: [7]})
|
||||
|
||||
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(self):
|
||||
return sorted(issue.all_ids(self.root))
|
||||
|
||||
def depends_of(self, id):
|
||||
return issue.load(self.root, id).depends
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 1. the default fills the graph
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class DepsAreTheDefaultTest(PullDepsTestCase):
|
||||
|
||||
def test_a_bare_pull_fills_depends(self):
|
||||
"""The acceptance criterion, and the whole point: no flag, and the file
|
||||
knows what blocks it."""
|
||||
self.blocked_pair()
|
||||
self.run_pull("10")
|
||||
self.assertEqual(self.depends_of("second-thing"), ["first-thing"])
|
||||
|
||||
def test_a_bare_pull_stores_the_blocker(self):
|
||||
"""`depends:` pointing at a file that is not there would be worse than
|
||||
an empty one — the blocker comes with it."""
|
||||
self.blocked_pair()
|
||||
self.run_pull("10")
|
||||
self.assertIn("first-thing", self.stored())
|
||||
|
||||
def test_the_walk_is_recursive(self):
|
||||
"""A blocker's blocker is context too, down to --depth (default 3)."""
|
||||
self.serve(listed=[payload(n, "Thing %d" % n) for n in range(1, 6)],
|
||||
deps={1: [2], 2: [3], 3: [4], 4: [5]})
|
||||
self.run_pull("1")
|
||||
self.assertEqual(self.stored(), ["thing-1", "thing-2", "thing-3", "thing-4"],
|
||||
"the default depth of 3 was not what was walked")
|
||||
|
||||
def test_depth_bounds_the_walk(self):
|
||||
self.serve(listed=[payload(n, "Thing %d" % n) for n in range(1, 6)],
|
||||
deps={1: [2], 2: [3], 3: [4], 4: [5]})
|
||||
self.run_pull("1", "--depth", "1")
|
||||
self.assertEqual(self.stored(), ["thing-1", "thing-2"])
|
||||
|
||||
def test_the_graph_hint_is_printed_when_there_is_a_graph(self):
|
||||
self.blocked_pair()
|
||||
out, _ = self.run_pull("10")
|
||||
self.assertIn("issue_tree.py", out)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 2. --no-deps is the way out, and it is free
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class NoDepsOptsOutTest(PullDepsTestCase):
|
||||
|
||||
def test_no_deps_leaves_depends_empty(self):
|
||||
self.blocked_pair()
|
||||
self.run_pull("10", "--no-deps")
|
||||
self.assertEqual(self.depends_of("second-thing"), [])
|
||||
|
||||
def test_no_deps_does_not_pull_the_blocker(self):
|
||||
self.blocked_pair()
|
||||
self.run_pull("10", "--no-deps")
|
||||
self.assertEqual(self.stored(), ["second-thing"])
|
||||
|
||||
def test_no_deps_spends_no_extra_request(self):
|
||||
"""The other half of the criterion: not the links, not the blocker.
|
||||
One issue asked for, one request made."""
|
||||
self.blocked_pair()
|
||||
self.run_pull("10", "--no-deps")
|
||||
self.assertEqual(self.fake.paths("/dependencies"), [])
|
||||
self.assertEqual(self.fake.issue_gets(), ["%s/issues/10" % BASE])
|
||||
|
||||
def test_no_deps_prints_no_graph_hint(self):
|
||||
self.blocked_pair()
|
||||
out, _ = self.run_pull("10", "--no-deps")
|
||||
self.assertNotIn("issue_tree.py", out)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 3. --deps is still accepted, and means nothing
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class DepsFlagIsANoOpTest(PullDepsTestCase):
|
||||
|
||||
def test_the_flag_is_still_accepted(self):
|
||||
"""Existing calls and the /tea:sync command tables must not break."""
|
||||
self.blocked_pair()
|
||||
self.run_pull("10", "--deps")
|
||||
self.assertEqual(self.depends_of("second-thing"), ["first-thing"])
|
||||
|
||||
def test_it_changes_nothing_about_the_run(self):
|
||||
self.blocked_pair()
|
||||
self.run_pull("10", "--deps")
|
||||
with_flag = (self.stored(), self.depends_of("second-thing"),
|
||||
list(self.fake.calls))
|
||||
|
||||
self.setUp()
|
||||
self.blocked_pair()
|
||||
self.run_pull("10")
|
||||
self.assertEqual((self.stored(), self.depends_of("second-thing"),
|
||||
list(self.fake.calls)), with_flag)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 4. one request per stored issue
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class TheCostIsOneRequestPerIssueTest(PullDepsTestCase):
|
||||
|
||||
def test_the_links_are_fetched_once_per_issue(self):
|
||||
"""They fill `depends:` AND steer the walk; fetching them twice is
|
||||
double the price the docstring quotes."""
|
||||
self.blocked_pair()
|
||||
self.run_pull("10")
|
||||
self.assertEqual(self.fake.paths("/dependencies"),
|
||||
["%s/issues/10/dependencies" % BASE,
|
||||
"%s/issues/7/dependencies" % BASE])
|
||||
|
||||
def test_a_bulk_pull_costs_one_per_issue(self):
|
||||
"""The number the docstring quotes: one list request, then one link
|
||||
request per issue that lands in the store."""
|
||||
self.serve(listed=[payload(n, "Thing %d" % n) for n in range(1, 21)])
|
||||
self.run_pull("-q", "x")
|
||||
self.assertEqual(len(self.fake.paths("/dependencies")), 20)
|
||||
self.assertEqual(len(self.fake.paths("/issues")), 1)
|
||||
|
||||
def test_a_cached_issue_costs_its_links_and_nothing_else(self):
|
||||
"""--cached stops the body and the thread, not the graph: a cached
|
||||
issue's blockers can be missing from disk even when it is not."""
|
||||
self.blocked_pair()
|
||||
self.run_pull("10", "--no-deps") # only #10 on disk
|
||||
self.fake.calls = []
|
||||
self.run_pull("10", "--cached")
|
||||
self.assertEqual(self.fake.paths("/dependencies"),
|
||||
["%s/issues/10/dependencies" % BASE,
|
||||
"%s/issues/7/dependencies" % BASE])
|
||||
self.assertIn("first-thing", self.stored())
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 5. filter mode follows blockers out of the selection
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class FilterModeFollowsOutwardTest(PullDepsTestCase):
|
||||
|
||||
def test_a_blocker_outside_the_filter_lands_in_the_store(self):
|
||||
"""Documented as deliberate: a blocker is followed because a stored
|
||||
issue named it, not because the filter selected it."""
|
||||
self.serve(listed=[payload(1, "Selected thing")],
|
||||
extra=[payload(99, "Outside thing")],
|
||||
deps={1: [99]})
|
||||
self.run_pull("-q", "x")
|
||||
self.assertEqual(self.stored(), ["outside-thing", "selected-thing"])
|
||||
self.assertEqual(self.depends_of("selected-thing"), ["outside-thing"])
|
||||
|
||||
def test_a_blocker_does_not_spend_the_limit(self):
|
||||
"""--limit counts the selection's writes; the graph is not part of the
|
||||
selection, so the store can legitimately hold more than N."""
|
||||
self.serve(listed=[payload(n, "Thing %d" % n) for n in range(1, 5)],
|
||||
extra=[payload(100 + n, "Blocker %d" % n) for n in range(1, 5)],
|
||||
deps={n: [100 + n] for n in range(1, 5)})
|
||||
self.run_pull("-q", "x", "--limit", "2")
|
||||
self.assertEqual(self.stored(),
|
||||
["blocker-1", "blocker-2", "thing-1", "thing-2"])
|
||||
|
||||
def test_a_closed_blocker_is_dropped_with_the_edge_to_it(self):
|
||||
"""The documented exception. Closed is not a unit of work, so filter
|
||||
mode drops it like any other closed issue — and `depends:` must not be
|
||||
left pointing at a file that is not there."""
|
||||
self.serve(listed=[payload(1, "Selected thing")],
|
||||
extra=[payload(99, "Closed blocker", state="closed")],
|
||||
deps={1: [99]})
|
||||
self.run_pull("-q", "x")
|
||||
self.assertEqual(self.stored(), ["selected-thing"])
|
||||
self.assertEqual(self.depends_of("selected-thing"), [])
|
||||
|
||||
def test_a_closed_blocker_is_stored_in_key_mode(self):
|
||||
"""An address is not a bulk read: `pull.py 1` has no closed rule."""
|
||||
self.serve(listed=[payload(1, "Selected thing")],
|
||||
extra=[payload(99, "Closed blocker", state="closed")],
|
||||
deps={1: [99]})
|
||||
self.run_pull("1")
|
||||
self.assertEqual(self.stored(), ["closed-blocker", "selected-thing"])
|
||||
|
||||
def test_a_dropped_closed_issue_costs_no_link_request(self):
|
||||
"""Nothing was stored for it, so there is no unit of work to complete
|
||||
— and its own blockers are not dragged in behind it."""
|
||||
self.serve(listed=[payload(1, "Closed thing", state="closed"),
|
||||
payload(2, "Open thing")],
|
||||
extra=[payload(50, "Blocker of the closed one")],
|
||||
deps={1: [50]})
|
||||
self.run_pull("-q", "x", "--state", "all")
|
||||
self.assertEqual(self.fake.paths("/dependencies"),
|
||||
["%s/issues/2/dependencies" % BASE])
|
||||
self.assertEqual(self.stored(), ["open-thing"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,349 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
`pull.py --limit N` bounds the WRITE, not the selection.
|
||||
|
||||
The bug this file exists to keep dead: the limit used to cut the list of
|
||||
payloads before pull.py dropped the closed ones, so a milestone whose first
|
||||
issues are closed spent the budget on issues that never reached disk —
|
||||
`--limit 20` wrote twelve, and the docstring promised twenty.
|
||||
|
||||
What is asserted, in the order the fix has to hold it:
|
||||
|
||||
1. **The count is of files.** N issues under the filter that would be stored →
|
||||
exactly N files, however many closed ones were enumerated on the way.
|
||||
2. **Pagination serves the budget.** More pages are requested while the budget
|
||||
is unfilled, and the page after the one that fills it is never requested.
|
||||
3. **The scan is bounded.** A filter that matches almost only closed issues
|
||||
stops after `_gitea.PAGE_SLACK` times the ideal page count, says so, and
|
||||
returns short — it does not walk the tracker.
|
||||
4. **`remote.py` is unchanged.** Its `--limit` still caps the listing, closed
|
||||
issues included, because it writes nothing there is a limit for.
|
||||
|
||||
The transport is stubbed at `_gitea.api`, the way the other suites do it, and
|
||||
the stub serves `page=` / `limit=` itself so the request pattern is a real
|
||||
observation and not an assumption. No network, and no test writes to the
|
||||
developer's store: each one builds its own in a `tempfile.TemporaryDirectory()`.
|
||||
"""
|
||||
import contextlib
|
||||
import io
|
||||
import os
|
||||
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
|
||||
import remote # noqa: E402
|
||||
|
||||
REPO = "claude-skills/tea"
|
||||
BASE = "repos/%s" % REPO
|
||||
|
||||
BODY = """## Summary
|
||||
Прозаическое описание задачи.
|
||||
|
||||
## Spec
|
||||
skills/issue/references/format.md
|
||||
|
||||
## Acceptance criteria
|
||||
- [ ] что-нибудь работает
|
||||
"""
|
||||
|
||||
|
||||
def payload(number, state="open", title=None, comments=0):
|
||||
return {"number": number, "title": title or "Issue number %d" % number,
|
||||
"body": BODY, "state": state, "comments": comments,
|
||||
"labels": [{"name": "type/task"}], "assignees": [], "milestone": None,
|
||||
"ref": "main", "updated_at": "2026-08-10T00:00:00Z",
|
||||
"html_url": "https://git.example/%s/issues/%d" % (REPO, number),
|
||||
"repository": {"full_name": REPO}}
|
||||
|
||||
|
||||
def alternating(count, first="closed"):
|
||||
"""`count` issues, every other one closed. The shape of the bug report:
|
||||
closed issues sitting in front of the open ones, in page order."""
|
||||
other = "open" if first == "closed" else "closed"
|
||||
return [payload(n, first if n % 2 else other) for n in range(1, count + 1)]
|
||||
|
||||
|
||||
class FakeTracker(object):
|
||||
"""`tea api` answered from a list, with real pagination.
|
||||
|
||||
It slices on the `page=` and `limit=` it was given rather than ignoring
|
||||
them, so "which pages were requested" is something the test can read off
|
||||
`self.list_pages` instead of inferring."""
|
||||
|
||||
def __init__(self, payloads):
|
||||
self.payloads = list(payloads)
|
||||
self.list_pages = [] # (page, per_page), in request order
|
||||
|
||||
def api(self, login, endpoint, method="GET", payload=None,
|
||||
payload_name=None, out_root=None, allow_fail=False):
|
||||
path, _, qs = endpoint.partition("?")
|
||||
q = urllib.parse.parse_qs(qs)
|
||||
|
||||
if path == "%s/issues" % BASE and method == "GET":
|
||||
page, per = int(q["page"][0]), int(q["limit"][0])
|
||||
self.list_pages.append((page, per))
|
||||
return self.payloads[(page - 1) * per:(page - 1) * per + per]
|
||||
|
||||
if path.endswith("/comments"):
|
||||
return []
|
||||
|
||||
if path.endswith("/dependencies"):
|
||||
return []
|
||||
|
||||
if "/issues/" in path and method == "GET":
|
||||
n = int(path.rsplit("/", 1)[1])
|
||||
for p in self.payloads:
|
||||
if p["number"] == n:
|
||||
return p
|
||||
return None
|
||||
|
||||
raise AssertionError("unstubbed call: %s %s" % (method, endpoint))
|
||||
|
||||
|
||||
class PullLimitTestCase(unittest.TestCase):
|
||||
"""A temp store, a fake tracker, no git and no network."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory(prefix="tea-limit-")
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.root = os.path.join(self.tmp.name, "tmp", "issues")
|
||||
os.makedirs(self.root)
|
||||
p = mock.patch.object(_gitea, "require_login", lambda: "test-login")
|
||||
p.start()
|
||||
self.addCleanup(p.stop)
|
||||
|
||||
# -- runners -----------------------------------------------------------
|
||||
|
||||
def serve(self, payloads):
|
||||
self.fake = FakeTracker(payloads)
|
||||
p = mock.patch.object(_gitea, "api", self.fake.api)
|
||||
p.start()
|
||||
self.addCleanup(p.stop)
|
||||
return self.fake
|
||||
|
||||
def run_pull(self, *argv):
|
||||
return self._run(pull, "pull.py", argv)
|
||||
|
||||
def run_remote(self, *argv):
|
||||
return self._run(remote, "remote.py", argv)
|
||||
|
||||
def _run(self, mod, name, argv):
|
||||
out, err = io.StringIO(), io.StringIO()
|
||||
args = [name, "--repo", REPO, "--out", self.root] + list(argv)
|
||||
with mock.patch.object(sys, "argv", args), \
|
||||
contextlib.redirect_stdout(out), \
|
||||
contextlib.redirect_stderr(err):
|
||||
mod.main()
|
||||
return out.getvalue(), err.getvalue()
|
||||
|
||||
# -- assertions --------------------------------------------------------
|
||||
|
||||
def stored(self):
|
||||
return sorted(issue.all_ids(self.root))
|
||||
|
||||
def assertStoredCount(self, n, why=""):
|
||||
got = self.stored()
|
||||
self.assertEqual(len(got), n, "%d issue(s) in the store, wanted %d%s: %s"
|
||||
% (len(got), n, why and " — " + why, got))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 1. the count is of files
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class LimitCountsWritesTest(PullLimitTestCase):
|
||||
|
||||
def test_closed_issues_do_not_spend_the_budget(self):
|
||||
"""The regression. Half the selection is closed and stands in front of
|
||||
the open ones; the limit still buys ten files."""
|
||||
self.serve(alternating(40))
|
||||
self.run_pull("-q", "x", "--state", "all", "--limit", "10")
|
||||
self.assertStoredCount(10)
|
||||
|
||||
def test_only_open_issues_landed(self):
|
||||
self.serve(alternating(40))
|
||||
self.run_pull("-q", "x", "--state", "all", "--limit", "10")
|
||||
for id in self.stored():
|
||||
self.assertEqual(issue.load(self.root, id).state, "open")
|
||||
|
||||
def test_the_dropped_ones_are_still_reported(self):
|
||||
"""Enumerated-and-dropped is not silence: the closed ones seen on the
|
||||
pages that were fetched are counted on stderr."""
|
||||
self.serve(alternating(40))
|
||||
_, err = self.run_pull("-q", "x", "--state", "all", "--limit", "10")
|
||||
self.assertIn("closed issue(s) enumerated, not stored", err)
|
||||
|
||||
def test_a_closed_issue_already_in_the_store_spends_it(self):
|
||||
"""It is refreshed rather than dropped — that is a write, so it counts.
|
||||
The limit is on what the store holds when the run ends, and this issue
|
||||
is in it."""
|
||||
kept = issue.Issue(id="already-here", title="Already here", body=BODY,
|
||||
labels=["type/task"], origin="gitea",
|
||||
extra={"gitea": gmap.remote_key(REPO, 1)})
|
||||
issue.save(self.root, kept)
|
||||
_gitea.save_map(self.root, {gmap.remote_key(REPO, 1): "already-here"})
|
||||
|
||||
self.serve(alternating(40)) # #1 is closed, and is on disk
|
||||
self.run_pull("-q", "x", "--state", "all", "--limit", "10")
|
||||
self.assertStoredCount(10)
|
||||
self.assertEqual(issue.load(self.root, "already-here").state, "closed",
|
||||
"a stored issue must learn it was closed")
|
||||
|
||||
def test_state_closed_writes_closed_ones(self):
|
||||
"""Nothing above may leak into the mode where closed IS the selection."""
|
||||
self.serve([payload(n, "closed") for n in range(1, 21)])
|
||||
self.run_pull("-q", "x", "--state", "closed", "--limit", "6")
|
||||
self.assertStoredCount(6)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 2. pagination serves the budget
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class PaginationFollowsTheBudgetTest(PullLimitTestCase):
|
||||
|
||||
def test_more_pages_are_fetched_until_the_budget_is_full(self):
|
||||
"""One page of ten holds five open issues, so ten files cost two."""
|
||||
self.serve(alternating(40))
|
||||
self.run_pull("-q", "x", "--state", "all", "--limit", "10")
|
||||
self.assertStoredCount(10)
|
||||
self.assertEqual([p for p, _ in self.fake.list_pages], [1, 2])
|
||||
|
||||
def test_the_page_after_the_last_needed_one_is_never_requested(self):
|
||||
"""The budget fills inside page 2; page 3 exists and must not be asked
|
||||
for. Bounding the write must not become fetching the whole repo."""
|
||||
self.serve(alternating(200))
|
||||
self.run_pull("-q", "x", "--state", "all", "--limit", "10")
|
||||
self.assertEqual(len(self.fake.list_pages), 2,
|
||||
"extra pages requested: %r" % (self.fake.list_pages,))
|
||||
|
||||
def test_an_unfiltered_selection_still_costs_one_page(self):
|
||||
"""Nothing is dropped, so nothing changes: the old arithmetic holds."""
|
||||
self.serve([payload(n) for n in range(1, 60)])
|
||||
self.run_pull("-q", "x", "--limit", "10")
|
||||
self.assertStoredCount(10)
|
||||
self.assertEqual(len(self.fake.list_pages), 1)
|
||||
|
||||
def test_running_out_of_pages_gives_a_short_answer(self):
|
||||
"""Six issues, three of them open, `--limit 10`: three files, no crash,
|
||||
and no page beyond the last."""
|
||||
self.serve(alternating(6))
|
||||
self.run_pull("-q", "x", "--state", "all", "--limit", "10")
|
||||
self.assertStoredCount(3)
|
||||
self.assertEqual(len(self.fake.list_pages), 1)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 3. the scan is bounded
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class ScanIsBoundedTest(PullLimitTestCase):
|
||||
|
||||
def test_a_selection_of_only_closed_issues_stops_at_the_page_budget(self):
|
||||
self.serve([payload(n, "closed") for n in range(1, 501)])
|
||||
_, err = self.run_pull("-q", "x", "--state", "all", "--limit", "10")
|
||||
self.assertStoredCount(0)
|
||||
self.assertEqual(len(self.fake.list_pages), _gitea.PAGE_SLACK,
|
||||
"the scan walked past its budget: %r" % (self.fake.list_pages,))
|
||||
self.assertIn("short of --limit", err)
|
||||
|
||||
def test_a_full_budget_does_not_warn(self):
|
||||
"""The warning means "there may be more"; it must not fire on a run
|
||||
that got everything it asked for."""
|
||||
self.serve(alternating(40))
|
||||
_, err = self.run_pull("-q", "x", "--state", "all", "--limit", "10")
|
||||
self.assertNotIn("short of --limit", err)
|
||||
|
||||
def test_a_selection_that_ran_out_does_not_warn(self):
|
||||
"""Six issues in the repo and the server said so — that is an answer,
|
||||
not a truncation."""
|
||||
self.serve(alternating(6))
|
||||
_, err = self.run_pull("-q", "x", "--state", "all", "--limit", "10")
|
||||
self.assertNotIn("short of --limit", err)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 4. remote.py is the deliberate exception
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class RemoteListingIsUnchangedTest(PullLimitTestCase):
|
||||
|
||||
def test_the_listing_limit_still_counts_lines_not_writes(self):
|
||||
"""remote.py writes nothing, so there is no write to bound: ten lines
|
||||
out, closed ones among them, one request."""
|
||||
self.serve(alternating(40))
|
||||
out, _ = self.run_remote("-q", "x", "--state", "all", "--limit", "10")
|
||||
numbered = [l for l in out.splitlines() if l.startswith("#")]
|
||||
self.assertEqual(len(numbered), 10)
|
||||
self.assertTrue(any("closed" in l for l in numbered),
|
||||
"a listing that hides closed issues is not a listing")
|
||||
self.assertEqual(len(self.fake.list_pages), 1)
|
||||
|
||||
def test_it_leaves_the_store_alone(self):
|
||||
self.serve(alternating(40))
|
||||
self.run_remote("-q", "x", "--state", "all", "--limit", "10")
|
||||
self.assertStoredCount(0, "discovery wrote to the store")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# the transport on its own
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class ListIssuesKeepTest(PullLimitTestCase):
|
||||
"""`_gitea.list_issues` without a caller in front of it — the counting rule
|
||||
is the transport's, and it is testable without a store."""
|
||||
|
||||
def list(self, payloads, **kw):
|
||||
self.serve(payloads)
|
||||
return _gitea.list_issues("test-login", BASE, state="all", **kw)
|
||||
|
||||
def test_without_keep_the_limit_caps_the_selection(self):
|
||||
got, _ = self.list(alternating(40), limit=10)
|
||||
self.assertEqual(len(got), 10)
|
||||
|
||||
def test_with_keep_the_limit_caps_the_kept(self):
|
||||
got, _ = self.list(alternating(40), limit=10,
|
||||
keep=lambda p: p["state"] == "open")
|
||||
self.assertEqual(len([p for p in got if p["state"] == "open"]), 10)
|
||||
|
||||
def test_the_rejected_ones_come_back_too(self):
|
||||
"""They were enumerated. The caller reports them; the transport does
|
||||
not get to throw away what it did not count."""
|
||||
got, _ = self.list(alternating(40), limit=10,
|
||||
keep=lambda p: p["state"] == "open")
|
||||
self.assertTrue([p for p in got if p["state"] == "closed"])
|
||||
|
||||
def test_a_limit_below_one_is_refused(self):
|
||||
"""The page arithmetic divides by the page size, and a limit of zero
|
||||
used to make that a traceback. It is a usage error, so it reads like
|
||||
one."""
|
||||
with self.assertRaises(SystemExit):
|
||||
self.list(alternating(4), limit=0)
|
||||
|
||||
def test_pull_requests_never_count(self):
|
||||
"""`matches` drops them, so they cannot spend the budget either."""
|
||||
mixed = []
|
||||
for n in range(1, 41):
|
||||
p = payload(n)
|
||||
if n % 2:
|
||||
p["pull_request"] = {"merged": False}
|
||||
mixed.append(p)
|
||||
got, _ = self.list(mixed, limit=10, keep=lambda p: True)
|
||||
self.assertEqual(len(got), 10)
|
||||
self.assertFalse([p for p in got if p.get("pull_request")])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -97,7 +97,7 @@ class FakeGitea(object):
|
||||
# -- the seam ----------------------------------------------------------
|
||||
|
||||
def api(self, login, endpoint, method="GET", payload=None,
|
||||
payload_name=None, out_root=None, allow_fail=False):
|
||||
payload_name=None, allow_fail=False):
|
||||
self.calls.append((method, endpoint, payload))
|
||||
path = endpoint.split("?")[0]
|
||||
|
||||
@@ -218,7 +218,7 @@ class AddDependencyTest(unittest.TestCase):
|
||||
return {"number": 102}
|
||||
|
||||
with mock.patch.object(_gitea, "api", fake_api):
|
||||
ok = _gitea.add_dependency("l", BASE, 102, REPO, 101, out_root=None)
|
||||
ok = _gitea.add_dependency("l", BASE, 102, REPO, 101)
|
||||
|
||||
self.assertTrue(ok)
|
||||
method, endpoint, payload = calls[0]
|
||||
|
||||
@@ -24,6 +24,7 @@ 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")
|
||||
AUTH_SCRIPTS = os.path.join(REPO, "skills", "auth", "scripts")
|
||||
|
||||
sys.path.insert(0, ISSUE_SCRIPTS)
|
||||
import issue # noqa: E402
|
||||
@@ -110,6 +111,8 @@ class FakeRepo(object):
|
||||
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)
|
||||
# the transport resolves the login pin through skills/auth/scripts
|
||||
shutil.copytree(AUTH_SCRIPTS, self.path("skills", "auth", "scripts"), ignore=skip)
|
||||
os.makedirs(self.path("sub", "deeper"))
|
||||
|
||||
if with_store:
|
||||
|
||||
Reference in New Issue
Block a user