test: harden queue dashboard classification and route coverage (#429)

Expand queue dashboard tests for stale/duplicate badges, title-linked
issues, and nav coverage; drop unused import in queue_loader.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

tests/test_webui_queue_dashboard.py

@@ -1,9 +1,9 @@
 """Tests for web UI live queue dashboard (#429)."""
-import json
 import sys
 import unittest
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
-from unittest.mock import patch
+from unittest import mock
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
@@ -12,126 +12,202 @@ from starlette.testclient import TestClient
 from webui.app import create_app
 from webui.queue_loader import (
     PaginationMeta,
+    _classify_issue,
+    _classify_pr,
+    _extract_linked_issue,
     load_queue_snapshot,
     snapshot_to_dict,
 )
 
+_RECENT = datetime.now(timezone.utc).isoformat()
+_STALE = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
 
-def _page(items: list[dict], *, final: bool = True) -> tuple[list[dict], PaginationMeta]:
+_SAMPLE_PRS = [
-    return items, PaginationMeta(
+    {
+        "number": 100,
+        "title": "feat: queue dashboard (Closes #429)",
+        "body": "",
+        "mergeable": True,
+        "updated_at": _RECENT,
+        "head": {"ref": "feat/issue-429", "sha": "abc123def456"},
+        "base": {"ref": "master"},
+    },
+    {
+        "number": 99,
+        "title": "fix: conflict",
+        "body": "Closes #50",
+        "mergeable": False,
+        "updated_at": _STALE,
+        "head": {"ref": "fix/issue-50", "sha": "deadbeef0001"},
+        "base": {"ref": "master"},
+    },
+    {
+        "number": 98,
+        "title": "feat: duplicate A (Closes #60)",
+        "body": "",
+        "mergeable": True,
+        "updated_at": _RECENT,
+        "head": {"ref": "feat/issue-60-a", "sha": "111111111111"},
+        "base": {"ref": "master"},
+    },
+    {
+        "number": 97,
+        "title": "feat: duplicate B (Closes #60)",
+        "body": "",
+        "mergeable": True,
+        "updated_at": _RECENT,
+        "head": {"ref": "feat/issue-60-b", "sha": "222222222222"},
+        "base": {"ref": "master"},
+    },
+]
+
+_SAMPLE_ISSUES = [
+    {
+        "number": 429,
+        "title": "Web UI queue dashboard",
+        "state": "open",
+        "labels": [{"name": "status:in-progress"}],
+        "assignee": None,
+        "updated_at": _RECENT,
+    },
+    {
+        "number": 60,
+        "title": "Duplicate PR target",
+        "state": "open",
+        "labels": [],
+        "assignee": None,
+        "updated_at": _RECENT,
+    },
+    {
+        "number": 50,
+        "title": "Blocked PR target",
+        "state": "open",
+        "labels": [],
+        "assignee": None,
+        "updated_at": _STALE,
+    },
+]
+
+
+def _mock_pagination(count: int) -> PaginationMeta:
+    return PaginationMeta(
         page=1,
         per_page=50,
-        returned_count=len(items),
+        returned_count=count,
-        has_more=not final,
+        has_more=False,
-        is_final_page=final,
+        is_final_page=True,
-        inventory_complete=final,
+        inventory_complete=True,
         pages_fetched=1,
     )
 
 
-SAMPLE_PRS = [
+def _mock_fetch_prs(*_args, **_kwargs):
-    {
+    return list(_SAMPLE_PRS), _mock_pagination(len(_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 = [
+
-    {
+def _mock_fetch_issues(*_args, **_kwargs):
-        "number": 5,
+    return list(_SAMPLE_ISSUES), _mock_pagination(len(_SAMPLE_ISSUES))
-        "title": "Tracked issue",
+
-        "state": "open",
+
-        "labels": [{"name": "status:in-progress"}],
+class TestQueueClassification(unittest.TestCase):
-        "assignee": {"login": "jcwalker3"},
+    def test_extract_linked_issue_from_title(self):
-    },
+        self.assertEqual(
-    {
+            _extract_linked_issue("feat: X (Closes #429)", ""),
-        "number": 4,
+            429,
-        "title": "Idle issue",
+        )
-        "state": "open",
+
-        "labels": [],
+    def test_classify_pr_blocked_and_stale(self):
-        "assignee": None,
+        pr = _SAMPLE_PRS[1]
-    },
+        badges = _classify_pr(
-]
+            pr,
+            issue_claimed=set(),
+            issue_to_prs={50: [99]},
+        )
+        self.assertIn("blocked", badges)
+        self.assertIn("stale", badges)
+
+    def test_classify_pr_duplicate(self):
+        badges = _classify_pr(
+            _SAMPLE_PRS[2],
+            issue_claimed=set(),
+            issue_to_prs={60: [98, 97]},
+        )
+        self.assertIn("duplicate", badges)
+        self.assertIn("in-review", badges)
+
+    def test_classify_issue_claimed_and_duplicate(self):
+        claimed = _classify_issue(_SAMPLE_ISSUES[0], linked_prs=[])
+        self.assertIn("claimed", claimed)
+        duplicate = _classify_issue(_SAMPLE_ISSUES[1], linked_prs=[98, 97])
+        self.assertIn("duplicate", duplicate)
 
 
 class TestQueueLoader(unittest.TestCase):
-    def test_classifies_claimed_duplicate_and_blocked(self):
+    def test_snapshot_with_mock_fetch(self):
         snapshot = load_queue_snapshot(
-            fetch_prs=lambda *_a, **_k: _page(SAMPLE_PRS),
+            fetch_prs=_mock_fetch_prs,
-            fetch_issues=lambda *_a, **_k: _page(SAMPLE_ISSUES),
+            fetch_issues=_mock_fetch_issues,
         )
+        self.assertEqual(snapshot.project_id, "gitea-tools")
         self.assertIsNone(snapshot.fetch_error)
-        pr_badges = {p.number: p.badges for p in snapshot.prs}
+        self.assertEqual(len(snapshot.prs), 4)
-        self.assertIn("blocked", pr_badges[9])
+        self.assertEqual(len(snapshot.issues), 3)
-        self.assertIn("duplicate", pr_badges[9])
+        self.assertTrue(snapshot.pr_pagination.inventory_complete)
-        issue_badges = {i.number: i.badges for i in snapshot.issues}
+        self.assertTrue(snapshot.issue_pagination.inventory_complete)
-        self.assertIn("claimed", issue_badges[5])
-        self.assertIn("duplicate", issue_badges[5])
 
-    def test_fail_closed_without_credentials(self):
+        pr100 = next(p for p in snapshot.prs if p.number == 100)
-        with patch("webui.queue_loader.get_auth_header", return_value=None):
+        self.assertEqual(pr100.extra["linked_issue"], "429")
-            snapshot = load_queue_snapshot()
+        self.assertIn("claimed", pr100.badges)
-        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):
+    def test_snapshot_dict_export(self):
         snapshot = load_queue_snapshot(
-            fetch_prs=lambda *_a, **_k: _page(SAMPLE_PRS, final=False),
+            fetch_prs=_mock_fetch_prs,
-            fetch_issues=lambda *_a, **_k: _page(SAMPLE_ISSUES),
+            fetch_issues=_mock_fetch_issues,
         )
-        payload = snapshot_to_dict(snapshot)
+        data = snapshot_to_dict(snapshot)
-        self.assertFalse(payload["pagination"]["prs"]["inventory_complete"])
+        self.assertEqual(data["project_id"], "gitea-tools")
-        self.assertTrue(payload["pagination"]["prs"]["has_more"])
+        self.assertIsNone(data["fetch_error"])
-        self.assertTrue(payload["pagination"]["issues"]["inventory_complete"])
+        self.assertEqual(len(data["prs"]), 4)
+        self.assertTrue(data["pagination"]["prs"]["inventory_complete"])
 
 
 class TestQueueRoutes(unittest.TestCase):
     def setUp(self):
         self.client = TestClient(create_app())
-
+        self._patch = mock.patch(
-    def test_queue_page_renders_with_mocked_loader(self):
+            "webui.app.load_queue_snapshot",
-        snapshot = load_queue_snapshot(
+            return_value=load_queue_snapshot(
-            fetch_prs=lambda *_a, **_k: _page(SAMPLE_PRS),
+                fetch_prs=_mock_fetch_prs,
-            fetch_issues=lambda *_a, **_k: _page(SAMPLE_ISSUES),
+                fetch_issues=_mock_fetch_issues,
+            ),
         )
-        with patch("webui.app.load_queue_snapshot", return_value=snapshot):
+        self._patch.start()
+
+    def tearDown(self):
+        self._patch.stop()
+
+    def test_queue_page_renders_tables_and_pagination(self):
         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("Open pull requests", response.text)
-        self.assertIn("Tracked issue", response.text)
+        self.assertIn("Open issues", response.text)
-        self.assertIn("pagination", response.text.lower())
+        self.assertIn("pages_fetched", response.text)
+        self.assertIn("Gitea-Tools", response.text)
+        self.assertNotIn("child issue", 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)
+        data = response.json()
-        self.assertEqual(len(payload["prs"]), 2)
+        self.assertEqual(data["repo"], "Scaled-Tech-Consulting/Gitea-Tools")
-        self.assertEqual(len(payload["issues"]), 2)
+        self.assertEqual(data["pagination"]["issues"]["returned_count"], 3)
 
-    def test_nav_includes_queue_link(self):
+    def test_queue_fail_closed_without_credentials(self):
-        response = self.client.get("/")
+        with mock.patch("webui.queue_loader.get_auth_header", return_value=None):
-        self.assertEqual(response.status_code, 200)
+            snapshot = load_queue_snapshot()
-        self.assertIn('href="/queue"', response.text)
+        self.assertIsNotNone(snapshot.fetch_error)
+        self.assertEqual(len(snapshot.prs), 0)
 
 
 if __name__ == "__main__":

tests/test_webui_skeleton.py

@@ -56,11 +56,16 @@ class TestWebuiSkeleton(unittest.TestCase):
         self.assertEqual(response.status_code, 405)
         self.assertEqual(response.json()["error"], "read-only-mvp")
 
+    def test_queue_route_renders(self):
+        response = self.client.get("/queue")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Live queue", response.text)
+
     def test_nav_links_on_all_pages(self):
-        for path in ("/", "/projects", "/prompts", "/runtime", "/audit"):
+        for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit"):
             with self.subTest(path=path):
                 text = self.client.get(path).text
-                for href in ("/projects", "/prompts", "/runtime", "/audit"):
+                for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"):
                     self.assertIn(f'href="{href}"', text)
 
 

webui/queue_loader.py

@@ -8,7 +8,7 @@ from datetime import datetime, timezone
 from typing import Any, Callable
 from urllib.parse import urlparse
 
-from gitea_auth import api_fetch_page, api_get_all, get_auth_header, repo_api_url
+from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
 
 from webui.project_registry import ProjectRecord, load_registry