sysadmin
ad59053cd7
The author allocator could select an ineligible issue for two independent reasons, both fixed here. Defect 1 — candidate truncation before ranking. The loader fetched the complete open inventory via api_get_all, then sliced it to `limit` items before constructing any WorkCandidate. Because every status:ready issue ties at priority 20 and the tie breaks on lowest number, the slice — not the ranking — decided the winner, so the same query returned different answers at different `limit` values. Candidate construction now consumes the full listing; `limit` bounds the reported skip list only, and any such truncation is reported explicitly rather than silently. Defect 2 — dependency state inferred from body substrings. Only "blocked on #" and "downstream of #" were recognized, so the repository's canonical `Depends: #N, #N` field never set dependency_unmet and dependency-blocked issues were emitted as eligible. The new allocator_dependencies module parses the canonical declaration into structured references and resolves each against live issue state: open means unmet, closed means met, and unavailable evidence fails closed. The complete open listing proves openness without extra calls; anything absent from it is confirmed by a cached targeted lookup instead of being assumed closed. The legacy "blocked on #" marker still parses. A failed listing now marks the inventory incomplete and the tool fails closed instead of ranking a partial set. Selection already advanced past a skipped candidate; with dependencies resolved correctly that fall-through now actually engages, and is covered by a regression. Tests: 35 new cases across parser, resolver, loader, and an MCP-level gitea_allocate_next_work regression, including limit-invariant selection over a 73-candidate inventory and a guard proving pre-ranking truncation would change the winner. No issue number is special-cased.
228 lines
9.6 KiB
Python
228 lines
9.6 KiB
Python
"""MCP-level allocator inventory and dependency regressions (#758).
|
|
|
|
Exercises ``_allocator_candidates_from_gitea`` and the ``gitea_allocate_next_work``
|
|
tool end to end against a faked Gitea API, proving:
|
|
|
|
* the complete open-issue inventory is ranked (no pre-ranking truncation);
|
|
* ``limit`` cannot change which candidate wins;
|
|
* canonical ``Depends:`` declarations are resolved from live issue state;
|
|
* unavailable dependency evidence fails closed;
|
|
* an incomplete listing fails closed instead of ranking a partial set.
|
|
|
|
Issue numbers here are synthetic; no production number is special-cased.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import gitea_mcp_server as srv
|
|
from control_plane_db import ControlPlaneDB
|
|
|
|
FAKE_AUTH = "token REDACTED"
|
|
ORG = "Scaled-Tech-Consulting"
|
|
REPO = "Gitea-Tools"
|
|
|
|
|
|
def _issue(number: int, *, body: str = "", labels=("status:ready",)) -> dict:
|
|
return {
|
|
"number": number,
|
|
"title": f"issue {number}",
|
|
"body": body,
|
|
"labels": [{"name": name} for name in labels],
|
|
"state": "open",
|
|
}
|
|
|
|
|
|
def _depends_body(*refs: int) -> str:
|
|
joined = ", ".join(f"#{r}" for r in refs)
|
|
return f"## Dependencies and linkage\n\n* Parent: #999 · Depends: {joined}\n"
|
|
|
|
|
|
class _FakeGitea:
|
|
"""Minimal stand-in for the two Gitea list endpoints plus issue lookups."""
|
|
|
|
def __init__(self, issues, *, closed=(), unavailable=(), fail_issue_list=False):
|
|
self.issues = list(issues)
|
|
self.closed = set(closed)
|
|
self.unavailable = set(unavailable)
|
|
self.fail_issue_list = fail_issue_list
|
|
self.lookups: list[int] = []
|
|
|
|
def api_get_all(self, url, _auth, **_kw):
|
|
if "/pulls" in url:
|
|
return []
|
|
if self.fail_issue_list:
|
|
raise RuntimeError("issue listing failed")
|
|
return list(self.issues)
|
|
|
|
def api_request(self, _method, url, _auth, **_kw):
|
|
number = int(url.rsplit("/", 1)[-1])
|
|
self.lookups.append(number)
|
|
if number in self.unavailable:
|
|
raise RuntimeError("lookup failed")
|
|
if number in self.closed:
|
|
return {"number": number, "state": "closed"}
|
|
return {"number": number, "state": "open"}
|
|
|
|
|
|
class AllocatorInventoryTest(unittest.TestCase):
|
|
"""Direct tests of the candidate loader."""
|
|
|
|
def _load(self, fake, **kwargs):
|
|
with patch("gitea_mcp_server._resolve", return_value=("h", ORG, REPO)), patch(
|
|
"gitea_mcp_server._auth", return_value=FAKE_AUTH
|
|
), patch("gitea_mcp_server.api_get_all", side_effect=fake.api_get_all), patch(
|
|
"gitea_mcp_server.api_request", side_effect=fake.api_request
|
|
):
|
|
return srv._allocator_candidates_from_gitea(
|
|
remote="prgs", host=None, org=ORG, repo=REPO, **kwargs
|
|
)
|
|
|
|
def test_full_inventory_above_fifty_is_ranked(self) -> None:
|
|
"""AC1: all 73 open issues become candidates, not the first 50."""
|
|
fake = _FakeGitea([_issue(n) for n in range(600, 673)])
|
|
candidates, _reasons, complete = self._load(fake)
|
|
self.assertTrue(complete)
|
|
self.assertEqual(len(candidates), 73)
|
|
self.assertEqual(min(c.number for c in candidates), 600)
|
|
self.assertEqual(max(c.number for c in candidates), 672)
|
|
|
|
def test_open_dependency_marks_candidate_unmet(self) -> None:
|
|
"""AC4/AC5/AC6: canonical Depends on an open issue blocks the candidate."""
|
|
fake = _FakeGitea(
|
|
[_issue(600, body=_depends_body(601, 602)), _issue(601), _issue(602)]
|
|
)
|
|
candidates, _reasons, _complete = self._load(fake)
|
|
blocked = next(c for c in candidates if c.number == 600)
|
|
self.assertTrue(blocked.dependency_unmet)
|
|
self.assertIn("#601", blocked.dependency_reason)
|
|
|
|
def test_closed_dependency_is_eligible(self) -> None:
|
|
"""A dependency absent from the open list is confirmed closed, not assumed."""
|
|
fake = _FakeGitea([_issue(600, body=_depends_body(500))], closed={500})
|
|
candidates, _reasons, _complete = self._load(fake)
|
|
candidate = next(c for c in candidates if c.number == 600)
|
|
self.assertFalse(candidate.dependency_unmet)
|
|
self.assertIn(500, fake.lookups) # proved live, not inferred
|
|
|
|
def test_unavailable_dependency_evidence_fails_closed(self) -> None:
|
|
"""AC7: an unreachable dependency must block, never pass."""
|
|
fake = _FakeGitea([_issue(600, body=_depends_body(500))], unavailable={500})
|
|
candidates, _reasons, _complete = self._load(fake)
|
|
candidate = next(c for c in candidates if c.number == 600)
|
|
self.assertTrue(candidate.dependency_unmet)
|
|
self.assertIn("fail closed", candidate.dependency_reason)
|
|
|
|
def test_dependency_state_lookups_are_cached(self) -> None:
|
|
"""Repeated references resolve with a single live lookup."""
|
|
fake = _FakeGitea(
|
|
[_issue(n, body=_depends_body(500)) for n in range(600, 610)],
|
|
closed={500},
|
|
)
|
|
self._load(fake)
|
|
self.assertEqual(fake.lookups.count(500), 1)
|
|
|
|
def test_open_dependency_needs_no_lookup(self) -> None:
|
|
"""The complete open listing already proves openness."""
|
|
fake = _FakeGitea([_issue(600, body=_depends_body(601)), _issue(601)])
|
|
self._load(fake)
|
|
self.assertNotIn(601, fake.lookups)
|
|
|
|
def test_failed_issue_listing_reports_incomplete(self) -> None:
|
|
"""AC3: a failed listing must not silently yield a short inventory."""
|
|
fake = _FakeGitea([], fail_issue_list=True)
|
|
_candidates, reasons, complete = self._load(fake)
|
|
self.assertFalse(complete)
|
|
self.assertTrue(any("failed to list open issues" in r for r in reasons))
|
|
|
|
|
|
class AllocateNextWorkToolTest(unittest.TestCase):
|
|
"""End-to-end tests of the gitea_allocate_next_work MCP tool."""
|
|
|
|
def setUp(self) -> None:
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
|
|
|
|
def tearDown(self) -> None:
|
|
self._tmp.cleanup()
|
|
|
|
def _allocate(self, fake, **kwargs):
|
|
with patch("gitea_mcp_server._profile_operation_gate", return_value=None), patch(
|
|
"gitea_mcp_server._resolve", return_value=("h", ORG, REPO)
|
|
), patch("gitea_mcp_server._auth", return_value=FAKE_AUTH), patch(
|
|
"gitea_mcp_server.get_profile",
|
|
return_value={"profile_name": "prgs-author", "role": "author"},
|
|
), patch(
|
|
"gitea_mcp_server._authenticated_username", return_value="jcwalker3"
|
|
), patch(
|
|
"gitea_mcp_server._control_plane_db_or_error", return_value=(self.db, [])
|
|
), patch(
|
|
"gitea_mcp_server.api_get_all", side_effect=fake.api_get_all
|
|
), patch(
|
|
"gitea_mcp_server.api_request", side_effect=fake.api_request
|
|
), patch(
|
|
"gitea_mcp_server.sentry_observability.monitor_checkin", return_value=None
|
|
):
|
|
return srv.gitea_allocate_next_work(
|
|
remote="prgs", org=ORG, repo=REPO, role="author", **kwargs
|
|
)
|
|
|
|
def test_limit_does_not_change_selection(self) -> None:
|
|
"""AC2/AC11: the winner is identical at limit=1 and limit=300."""
|
|
issues = [_issue(n) for n in range(600, 673)] # 73 candidates
|
|
low = self._allocate(_FakeGitea(issues), limit=1)
|
|
high = self._allocate(_FakeGitea(issues), limit=300)
|
|
self.assertEqual(low["selected"]["number"], high["selected"]["number"])
|
|
self.assertEqual(low["selected"]["number"], 600)
|
|
self.assertEqual(low["candidate_count"], 73)
|
|
self.assertEqual(high["candidate_count"], 73)
|
|
|
|
def test_dependency_blocked_winner_falls_through(self) -> None:
|
|
"""AC8: a blocked highest-ranked issue yields the next eligible one."""
|
|
issues = [
|
|
_issue(600, body=_depends_body(601)),
|
|
_issue(601),
|
|
_issue(602),
|
|
]
|
|
result = self._allocate(_FakeGitea(issues))
|
|
# 600 is blocked by open 601; 601 itself is a valid candidate.
|
|
self.assertEqual(result["selected"]["number"], 601)
|
|
skipped = {s["number"] for s in result["skipped"]}
|
|
self.assertIn(600, skipped)
|
|
|
|
def test_limit_truncates_only_the_reported_skip_list(self) -> None:
|
|
"""A shortened report is labelled, never presented as full coverage."""
|
|
issues = [_issue(n, body=_depends_body(999)) for n in range(600, 640)]
|
|
issues.append(_issue(999)) # open dependency blocks all of the above
|
|
issues.append(_issue(700)) # the one eligible candidate
|
|
result = self._allocate(_FakeGitea(issues), limit=5)
|
|
self.assertTrue(result["skipped_report_truncated"])
|
|
self.assertEqual(len(result["skipped"]), 5)
|
|
self.assertGreater(result["skipped_total"], 5)
|
|
self.assertEqual(result["limit_applies_to"], "reported_skip_list_only")
|
|
|
|
def test_incomplete_inventory_fails_closed(self) -> None:
|
|
"""AC3: no selection is made from a partial candidate set."""
|
|
result = self._allocate(_FakeGitea([], fail_issue_list=True))
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["inventory_complete"])
|
|
self.assertIsNone(result["assignment"])
|
|
self.assertTrue(
|
|
any("fail closed" in r for r in result["reasons"]),
|
|
result["reasons"],
|
|
)
|
|
|
|
def test_selection_policy_is_reported(self) -> None:
|
|
"""AC10: tie-breaking is stated in the result, not left implicit."""
|
|
result = self._allocate(_FakeGitea([_issue(600)]))
|
|
self.assertIn("selection_policy", result)
|
|
self.assertIn("number asc", result["selection_policy"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|