feat: add lease and collision visibility to web UI (Closes #433)

Adds read-only /leases and /api/leases routes for issue claim inventory,
reviewer PR lease comments, duplicate PR/branch warnings, and local issue
lock visibility without acquire/release mutations.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
2026-07-07 15:29:56 -04:00
co-authored by Claude Opus 4.8
parent ee8e9a0247
commit 09e83fd034
6 changed files with 583 additions and 11 deletions
+94
View File
@@ -0,0 +1,94 @@
"""Tests for web UI lease visibility (#433)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
from webui.app import create_app
from webui.lease_loader import (
_duplicate_branch_warnings,
_duplicate_pr_warnings,
load_lease_snapshot,
parse_reviewer_lease_comment,
snapshot_to_dict,
)
class TestLeaseLoader(unittest.TestCase):
def test_parse_reviewer_lease_comment(self):
body = (
"<!-- mcp-review-lease:v1 -->\n"
"pr: #42\n"
"reviewer_identity: sysadmin\n"
"profile: prgs-reviewer\n"
"phase: validating\n"
"expires_at: 2026-07-07T20:00:00Z\n"
)
parsed = parse_reviewer_lease_comment(body)
self.assertIsNotNone(parsed)
assert parsed is not None
self.assertEqual(parsed["pr_number"], 42)
self.assertEqual(parsed["phase"], "validating")
def test_duplicate_pr_warnings(self):
raw_prs = [
{"number": 1, "title": "Closes #99", "body": ""},
{"number": 2, "title": "fixes #99", "body": ""},
]
warnings = _duplicate_pr_warnings(raw_prs)
self.assertEqual(len(warnings), 1)
self.assertEqual(warnings[0].issue_number, 99)
self.assertEqual(warnings[0].pr_numbers, (1, 2))
def test_duplicate_branch_warnings(self):
warnings = _duplicate_branch_warnings([
"feat/issue-12-a",
"feat/issue-12-b",
"master",
])
self.assertEqual(len(warnings), 1)
self.assertEqual(warnings[0].issue_number, 12)
def test_load_snapshot_with_injected_fetch(self):
def fetch_prs(_h, _o, _r, _a):
return ([], None)
def fetch_issues(_h, _o, _r, _a):
return ([], None)
snapshot = load_lease_snapshot(
fetch_prs=fetch_prs,
fetch_issues=fetch_issues,
fetch_comments=lambda *_a, **_k: [],
issue_lock_path=str(Path("/nonexistent/lock.json")),
)
data = snapshot_to_dict(snapshot)
self.assertIn("claim_inventory", data)
self.assertIn("collision_history", data)
class TestLeaseRoutes(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
def test_leases_page_renders(self):
response = self.client.get("/leases")
self.assertEqual(response.status_code, 200)
self.assertIn("Leases", response.text)
self.assertIn("Collision warnings", response.text)
self.assertNotIn("child issue", response.text.lower())
def test_api_leases_json(self):
response = self.client.get("/api/leases")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIn("claim_inventory", data)
self.assertIn("duplicate_prs", data)
self.assertIn("reviewer_leases", data)
if __name__ == "__main__":
unittest.main()
+9 -5
View File
@@ -46,10 +46,14 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertEqual(response.status_code, 200)
self.assertIn("Gitea-Tools", response.text)
def test_leases_is_implemented(self):
response = self.client.get("/leases")
self.assertEqual(response.status_code, 200)
self.assertIn("Leases", response.text)
self.assertIn("Collision warnings", response.text)
def test_extra_stub_routes(self):
for path in ("/worktrees", "/leases"):
with self.subTest(path=path):
self.assertEqual(self.client.get(path).status_code, 200)
self.assertEqual(self.client.get("/worktrees").status_code, 200)
def test_post_is_rejected(self):
response = self.client.post("/health")
@@ -62,10 +66,10 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertIn("Live queue", response.text)
def test_nav_links_on_all_pages(self):
for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit"):
for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
with self.subTest(path=path):
text = self.client.get(path).text
for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"):
for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
self.assertIn(f'href="{href}"', text)