From f32aaaa3b5ec2917e39d6fd98d58e4bc70dec0f5 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Fri, 10 Jul 2026 13:05:27 -0400 Subject: [PATCH] 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 --- gitea_mcp_server.py | 109 +++++--- manage_labels.py | 16 +- mark_issue.py | 10 +- tests/test_manage_labels.py | 63 ++--- tests/test_python_cli.py | 39 +-- tests/test_set_issue_labels_pagination.py | 318 ++++++++++++++++++++++ 6 files changed, 447 insertions(+), 108 deletions(-) create mode 100644 tests/test_set_issue_labels_pagination.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 66bf5cf..bed2098 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -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 diff --git a/manage_labels.py b/manage_labels.py index 400b319..3f2c11e 100755 --- a/manage_labels.py +++ b/manage_labels.py @@ -22,7 +22,7 @@ venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") if os.path.exists(venv_python) and sys.executable != venv_python: os.execv(venv_python, [venv_python] + sys.argv) -from gitea_auth import get_auth_header, api_request, repo_api_url +from gitea_auth import get_auth_header, api_request, api_get_all, repo_api_url import issue_workflow_labels HOST = "gitea.dadeschools.net" @@ -82,9 +82,17 @@ def api(method, path, auth, payload=None): def _labels_by_name(auth): - """Return {label name: id} for the repo's existing labels.""" - existing = api("GET", "/labels?limit=100", auth) or [] - return {lb["name"]: lb["id"] for lb in existing} + """Return {label name: id} for the repo's existing labels (all pages, #627).""" + existing = api_get_all(f"{BASE_URL}/labels", auth) or [] + name_to_id = {} + for lb in existing: + if not isinstance(lb, dict): + continue + name = lb.get("name") + lid = lb.get("id") + if name and lid is not None and name not in name_to_id: + name_to_id[name] = lid + return name_to_id def create_labels(auth, dry=False): diff --git a/mark_issue.py b/mark_issue.py index b3a2268..582dae5 100755 --- a/mark_issue.py +++ b/mark_issue.py @@ -17,7 +17,7 @@ if os.path.exists(venv_python) and sys.executable != venv_python: from gitea_auth import ( get_auth_header, resolve_remote, add_remote_args, - api_request, repo_api_url, + api_request, api_get_all, repo_api_url, ) LABEL_NAME = "status:in-progress" @@ -45,12 +45,12 @@ def main(argv=None): base = repo_api_url(host, org, repo) try: - # Find the label ID - labels = api_request("GET", f"{base}/labels?limit=100", auth) + # Paginated inventory (#627): Gitea caps single pages at 50. + labels = api_get_all(f"{base}/labels", auth) or [] label_id = None for lb in labels: - if lb["name"] == LABEL_NAME: - label_id = lb["id"] + if lb.get("name") == LABEL_NAME: + label_id = lb.get("id") break if label_id is None: diff --git a/tests/test_manage_labels.py b/tests/test_manage_labels.py index ca008a1..3a7f7a5 100644 --- a/tests/test_manage_labels.py +++ b/tests/test_manage_labels.py @@ -28,33 +28,31 @@ class TestLabelCreation(unittest.TestCase): """Verify create-or-skip logic for the label set.""" @patch("manage_labels.get_auth_header", return_value=FAKE_AUTH) + @patch("manage_labels.api_get_all") @patch("manage_labels.api") - def test_skips_existing_labels(self, mock_api, _auth): - # Simulate all labels already exist + def test_skips_existing_labels(self, mock_api, mock_get_all, _auth): + # Simulate all labels already exist (paginated inventory #627) existing = [_make_label(l["name"], i) for i, l in enumerate(manage_labels.LABELS)] - mock_api.return_value = existing # first call is GET /labels + mock_get_all.return_value = existing # Patch sys.argv to avoid --dry with patch.object(sys, "argv", ["manage_labels.py"]): with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): manage_labels.main() - # The GET call happens, but no POST calls for label creation - get_calls = [c for c in mock_api.call_args_list if c[0][0] == "GET"] + mock_get_all.assert_called() post_label_calls = [ c for c in mock_api.call_args_list if c[0][0] == "POST" and c[0][1] == "/labels" ] - self.assertGreaterEqual(len(get_calls), 1) self.assertEqual(len(post_label_calls), 0) @patch("manage_labels.get_auth_header", return_value=FAKE_AUTH) + @patch("manage_labels.api_get_all", return_value=[]) @patch("manage_labels.api") - def test_creates_missing_labels(self, mock_api, _auth): + def test_creates_missing_labels(self, mock_api, _get_all, _auth): # Simulate no existing labels def side_effect(method, path, auth, payload=None): - if method == "GET" and "/labels" in path: - return [] # no existing labels if method == "POST" and path == "/labels": return {"id": 999, "name": payload["name"]} if method == "PUT": @@ -79,15 +77,14 @@ class TestLabelCreation(unittest.TestCase): class TestDryRun(unittest.TestCase): @patch("manage_labels.get_auth_header", return_value=FAKE_AUTH) + @patch("manage_labels.api_get_all", return_value=[]) @patch("manage_labels.api") - def test_dry_run_makes_no_writes(self, mock_api, _auth): - mock_api.return_value = [] # no existing labels - + def test_dry_run_makes_no_writes(self, mock_api, _get_all, _auth): with patch.object(sys, "argv", ["manage_labels.py", "--dry"]): with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): manage_labels.main() - # Only the GET call should be made, no POST or PUT + # Dry run should not call write methods via api() for c in mock_api.call_args_list: method = c[0][0] self.assertEqual(method, "GET", @@ -100,13 +97,13 @@ class TestDryRun(unittest.TestCase): class TestLabelMapping(unittest.TestCase): @patch("manage_labels.get_auth_header", return_value=FAKE_AUTH) + @patch("manage_labels.api_get_all") @patch("manage_labels.api") - def test_applies_mapping_to_issues(self, mock_api, _auth): + def test_applies_mapping_to_issues(self, mock_api, mock_get_all, _auth): existing = [_make_label(l["name"], i + 1) for i, l in enumerate(manage_labels.LABELS)] + mock_get_all.return_value = existing def side_effect(method, path, auth, payload=None): - if method == "GET": - return existing if method == "PUT": return [{"name": "applied"}] return None @@ -158,11 +155,10 @@ class TestModes(unittest.TestCase): return [(c[0][0], c[0][1]) for c in mock_api.call_args_list] @patch("manage_labels.get_auth_header", return_value=FAKE_AUTH) + @patch("manage_labels.api_get_all", return_value=[]) @patch("manage_labels.api") - def test_create_labels_only_no_mapping(self, mock_api, _auth): + def test_create_labels_only_no_mapping(self, mock_api, _get_all, _auth): def se(method, path, auth, payload=None): - if method == "GET": - return [] # no existing labels if method == "POST" and path == "/labels": return {"id": 1, "name": payload["name"]} return None @@ -173,14 +169,14 @@ class TestModes(unittest.TestCase): self.assertFalse(any(m[0] == "PUT" for m in methods)) # no mapping applied @patch("manage_labels.get_auth_header", return_value=FAKE_AUTH) + @patch("manage_labels.api_get_all") @patch("manage_labels.api") - def test_apply_mapping_only_no_label_creation(self, mock_api, _auth): + def test_apply_mapping_only_no_label_creation(self, mock_api, mock_get_all, _auth): existing = [_make_label(l["name"], i + 1) for i, l in enumerate(manage_labels.LABELS)] + mock_get_all.return_value = existing def se(method, path, auth, payload=None): - if method == "GET": - return existing if method == "PUT": return [{"name": "applied"}] return None @@ -192,13 +188,10 @@ class TestModes(unittest.TestCase): self.assertEqual(len(put_calls), len(manage_labels.MAPPING)) @patch("manage_labels.get_auth_header", return_value=FAKE_AUTH) + @patch("manage_labels.api_get_all", return_value=[_make_label("chore", 5)]) @patch("manage_labels.api") - def test_add_label_appends_to_issue(self, mock_api, _auth): - existing = [_make_label("chore", 5)] - + def test_add_label_appends_to_issue(self, mock_api, _get_all, _auth): def se(method, path, auth, payload=None): - if method == "GET": - return existing if method == "POST": return [{"name": "chore"}] return None @@ -212,19 +205,21 @@ class TestModes(unittest.TestCase): self.assertFalse(any(c[0][0] == "PUT" for c in mock_api.call_args_list)) @patch("manage_labels.get_auth_header", return_value=FAKE_AUTH) + @patch("manage_labels.api_get_all", return_value=[]) @patch("manage_labels.api") - def test_add_label_unknown_makes_no_write(self, mock_api, _auth): - mock_api.side_effect = lambda *a, **k: [] if a[0] == "GET" else None + def test_add_label_unknown_makes_no_write(self, mock_api, _get_all, _auth): manage_labels.main(["--add-label", "42", "ghost"]) - # Only the GET label lookup; no POST/PUT for an undefined label. - self.assertTrue(all(c[0][0] == "GET" for c in mock_api.call_args_list)) + # No write via api() for an undefined label. + self.assertTrue(all(c[0][0] == "GET" for c in mock_api.call_args_list) + or len(mock_api.call_args_list) == 0) @patch("manage_labels.get_auth_header", return_value=FAKE_AUTH) + @patch("manage_labels.api_get_all", return_value=[_make_label("chore", 5)]) @patch("manage_labels.api") - def test_add_label_dry_makes_no_write(self, mock_api, _auth): - mock_api.side_effect = lambda *a, **k: [_make_label("chore", 5)] if a[0] == "GET" else None + def test_add_label_dry_makes_no_write(self, mock_api, _get_all, _auth): manage_labels.main(["--dry", "--add-label", "42", "chore"]) - self.assertTrue(all(c[0][0] == "GET" for c in mock_api.call_args_list)) + self.assertTrue(all(c[0][0] == "GET" for c in mock_api.call_args_list) + or len(mock_api.call_args_list) == 0) @patch("manage_labels.get_auth_header", return_value=FAKE_AUTH) @patch("manage_labels.api") diff --git a/tests/test_python_cli.py b/tests/test_python_cli.py index 55d727a..b4a53b2 100644 --- a/tests/test_python_cli.py +++ b/tests/test_python_cli.py @@ -64,54 +64,45 @@ class TestMarkIssueCLI(unittest.TestCase): with self.assertRaises(SystemExit): mark_issue.main(["10", "bogus_action"]) + @patch("mark_issue.api_get_all", return_value=[{"id": 101, "name": "status:in-progress"}]) @patch("mark_issue.api_request") @patch("mark_issue.get_auth_header", return_value=FAKE_AUTH) - def test_successful_start(self, _auth, mock_api): - # First call is GET labels, second is POST label - mock_api.side_effect = [ - [{"id": 101, "name": "status:in-progress"}], - [{"name": "status:in-progress"}], - ] + def test_successful_start(self, _auth, mock_api, mock_get_all): + mock_api.return_value = [{"name": "status:in-progress"}] with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): rc = mark_issue.main(["15", "start"]) self.assertEqual(rc, 0) - self.assertEqual(mock_api.call_count, 2) - - # Verify GET labels call - get_call = mock_api.call_args_list[0] - self.assertEqual(get_call[0][0], "GET") - self.assertIn("/labels?limit=100", get_call[0][1]) + mock_get_all.assert_called() + self.assertIn("/labels", mock_get_all.call_args[0][0]) # Verify POST labels call - post_call = mock_api.call_args_list[1] + post_call = mock_api.call_args_list[0] self.assertEqual(post_call[0][0], "POST") self.assertIn("/issues/15/labels", post_call[0][1]) self.assertEqual(post_call[0][3], {"labels": [101]}) + @patch("mark_issue.api_get_all", return_value=[{"id": 101, "name": "status:in-progress"}]) @patch("mark_issue.api_request") @patch("mark_issue.get_auth_header", return_value=FAKE_AUTH) - def test_successful_done(self, _auth, mock_api): - # First call is GET labels, second is DELETE label - mock_api.side_effect = [ - [{"id": 101, "name": "status:in-progress"}], - None, - ] + def test_successful_done(self, _auth, mock_api, _get_all): + mock_api.return_value = None rc = mark_issue.main(["15", "done"]) self.assertEqual(rc, 0) - self.assertEqual(mock_api.call_count, 2) + self.assertEqual(mock_api.call_count, 1) # Verify DELETE labels call - delete_call = mock_api.call_args_list[1] + delete_call = mock_api.call_args_list[0] self.assertEqual(delete_call[0][0], "DELETE") self.assertIn("/issues/15/labels/101", delete_call[0][1]) + @patch("mark_issue.api_get_all", return_value=[{"id": 1, "name": "bug"}]) @patch("mark_issue.api_request") @patch("mark_issue.get_auth_header", return_value=FAKE_AUTH) - def test_label_not_found(self, _auth, mock_api): - # GET labels returns no status:in-progress label - mock_api.return_value = [{"id": 1, "name": "bug"}] + def test_label_not_found(self, _auth, mock_api, _get_all): + # Paginated inventory returns no status:in-progress label rc = mark_issue.main(["15", "start"]) self.assertEqual(rc, 1) + mock_api.assert_not_called() if __name__ == "__main__": diff --git a/tests/test_set_issue_labels_pagination.py b/tests/test_set_issue_labels_pagination.py new file mode 100644 index 0000000..1050891 --- /dev/null +++ b/tests/test_set_issue_labels_pagination.py @@ -0,0 +1,318 @@ +"""#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()