From f5370a94d3739908dbb6acbd50996568ccb8dafd Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 13:55:58 -0400 Subject: [PATCH] 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) --- tests/test_webui_queue_dashboard.py | 252 ++++++++++++++++++---------- tests/test_webui_skeleton.py | 9 +- webui/queue_loader.py | 2 +- 3 files changed, 172 insertions(+), 91 deletions(-) diff --git a/tests/test_webui_queue_dashboard.py b/tests/test_webui_queue_dashboard.py index f4671ac..cc9ea97 100644 --- a/tests/test_webui_queue_dashboard.py +++ b/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]: - return items, PaginationMeta( +_SAMPLE_PRS = [ + { + "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), - has_more=not final, - is_final_page=final, - inventory_complete=final, + returned_count=count, + has_more=False, + is_final_page=True, + inventory_complete=True, 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"}, - }, -] +def _mock_fetch_prs(*_args, **_kwargs): + return list(_SAMPLE_PRS), _mock_pagination(len(_SAMPLE_PRS)) -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, - }, -] + +def _mock_fetch_issues(*_args, **_kwargs): + return list(_SAMPLE_ISSUES), _mock_pagination(len(_SAMPLE_ISSUES)) + + +class TestQueueClassification(unittest.TestCase): + def test_extract_linked_issue_from_title(self): + self.assertEqual( + _extract_linked_issue("feat: X (Closes #429)", ""), + 429, + ) + + def test_classify_pr_blocked_and_stale(self): + 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_issues=lambda *_a, **_k: _page(SAMPLE_ISSUES), + fetch_prs=_mock_fetch_prs, + 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.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]) + self.assertEqual(len(snapshot.prs), 4) + self.assertEqual(len(snapshot.issues), 3) + self.assertTrue(snapshot.pr_pagination.inventory_complete) + self.assertTrue(snapshot.issue_pagination.inventory_complete) - 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, ()) + pr100 = next(p for p in snapshot.prs if p.number == 100) + self.assertEqual(pr100.extra["linked_issue"], "429") + self.assertIn("claimed", pr100.badges) - 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_issues=lambda *_a, **_k: _page(SAMPLE_ISSUES), + fetch_prs=_mock_fetch_prs, + fetch_issues=_mock_fetch_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"]) + data = snapshot_to_dict(snapshot) + self.assertEqual(data["project_id"], "gitea-tools") + self.assertIsNone(data["fetch_error"]) + 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()) - - 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), + self._patch = mock.patch( + "webui.app.load_queue_snapshot", + return_value=load_queue_snapshot( + fetch_prs=_mock_fetch_prs, + fetch_issues=_mock_fetch_issues, + ), ) - with patch("webui.app.load_queue_snapshot", return_value=snapshot): - response = self.client.get("/queue") + 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("Tracked issue", response.text) - self.assertIn("pagination", response.text.lower()) + self.assertIn("Open pull requests", response.text) + self.assertIn("Open issues", response.text) + 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") + 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) + data = response.json() + self.assertEqual(data["repo"], "Scaled-Tech-Consulting/Gitea-Tools") + self.assertEqual(data["pagination"]["issues"]["returned_count"], 3) - def test_nav_includes_queue_link(self): - response = self.client.get("/") - self.assertEqual(response.status_code, 200) - self.assertIn('href="/queue"', response.text) + def test_queue_fail_closed_without_credentials(self): + with mock.patch("webui.queue_loader.get_auth_header", return_value=None): + snapshot = load_queue_snapshot() + self.assertIsNotNone(snapshot.fetch_error) + self.assertEqual(len(snapshot.prs), 0) if __name__ == "__main__": diff --git a/tests/test_webui_skeleton.py b/tests/test_webui_skeleton.py index 937b095..5dd23a1 100644 --- a/tests/test_webui_skeleton.py +++ b/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) diff --git a/webui/queue_loader.py b/webui/queue_loader.py index 57ebeeb..a70e58d 100644 --- a/webui/queue_loader.py +++ b/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