fix(governance): audit merged PRs after branch cleanup
This commit is contained in:
@@ -243,13 +243,25 @@ def validate_pull_request(
|
|||||||
pr: dict[str, Any], task: RemoteTask
|
pr: dict[str, Any], task: RemoteTask
|
||||||
) -> list[AuditFinding]:
|
) -> list[AuditFinding]:
|
||||||
findings: list[AuditFinding] = []
|
findings: list[AuditFinding] = []
|
||||||
if not task.work_branch or pull_request_head(pr) != task.work_branch:
|
body = pr.get("body")
|
||||||
|
body = body if isinstance(body, str) else ""
|
||||||
|
head = pull_request_head(pr)
|
||||||
|
head_matches = bool(task.work_branch) and head == task.work_branch
|
||||||
|
if (
|
||||||
|
not head_matches
|
||||||
|
and bool(pr.get("merged"))
|
||||||
|
and re.fullmatch(r"refs/pull/\d+/head", head)
|
||||||
|
and bool(task.work_branch)
|
||||||
|
and task.work_branch in body
|
||||||
|
):
|
||||||
|
# Gitea rewrites head.ref after a merged source branch is deleted. The
|
||||||
|
# immutable PR body still records the claimed work branch.
|
||||||
|
head_matches = True
|
||||||
|
if not head_matches:
|
||||||
findings.append(
|
findings.append(
|
||||||
AuditFinding(task.number, "pull-request", "PR head 与 claim 工作分支不一致。")
|
AuditFinding(task.number, "pull-request", "PR head 与 claim 工作分支不一致。")
|
||||||
)
|
)
|
||||||
body = pr.get("body")
|
required_values = [f"docs/tasks/{task.task_id}.md", task.work_branch or ""]
|
||||||
body = body if isinstance(body, str) else ""
|
|
||||||
required_values = [f"docs/tasks/{task.task_id}.md"]
|
|
||||||
if task.context_ref:
|
if task.context_ref:
|
||||||
required_values.append(task.context_ref)
|
required_values.append(task.context_ref)
|
||||||
required_values.extend(task.write_paths or [])
|
required_values.extend(task.write_paths or [])
|
||||||
|
|||||||
@@ -23,6 +23,8 @@ from audit_gitea_coordination import (
|
|||||||
parse_datetime,
|
parse_datetime,
|
||||||
parse_write_paths,
|
parse_write_paths,
|
||||||
select_latest_claim,
|
select_latest_claim,
|
||||||
|
RemoteTask,
|
||||||
|
validate_pull_request,
|
||||||
)
|
)
|
||||||
from setup_gitea_labels import LABELS, NoRedirect, build_plan
|
from setup_gitea_labels import LABELS, NoRedirect, build_plan
|
||||||
from test_gitea_claim_race import PROBE_PREFIX, new_probe_branch
|
from test_gitea_claim_race import PROBE_PREFIX, new_probe_branch
|
||||||
@@ -180,6 +182,31 @@ class OfflineRuleTests(unittest.TestCase):
|
|||||||
|
|
||||||
|
|
||||||
class GiteaHelperTests(unittest.TestCase):
|
class GiteaHelperTests(unittest.TestCase):
|
||||||
|
def test_merged_pr_accepts_gitea_deleted_branch_ref(self) -> None:
|
||||||
|
sha = "a" * 40
|
||||||
|
task = RemoteTask(
|
||||||
|
number=1,
|
||||||
|
task_id="T-123",
|
||||||
|
state="closed",
|
||||||
|
status="status/done",
|
||||||
|
labels={"status/done"},
|
||||||
|
body="",
|
||||||
|
work_branch="agent/worker-1/T-123",
|
||||||
|
context_ref=sha,
|
||||||
|
write_paths=["docs/tasks/T-123.md"],
|
||||||
|
)
|
||||||
|
pr = {
|
||||||
|
"merged": True,
|
||||||
|
"head": {"ref": "refs/pull/1/head"},
|
||||||
|
"body": (
|
||||||
|
"Closes #1\n"
|
||||||
|
"task_file: docs/tasks/T-123.md\n"
|
||||||
|
f"context_ref: {sha}\n"
|
||||||
|
"work_branch: agent/worker-1/T-123\n"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
self.assertEqual([], validate_pull_request(pr, task))
|
||||||
|
|
||||||
def test_waiting_status_is_not_claimable_or_active(self) -> None:
|
def test_waiting_status_is_not_claimable_or_active(self) -> None:
|
||||||
self.assertIn("status/waiting", STATUS_LABELS)
|
self.assertIn("status/waiting", STATUS_LABELS)
|
||||||
self.assertNotIn("status/waiting", READY_OR_ACTIVE_LABELS)
|
self.assertNotIn("status/waiting", READY_OR_ACTIVE_LABELS)
|
||||||
|
|||||||
Reference in New Issue
Block a user