#!/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()