sysadmin
22a1c4c2df
When Gitea credentials are missing or the queue fetch fails closed, show "Not loaded" instead of "No open items." and keep pagination marked unavailable. Successful empty inventories still show the empty state with complete pagination proof. Closes #458
288 lines
9.0 KiB
Python
288 lines
9.0 KiB
Python
"""Tests for web UI live queue dashboard (#429)."""
|
|
import sys
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
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,
|
|
QueueSnapshot,
|
|
_classify_issue,
|
|
_classify_pr,
|
|
_extract_linked_issue,
|
|
load_queue_snapshot,
|
|
snapshot_to_dict,
|
|
)
|
|
from webui.queue_views import render_queue_page
|
|
|
|
_RECENT = datetime.now(timezone.utc).isoformat()
|
|
_STALE = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
|
|
|
|
_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=count,
|
|
has_more=False,
|
|
is_final_page=True,
|
|
inventory_complete=True,
|
|
pages_fetched=1,
|
|
)
|
|
|
|
|
|
def _mock_fetch_prs(*_args, **_kwargs):
|
|
return list(_SAMPLE_PRS), _mock_pagination(len(_SAMPLE_PRS))
|
|
|
|
|
|
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_snapshot_with_mock_fetch(self):
|
|
snapshot = load_queue_snapshot(
|
|
fetch_prs=_mock_fetch_prs,
|
|
fetch_issues=_mock_fetch_issues,
|
|
)
|
|
self.assertEqual(snapshot.project_id, "gitea-tools")
|
|
self.assertIsNone(snapshot.fetch_error)
|
|
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)
|
|
|
|
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_dict_export(self):
|
|
snapshot = load_queue_snapshot(
|
|
fetch_prs=_mock_fetch_prs,
|
|
fetch_issues=_mock_fetch_issues,
|
|
)
|
|
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())
|
|
self._patch = mock.patch(
|
|
"webui.app.load_queue_snapshot",
|
|
return_value=load_queue_snapshot(
|
|
fetch_prs=_mock_fetch_prs,
|
|
fetch_issues=_mock_fetch_issues,
|
|
),
|
|
)
|
|
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("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):
|
|
response = self.client.get("/api/queue")
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertEqual(data["repo"], "Scaled-Tech-Consulting/Gitea-Tools")
|
|
self.assertEqual(data["pagination"]["issues"]["returned_count"], 3)
|
|
|
|
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)
|
|
|
|
|
|
def _empty_pagination() -> PaginationMeta:
|
|
return PaginationMeta(
|
|
page=1,
|
|
per_page=50,
|
|
returned_count=0,
|
|
has_more=False,
|
|
is_final_page=True,
|
|
inventory_complete=True,
|
|
pages_fetched=1,
|
|
)
|
|
|
|
|
|
def _empty_fetch(*_args, **_kwargs):
|
|
return [], _empty_pagination()
|
|
|
|
|
|
class TestQueueFailClosedUx(unittest.TestCase):
|
|
"""Regression tests for #458 fail-closed empty-state copy."""
|
|
|
|
def test_fail_closed_view_suppresses_empty_queue_copy(self):
|
|
snapshot = QueueSnapshot(
|
|
project_id="gitea-tools",
|
|
repo_label="Scaled-Tech-Consulting/Gitea-Tools",
|
|
prs=(),
|
|
issues=(),
|
|
pr_pagination=None,
|
|
issue_pagination=None,
|
|
fetch_error="Gitea credentials unavailable for gitea.prgs.cc",
|
|
)
|
|
html = render_queue_page(snapshot)
|
|
self.assertIn("Queue unavailable", html)
|
|
self.assertIn("Not loaded", html)
|
|
self.assertNotIn("No open items.", html)
|
|
self.assertIn("pagination:</strong> unavailable", html)
|
|
|
|
def test_fail_closed_route_does_not_show_empty_queue_copy(self):
|
|
client = TestClient(create_app())
|
|
snapshot = load_queue_snapshot(
|
|
fetch_prs=_empty_fetch,
|
|
fetch_issues=_empty_fetch,
|
|
)
|
|
snapshot = QueueSnapshot(
|
|
project_id=snapshot.project_id,
|
|
repo_label=snapshot.repo_label,
|
|
prs=(),
|
|
issues=(),
|
|
pr_pagination=None,
|
|
issue_pagination=None,
|
|
fetch_error="Gitea credentials unavailable for gitea.prgs.cc",
|
|
)
|
|
with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
|
|
response = client.get("/queue")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("Queue unavailable", response.text)
|
|
self.assertNotIn("No open items.", response.text)
|
|
|
|
def test_successful_empty_inventory_shows_empty_copy(self):
|
|
snapshot = load_queue_snapshot(
|
|
fetch_prs=_empty_fetch,
|
|
fetch_issues=_empty_fetch,
|
|
)
|
|
self.assertIsNone(snapshot.fetch_error)
|
|
self.assertEqual(len(snapshot.prs), 0)
|
|
self.assertEqual(len(snapshot.issues), 0)
|
|
self.assertTrue(snapshot.pr_pagination.inventory_complete)
|
|
|
|
html = render_queue_page(snapshot)
|
|
self.assertNotIn("Queue unavailable", html)
|
|
self.assertEqual(html.count("No open items."), 2)
|
|
self.assertIn("pagination (complete)", html)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |