"""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()