290 lines
9.1 KiB
Python
290 lines
9.1 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.dict("os.environ", {"WEBUI_TEST_OFFLINE": "0"}):
|
|
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()
|