sysadmin
7f2b9f36de
Read-only MCP tool gitea_workflow_dashboard plus mcp-menu entry so operators and LLMs can see PR/issue queues, leases, terminal locks, blockers, and exact next-safe prompts without reconstructing state from comments. Never assigns work or presents blocked/terminal-locked items as safe.
275 lines
11 KiB
Python
275 lines
11 KiB
Python
"""Hermetic tests for workflow dashboard (#605).
|
|
|
|
Covers terminal-blocked queue shapes in the spirit of #593/#592/#587 where an
|
|
active terminal-review lock must suppress other PRs as safe review/merge work.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from allocator_service import WorkCandidate
|
|
from workflow_dashboard import (
|
|
DASHBOARD_VERSION,
|
|
build_workflow_dashboard,
|
|
format_human_summary,
|
|
)
|
|
|
|
|
|
def _issue(
|
|
number: int,
|
|
*,
|
|
title: str = "",
|
|
labels: tuple[str, ...] = ("status:ready",),
|
|
priority: int = 20,
|
|
blocked: bool = False,
|
|
dependency_unmet: bool = False,
|
|
dependency_reason: str | None = None,
|
|
claimed: bool = False,
|
|
) -> WorkCandidate:
|
|
return WorkCandidate(
|
|
kind="issue",
|
|
number=number,
|
|
title=title or f"issue {number}",
|
|
labels=labels,
|
|
priority=priority,
|
|
blocked=blocked,
|
|
dependency_unmet=dependency_unmet,
|
|
dependency_reason=dependency_reason,
|
|
already_claimed_elsewhere=claimed,
|
|
)
|
|
|
|
|
|
def _pr(
|
|
number: int,
|
|
*,
|
|
title: str = "",
|
|
head_sha: str = "abc123",
|
|
request_changes: bool = False,
|
|
approved: bool = False,
|
|
mergeable: bool = False,
|
|
contaminated: bool = False,
|
|
approval_stale: bool = False,
|
|
priority: int = 5,
|
|
) -> WorkCandidate:
|
|
return WorkCandidate(
|
|
kind="pr",
|
|
number=number,
|
|
title=title or f"pr {number}",
|
|
head_sha=head_sha,
|
|
request_changes_current_head=request_changes,
|
|
approval_on_current_head=approved,
|
|
mergeable=mergeable,
|
|
approval_contaminated=contaminated,
|
|
approval_stale=approval_stale,
|
|
priority=priority,
|
|
)
|
|
|
|
|
|
class TestWorkflowDashboard(unittest.TestCase):
|
|
def test_version_and_read_only_payload(self):
|
|
snap = build_workflow_dashboard(
|
|
candidates=[_issue(605)],
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
)
|
|
payload = snap.as_dict()
|
|
self.assertTrue(payload["read_only"])
|
|
self.assertEqual(payload["dashboard_version"], DASHBOARD_VERSION)
|
|
self.assertTrue(payload["success"])
|
|
self.assertTrue(payload["inventory_complete"])
|
|
self.assertIn("human_summary", payload)
|
|
|
|
def test_never_marks_blocked_as_safe(self):
|
|
candidates = [
|
|
_issue(10, blocked=True, labels=("status:blocked",)),
|
|
_issue(11, dependency_unmet=True, dependency_reason="depends on #9"),
|
|
_issue(12, claimed=True),
|
|
_issue(605, labels=("status:ready",)),
|
|
]
|
|
snap = build_workflow_dashboard(candidates=candidates)
|
|
blocked_numbers = {e.number for e in snap.blocked_items}
|
|
self.assertIn(10, blocked_numbers)
|
|
self.assertIn(11, blocked_numbers)
|
|
self.assertIn(12, blocked_numbers)
|
|
for entry in snap.blocked_items:
|
|
self.assertFalse(entry.as_dict()["is_safe"])
|
|
self.assertEqual(entry.safe_for_roles, ())
|
|
self.assertIsNotNone(entry.block_reason)
|
|
|
|
author = snap.next_safe_by_role["author"]
|
|
self.assertEqual(author.status, "safe")
|
|
self.assertEqual(author.target_number, 605)
|
|
self.assertNotIn(author.target_number, blocked_numbers)
|
|
summary = format_human_summary(snap)
|
|
self.assertIn("NOT safe", summary)
|
|
self.assertIn("issue#10", summary.replace(" ", ""))
|
|
|
|
def test_author_prefers_oldest_ready_issue(self):
|
|
candidates = [
|
|
_issue(620, labels=("status:ready",)),
|
|
_issue(605, labels=("status:ready",)),
|
|
_issue(610, labels=("status:ready",)),
|
|
]
|
|
snap = build_workflow_dashboard(candidates=candidates)
|
|
author = snap.next_safe_by_role["author"]
|
|
self.assertEqual(author.status, "safe")
|
|
self.assertEqual(author.target_number, 605)
|
|
self.assertIn("gitea_allocate_next_work", author.prompt)
|
|
self.assertIn("role='author'", author.prompt)
|
|
|
|
def test_review_and_merge_ready_buckets(self):
|
|
candidates = [
|
|
_pr(100, head_sha="r1"), # review-ready
|
|
_pr(101, approved=True, mergeable=True, head_sha="m1", priority=8),
|
|
_pr(102, request_changes=True, head_sha="a1", priority=10),
|
|
]
|
|
snap = build_workflow_dashboard(candidates=candidates)
|
|
self.assertEqual([e.number for e in snap.review_ready_prs], [100])
|
|
self.assertEqual([e.number for e in snap.merge_ready_prs], [101])
|
|
self.assertEqual([e.number for e in snap.author_remediation], [102])
|
|
|
|
reviewer = snap.next_safe_by_role["reviewer"]
|
|
self.assertEqual(reviewer.status, "safe")
|
|
self.assertEqual(reviewer.target_number, 100)
|
|
self.assertEqual(reviewer.head_sha, "r1")
|
|
|
|
merger = snap.next_safe_by_role["merger"]
|
|
self.assertEqual(merger.status, "safe")
|
|
self.assertEqual(merger.target_number, 101)
|
|
self.assertEqual(merger.head_sha, "m1")
|
|
|
|
author = snap.next_safe_by_role["author"]
|
|
self.assertEqual(author.status, "safe")
|
|
self.assertEqual(author.target_number, 102)
|
|
|
|
def test_terminal_lock_blocks_other_prs_as_safe(
|
|
self,
|
|
):
|
|
"""#593/#592/#587-style: terminal lock ⇒ other PRs are not safe."""
|
|
candidates = [
|
|
_pr(587, head_sha="deadbeef", priority=5),
|
|
_pr(592, approved=True, mergeable=True, head_sha="cafebabe", priority=8),
|
|
_pr(593, head_sha="terminalhead", priority=9),
|
|
_issue(605, labels=("status:ready",)),
|
|
]
|
|
snap = build_workflow_dashboard(
|
|
candidates=candidates,
|
|
terminal_pr=593,
|
|
terminal_lock={"terminal_pr": 593, "active": True, "state": "locked"},
|
|
)
|
|
|
|
# Non-terminal PRs must appear blocked, never in safe buckets.
|
|
blocked_prs = {
|
|
e.number for e in snap.blocked_items if e.kind == "pr"
|
|
}
|
|
self.assertIn(587, blocked_prs)
|
|
self.assertIn(592, blocked_prs)
|
|
self.assertNotIn(593, blocked_prs) # terminal PR itself may still be routeable
|
|
|
|
# Terminal PR itself may remain review-ready; others must not.
|
|
self.assertEqual([e.number for e in snap.review_ready_prs], [593])
|
|
self.assertEqual(snap.merge_ready_prs, [])
|
|
self.assertNotIn(587, [e.number for e in snap.review_ready_prs])
|
|
self.assertNotIn(592, [e.number for e in snap.merge_ready_prs])
|
|
|
|
for entry in snap.blocked_items:
|
|
if entry.number in (587, 592):
|
|
self.assertIn("terminal-review lock", entry.block_reason or "")
|
|
self.assertEqual(entry.safe_for_roles, ())
|
|
self.assertFalse(entry.as_dict()["is_safe"])
|
|
|
|
reviewer = snap.next_safe_by_role["reviewer"]
|
|
# Reviewer may only target the terminal PR — never 587/592.
|
|
self.assertEqual(reviewer.status, "safe")
|
|
self.assertEqual(reviewer.target_number, 593)
|
|
self.assertEqual(reviewer.head_sha, "terminalhead")
|
|
self.assertNotEqual(reviewer.target_number, 587)
|
|
self.assertNotEqual(reviewer.target_number, 592)
|
|
|
|
merger = snap.next_safe_by_role["merger"]
|
|
# Merge-ready #592 is NOT safe while terminal lock is on #593.
|
|
self.assertNotEqual(merger.target_number, 592)
|
|
self.assertIn("593", merger.prompt)
|
|
self.assertIn(
|
|
merger.status,
|
|
("blocked_terminal", "idle", "safe"),
|
|
)
|
|
if merger.status == "safe":
|
|
self.assertEqual(merger.target_number, 593)
|
|
|
|
# Author issue work remains visible (issues are not terminal-blocked).
|
|
author = snap.next_safe_by_role["author"]
|
|
self.assertEqual(author.status, "safe")
|
|
self.assertEqual(author.target_number, 605)
|
|
|
|
summary = format_human_summary(snap)
|
|
self.assertIn("Terminal review lock: ACTIVE on PR #593", summary)
|
|
self.assertIn("Do not treat other open PRs as safe", summary)
|
|
|
|
def test_incomplete_inventory_fails_closed(self):
|
|
snap = build_workflow_dashboard(
|
|
candidates=[_issue(605)],
|
|
inventory_complete=False,
|
|
inventory_reasons=["page truncated"],
|
|
)
|
|
payload = snap.as_dict()
|
|
self.assertFalse(payload["inventory_complete"])
|
|
self.assertEqual(payload["review_ready_prs"], [])
|
|
self.assertEqual(payload["merge_ready_prs"], [])
|
|
for action in snap.next_safe_by_role.values():
|
|
self.assertEqual(action.status, "none")
|
|
self.assertIsNone(action.target_number)
|
|
self.assertIn("inventory incomplete", action.prompt.lower())
|
|
self.assertFalse(action.as_dict()["is_safe"])
|
|
|
|
def test_leases_partition_active_vs_stale(self):
|
|
leases = [
|
|
{"lease_id": "L1", "role": "author", "status": "active", "work_number": 605},
|
|
{"lease_id": "L2", "role": "reviewer", "status": "expired", "work_number": 99},
|
|
{"lease_id": "L3", "role": "merger", "stale": True, "work_number": 88},
|
|
]
|
|
snap = build_workflow_dashboard(candidates=[], leases=leases)
|
|
self.assertEqual(len(snap.active_leases_by_role["author"]), 1)
|
|
self.assertEqual(len(snap.stale_or_expired_leases), 2)
|
|
|
|
def test_discussion_and_controller_needed(self):
|
|
candidates = [
|
|
_issue(1, labels=("discussion", "type:discussion")),
|
|
_pr(2, contaminated=True, head_sha="x"),
|
|
]
|
|
snap = build_workflow_dashboard(candidates=candidates)
|
|
self.assertEqual([e.number for e in snap.discussion_issues], [1])
|
|
self.assertTrue(any(e.number == 2 for e in snap.controller_needed))
|
|
recon = snap.next_safe_by_role["reconciler"]
|
|
self.assertEqual(recon.status, "safe")
|
|
self.assertEqual(recon.target_number, 2)
|
|
|
|
def test_human_summary_includes_exact_prompts(self):
|
|
snap = build_workflow_dashboard(
|
|
candidates=[_issue(605)],
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
)
|
|
text = format_human_summary(snap)
|
|
self.assertIn("gitea_allocate_next_work", text)
|
|
self.assertIn("prgs/Scaled-Tech-Consulting/Gitea-Tools", text)
|
|
self.assertIn("never self-selects", text.lower())
|
|
self.assertIn("Primary next:", text)
|
|
|
|
def test_missing_pr_head_sha_is_blocked(self):
|
|
candidates = [_pr(50, head_sha="")]
|
|
# WorkCandidate allows empty head; dashboard must block it.
|
|
c = candidates[0]
|
|
c.head_sha = ""
|
|
snap = build_workflow_dashboard(candidates=[c])
|
|
self.assertEqual(len(snap.blocked_items), 1)
|
|
self.assertIn("head_sha", snap.blocked_items[0].block_reason or "")
|
|
self.assertEqual(snap.review_ready_prs, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|