feat: add live PR/issue queue dashboard to web UI (Closes #429)

Adds read-only /queue and /api/queue routes that load open PRs and issues from the default registry project via gitea_auth, surface pagination proof, and classify items with claimed/blocked/in-review/duplicate badges. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 15:05:41 -04:00
parent 3599a9e12a
commit c91e3642de
6 changed files with 676 additions and 1 deletions
@@ -0,0 +1,138 @@
+"""Tests for web UI live queue dashboard (#429)."""
+import json
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.testclient import TestClient
+
+from webui.app import create_app
+from webui.queue_loader import (
+    PaginationMeta,
+    load_queue_snapshot,
+    snapshot_to_dict,
+)
+
+
+def _page(items: list[dict], *, final: bool = True) -> tuple[list[dict], PaginationMeta]:
+    return items, PaginationMeta(
+        page=1,
+        per_page=50,
+        returned_count=len(items),
+        has_more=not final,
+        is_final_page=final,
+        inventory_complete=final,
+        pages_fetched=1,
+    )
+
+
+SAMPLE_PRS = [
+    {
+        "number": 10,
+        "title": "feat: example",
+        "mergeable": True,
+        "body": "Closes #5",
+        "head": {"ref": "feat/x", "sha": "abc123def456"},
+        "base": {"ref": "master"},
+    },
+    {
+        "number": 9,
+        "title": "fix: conflict",
+        "mergeable": False,
+        "body": "Closes #5",
+        "head": {"ref": "feat/y", "sha": "deadbeef0000"},
+        "base": {"ref": "master"},
+    },
+]
+
+SAMPLE_ISSUES = [
+    {
+        "number": 5,
+        "title": "Tracked issue",
+        "state": "open",
+        "labels": [{"name": "status:in-progress"}],
+        "assignee": {"login": "jcwalker3"},
+    },
+    {
+        "number": 4,
+        "title": "Idle issue",
+        "state": "open",
+        "labels": [],
+        "assignee": None,
+    },
+]
+
+
+class TestQueueLoader(unittest.TestCase):
+    def test_classifies_claimed_duplicate_and_blocked(self):
+        snapshot = load_queue_snapshot(
+            fetch_prs=lambda *_a, **_k: _page(SAMPLE_PRS),
+            fetch_issues=lambda *_a, **_k: _page(SAMPLE_ISSUES),
+        )
+        self.assertIsNone(snapshot.fetch_error)
+        pr_badges = {p.number: p.badges for p in snapshot.prs}
+        self.assertIn("blocked", pr_badges[9])
+        self.assertIn("duplicate", pr_badges[9])
+        issue_badges = {i.number: i.badges for i in snapshot.issues}
+        self.assertIn("claimed", issue_badges[5])
+        self.assertIn("duplicate", issue_badges[5])
+
+    def test_fail_closed_without_credentials(self):
+        with patch("webui.queue_loader.get_auth_header", return_value=None):
+            snapshot = load_queue_snapshot()
+        self.assertIsNotNone(snapshot.fetch_error)
+        self.assertIn("credentials unavailable", snapshot.fetch_error.lower())
+        self.assertEqual(snapshot.prs, ())
+        self.assertEqual(snapshot.issues, ())
+
+    def test_snapshot_json_includes_pagination(self):
+        snapshot = load_queue_snapshot(
+            fetch_prs=lambda *_a, **_k: _page(SAMPLE_PRS, final=False),
+            fetch_issues=lambda *_a, **_k: _page(SAMPLE_ISSUES),
+        )
+        payload = snapshot_to_dict(snapshot)
+        self.assertFalse(payload["pagination"]["prs"]["inventory_complete"])
+        self.assertTrue(payload["pagination"]["prs"]["has_more"])
+        self.assertTrue(payload["pagination"]["issues"]["inventory_complete"])
+
+
+class TestQueueRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+
+    def test_queue_page_renders_with_mocked_loader(self):
+        snapshot = load_queue_snapshot(
+            fetch_prs=lambda *_a, **_k: _page(SAMPLE_PRS),
+            fetch_issues=lambda *_a, **_k: _page(SAMPLE_ISSUES),
+        )
+        with patch("webui.app.load_queue_snapshot", return_value=snapshot):
+            response = self.client.get("/queue")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Live queue", response.text)
+        self.assertIn("feat: example", response.text)
+        self.assertIn("Tracked issue", response.text)
+        self.assertIn("pagination", response.text.lower())
+
+    def test_api_queue_json(self):
+        snapshot = load_queue_snapshot(
+            fetch_prs=lambda *_a, **_k: _page(SAMPLE_PRS),
+            fetch_issues=lambda *_a, **_k: _page(SAMPLE_ISSUES),
+        )
+        with patch("webui.app.load_queue_snapshot", return_value=snapshot):
+            response = self.client.get("/api/queue")
+        self.assertEqual(response.status_code, 200)
+        payload = json.loads(response.text)
+        self.assertEqual(len(payload["prs"]), 2)
+        self.assertEqual(len(payload["issues"]), 2)
+
+    def test_nav_includes_queue_link(self):
+        response = self.client.get("/")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn('href="/queue"', response.text)
+
+
+if __name__ == "__main__":
+    unittest.main()