141 lines
4.9 KiB
Python
141 lines
4.9 KiB
Python
import requests
|
|
from pydantic import BaseModel
|
|
from rich.console import Console
|
|
|
|
from typing import Any
|
|
|
|
from nixprstatus.output import OutputFormat
|
|
|
|
DEFAULT_HEADERS = {
|
|
"Accept": "application/vnd.github.text+json",
|
|
}
|
|
DEFAULT_BRANCHES = ["master", "nixos-unstable-small", "nixos-unstable", "nixos-25.05"]
|
|
STABLE_BRANCH = "nixos-25.05"
|
|
BACKPORT_LABEL = "backport release-25.05"
|
|
|
|
|
|
class PRStatus(BaseModel):
|
|
title: str
|
|
merged: bool
|
|
branches: dict[str, bool]
|
|
|
|
def print(self, format: OutputFormat = OutputFormat.CONSOLE) -> None:
|
|
match format:
|
|
case OutputFormat.JSON:
|
|
print(self.model_dump_json())
|
|
case OutputFormat.CONSOLE:
|
|
console = Console(highlight=False)
|
|
console.print(f"{self.title}\n")
|
|
merged = ":white_check_mark: merged" if self.merged else ":x: merged"
|
|
console.print(merged)
|
|
|
|
for branch in self.branches:
|
|
output = (
|
|
f":white_check_mark: {branch}"
|
|
if self.branches[branch]
|
|
else f":x: {branch}"
|
|
)
|
|
console.print(output)
|
|
case _:
|
|
raise ValueError(f"Unknown format: {format}")
|
|
|
|
|
|
def commit_in_branch(commit_sha: str, branch: str) -> bool:
|
|
url = f"https://api.github.com/repos/NixOS/nixpkgs/compare/{branch}...{commit_sha}"
|
|
commit_response = requests.get(url, headers=DEFAULT_HEADERS)
|
|
commit_response.raise_for_status()
|
|
status = commit_response.json().get("status")
|
|
|
|
if status in ["identical", "behind"]:
|
|
return True
|
|
return False
|
|
|
|
|
|
def commits_since(first_ref: str, last_ref: str) -> int:
|
|
url = f"https://api.github.com/repos/NixOS/nixpkgs/compare/{first_ref}...{last_ref}"
|
|
commit_response = requests.get(url, headers=DEFAULT_HEADERS)
|
|
commit_response.raise_for_status()
|
|
|
|
behind_by = commit_response.json()["behind_by"]
|
|
if not isinstance(behind_by, int):
|
|
raise TypeError("behind_by unexpected type")
|
|
return behind_by
|
|
|
|
|
|
def get_pr(pr: int) -> dict[str, Any]:
|
|
url = f"https://api.github.com/repos/NixOS/nixpkgs/pulls/{pr}"
|
|
pr_response = requests.get(url, headers=DEFAULT_HEADERS)
|
|
pr_response.raise_for_status()
|
|
|
|
resp = pr_response.json()
|
|
if not isinstance(resp, dict):
|
|
raise TypeError("PR response has unexpected type")
|
|
return resp
|
|
|
|
|
|
def pr_merge_status(
|
|
pr: int, branches: list[str] | None = DEFAULT_BRANCHES, check_backport: bool = True
|
|
) -> PRStatus:
|
|
if not branches:
|
|
branches = DEFAULT_BRANCHES
|
|
url = f"https://api.github.com/repos/NixOS/nixpkgs/pulls/{pr}"
|
|
pr_response = requests.get(url, headers=DEFAULT_HEADERS)
|
|
pr_response.raise_for_status()
|
|
|
|
pr_data = pr_response.json()
|
|
title = pr_data["title"]
|
|
|
|
merged = pr_data["merged"]
|
|
if merged is False:
|
|
return PRStatus(
|
|
title=title, merged=False, branches={branch: False for branch in branches}
|
|
)
|
|
|
|
commit_sha = pr_data.get("merge_commit_sha")
|
|
|
|
# Check for backport label
|
|
has_backport_label = BACKPORT_LABEL in [
|
|
label.get("name") for label in pr_data["labels"]
|
|
]
|
|
|
|
results = {}
|
|
|
|
# Check if base branch is in our list, if it is
|
|
# no need to call commit_in_branch
|
|
merge_base_branch = pr_data.get("base", {}).get("ref")
|
|
if merge_base_branch in branches:
|
|
results[merge_base_branch] = True
|
|
branches.remove(merge_base_branch)
|
|
|
|
for branch in branches:
|
|
in_branch = commit_in_branch(commit_sha, branch)
|
|
results[branch] = in_branch
|
|
|
|
if check_backport and has_backport_label and STABLE_BRANCH in branches:
|
|
# Check comments for message about backport
|
|
comment_url = f"https://api.github.com/repos/NixOS/nixpkgs/issues/{pr}/comments"
|
|
comment_response = requests.get(comment_url, headers=DEFAULT_HEADERS)
|
|
comment_response.raise_for_status()
|
|
|
|
for comment in comment_response.json():
|
|
body = comment.get("body_text", "")
|
|
if "Successfully created backport PR" in body:
|
|
backport_pr = body.split("\n")[-1].replace("#", "")
|
|
|
|
# Check if backport pr has been merged
|
|
backport_url = (
|
|
f"https://api.github.com/repos/NixOS/nixpkgs/pulls/{backport_pr}"
|
|
)
|
|
backport_response = requests.get(backport_url, headers=DEFAULT_HEADERS)
|
|
backport_sha = backport_response.json().get("merge_commit_sha")
|
|
if backport_sha is None:
|
|
results[f"{STABLE_BRANCH} (#{backport_pr})"] = False
|
|
return PRStatus(title=title, merged=True, branches=results)
|
|
|
|
results.pop(STABLE_BRANCH)
|
|
results[f"{STABLE_BRANCH} (#{backport_pr})"] = commit_in_branch(
|
|
backport_sha, STABLE_BRANCH
|
|
)
|
|
|
|
return PRStatus(title=title, merged=True, branches=results)
|