Gitea-Tools/tests/test_webui_queue_dashboard.py

"""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,
    _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()

_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)


if __name__ == "__main__":
    unittest.main()