Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 74 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

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

Browse Source 
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]>
This commit is contained in:
sysadmin
2026-07-07 15:05:42 -04:00
co-authored by Claude Opus 4.8
parent c91e3642de
commit f5370a94d3
3 changed files with 172 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_webui_queue_dashboard.py
+164 -88
View File
@@ -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__":