193 lines
6.9 KiB
Python
193 lines
6.9 KiB
Python
"""Tests for web UI workflow traffic-control view (#640)."""
|
|
|
|
import sys
|
|
import unittest
|
|
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.traffic_loader import (
|
|
TrafficItem,
|
|
TrafficSnapshot,
|
|
load_traffic_snapshot,
|
|
snapshot_to_dict,
|
|
)
|
|
from webui.traffic_views import render_traffic_page
|
|
from allocator_service import WorkCandidate
|
|
|
|
|
|
class TestTrafficClassification(unittest.TestCase):
|
|
def test_runnable_candidate_classification(self):
|
|
cand = WorkCandidate(
|
|
kind="issue",
|
|
number=640,
|
|
state="open",
|
|
labels=("status:ready",),
|
|
title="Web Console: Workflow traffic-control view (Phase 1)",
|
|
priority=20,
|
|
)
|
|
snap = load_traffic_snapshot(candidates=[cand])
|
|
self.assertEqual(len(snap.runnable), 1)
|
|
self.assertEqual(snap.runnable[0].number, 640)
|
|
self.assertTrue(snap.runnable[0].is_safe)
|
|
self.assertEqual(snap.runnable[0].traffic_state, "runnable")
|
|
|
|
def test_blocked_dependency_candidate_classification(self):
|
|
cand = WorkCandidate(
|
|
kind="issue",
|
|
number=643,
|
|
state="open",
|
|
labels=("status:ready",),
|
|
title="Web Console: Requests & intent preview (Phase 2)",
|
|
priority=20,
|
|
dependency_unmet=True,
|
|
dependency_reason="issue#643 depends on unresolved issue(s) #640; they are not closed",
|
|
)
|
|
snap = load_traffic_snapshot(candidates=[cand])
|
|
self.assertEqual(len(snap.blocked), 1)
|
|
self.assertEqual(snap.blocked[0].number, 643)
|
|
self.assertFalse(snap.blocked[0].is_safe)
|
|
self.assertEqual(snap.blocked[0].traffic_state, "blocked")
|
|
self.assertIn("depends on unresolved issue(s) #640", snap.blocked[0].block_reason)
|
|
|
|
def test_leased_candidate_classification(self):
|
|
cand = WorkCandidate(
|
|
kind="issue",
|
|
number=640,
|
|
state="open",
|
|
labels=("status:in-progress",),
|
|
title="Web Console: Workflow traffic-control view (Phase 1)",
|
|
priority=20,
|
|
)
|
|
lease = {
|
|
"kind": "issue",
|
|
"number": 640,
|
|
"session_id": "prgs-author-12345",
|
|
"role": "author",
|
|
"status": "active",
|
|
}
|
|
snap = load_traffic_snapshot(candidates=[cand], leases=[lease])
|
|
self.assertEqual(len(snap.leased), 1)
|
|
self.assertEqual(snap.leased[0].number, 640)
|
|
self.assertEqual(snap.leased[0].traffic_state, "leased")
|
|
self.assertIsNotNone(snap.leased[0].lease_info)
|
|
|
|
def test_needs_controller_candidate_classification(self):
|
|
cand = WorkCandidate(
|
|
kind="issue",
|
|
number=700,
|
|
state="open",
|
|
labels=("status:blocked",),
|
|
title="Controller intervention needed",
|
|
priority=10,
|
|
blocked=True,
|
|
)
|
|
snap = load_traffic_snapshot(candidates=[cand])
|
|
self.assertEqual(len(snap.needs_controller), 1)
|
|
self.assertEqual(snap.needs_controller[0].number, 700)
|
|
|
|
|
|
class TestTrafficLoader(unittest.TestCase):
|
|
def test_snapshot_to_dict_export(self):
|
|
cand = WorkCandidate(
|
|
kind="issue",
|
|
number=640,
|
|
state="open",
|
|
labels=("status:ready",),
|
|
title="Traffic control test",
|
|
priority=20,
|
|
)
|
|
snap = load_traffic_snapshot(candidates=[cand])
|
|
data = snapshot_to_dict(snap)
|
|
self.assertEqual(data["project_id"], "gitea-tools")
|
|
self.assertEqual(len(data["runnable"]), 1)
|
|
self.assertTrue(data["inventory_complete"])
|
|
|
|
def test_fail_closed_error_handling(self):
|
|
with mock.patch("webui.traffic_loader.load_queue_snapshot", side_effect=RuntimeError("Gitea connection failed")):
|
|
snap = load_traffic_snapshot()
|
|
self.assertIsNotNone(snap.fetch_error)
|
|
self.assertIn("Failed to load traffic state", snap.fetch_error)
|
|
self.assertEqual(len(snap.runnable), 0)
|
|
self.assertFalse(snap.inventory_complete)
|
|
|
|
|
|
class TestTrafficRoutesAndRendering(unittest.TestCase):
|
|
def setUp(self):
|
|
self.client = TestClient(create_app())
|
|
|
|
def test_traffic_html_page_rendering(self):
|
|
cand1 = WorkCandidate(
|
|
kind="issue",
|
|
number=640,
|
|
state="open",
|
|
labels=("status:ready",),
|
|
title="Traffic View Implementation",
|
|
priority=20,
|
|
)
|
|
cand2 = WorkCandidate(
|
|
kind="issue",
|
|
number=643,
|
|
state="open",
|
|
labels=("status:ready",),
|
|
title="Dependent Feature",
|
|
priority=20,
|
|
dependency_unmet=True,
|
|
dependency_reason="issue#643 depends on unresolved issue(s) #640; they are not closed",
|
|
)
|
|
snap = load_traffic_snapshot(candidates=[cand1, cand2])
|
|
with mock.patch("webui.app.load_traffic_snapshot", return_value=snap):
|
|
response = self.client.get("/traffic")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("Workflow Traffic Control", response.text)
|
|
self.assertIn("1. Runnable Lanes", response.text)
|
|
self.assertIn("3. Blocked Items", response.text)
|
|
self.assertIn("Traffic View Implementation", response.text)
|
|
self.assertIn("depends on unresolved issue(s) #640", response.text)
|
|
|
|
def test_api_traffic_json_route(self):
|
|
cand = WorkCandidate(
|
|
kind="issue",
|
|
number=640,
|
|
state="open",
|
|
labels=("status:ready",),
|
|
title="Traffic View API Test",
|
|
priority=20,
|
|
)
|
|
snap = load_traffic_snapshot(candidates=[cand])
|
|
with mock.patch("webui.app.load_traffic_snapshot", return_value=snap):
|
|
response = self.client.get("/api/traffic")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()
|
|
self.assertEqual(data["project_id"], "gitea-tools")
|
|
self.assertEqual(len(data["runnable"]), 1)
|
|
self.assertEqual(data["runnable"][0]["number"], 640)
|
|
|
|
def test_render_traffic_fail_closed_page(self):
|
|
snap = TrafficSnapshot(
|
|
project_id="gitea-tools",
|
|
repo_label="Scaled-Tech-Consulting/Gitea-Tools",
|
|
runnable=(),
|
|
leased=(),
|
|
blocked=(),
|
|
needs_controller=(),
|
|
terminal_complete=(),
|
|
next_roles=(),
|
|
fetch_error="Gitea credentials unavailable for gitea.prgs.cc",
|
|
inventory_complete=False,
|
|
)
|
|
html = render_traffic_page(snap)
|
|
self.assertIn("Traffic data unavailable", html)
|
|
self.assertIn("Fail closed", html)
|
|
self.assertNotIn("1. Runnable Lanes", html)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|