sysadminandClaude Opus 4.8
ed0e8c82de
status:pr-open was applied by gitea_create_pr and never removed again. Every
terminal path finished without touching it, so a repository audit found 40
closed issues still advertising an open PR.
Add terminal_pr_label_cleanup.py as the single authoritative rule and route
every sanctioned terminal path through it, so the paths cannot drift:
- merge (gitea_merge_pr)
- close without merge (gitea_edit_pr)
- supersession/abandonment (gitea_reconcile_superseded_by_merged_pr)
- already-landed reconciliation (gitea_reconcile_already_landed_pr)
- controller closure (gitea_close_issue)
- retry/recovery (new gitea_cleanup_terminal_pr_labels)
The rule removes only status:pr-open, preserves every other label, allows an
empty resulting set, is a no-op when the label is absent (so retries are
safe), and confirms the outcome by read-after-write rather than assumption.
Controller closure runs the cleanup before the state change and fails closed
if it cannot be completed and verified; closing first would bake in the stale
label with no later step to catch it. Post-merge cleanup never blocks the
merge, which already happened, and reports failures with a safe next action.
Also:
- gitea_assess_terminal_label_hygiene: read-only terminal validation that
reports residual status:pr-open, exempting issues with a genuinely open PR.
- _put_issue_label_names now accepts Gitea's empty response body when the
requested set is empty, so clearing the last label works.
- test_audit's close_issue fixture keys on the request instead of call order,
since closing now also reads labels for the cleanup and its read-back.
Docs: label-taxonomy terminal-transition section, runbook pointer, and the
review-merge / reconcile-landed final-report terminal-label requirements.
Suite: 4045 passed, 11 failed, 6 skipped. The same 11 failures reproduce on
clean master df31674 (4010 passed, 11 failed) and are pre-existing.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
556 lines
24 KiB
Python
556 lines
24 KiB
Python
"""#780: ``status:pr-open`` must not survive a terminal PR transition.
|
|
|
|
The leak this file locks down: ``gitea_create_pr`` applied ``status:pr-open``
|
|
and no terminal path ever removed it, so a repository audit found 40 closed
|
|
issues still advertising an open PR that had long since merged or closed.
|
|
|
|
Coverage mirrors the issue's acceptance criteria: merge, close-without-merge,
|
|
supersession, already-landed reconciliation, controller closure, retry /
|
|
idempotency, unrelated-label preservation, the only-label (empty set) case,
|
|
and terminal validation of any residual label.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import mcp_server
|
|
import terminal_pr_label_cleanup as tplc
|
|
|
|
|
|
FAKE_AUTH = "token test-token"
|
|
PR_OPEN = tplc.PR_OPEN_LABEL
|
|
|
|
|
|
def _lb(name: str, lid: int) -> dict:
|
|
return {"id": lid, "name": name, "color": "000000"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pure rule: planning
|
|
# ---------------------------------------------------------------------------
|
|
class TestPlanPrOpenCleanup(unittest.TestCase):
|
|
|
|
def test_removes_only_the_pr_open_label(self):
|
|
plan = tplc.plan_pr_open_cleanup(
|
|
["type:bug", PR_OPEN, "workflow-hardening"],
|
|
terminal_reason=tplc.MERGED,
|
|
)
|
|
self.assertTrue(plan["cleanup_required"])
|
|
self.assertEqual(plan["removed"], [PR_OPEN])
|
|
self.assertEqual(plan["labels_after"], ["type:bug", "workflow-hardening"])
|
|
|
|
def test_preserves_unrelated_labels_in_original_order(self):
|
|
labels = ["workflow-hardening", "type:bug", PR_OPEN, "role:author", "leases"]
|
|
plan = tplc.plan_pr_open_cleanup(labels, terminal_reason=tplc.MERGED)
|
|
self.assertEqual(
|
|
plan["labels_after"],
|
|
["workflow-hardening", "type:bug", "role:author", "leases"],
|
|
)
|
|
self.assertNotIn(PR_OPEN, plan["labels_after"])
|
|
|
|
def test_only_label_yields_empty_set(self):
|
|
plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=tplc.MERGED)
|
|
self.assertTrue(plan["cleanup_required"])
|
|
self.assertEqual(plan["labels_after"], [])
|
|
self.assertTrue(plan["empty_label_set"])
|
|
|
|
def test_absent_label_is_an_idempotent_noop(self):
|
|
plan = tplc.plan_pr_open_cleanup(
|
|
["type:bug", "status:done"], terminal_reason=tplc.RETRY_RECOVERY
|
|
)
|
|
self.assertFalse(plan["cleanup_required"])
|
|
self.assertTrue(plan["idempotent_noop"])
|
|
self.assertEqual(plan["labels_after"], ["type:bug", "status:done"])
|
|
|
|
def test_accepts_gitea_label_objects(self):
|
|
plan = tplc.plan_pr_open_cleanup(
|
|
{"labels": [{"name": PR_OPEN}, {"name": "type:bug"}]},
|
|
terminal_reason=tplc.SUPERSEDED,
|
|
)
|
|
self.assertEqual(plan["labels_after"], ["type:bug"])
|
|
|
|
def test_every_terminal_reason_is_planable(self):
|
|
for reason in tplc.TERMINAL_REASONS:
|
|
plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=reason)
|
|
self.assertEqual(plan["terminal_reason"], reason)
|
|
self.assertTrue(plan["terminal_reason_description"])
|
|
|
|
def test_unknown_terminal_reason_fails_closed(self):
|
|
with self.assertRaises(ValueError):
|
|
tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason="whenever")
|
|
|
|
def test_reason_aliases_normalize(self):
|
|
self.assertEqual(tplc.canonical_terminal_reason("merge"), tplc.MERGED)
|
|
self.assertEqual(
|
|
tplc.canonical_terminal_reason("already-landed"), tplc.ALREADY_LANDED
|
|
)
|
|
self.assertEqual(
|
|
tplc.canonical_terminal_reason("controller-closure"),
|
|
tplc.CONTROLLER_CLOSURE,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pure rule: read-after-write verification
|
|
# ---------------------------------------------------------------------------
|
|
class TestVerifyPrOpenCleanup(unittest.TestCase):
|
|
|
|
def test_verified_when_observed_matches_plan(self):
|
|
plan = tplc.plan_pr_open_cleanup(
|
|
["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
|
|
)
|
|
result = tplc.verify_pr_open_cleanup(["type:bug"], plan=plan)
|
|
self.assertTrue(result["verified"])
|
|
self.assertFalse(result["residual"])
|
|
self.assertEqual(result["reasons"], [])
|
|
|
|
def test_residual_label_is_reported(self):
|
|
plan = tplc.plan_pr_open_cleanup(
|
|
["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
|
|
)
|
|
result = tplc.verify_pr_open_cleanup(["type:bug", PR_OPEN], plan=plan)
|
|
self.assertFalse(result["verified"])
|
|
self.assertTrue(result["residual"])
|
|
self.assertIn(PR_OPEN, result["reasons"][0])
|
|
self.assertTrue(result["safe_next_action"])
|
|
|
|
def test_dropped_unrelated_label_is_reported(self):
|
|
plan = tplc.plan_pr_open_cleanup(
|
|
["type:bug", "leases", PR_OPEN], terminal_reason=tplc.MERGED
|
|
)
|
|
result = tplc.verify_pr_open_cleanup(["type:bug"], plan=plan)
|
|
self.assertFalse(result["verified"])
|
|
self.assertEqual(result["unexpected_removals"], ["leases"])
|
|
|
|
def test_unexpected_added_label_is_reported(self):
|
|
plan = tplc.plan_pr_open_cleanup(
|
|
["type:bug", PR_OPEN], terminal_reason=tplc.MERGED
|
|
)
|
|
result = tplc.verify_pr_open_cleanup(["type:bug", "surprise"], plan=plan)
|
|
self.assertFalse(result["verified"])
|
|
self.assertEqual(result["unexpected_additions"], ["surprise"])
|
|
|
|
def test_empty_observed_set_verifies_for_only_label_case(self):
|
|
plan = tplc.plan_pr_open_cleanup([PR_OPEN], terminal_reason=tplc.MERGED)
|
|
result = tplc.verify_pr_open_cleanup([], plan=plan)
|
|
self.assertTrue(result["verified"])
|
|
self.assertTrue(result["empty_label_set"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Terminal validation
|
|
# ---------------------------------------------------------------------------
|
|
class TestDetectResidualPrOpen(unittest.TestCase):
|
|
|
|
def test_clean_repository(self):
|
|
issues = [
|
|
{"number": 1, "state": "closed", "labels": [{"name": "type:bug"}]},
|
|
{"number": 2, "state": "open", "labels": []},
|
|
]
|
|
result = tplc.detect_residual_pr_open(issues)
|
|
self.assertTrue(result["clean"])
|
|
self.assertEqual(result["residual_count"], 0)
|
|
self.assertEqual(result["checked_count"], 2)
|
|
|
|
def test_reports_each_stale_issue(self):
|
|
issues = [
|
|
{"number": 626, "state": "closed", "labels": [{"name": PR_OPEN}]},
|
|
{"number": 772, "state": "closed", "labels": [{"name": PR_OPEN}]},
|
|
{"number": 9, "state": "open", "labels": [{"name": "type:bug"}]},
|
|
]
|
|
result = tplc.detect_residual_pr_open(issues)
|
|
self.assertFalse(result["clean"])
|
|
self.assertEqual(result["residual_count"], 2)
|
|
self.assertEqual(
|
|
[entry["number"] for entry in result["residual_issues"]], [626, 772]
|
|
)
|
|
self.assertTrue(result["safe_next_action"])
|
|
|
|
def test_issue_with_a_live_open_pr_is_not_residual(self):
|
|
issues = [{"number": 42, "state": "open", "labels": [{"name": PR_OPEN}]}]
|
|
result = tplc.detect_residual_pr_open(issues, open_pr_issue_numbers=[42])
|
|
self.assertTrue(result["clean"])
|
|
self.assertEqual(result["exempt_open_pr_issues"], [42])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Executor: one authoritative rule, with read-after-write proof
|
|
# ---------------------------------------------------------------------------
|
|
class _ExecutorHarness(unittest.TestCase):
|
|
"""Drives mcp_server.clear_pr_open_label against a fake Gitea."""
|
|
|
|
def setUp(self):
|
|
self.issue_labels: dict[int, list[str]] = {}
|
|
self.repo_labels = {
|
|
PR_OPEN: 4,
|
|
"type:bug": 1,
|
|
"workflow-hardening": 2,
|
|
"leases": 3,
|
|
"status:done": 5,
|
|
}
|
|
self.puts: list[tuple[int, list[int]]] = []
|
|
|
|
patch("mcp_server._resolve", return_value=("h", "o", "r")).start()
|
|
patch("mcp_server._auth", return_value=FAKE_AUTH).start()
|
|
patch(
|
|
"mcp_server.repo_api_url",
|
|
return_value="https://gitea.example/api/v1/repos/o/r",
|
|
).start()
|
|
patch("gitea_audit.audit_enabled", return_value=False).start()
|
|
patch("mcp_server.api_request", side_effect=self._api).start()
|
|
# api_get_all resolves api_request inside gitea_auth, so patching the
|
|
# mcp_server binding alone would let the label inventory hit the network.
|
|
patch("mcp_server.api_get_all", side_effect=self._api_get_all).start()
|
|
self.addCleanup(patch.stopall)
|
|
|
|
def _api_get_all(self, url, auth, **_kwargs):
|
|
if "/labels" in url:
|
|
return [_lb(name, lid) for name, lid in self.repo_labels.items()]
|
|
raise AssertionError(f"unexpected paginated GET: {url}")
|
|
|
|
def _api(self, method, url, auth, payload=None):
|
|
if method == "GET" and "/issues/" in url:
|
|
num = int(url.rsplit("/issues/", 1)[1].split("?")[0])
|
|
return {
|
|
"number": num,
|
|
"labels": [
|
|
{"name": n, "id": self.repo_labels[n]}
|
|
for n in self.issue_labels.get(num, [])
|
|
],
|
|
}
|
|
if method == "PUT" and url.endswith("/labels"):
|
|
num = int(url.rsplit("/issues/", 1)[1].split("/")[0])
|
|
ids = payload["labels"]
|
|
by_id = {lid: name for name, lid in self.repo_labels.items()}
|
|
names = [by_id[i] for i in ids]
|
|
self.puts.append((num, ids))
|
|
self.issue_labels[num] = names
|
|
return [_lb(n, self.repo_labels[n]) for n in names]
|
|
raise AssertionError(f"unexpected API call: {method} {url}")
|
|
|
|
def _clear(self, numbers, reason=tplc.MERGED):
|
|
return mcp_server.clear_pr_open_label(
|
|
numbers, "prgs", None, None, None, terminal_reason=reason
|
|
)
|
|
|
|
|
|
class TestClearPrOpenLabel(_ExecutorHarness):
|
|
|
|
def test_removes_label_and_preserves_the_rest(self):
|
|
self.issue_labels[780] = ["type:bug", PR_OPEN, "workflow-hardening"]
|
|
summary = self._clear([780])
|
|
self.assertTrue(summary["clean"])
|
|
self.assertEqual(summary["removed"], [780])
|
|
self.assertEqual(
|
|
self.issue_labels[780], ["type:bug", "workflow-hardening"]
|
|
)
|
|
|
|
def test_only_label_results_in_empty_set(self):
|
|
self.issue_labels[626] = [PR_OPEN]
|
|
summary = self._clear([626])
|
|
self.assertTrue(summary["clean"])
|
|
self.assertEqual(self.issue_labels[626], [])
|
|
self.assertEqual(self.puts, [(626, [])])
|
|
self.assertTrue(summary["results"][0]["empty_label_set"])
|
|
|
|
def test_read_after_write_proof_is_returned(self):
|
|
self.issue_labels[780] = ["type:bug", PR_OPEN]
|
|
summary = self._clear([780])
|
|
entry = summary["results"][0]
|
|
self.assertTrue(entry["verified"])
|
|
self.assertEqual(entry["labels_before"], ["type:bug", PR_OPEN])
|
|
self.assertEqual(entry["labels_after"], ["type:bug"])
|
|
self.assertEqual(entry["verification"]["observed_labels"], ["type:bug"])
|
|
|
|
def test_repeated_cleanup_is_harmless(self):
|
|
self.issue_labels[780] = ["type:bug", PR_OPEN]
|
|
first = self._clear([780])
|
|
second = self._clear([780], reason=tplc.RETRY_RECOVERY)
|
|
third = self._clear([780], reason=tplc.RETRY_RECOVERY)
|
|
self.assertTrue(first["clean"] and second["clean"] and third["clean"])
|
|
self.assertEqual(second["already_absent"], [780])
|
|
self.assertEqual(third["already_absent"], [780])
|
|
# Exactly one mutation across three calls.
|
|
self.assertEqual(len(self.puts), 1)
|
|
self.assertEqual(self.issue_labels[780], ["type:bug"])
|
|
|
|
def test_noop_path_never_reads_the_label_inventory(self):
|
|
self.issue_labels[780] = ["type:bug"]
|
|
with patch("mcp_server._repo_label_id_map") as mock_map:
|
|
summary = self._clear([780])
|
|
self.assertTrue(summary["clean"])
|
|
mock_map.assert_not_called()
|
|
|
|
def test_duplicate_issue_numbers_are_collapsed(self):
|
|
self.issue_labels[780] = ["type:bug", PR_OPEN]
|
|
summary = self._clear([780, 780, "780"])
|
|
self.assertEqual(summary["checked"], [780])
|
|
self.assertEqual(len(self.puts), 1)
|
|
|
|
def test_no_issue_numbers_is_a_clean_noop(self):
|
|
summary = self._clear([])
|
|
self.assertTrue(summary["clean"])
|
|
self.assertEqual(summary["checked"], [])
|
|
|
|
def test_failed_mutation_is_reported_not_swallowed(self):
|
|
self.issue_labels[780] = ["type:bug", PR_OPEN]
|
|
|
|
def boom(*_a, **_kw):
|
|
raise RuntimeError("gitea exploded")
|
|
|
|
with patch("mcp_server._put_issue_label_names", side_effect=boom):
|
|
summary = self._clear([780])
|
|
self.assertFalse(summary["clean"])
|
|
self.assertEqual(summary["failed"], [780])
|
|
self.assertTrue(summary["safe_next_action"])
|
|
self.assertIn(PR_OPEN, self.issue_labels[780])
|
|
|
|
def test_residual_label_after_write_fails_verification(self):
|
|
self.issue_labels[780] = ["type:bug", PR_OPEN]
|
|
real_api = self._api
|
|
|
|
# Simulate a write that reports success but leaves the label behind.
|
|
def sticky(method, url, auth, payload=None):
|
|
if method == "PUT" and url.endswith("/labels"):
|
|
num = int(url.rsplit("/issues/", 1)[1].split("/")[0])
|
|
self.puts.append((num, payload["labels"]))
|
|
return [_lb("type:bug", 1), _lb(PR_OPEN, 4)]
|
|
return real_api(method, url, auth, payload)
|
|
|
|
with patch("mcp_server.api_request", side_effect=sticky):
|
|
summary = self._clear([780])
|
|
self.assertFalse(summary["clean"])
|
|
self.assertEqual(summary["failed"], [780])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Terminal workflow paths
|
|
# ---------------------------------------------------------------------------
|
|
class TestTerminalPathsUseTheSharedRule(_ExecutorHarness):
|
|
"""Merge, close-without-merge, supersession and already-landed."""
|
|
|
|
def test_merge_path_clears_the_label_for_linked_issues(self):
|
|
self.issue_labels[780] = ["type:bug", PR_OPEN]
|
|
merged_pr = {
|
|
"title": "fix: terminal label cleanup",
|
|
"body": "Closes #780",
|
|
"head": {"ref": "fix/issue-780-terminal-pr-open-label-cleanup"},
|
|
}
|
|
with patch(
|
|
"mcp_server.release_in_progress_label", return_value={780: "released"}
|
|
):
|
|
result = mcp_server.cleanup_in_progress_for_pr(
|
|
merged_pr, "prgs", None, None, None, terminal_reason=tplc.MERGED
|
|
)
|
|
cleanup = result["pr_open_label_cleanup"]
|
|
self.assertTrue(cleanup["clean"])
|
|
self.assertEqual(cleanup["terminal_reason"], tplc.MERGED)
|
|
self.assertEqual(self.issue_labels[780], ["type:bug"])
|
|
|
|
def test_close_without_merge_clears_the_label(self):
|
|
self.issue_labels[781] = ["type:bug", PR_OPEN, "leases"]
|
|
closed_pr = {
|
|
"title": "chore: abandoned",
|
|
"body": "Closes #781",
|
|
"head": {"ref": "chore/issue-781-abandoned"},
|
|
}
|
|
with patch(
|
|
"mcp_server.release_in_progress_label", return_value={781: "released"}
|
|
):
|
|
result = mcp_server.cleanup_in_progress_for_pr(
|
|
closed_pr,
|
|
"prgs",
|
|
None,
|
|
None,
|
|
None,
|
|
terminal_reason=tplc.CLOSED_WITHOUT_MERGE,
|
|
)
|
|
cleanup = result["pr_open_label_cleanup"]
|
|
self.assertTrue(cleanup["clean"])
|
|
self.assertEqual(cleanup["terminal_reason"], tplc.CLOSED_WITHOUT_MERGE)
|
|
self.assertEqual(self.issue_labels[781], ["type:bug", "leases"])
|
|
|
|
def test_pr_without_linked_issue_reports_an_empty_cleanup(self):
|
|
pr = {"title": "chore: no link", "body": "", "head": {"ref": "chore/none"}}
|
|
result = mcp_server.cleanup_in_progress_for_pr(
|
|
pr, "prgs", None, None, None, terminal_reason=tplc.MERGED
|
|
)
|
|
self.assertEqual(result["cleanup_status"], "no linked issue found")
|
|
self.assertTrue(result["pr_open_label_cleanup"]["clean"])
|
|
self.assertEqual(result["pr_open_label_cleanup"]["checked"], [])
|
|
|
|
def test_supersession_reason_is_recorded(self):
|
|
self.issue_labels[600] = [PR_OPEN, "type:bug"]
|
|
summary = self._clear([600], reason=tplc.SUPERSEDED)
|
|
self.assertTrue(summary["clean"])
|
|
self.assertEqual(summary["terminal_reason"], tplc.SUPERSEDED)
|
|
self.assertEqual(self.issue_labels[600], ["type:bug"])
|
|
|
|
def test_already_landed_reconciliation_reason_is_recorded(self):
|
|
self.issue_labels[601] = [PR_OPEN]
|
|
summary = self._clear([601], reason=tplc.ALREADY_LANDED)
|
|
self.assertTrue(summary["clean"])
|
|
self.assertEqual(summary["terminal_reason"], tplc.ALREADY_LANDED)
|
|
self.assertEqual(self.issue_labels[601], [])
|
|
|
|
def test_issue_780_regression_stale_label_survived_every_terminal_path(self):
|
|
"""Regression for the observed leak.
|
|
|
|
Before the fix each terminal path finished without touching
|
|
``status:pr-open``, so the audit found closed issues still carrying it.
|
|
Every path now routes through the one shared rule and leaves nothing
|
|
behind — while preserving each issue's other labels.
|
|
"""
|
|
stale = {
|
|
626: (["type:bug", PR_OPEN], tplc.CONTROLLER_CLOSURE),
|
|
772: (["workflow-hardening", PR_OPEN], tplc.MERGED),
|
|
768: ([PR_OPEN], tplc.CLOSED_WITHOUT_MERGE),
|
|
758: (["leases", PR_OPEN, "type:bug"], tplc.SUPERSEDED),
|
|
755: (["status:done", PR_OPEN], tplc.ALREADY_LANDED),
|
|
}
|
|
for number, (labels, _reason) in stale.items():
|
|
self.issue_labels[number] = list(labels)
|
|
|
|
for number, (_labels, reason) in stale.items():
|
|
summary = self._clear([number], reason=reason)
|
|
self.assertTrue(summary["clean"], msg=f"issue #{number}")
|
|
|
|
audit = tplc.detect_residual_pr_open(
|
|
[
|
|
{"number": num, "state": "closed", "labels": names}
|
|
for num, names in self.issue_labels.items()
|
|
]
|
|
)
|
|
self.assertTrue(audit["clean"])
|
|
self.assertEqual(audit["residual_count"], 0)
|
|
# Unrelated labels survived every path.
|
|
self.assertEqual(self.issue_labels[626], ["type:bug"])
|
|
self.assertEqual(self.issue_labels[772], ["workflow-hardening"])
|
|
self.assertEqual(self.issue_labels[768], [])
|
|
self.assertEqual(self.issue_labels[758], ["leases", "type:bug"])
|
|
self.assertEqual(self.issue_labels[755], ["status:done"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Controller closure
|
|
# ---------------------------------------------------------------------------
|
|
class TestControllerClosure(unittest.TestCase):
|
|
|
|
def test_close_issue_clears_label_before_closing_and_validates(self):
|
|
calls: list[str] = []
|
|
|
|
def fake_clear(numbers, *_a, **kwargs):
|
|
calls.append(f"clear:{kwargs['terminal_reason']}")
|
|
return {
|
|
"label": PR_OPEN,
|
|
"clean": True,
|
|
"checked": list(numbers),
|
|
"removed": list(numbers),
|
|
"already_absent": [],
|
|
"failed": [],
|
|
"results": [],
|
|
"reasons": [],
|
|
"safe_next_action": "",
|
|
"terminal_reason": kwargs["terminal_reason"],
|
|
}
|
|
|
|
def fake_api(method, url, auth, payload=None):
|
|
if method == "PATCH":
|
|
calls.append("patch:closed")
|
|
return {"state": "closed"}
|
|
return {"labels": [{"name": "type:bug"}]}
|
|
|
|
with patch("mcp_server.clear_pr_open_label", side_effect=fake_clear), \
|
|
patch("mcp_server.api_request", side_effect=fake_api), \
|
|
patch("mcp_server._profile_permission_block", return_value=None), \
|
|
patch("mcp_server.verify_preflight_purity", return_value=None), \
|
|
patch("mcp_server.release_in_progress_label", return_value={}), \
|
|
patch("mcp_server._resolve", return_value=("h", "o", "r")), \
|
|
patch("mcp_server._auth", return_value=FAKE_AUTH), \
|
|
patch("gitea_audit.audit_enabled", return_value=False):
|
|
result = mcp_server.gitea_close_issue(issue_number=780, remote="prgs")
|
|
|
|
self.assertTrue(result["success"])
|
|
# Cleanup precedes the state change: closing first would bake in the leak.
|
|
self.assertEqual(calls[0], f"clear:{tplc.CONTROLLER_CLOSURE}")
|
|
self.assertIn("patch:closed", calls)
|
|
self.assertTrue(result["terminal_label_validation"]["clean"])
|
|
|
|
def test_close_issue_fails_closed_when_cleanup_cannot_complete(self):
|
|
def fake_clear(numbers, *_a, **kwargs):
|
|
return {
|
|
"label": PR_OPEN,
|
|
"clean": False,
|
|
"checked": list(numbers),
|
|
"removed": [],
|
|
"already_absent": [],
|
|
"failed": list(numbers),
|
|
"results": [],
|
|
"reasons": ["label replacement failed: boom"],
|
|
"safe_next_action": "retry",
|
|
"terminal_reason": kwargs["terminal_reason"],
|
|
}
|
|
|
|
def fail_on_patch(method, url, auth, payload=None):
|
|
if method == "PATCH":
|
|
raise AssertionError("issue must not be closed when cleanup failed")
|
|
return {}
|
|
|
|
with patch("mcp_server.clear_pr_open_label", side_effect=fake_clear), \
|
|
patch("mcp_server.api_request", side_effect=fail_on_patch), \
|
|
patch("mcp_server._profile_permission_block", return_value=None), \
|
|
patch("mcp_server.verify_preflight_purity", return_value=None), \
|
|
patch("mcp_server._resolve", return_value=("h", "o", "r")), \
|
|
patch("mcp_server._auth", return_value=FAKE_AUTH), \
|
|
patch("gitea_audit.audit_enabled", return_value=False):
|
|
result = mcp_server.gitea_close_issue(issue_number=780, remote="prgs")
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(result["blocked"])
|
|
self.assertFalse(result["performed"])
|
|
self.assertIn("#780", result["message"])
|
|
self.assertTrue(result["safe_next_action"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Capability wiring
|
|
# ---------------------------------------------------------------------------
|
|
class TestCapabilityWiring(unittest.TestCase):
|
|
|
|
def test_task_is_registered_with_label_authority(self):
|
|
import task_capability_map
|
|
|
|
self.assertEqual(
|
|
task_capability_map.required_permission("cleanup_terminal_pr_labels"),
|
|
"gitea.issue.comment",
|
|
)
|
|
self.assertEqual(
|
|
task_capability_map.required_role("cleanup_terminal_pr_labels"),
|
|
"author",
|
|
)
|
|
self.assertEqual(
|
|
task_capability_map.tool_required_permission(
|
|
"gitea_cleanup_terminal_pr_labels"
|
|
),
|
|
"gitea.issue.comment",
|
|
)
|
|
|
|
def test_recovery_tool_rejects_an_unknown_reason_without_mutating(self):
|
|
with patch("mcp_server._profile_permission_block", return_value=None), \
|
|
patch("mcp_server.verify_preflight_purity", return_value=None), \
|
|
patch("mcp_server.clear_pr_open_label") as mock_clear:
|
|
result = mcp_server.gitea_cleanup_terminal_pr_labels(
|
|
issue_numbers=[780], terminal_reason="sometime", remote="prgs"
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["clean"])
|
|
mock_clear.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|