Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 74 Pull Requests 9 Actions Packages Projects Releases Wiki Activity
Files
bee24e2b102dbdc201932d2e5397a7d9e3be5fc5
Gitea-Tools/tests/test_set_issue_labels_pagination.py
T
sysadmin f32aaaa3b5 fix: paginate repository-label inventory for gitea_set_issue_labels (#627) 
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
2026-07-10 13:05:27 -04:00

319 lines
13 KiB
Python
Raw Blame History

"""#627: paginated repository-label resolution for gitea_set_issue_labels.
Reproduces the post-merge #601 reconciliation failure where later-page
labels (e.g. type:feature, workflow-hardening) were falsely rejected as
nonexistent because inventory used a single-page GET labels?limit=100.
"""
from __future__ import annotations
import os
import sys
import unittest
from unittest.mock import patch, call
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
import gitea_auth
FAKE_AUTH = "token test-token"
PAGE_SIZE = 50 # Gitea effective max; see gitea_auth.api_get_all
def _lb(name: str, lid: int) -> dict:
return {"id": lid, "name": name, "color": "000000"}
def _pages(labels: list[dict], page_size: int = PAGE_SIZE) -> list[list[dict]]:
if not labels:
return [[]]
pages = []
for i in range(0, len(labels), page_size):
pages.append(labels[i : i + page_size])
return pages
class TestRepoLabelIdMapPagination(unittest.TestCase):
"""_repo_label_id_map must use api_get_all (all pages)."""
@patch("mcp_server.api_get_all")
def test_fewer_than_one_page(self, mock_all):
labels = [_lb(f"l{i}", i) for i in range(3)]
mock_all.return_value = labels
m = mcp_server._repo_label_id_map("https://gitea.example/api/v1/repos/o/r", FAKE_AUTH)
self.assertEqual(m, {"l0": 0, "l1": 1, "l2": 2})
mock_all.assert_called_once()
self.assertIn("/labels", mock_all.call_args[0][0])
self.assertNotIn("limit=100", mock_all.call_args[0][0])
@patch("mcp_server.api_get_all")
def test_exactly_one_full_page(self, mock_all):
labels = [_lb(f"l{i:03d}", i) for i in range(PAGE_SIZE)]
mock_all.return_value = labels
m = mcp_server._repo_label_id_map("https://gitea.example/api/v1/repos/o/r", FAKE_AUTH)
self.assertEqual(len(m), PAGE_SIZE)
self.assertEqual(m["l000"], 0)
self.assertEqual(m[f"l{PAGE_SIZE - 1:03d}"], PAGE_SIZE - 1)
@patch("mcp_server.api_get_all")
def test_more_than_one_page_includes_later_labels(self, mock_all):
# Page 1: 50 early labels; page 2: type:feature + workflow-hardening (#601 style)
early = [_lb(f"early-{i:02d}", i) for i in range(PAGE_SIZE)]
late = [
_lb("type:feature", 1001),
_lb("workflow-hardening", 1002),
]
mock_all.return_value = early + late
m = mcp_server._repo_label_id_map("https://gitea.example/api/v1/repos/o/r", FAKE_AUTH)
self.assertEqual(len(m), PAGE_SIZE + 2)
self.assertEqual(m["type:feature"], 1001)
self.assertEqual(m["workflow-hardening"], 1002)
@patch("mcp_server.api_get_all")
def test_duplicate_names_keep_first_seen_id(self, mock_all):
mock_all.return_value = [
_lb("type:feature", 103),
_lb("type:feature", 130), # duplicate id later
_lb("workflow-hardening", 105),
_lb("workflow-hardening", 128),
]
m = mcp_server._repo_label_id_map("https://gitea.example/api/v1/repos/o/r", FAKE_AUTH)
self.assertEqual(m["type:feature"], 103)
self.assertEqual(m["workflow-hardening"], 105)
@patch("mcp_server.api_get_all", return_value={"not": "a list"})
def test_non_list_inventory_fails_closed(self, _all):
with self.assertRaises(RuntimeError) as ctx:
mcp_server._repo_label_id_map("https://gitea.example/api/v1/repos/o/r", FAKE_AUTH)
self.assertIn("expected a list", str(ctx.exception).lower())
@patch("mcp_server.api_get_all", side_effect=RuntimeError("page 2 failed: connection reset"))
def test_later_page_failure_propagates(self, _all):
with self.assertRaises(RuntimeError) as ctx:
mcp_server._repo_label_id_map("https://gitea.example/api/v1/repos/o/r", FAKE_AUTH)
self.assertIn("page 2 failed", str(ctx.exception))
class TestSetIssueLabelsPagination(unittest.TestCase):
"""gitea_set_issue_labels full-set replacement with multi-page inventory."""
def setUp(self):
self._remotes = patch.dict(
mcp_server.REMOTES,
{"prgs": {"host": "gitea.example.com", "org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools"}},
)
self._remotes.start()
# Allow mutation path without full preflight stack when possible
self._preflight = patch(
"mcp_server.verify_preflight_purity", return_value=None
)
self._preflight.start()
self._perm = patch(
"mcp_server._profile_permission_block", return_value=None
)
self._perm.start()
self._auth = patch(
"mcp_server._auth", return_value=FAKE_AUTH
)
self._auth.start()
self._audited = patch("mcp_server._audited")
mock_aud = self._audited.start()
mock_aud.return_value.__enter__ = lambda s: None
mock_aud.return_value.__exit__ = lambda s, *a: None
def tearDown(self):
self._remotes.stop()
self._preflight.stop()
self._perm.stop()
self._auth.stop()
self._audited.stop()
def _inventory_early_and_late(self) -> list[dict]:
early = [_lb(f"alpha-{i:02d}", i + 1) for i in range(PAGE_SIZE)]
late = [
_lb("type:feature", 9001),
_lb("workflow-hardening", 9002),
_lb("status:ready", 9003),
_lb("anti-stomp", 9004),
_lb("leases", 9005),
_lb("recovery", 9006),
]
return early + late
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all")
def test_requested_labels_split_across_pages(self, mock_all, mock_req):
inv = self._inventory_early_and_late()
mock_all.return_value = inv
requested = ["alpha-00", "type:feature", "workflow-hardening"]
mock_req.return_value = [_lb(n, inv_i["id"]) for n in requested
for inv_i in inv if inv_i["name"] == n]
res = mcp_server.gitea_set_issue_labels(
issue_number=601,
labels=requested,
remote="prgs",
)
self.assertEqual({lb["name"] for lb in res}, set(requested))
# PUT must use ids from both pages
put = mock_req.call_args
self.assertEqual(put[0][0], "PUT")
payload_ids = put[0][3]["labels"]
self.assertEqual(payload_ids, [1, 9001, 9002])
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all")
def test_missing_requested_label_rejected_before_put(self, mock_all, mock_req):
mock_all.return_value = [_lb("bug", 1), _lb("status:ready", 2)]
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_set_issue_labels(
issue_number=9,
labels=["bug", "does-not-exist"],
remote="prgs",
)
self.assertIn("do not exist", str(ctx.exception))
self.assertIn("does-not-exist", str(ctx.exception))
mock_req.assert_not_called()
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all")
def test_complete_set_preserves_later_page_labels(self, mock_all, mock_req):
inv = self._inventory_early_and_late()
mock_all.return_value = inv
# Full-set replacement removing only status:pr-open style stale label:
# keep later-page feature labels (the #601 reconciliation shape).
requested = [
"anti-stomp",
"leases",
"recovery",
"type:feature",
"workflow-hardening",
]
mock_req.return_value = [_lb(n, next(x["id"] for x in inv if x["name"] == n))
for n in requested]
res = mcp_server.gitea_set_issue_labels(
issue_number=601, labels=requested, remote="prgs"
)
self.assertEqual([lb["name"] for lb in res], requested)
payload_ids = mock_req.call_args[0][3]["labels"]
self.assertEqual(payload_ids, [9004, 9005, 9006, 9001, 9002])
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all")
def test_issue_601_regression_later_page_names_accepted(self, mock_all, mock_req):
"""Regression: #601 reconciler rejected type:feature + workflow-hardening.
Single-page inventory would omit them; paginated inventory must accept.
"""
early = [_lb(f"page1-{i:02d}", i + 1) for i in range(PAGE_SIZE)]
# Exactly the labels from the #601 residual set (minus status:pr-open)
late = [
_lb("type:feature", 103),
_lb("workflow-hardening", 105),
_lb("anti-stomp", 106),
_lb("leases", 109),
_lb("recovery", 110),
]
mock_all.return_value = early + late
requested = [
"anti-stomp",
"leases",
"recovery",
"type:feature",
"workflow-hardening",
]
mock_req.return_value = [
_lb(n, next(x["id"] for x in late if x["name"] == n)) for n in requested
]
res = mcp_server.gitea_set_issue_labels(
issue_number=601, labels=requested, remote="prgs"
)
names = {lb["name"] for lb in res}
self.assertEqual(names, set(requested))
self.assertNotIn("status:pr-open", names)
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all")
def test_duplicate_label_ids_resolve_deterministically(self, mock_all, mock_req):
mock_all.return_value = [
_lb("type:feature", 103),
_lb("type:feature", 130),
_lb("workflow-hardening", 105),
_lb("workflow-hardening", 128),
]
requested = ["type:feature", "workflow-hardening"]
mock_req.return_value = [_lb("type:feature", 103), _lb("workflow-hardening", 105)]
mcp_server.gitea_set_issue_labels(
issue_number=1, labels=requested, remote="prgs"
)
self.assertEqual(mock_req.call_args[0][3]["labels"], [103, 105])
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all")
def test_post_mutation_mismatch_fails_closed(self, mock_all, mock_req):
mock_all.return_value = [_lb("bug", 1), _lb("status:ready", 2)]
# Server returns incomplete set → verification must fail
mock_req.return_value = [_lb("bug", 1)]
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_set_issue_labels(
issue_number=9,
labels=["bug", "status:ready"],
remote="prgs",
)
self.assertIn("Post-mutation label verification failed", str(ctx.exception))
self.assertIn("status:ready", str(ctx.exception))
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all", side_effect=RuntimeError("Gitea 502 on page 2"))
def test_pagination_failure_fails_closed_before_put(self, _all, mock_req):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_set_issue_labels(
issue_number=9, labels=["bug"], remote="prgs"
)
self.assertIn("page 2", str(ctx.exception))
mock_req.assert_not_called()
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all")
def test_empty_label_set_clears_all(self, mock_all, mock_req):
mock_all.return_value = [_lb("bug", 1)]
mock_req.return_value = []
res = mcp_server.gitea_set_issue_labels(
issue_number=9, labels=[], remote="prgs"
)
self.assertEqual(res, [])
self.assertEqual(mock_req.call_args[0][3]["labels"], [])
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all")
def test_does_not_call_single_page_limit_100(self, mock_all, mock_req):
mock_all.return_value = [_lb("bug", 1)]
mock_req.return_value = [_lb("bug", 1)]
mcp_server.gitea_set_issue_labels(
issue_number=9, labels=["bug"], remote="prgs"
)
for c in mock_req.call_args_list:
url = c[0][1] if len(c[0]) > 1 else ""
self.assertNotIn("labels?limit=100", str(url))
mock_all.assert_called()
class TestApiGetAllPageCapDocumentsGiteaLimit(unittest.TestCase):
"""Sanity: api_get_all clamps page_size to 50 (root cause of limit=100 trap)."""
@patch("gitea_auth.api_request")
def test_page_size_clamped_to_fifty(self, mock_req):
mock_req.return_value = []
gitea_auth.api_get_all("https://gitea.example/api/v1/repos/o/r/labels",
FAKE_AUTH, page_size=100)
# First (only) call must use limit=50, not 100
url = mock_req.call_args[0][1]
self.assertIn("limit=50", url)
self.assertNotIn("limit=100", url)
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink