sysadminandClaude Opus 4.8
211890f361
Add a Phase 3 read-only console that resolves issue↔PR linkage with evidence (closes keyword, branch marker, body mention), surfaces the latest canonical handoff for a focused thread, and deep-links to Gitea only under the admin reveal opt-in. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
512 lines
20 KiB
Python
512 lines
20 KiB
Python
"""Tests for the Gitea issue↔PR linkage console (#645, Phase 3).
|
|
|
|
Covers the acceptance criteria of the issue:
|
|
|
|
* AC1 — issue↔PR linkage is visible for the selected project/repo, in both
|
|
directions, with the evidence that produced each edge.
|
|
* AC2 — the latest canonical handoff (CTH) is summarized for a focused thread.
|
|
* AC3 — an external Gitea link appears only under the admin reveal opt-in.
|
|
* AC4 — every case is driven by mocked Gitea payloads; no network.
|
|
|
|
Plus the invariants this console must not violate: a partial or failed read is
|
|
never rendered as "no link exists", an unfetched thread is never rendered as
|
|
"no handoff", redaction happens before display, and the surface stays read-only.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from tests.webui_testclient import TestClient
|
|
|
|
from canonical_thread_handoff import format_cth_body
|
|
from webui.app import create_app
|
|
from webui.linkage_loader import (
|
|
EVIDENCE_BRANCH,
|
|
EVIDENCE_CLOSES,
|
|
EVIDENCE_REFERENCE,
|
|
HANDOFF_LOADED,
|
|
HANDOFF_NOT_LOADED,
|
|
HANDOFF_UNAVAILABLE,
|
|
LinkageSnapshot,
|
|
load_linkage_snapshot,
|
|
resolve_linkage,
|
|
resolve_pr_links,
|
|
snapshot_to_dict,
|
|
summarize_handoff,
|
|
)
|
|
from webui.linkage_views import render_linkage_page
|
|
from webui.nav import nav_hrefs
|
|
from webui.queue_loader import PaginationMeta
|
|
|
|
|
|
def _pagination(*, complete: bool = True, count: int = 0) -> PaginationMeta:
|
|
return PaginationMeta(
|
|
page=1,
|
|
per_page=50,
|
|
returned_count=count,
|
|
has_more=not complete,
|
|
is_final_page=complete,
|
|
inventory_complete=complete,
|
|
pages_fetched=1,
|
|
)
|
|
|
|
|
|
def _pr(
|
|
number: int,
|
|
*,
|
|
title: str = "",
|
|
body: str = "",
|
|
head: str = "",
|
|
state: str = "open",
|
|
labels: tuple[str, ...] = (),
|
|
) -> dict:
|
|
return {
|
|
"number": number,
|
|
"title": title or f"pr {number}",
|
|
"body": body,
|
|
"state": state,
|
|
"head": {"ref": head},
|
|
"labels": [{"name": name} for name in labels],
|
|
}
|
|
|
|
|
|
def _issue(
|
|
number: int,
|
|
*,
|
|
title: str = "",
|
|
state: str = "open",
|
|
labels: tuple[str, ...] = (),
|
|
) -> dict:
|
|
return {
|
|
"number": number,
|
|
"title": title or f"issue {number}",
|
|
"state": state,
|
|
"labels": [{"name": name} for name in labels],
|
|
}
|
|
|
|
|
|
def _fetcher(items: list[dict], *, complete: bool = True):
|
|
def _fetch(*_args, **_kwargs):
|
|
return items, _pagination(complete=complete, count=len(items))
|
|
|
|
return _fetch
|
|
|
|
|
|
def _load(
|
|
issues: list[dict],
|
|
prs: list[dict],
|
|
*,
|
|
complete: bool = True,
|
|
**kwargs,
|
|
) -> LinkageSnapshot:
|
|
return load_linkage_snapshot(
|
|
fetch_prs=_fetcher(prs, complete=complete),
|
|
fetch_issues=_fetcher(issues, complete=complete),
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
def _cth(comment_id: int, *, created_at: str, status: str, next_owner: str) -> dict:
|
|
return {
|
|
"id": comment_id,
|
|
"created_at": created_at,
|
|
"user": {"login": "jcwalker3"},
|
|
"body": format_cth_body(
|
|
cth_type="Author Handoff",
|
|
status=status,
|
|
next_owner=next_owner,
|
|
current_blocker="none",
|
|
decision="implemented",
|
|
proof="full suite green",
|
|
next_action="review PR",
|
|
ready_to_paste_prompt="Review PR #902 now.",
|
|
),
|
|
}
|
|
|
|
|
|
class TestLinkageEvidence(unittest.TestCase):
|
|
"""AC1 — every edge records how it was found, and keeps all candidates."""
|
|
|
|
def test_closes_keyword_in_body_is_strongest_evidence(self):
|
|
links = resolve_pr_links(_pr(902, body="Closes #643"))
|
|
self.assertEqual([link.issue_number for link in links], [643])
|
|
self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
|
|
self.assertTrue(links[0].closes)
|
|
|
|
def test_closes_keyword_in_title_counts(self):
|
|
links = resolve_pr_links(_pr(902, title="feat(webui): preview (Closes #643)"))
|
|
self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
|
|
|
|
def test_canonical_branch_marker_links_without_a_keyword(self):
|
|
links = resolve_pr_links(_pr(902, head="feat/issue-643-request-preview"))
|
|
self.assertEqual([link.issue_number for link in links], [643])
|
|
self.assertEqual(links[0].evidence, (EVIDENCE_BRANCH,))
|
|
self.assertFalse(links[0].closes)
|
|
|
|
def test_non_canonical_branch_is_not_treated_as_a_marker(self):
|
|
self.assertEqual(resolve_pr_links(_pr(902, head="issue-643-preview")), ())
|
|
|
|
def test_bare_mention_is_recorded_as_the_weakest_evidence(self):
|
|
links = resolve_pr_links(_pr(902, body="context in #643"))
|
|
self.assertEqual(links[0].evidence, (EVIDENCE_REFERENCE,))
|
|
self.assertFalse(links[0].closes)
|
|
|
|
def test_several_evidence_kinds_merge_onto_one_edge(self):
|
|
links = resolve_pr_links(
|
|
_pr(902, body="Closes #643 — see #643", head="feat/issue-643-preview")
|
|
)
|
|
self.assertEqual(len(links), 1)
|
|
self.assertEqual(
|
|
links[0].evidence,
|
|
(EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE),
|
|
)
|
|
|
|
def test_stronger_evidence_sorts_first(self):
|
|
links = resolve_pr_links(_pr(902, body="Closes #643, related #700"))
|
|
self.assertEqual([link.issue_number for link in links], [643, 700])
|
|
|
|
def test_self_reference_is_not_linkage(self):
|
|
links = resolve_pr_links(_pr(902, body="supersedes #902"))
|
|
self.assertEqual(links, ())
|
|
|
|
def test_every_candidate_is_kept_never_collapsed_to_a_guess(self):
|
|
links = resolve_pr_links(_pr(902, body="Closes #643\nCloses #644"))
|
|
self.assertEqual([link.issue_number for link in links], [643, 644])
|
|
|
|
|
|
class TestLinkageIndex(unittest.TestCase):
|
|
def test_issue_direction_ignores_mention_only_edges(self):
|
|
index = resolve_linkage([_pr(902, body="context in #643")])
|
|
self.assertIsNone(index.issue_prs.get(643))
|
|
self.assertEqual(index.pr_links[902][0].evidence, (EVIDENCE_REFERENCE,))
|
|
|
|
def test_contested_issue_is_reported_when_two_prs_claim_it(self):
|
|
index = resolve_linkage(
|
|
[_pr(902, body="Closes #643"), _pr(903, head="feat/issue-643-again")]
|
|
)
|
|
self.assertEqual(index.contested_issues(), (643,))
|
|
self.assertEqual(index.issue_prs[643], (902, 903))
|
|
|
|
def test_single_claim_is_not_contested(self):
|
|
index = resolve_linkage([_pr(902, body="Closes #643")])
|
|
self.assertEqual(index.contested_issues(), ())
|
|
|
|
def test_ambiguous_when_two_issues_tie_at_the_strongest_evidence(self):
|
|
index = resolve_linkage([_pr(902, body="Closes #643\nCloses #644")])
|
|
self.assertTrue(index.ambiguous(902))
|
|
|
|
def test_weaker_candidate_alongside_a_stronger_one_is_not_ambiguous(self):
|
|
index = resolve_linkage([_pr(902, body="Closes #643, see #700")])
|
|
self.assertFalse(index.ambiguous(902))
|
|
self.assertEqual(index.primary_issue(902).issue_number, 643)
|
|
|
|
def test_malformed_pr_row_is_skipped_not_raised_on(self):
|
|
index = resolve_linkage([{"title": "no number"}, _pr(902, body="Closes #643")])
|
|
self.assertEqual(sorted(index.pr_links), [902])
|
|
|
|
|
|
class TestLinkageSnapshot(unittest.TestCase):
|
|
"""AC1 — linkage is visible per project/repo, in both directions."""
|
|
|
|
def test_both_directions_are_populated(self):
|
|
snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
|
|
self.assertTrue(snapshot.ok)
|
|
self.assertEqual([node.number for node in snapshot.issues], [643])
|
|
self.assertEqual(snapshot.issues[0].linked_prs, (902,))
|
|
self.assertEqual(snapshot.prs[0].links[0].issue_number, 643)
|
|
|
|
def test_repo_scope_comes_from_the_registry_project(self):
|
|
snapshot = _load([], [])
|
|
self.assertIn("/", snapshot.repo_label)
|
|
self.assertTrue(snapshot.project_id)
|
|
|
|
def test_unknown_project_fails_closed_with_a_reason(self):
|
|
snapshot = _load([_issue(643)], [], project_id="no-such-project")
|
|
self.assertFalse(snapshot.ok)
|
|
self.assertIn("not found in registry", snapshot.fetch_error)
|
|
self.assertEqual(snapshot.issues, ())
|
|
|
|
def test_orphan_pr_is_identifiable(self):
|
|
snapshot = _load([], [_pr(902), _pr(903, body="Closes #643")])
|
|
self.assertEqual([node.number for node in snapshot.orphan_prs], [902])
|
|
|
|
def test_state_scope_defaults_to_open_and_is_reported(self):
|
|
self.assertEqual(_load([], []).state_scope, "open")
|
|
self.assertEqual(_load([], [], state="all").state_scope, "all")
|
|
|
|
def test_unsupported_state_falls_back_to_open(self):
|
|
self.assertEqual(_load([], [], state="../etc").state_scope, "open")
|
|
|
|
def test_state_is_passed_through_to_the_fetchers(self):
|
|
seen: list[str] = []
|
|
|
|
def _fetch(*_args, **kwargs):
|
|
seen.append(kwargs.get("state", ""))
|
|
return [], _pagination()
|
|
|
|
load_linkage_snapshot(state="all", fetch_prs=_fetch, fetch_issues=_fetch)
|
|
self.assertEqual(seen, ["all", "all"])
|
|
|
|
|
|
class TestPartialInventoryIsNotAnAbsenceClaim(unittest.TestCase):
|
|
"""An empty edge list from a partial read must never read as 'no link'."""
|
|
|
|
def test_incomplete_pagination_marks_links_non_authoritative(self):
|
|
snapshot = _load([_issue(643)], [], complete=False)
|
|
self.assertFalse(snapshot.inventory_complete)
|
|
self.assertFalse(snapshot.issues[0].links_authoritative)
|
|
|
|
def test_complete_pagination_marks_links_authoritative(self):
|
|
snapshot = _load([_issue(643)], [], complete=True)
|
|
self.assertTrue(snapshot.inventory_complete)
|
|
self.assertTrue(snapshot.issues[0].links_authoritative)
|
|
|
|
def test_partial_window_renders_a_qualified_empty_cell(self):
|
|
html = render_linkage_page(_load([_issue(643)], [], complete=False))
|
|
self.assertIn("none found (partial inventory)", html)
|
|
|
|
def test_complete_window_renders_a_plain_none(self):
|
|
html = render_linkage_page(_load([_issue(643)], [], complete=True))
|
|
self.assertNotIn("partial inventory", html)
|
|
self.assertIn(">none<", html)
|
|
|
|
def test_missing_credentials_fail_closed_without_a_table(self):
|
|
with mock.patch(
|
|
"webui.linkage_loader._offline_test_mode", return_value=False
|
|
), mock.patch("webui.linkage_loader.get_auth_header", return_value=""):
|
|
snapshot = load_linkage_snapshot()
|
|
self.assertFalse(snapshot.ok)
|
|
self.assertIn("credentials unavailable", snapshot.fetch_error)
|
|
html = render_linkage_page(snapshot)
|
|
self.assertIn("Linkage unavailable", html)
|
|
self.assertNotIn("Issues → pull requests", html)
|
|
|
|
def test_fetch_failure_is_reported_not_raised(self):
|
|
def _boom(*_args, **_kwargs):
|
|
raise RuntimeError("gitea 502")
|
|
|
|
snapshot = load_linkage_snapshot(fetch_prs=_boom, fetch_issues=_boom)
|
|
self.assertFalse(snapshot.ok)
|
|
self.assertIn("Gitea fetch failed", snapshot.fetch_error)
|
|
|
|
|
|
class TestHandoffSummary(unittest.TestCase):
|
|
"""AC2 — the latest canonical handoff is summarized for a focused thread."""
|
|
|
|
def test_latest_cth_wins(self):
|
|
summary = summarize_handoff([
|
|
_cth(1, created_at="2026-07-24T10:00:00Z", status="in progress",
|
|
next_owner="author"),
|
|
_cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
|
|
next_owner="reviewer"),
|
|
])
|
|
self.assertEqual(summary.comment_id, 2)
|
|
self.assertEqual(summary.status, "PR-open")
|
|
self.assertEqual(summary.next_owner, "reviewer")
|
|
self.assertTrue(summary.cth_type_known)
|
|
|
|
def test_thread_without_a_cth_summarizes_to_none(self):
|
|
self.assertIsNone(summarize_handoff([{"id": 1, "body": "ordinary comment"}]))
|
|
|
|
def test_unknown_heading_is_reported_not_republished(self):
|
|
summary = summarize_handoff([
|
|
{
|
|
"id": 5,
|
|
"created_at": "2026-07-25T10:00:00Z",
|
|
"user": {"login": "someone"},
|
|
"body": "<!-- cth:v1 -->\n## CTH: Totally Made Up\n\nStatus: odd\n",
|
|
}
|
|
])
|
|
self.assertFalse(summary.cth_type_known)
|
|
self.assertEqual(summary.cth_type, "unrecognized")
|
|
self.assertNotIn("Totally Made Up", json.dumps(summary.to_dict()))
|
|
|
|
def test_focused_pr_loads_its_handoff(self):
|
|
snapshot = _load(
|
|
[_issue(643)],
|
|
[_pr(902, body="Closes #643")],
|
|
pr=902,
|
|
comment_source=lambda kind, number: [
|
|
_cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
|
|
next_owner="reviewer")
|
|
],
|
|
)
|
|
self.assertEqual(snapshot.handoff_status.state, HANDOFF_LOADED)
|
|
self.assertEqual(snapshot.focus, ("pr", 902))
|
|
self.assertEqual(snapshot.prs[0].handoff.status, "PR-open")
|
|
|
|
def test_unfocused_rows_report_not_loaded_never_none(self):
|
|
snapshot = _load(
|
|
[_issue(643)],
|
|
[_pr(902, body="Closes #643"), _pr(903)],
|
|
pr=902,
|
|
comment_source=lambda kind, number: [],
|
|
)
|
|
other = next(node for node in snapshot.prs if node.number == 903)
|
|
self.assertIsNone(other.handoff)
|
|
self.assertEqual(other.handoff_status.state, HANDOFF_NOT_LOADED)
|
|
self.assertIn("not loaded", render_linkage_page(snapshot))
|
|
|
|
def test_no_focus_means_no_thread_is_claimed_handoff_free(self):
|
|
snapshot = _load([_issue(643)], [])
|
|
self.assertEqual(snapshot.handoff_status.state, HANDOFF_NOT_LOADED)
|
|
self.assertIn("thread-scoped", snapshot.handoff_status.reason)
|
|
|
|
def test_comment_source_failure_degrades_only_the_handoff(self):
|
|
def _boom(_kind, _number):
|
|
raise RuntimeError("comments 500")
|
|
|
|
snapshot = _load(
|
|
[_issue(643)], [_pr(902, body="Closes #643")], pr=902, comment_source=_boom
|
|
)
|
|
self.assertTrue(snapshot.ok)
|
|
self.assertEqual(snapshot.handoff_status.state, HANDOFF_UNAVAILABLE)
|
|
self.assertEqual(snapshot.issues[0].linked_prs, (902,))
|
|
self.assertIn("unavailable", render_linkage_page(snapshot))
|
|
|
|
def test_loaded_thread_with_no_cth_says_so_explicitly(self):
|
|
snapshot = _load(
|
|
[_issue(643)],
|
|
[_pr(902, body="Closes #643")],
|
|
pr=902,
|
|
comment_source=lambda kind, number: [{"id": 1, "body": "hi"}],
|
|
)
|
|
self.assertIn(
|
|
"no Canonical Thread Handoff comment found", render_linkage_page(snapshot)
|
|
)
|
|
|
|
|
|
class TestDeepLinks(unittest.TestCase):
|
|
"""AC3 — an external Gitea link is emitted only when permitted."""
|
|
|
|
def test_deep_links_are_withheld_by_default(self):
|
|
with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": ""}):
|
|
snapshot = _load([_issue(643)], [])
|
|
html = render_linkage_page(snapshot)
|
|
self.assertFalse(snapshot.deep_links_enabled)
|
|
self.assertIsNone(snapshot.issues[0].deep_link)
|
|
self.assertIn("Gitea deep links are withheld", html)
|
|
|
|
def test_reveal_opt_in_emits_the_link(self):
|
|
with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}):
|
|
snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
|
|
html = render_linkage_page(snapshot)
|
|
self.assertTrue(snapshot.deep_links_enabled)
|
|
self.assertIn("/issues/643", snapshot.issues[0].deep_link)
|
|
self.assertIn("/pulls/902", snapshot.prs[0].deep_link)
|
|
self.assertIn(f'href="{snapshot.issues[0].deep_link}"', html)
|
|
|
|
|
|
class TestRedactionBoundary(unittest.TestCase):
|
|
def test_secret_shaped_title_is_redacted_before_display(self):
|
|
snapshot = _load(
|
|
[_issue(643, title="token=ghp_thisisnotarealsecretvalue0001")], []
|
|
)
|
|
payload = json.dumps(snapshot_to_dict(snapshot))
|
|
self.assertNotIn("ghp_thisisnotarealsecretvalue0001", payload)
|
|
self.assertNotIn(
|
|
"ghp_thisisnotarealsecretvalue0001", render_linkage_page(snapshot)
|
|
)
|
|
|
|
def test_handoff_fields_are_redacted(self):
|
|
comment = _cth(
|
|
2, created_at="2026-07-25T10:00:00Z", status="ok", next_owner="reviewer"
|
|
)
|
|
comment["body"] += "\nDecision: password=hunter2hunter2\n"
|
|
snapshot = _load(
|
|
[_issue(643)],
|
|
[_pr(902, body="Closes #643")],
|
|
pr=902,
|
|
comment_source=lambda kind, number: [comment],
|
|
)
|
|
self.assertNotIn("hunter2hunter2", json.dumps(snapshot_to_dict(snapshot)))
|
|
self.assertNotIn("hunter2hunter2", render_linkage_page(snapshot))
|
|
|
|
def test_html_escapes_markup_in_a_title(self):
|
|
snapshot = _load([_issue(643, title="<script>alert(1)</script>")], [])
|
|
html = render_linkage_page(snapshot)
|
|
self.assertNotIn("<script>alert(1)</script>", html)
|
|
self.assertIn("<script>", html)
|
|
|
|
|
|
class TestLinkageRoutes(unittest.TestCase):
|
|
def setUp(self):
|
|
self.snapshot = _load(
|
|
[_issue(643, labels=("status:ready",))],
|
|
[_pr(902, body="Closes #643", labels=("status:pr-open",))],
|
|
)
|
|
self.client = TestClient(create_app())
|
|
|
|
def test_page_renders_both_tables(self):
|
|
with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
|
|
response = self.client.get("/gitea")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("Issues → pull requests", response.text)
|
|
self.assertIn("Pull requests → issues", response.text)
|
|
self.assertIn("#643", response.text)
|
|
|
|
def test_api_exports_the_same_model(self):
|
|
with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
|
|
response = self.client.get("/api/v1/gitea/linkage")
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertTrue(payload["ok"])
|
|
self.assertEqual(payload["issues"][0]["linked_prs"], [902])
|
|
self.assertEqual(payload["prs"][0]["links"][0]["issue_number"], 643)
|
|
self.assertEqual(payload["schema_version"], 1)
|
|
|
|
def test_api_declares_the_evidence_vocabulary(self):
|
|
with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
|
|
payload = self.client.get("/api/v1/gitea/linkage").json()
|
|
names = {entry["name"] for entry in payload["evidence_kinds"]}
|
|
self.assertEqual(names, {EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE})
|
|
|
|
def test_api_fails_closed_with_a_non_200_when_the_read_failed(self):
|
|
failed = _load([], [], project_id="no-such-project")
|
|
with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
|
|
response = self.client.get("/api/v1/gitea/linkage")
|
|
self.assertEqual(response.status_code, 502)
|
|
self.assertFalse(response.json()["ok"])
|
|
|
|
def test_page_still_renders_when_the_read_failed(self):
|
|
failed = _load([], [], project_id="no-such-project")
|
|
with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
|
|
response = self.client.get("/gitea")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("Linkage unavailable", response.text)
|
|
|
|
def test_query_parameters_reach_the_loader(self):
|
|
with mock.patch(
|
|
"webui.app.load_linkage_snapshot", return_value=self.snapshot
|
|
) as loader:
|
|
self.client.get("/gitea?project=gitea-tools&state=all&pr=902")
|
|
loader.assert_called_once()
|
|
args, kwargs = loader.call_args
|
|
self.assertEqual(args[0], "gitea-tools")
|
|
self.assertEqual(kwargs["state"], "all")
|
|
self.assertEqual(kwargs["pr"], 902)
|
|
self.assertIsNone(kwargs["issue"])
|
|
|
|
def test_surface_stays_read_only(self):
|
|
for path in ("/gitea", "/api/v1/gitea/linkage"):
|
|
with self.subTest(path=path):
|
|
self.assertEqual(self.client.post(path).status_code, 405)
|
|
|
|
def test_nav_exposes_the_linkage_page_as_live(self):
|
|
self.assertIn("/gitea", nav_hrefs())
|
|
home = self.client.get("/").text
|
|
self.assertIn('href="/gitea"', home)
|
|
self.assertIn(">Gitea<", home)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|