feat: add workflow dashboard for queue, leases, and next safe action (Closes #605)
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.
This commit is contained in:
@@ -0,0 +1,274 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user