sysadminandClaude Opus 4.8
1948d3dc21
Address PR #885 REQUEST_CHANGES: full head_sha pins from queue signals, reviewer leases keyed by pr_number only, claim inventory via entries, live-path fixture tests, and traffic state vocabulary docs. Closes #640 (re-review at new head) Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
409 lines
14 KiB
Python
409 lines
14 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 TestTrafficLivePath(unittest.TestCase):
|
||
"""Live path tests: inject QueueSnapshot + LeaseSnapshot (no candidates=).
|
||
|
||
Covers the production ``load_traffic_snapshot()`` branch that ``/traffic``
|
||
and ``/api/traffic`` actually execute (#640 B1–B5).
|
||
"""
|
||
|
||
FULL_SHA = "069a9af7e6aa2c2994e07199d1b0814819457017"
|
||
|
||
def _queue(
|
||
self,
|
||
*,
|
||
prs=(),
|
||
issues=(),
|
||
):
|
||
from webui.queue_loader import QueueSnapshot
|
||
|
||
return QueueSnapshot(
|
||
project_id="gitea-tools",
|
||
repo_label="Scaled-Tech-Consulting/Gitea-Tools",
|
||
prs=tuple(prs),
|
||
issues=tuple(issues),
|
||
pr_pagination=None,
|
||
issue_pagination=None,
|
||
fetch_error=None,
|
||
)
|
||
|
||
def _lease(
|
||
self,
|
||
*,
|
||
claim_inventory=None,
|
||
reviewer_leases=(),
|
||
):
|
||
from webui.lease_loader import LeaseSnapshot
|
||
|
||
return LeaseSnapshot(
|
||
project_id="gitea-tools",
|
||
repo_label="Scaled-Tech-Consulting/Gitea-Tools",
|
||
issue_lock=None,
|
||
claim_inventory=claim_inventory or {"entries": [], "counts": {}},
|
||
reviewer_leases=tuple(reviewer_leases),
|
||
duplicate_prs=(),
|
||
duplicate_branches=(),
|
||
collision_history=(),
|
||
fetch_error=None,
|
||
)
|
||
|
||
def test_live_pr_uses_full_head_sha_and_is_runnable(self):
|
||
from webui.queue_loader import QueueItem
|
||
|
||
pr = QueueItem(
|
||
number=885,
|
||
title="traffic control",
|
||
badges=("in-review",),
|
||
extra={"head_sha": self.FULL_SHA[:12], "linked_issue": "640"},
|
||
signals={
|
||
"head_sha": self.FULL_SHA,
|
||
"mergeable": True,
|
||
"labels": (),
|
||
"linked_issue": 640,
|
||
},
|
||
)
|
||
q = self._queue(prs=[pr])
|
||
l = self._lease()
|
||
snap = load_traffic_snapshot(
|
||
fetch_queue_snapshot=lambda: q,
|
||
fetch_lease_snapshot=lambda: l,
|
||
)
|
||
self.assertIsNone(snap.fetch_error)
|
||
self.assertEqual(len(snap.runnable), 1)
|
||
item = snap.runnable[0]
|
||
self.assertEqual(item.kind, "pr")
|
||
self.assertEqual(item.number, 885)
|
||
self.assertEqual(item.head_sha, self.FULL_SHA)
|
||
self.assertNotEqual(item.head_sha, self.FULL_SHA[:12])
|
||
self.assertIsNone(item.block_reason)
|
||
self.assertEqual(len(snap.blocked), 0)
|
||
|
||
def test_live_pr_without_head_sha_is_blocked(self):
|
||
from webui.queue_loader import QueueItem
|
||
|
||
pr = QueueItem(
|
||
number=1,
|
||
title="missing pin",
|
||
badges=("open",),
|
||
extra={"head_sha": ""},
|
||
signals={"head_sha": "", "mergeable": True, "labels": ()},
|
||
)
|
||
snap = load_traffic_snapshot(
|
||
fetch_queue_snapshot=lambda: self._queue(prs=[pr]),
|
||
fetch_lease_snapshot=lambda: self._lease(),
|
||
)
|
||
self.assertEqual(len(snap.blocked) + len(snap.needs_controller), 1)
|
||
item = (snap.blocked or snap.needs_controller)[0]
|
||
self.assertIn("missing head_sha", (item.block_reason or "").lower())
|
||
|
||
def test_reviewer_lease_keys_by_pr_not_linked_issue(self):
|
||
from webui.queue_loader import QueueItem
|
||
|
||
pr = QueueItem(
|
||
number=885,
|
||
title="leased pr",
|
||
badges=("in-review",),
|
||
extra={"head_sha": self.FULL_SHA[:12]},
|
||
signals={"head_sha": self.FULL_SHA, "mergeable": True, "labels": ()},
|
||
)
|
||
issue = QueueItem(
|
||
number=640,
|
||
title="linked issue",
|
||
badges=("open",),
|
||
extra={},
|
||
signals={"labels": ()},
|
||
)
|
||
# Marker-shaped record: has both pr_number and issue_number; must
|
||
# attach to the PR only (B2).
|
||
reviewer_lease = {
|
||
"pr_number": 885,
|
||
"issue_number": 640,
|
||
"phase": "validating",
|
||
"reviewer_identity": "sysadmin",
|
||
"session_id": "review-sess-1",
|
||
}
|
||
snap = load_traffic_snapshot(
|
||
fetch_queue_snapshot=lambda: self._queue(prs=[pr], issues=[issue]),
|
||
fetch_lease_snapshot=lambda: self._lease(reviewer_leases=[reviewer_lease]),
|
||
)
|
||
leased_prs = [i for i in snap.leased if i.kind == "pr" and i.number == 885]
|
||
self.assertEqual(len(leased_prs), 1)
|
||
self.assertEqual(leased_prs[0].lease_info.get("pr_number"), 885)
|
||
# Issue 640 must not inherit the reviewer lease just because issue_number
|
||
# is present on the marker.
|
||
for item in list(snap.leased) + list(snap.runnable) + list(snap.blocked):
|
||
if item.kind == "issue" and item.number == 640:
|
||
self.assertIsNone(
|
||
item.lease_info,
|
||
"reviewer lease must not attach to linked issue #640",
|
||
)
|
||
break
|
||
else:
|
||
self.fail("expected issue #640 in traffic snapshot")
|
||
|
||
def test_claim_inventory_entries_key_marks_issue_leased(self):
|
||
from webui.queue_loader import QueueItem
|
||
|
||
issue = QueueItem(
|
||
number=640,
|
||
title="claimed issue",
|
||
badges=("claimed",),
|
||
extra={},
|
||
signals={"labels": ("status:in-progress",)},
|
||
)
|
||
inventory = {
|
||
"entries": [
|
||
{
|
||
"issue_number": 640,
|
||
"status": "active",
|
||
"latest_heartbeat": {"session_id": "author-sess-9"},
|
||
"reasons": ["claim has structured heartbeat proof"],
|
||
}
|
||
],
|
||
"counts": {"active": 1},
|
||
"in_progress_total": 1,
|
||
}
|
||
snap = load_traffic_snapshot(
|
||
fetch_queue_snapshot=lambda: self._queue(issues=[issue]),
|
||
fetch_lease_snapshot=lambda: self._lease(claim_inventory=inventory),
|
||
)
|
||
leased_issues = [i for i in snap.leased if i.kind == "issue" and i.number == 640]
|
||
self.assertEqual(len(leased_issues), 1)
|
||
self.assertEqual(leased_issues[0].traffic_state, "leased")
|
||
|
||
def test_active_claims_key_is_ignored(self):
|
||
"""B3 regression: fictional ``active_claims`` must not create lease_info."""
|
||
from webui.queue_loader import QueueItem
|
||
|
||
issue = QueueItem(
|
||
number=640,
|
||
title="open issue",
|
||
badges=("open",),
|
||
extra={},
|
||
signals={"labels": ()},
|
||
)
|
||
# Only the broken key — must NOT produce lease_info. Entries-less
|
||
# inventory is empty (entries is the real claim_inventory key).
|
||
inventory = {
|
||
"active_claims": [
|
||
{
|
||
"kind": "issue",
|
||
"number": 640,
|
||
"issue_number": 640,
|
||
"status": "active",
|
||
},
|
||
],
|
||
"counts": {},
|
||
}
|
||
snap = load_traffic_snapshot(
|
||
fetch_queue_snapshot=lambda: self._queue(issues=[issue]),
|
||
fetch_lease_snapshot=lambda: self._lease(claim_inventory=inventory),
|
||
)
|
||
items = [
|
||
i
|
||
for i in (
|
||
list(snap.runnable)
|
||
+ list(snap.leased)
|
||
+ list(snap.blocked)
|
||
+ list(snap.needs_controller)
|
||
)
|
||
if i.kind == "issue" and i.number == 640
|
||
]
|
||
self.assertEqual(len(items), 1)
|
||
self.assertIsNone(
|
||
items[0].lease_info,
|
||
"active_claims is not a real inventory key; entries-only",
|
||
)
|
||
|
||
|
||
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()
|