feat: expose actionable pull request checks (Closes #517)
CI / lint (pull_request) Successful in 1m5s
CI / build-release (pull_request) Successful in 5s
CI / release-candidate (pull_request) Has been skipped

This commit is contained in:
timmy committed 2026-08-10 22:31:50 +00:00
1 parent 7555cefccb
commit 9724358c89
8 files changed
+278 -8

No files matched your search

+48 -1
View File
@@ -4,7 +4,7 @@ import re
import shlex
from contextlib import asynccontextmanager
from typing import Any
from urllib.parse import urlsplit
from urllib.parse import urljoin, urlsplit
import httpx
@@ -600,6 +600,51 @@ def _safe_web_url(value: Any) -> str:
return value if parsed.scheme in {"http", "https"} and parsed.netloc else ""
def _safe_gitea_web_url(value: Any) -> str:
if not isinstance(value, str) or not value.strip():
return ""
resolved = urljoin(f"{GITEA_URL}/", value.strip())
parsed = urlsplit(resolved)
configured = urlsplit(GITEA_URL)
base_path = configured.path.rstrip("/")
if (
parsed.scheme not in {"http", "https"}
or parsed.scheme != configured.scheme
or parsed.netloc != configured.netloc
or (base_path and parsed.path != base_path and not parsed.path.startswith(f"{base_path}/"))
):
return ""
return resolved
def _normalize_commit_checks(status: Any) -> list[dict]:
entries = status.get("statuses") if isinstance(status, dict) else None
if not isinstance(entries, list):
return []
rank = {"error": 0, "failure": 0, "pending": 1, "warning": 2, "success": 3}
checks = []
for index, entry in enumerate(entries):
if not isinstance(entry, dict):
continue
name = entry.get("context")
if not isinstance(name, str) or not name.strip():
continue
state = entry.get("status", entry.get("state", "unknown"))
state = state.lower() if isinstance(state, str) else "unknown"
if state not in rank:
state = "unknown"
description = entry.get("description")
checks.append({
"name": name.strip()[:120],
"state": state,
"description": description.strip()[:240] if isinstance(description, str) else "",
"url": _safe_gitea_web_url(entry.get("target_url")),
"_index": index,
})
checks.sort(key=lambda check: (rank.get(check["state"], 2), check["_index"]))
return [{key: value for key, value in check.items() if key != "_index"} for check in checks[:20]]
def _normalize_notifications(threads: Any) -> list[dict]:
if not isinstance(threads, list):
raise ValueError("Gitea notification response was not a list")
@@ -1741,6 +1786,7 @@ async def pull_completion_review(repository: str, number: int) -> dict:
"mergeable": pull.get("mergeable") is True,
"merged": pull.get("merged") is True,
"ci_state": status.get("state", "unknown") if isinstance(status, dict) else "unknown",
"checks": _normalize_commit_checks(status),
"files": [
{
"filename": item.get("filename", ""),
@@ -1873,6 +1919,7 @@ async def pull_review_detail(repository: str, number: int) -> dict:
"author": user.get("login", ""),
"head_sha": sha,
"ci_state": status.get("state", "unknown") if isinstance(status, dict) else "unknown",
"checks": _normalize_commit_checks(status),
"files": normalized_files,
"reviews": normalized_reviews,
}