Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

fix: paginate repository-label inventory for gitea_set_issue_labels (#627)

Browse Source 
Replace single-page GET labels?limit=100 with api_get_all-backed
_repo_label_id_map so later-page labels (e.g. type:feature,
workflow-hardening) are not falsely rejected during full-set replacement.

Also add post-mutation verification, fix related MCP/CLI label inventory
call sites, and add multi-page regression tests for the #601 reconciler
failure mode.

Closes #627
This commit is contained in:
sysadmin
2026-07-10 13:05:27 -04:00
parent 5347b313ea
commit f32aaaa3b5
6 changed files with 447 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+68 -41
View File
@@ -1216,12 +1216,32 @@ def extract_linked_issue_numbers(text: str | None, branch_name: str | None = Non
return sorted(list(issues))
def _repo_label_id_map(base: str, auth: str) -> dict[str, int]:
labels = api_get_all(f"{base}/labels", auth) or []
return {
str(lb["name"]): int(lb["id"])
for lb in labels
if isinstance(lb, dict) and lb.get("name") and lb.get("id") is not None
}
"""Map repository label names to IDs across **all** label pages (#627).
Uses :func:`api_get_all` so inventories larger than Gitea's per-page cap
(50) are complete. Duplicate names keep the **first-seen** id for
deterministic resolution (fail-open for attach; names still resolve).
"""
labels = api_get_all(f"{base}/labels", auth)
if labels is None:
labels = []
if not isinstance(labels, list):
raise RuntimeError(
"failed to list repository labels: expected a list page sequence, "
f"got {type(labels).__name__}"
)
name_to_id: dict[str, int] = {}
for lb in labels:
if not isinstance(lb, dict):
continue
name = lb.get("name")
lid = lb.get("id")
if not name or lid is None:
continue
key = str(name)
if key not in name_to_id:
name_to_id[key] = int(lid)
return name_to_id
def _issue_label_names(base: str, auth: str, issue_number: int) -> list[str]:
issue = api_request("GET", f"{base}/issues/{issue_number}", auth) or {}
@@ -1235,20 +1255,46 @@ def _put_issue_label_names(
names: list[str],
label_ids_by_name: dict[str, int] | None = None,
) -> list[dict]:
"""Full-set label replacement with complete inventory + post-mutation check.
Missing requested names fail closed before PUT. After PUT, the returned
label set must match the requested names (order-independent) so callers
never silently drop labels (#627).
"""
by_name = label_ids_by_name or _repo_label_id_map(base, auth)
missing = [name for name in names if name not in by_name]
if missing:
raise RuntimeError(
f"The following labels do not exist on the repository: {missing}. "
"Create the canonical workflow labels first."
"Please create them first using gitea_create_label."
)
ids = [by_name[name] for name in names]
return api_request(
res = api_request(
"PUT",
f"{base}/issues/{issue_number}/labels",
auth,
{"labels": ids},
)
if not isinstance(res, list):
raise RuntimeError(
"Post-mutation label verification failed: expected a list of labels "
f"from Gitea, got {type(res).__name__}."
)
final_names = {
str(lb.get("name"))
for lb in res
if isinstance(lb, dict) and lb.get("name")
}
expected = {str(n) for n in names}
if final_names != expected:
missing_after = sorted(expected - final_names)
extra_after = sorted(final_names - expected)
raise RuntimeError(
"Post-mutation label verification failed: "
f"missing={missing_after} unexpected={extra_after}. "
"Full-set replacement did not match the requested label set."
)
return res
def _transition_issue_status(
*,
@@ -1326,12 +1372,9 @@ def release_in_progress_label(issue_numbers: list[int], remote: str, host: str |
base = repo_api_url(h, o, r)
try:
labels = api_request("GET", f"{base}/labels?limit=100", auth)
label_id = None
for lb in labels:
if lb["name"] == "status:in-progress":
label_id = lb["id"]
break
# Paginated inventory (#627): status labels must resolve even when
# the repo has more labels than one Gitea page.
label_id = _repo_label_id_map(base, auth).get("status:in-progress")
except Exception as exc:
return {num: f"error fetching repo labels: {_redact(str(exc))}" for num in issue_numbers}
@@ -10286,11 +10329,8 @@ def gitea_cleanup_stale_claims(
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
labels = api_request("GET", f"{base}/labels?limit=100", auth)
label_id = next(
(lb["id"] for lb in labels if lb.get("name") == "status:in-progress"),
None,
)
# Paginated inventory (#627) — do not use single-page labels?limit=100.
label_id = _repo_label_id_map(base, auth).get("status:in-progress")
if label_id is None:
raise RuntimeError("Label 'status:in-progress' not found")
@@ -10448,29 +10488,16 @@ def gitea_set_issue_labels(
auth = _auth(h)
base = repo_api_url(h, o, r)
# 1. Fetch existing labels on the repo to resolve names -> IDs
existing = api_request("GET", f"{base}/labels?limit=100", auth)
name_to_id = {lb["name"]: lb["id"] for lb in existing}
# 2. Check if any requested labels do not exist, and raise error
label_ids = []
missing_labels = []
for name in labels:
if name in name_to_id:
label_ids.append(name_to_id[name])
else:
missing_labels.append(name)
if missing_labels:
raise RuntimeError(
f"The following labels do not exist on the repository: {missing_labels}. "
"Please create them first using gitea_create_label."
)
# 3. PUT the labels to the issue
# Full-set replacement via paginated name→id map + post-mutation verify (#627).
# Never use single-page GET labels?limit=100 — Gitea caps pages at 50.
with _audited("set_issue_labels", host=h, remote=remote, org=o, repo=r,
issue_number=issue_number, request_metadata={"labels": labels}):
res = api_request("PUT", f"{base}/issues/{issue_number}/labels", auth, {"labels": label_ids})
issue_number=issue_number, request_metadata={"labels": list(labels)}):
res = _put_issue_label_names(
base=base,
auth=auth,
issue_number=issue_number,
names=list(labels),
)
return res