clan-infra/pkgs/clan-merge/clan_merge/__init__.py

142 lines
4.1 KiB
Python
Raw Normal View History

2023-07-26 09:17:26 +00:00
import argparse
import json
import urllib.request
from os import environ
from typing import Optional
def load_token() -> str:
GITEA_TOKEN_FILE = environ.get("GITEA_TOKEN_FILE", default=None)
assert GITEA_TOKEN_FILE is not None
with open(GITEA_TOKEN_FILE, "r") as f:
return f.read().strip()
2023-08-07 08:40:37 +00:00
def pr_message(pr: dict) -> str:
return f"PR {pr['number']} in repo {pr['base']['repo']['name']} from user {pr['user']['login']}: {pr['title']}"
2023-07-26 09:17:26 +00:00
def is_ci_green(pr: dict) -> bool:
2023-08-07 08:40:37 +00:00
print(f"Checking CI status for {pr_message(pr)}")
repo = pr["base"]["repo"]["name"]
2023-07-26 09:17:26 +00:00
url = (
"https://git.clan.lol/api/v1/repos/clan/"
+ repo
2023-07-26 09:17:26 +00:00
+ "/commits/"
+ pr["head"]["sha"]
+ "/status"
)
response = urllib.request.urlopen(url)
data = json.loads(response.read())
# check for all commit statuses to have status "success"
if not data["statuses"]:
2023-08-07 08:40:37 +00:00
print(f"No CI status for {pr_message(pr)}")
return False
2023-07-26 09:17:26 +00:00
for status in data["statuses"]:
if status["status"] != "success":
return False
return True
2023-07-26 12:33:36 +00:00
def decide_merge(pr: dict, allowed_users: list[str], bot_name: str) -> bool:
assignees = pr["assignees"] if pr["assignees"] else []
2023-07-26 09:17:26 +00:00
if (
pr["user"]["login"] in allowed_users
and pr["mergeable"] is True
and not pr["title"].startswith("WIP:")
and pr["state"] == "open"
2023-07-26 12:33:36 +00:00
# check if bot is assigned
and any(assignee["login"] == bot_name for assignee in assignees)
2023-07-26 09:17:26 +00:00
and is_ci_green(pr)
):
return True
return False
# python equivalent for: curl -X 'GET' https://git.clan.lol/api/v1/repos/clan/{repo}/pulls
def list_prs(repo: str) -> list:
url = "https://git.clan.lol/api/v1/repos/clan/" + repo + "/pulls"
response = urllib.request.urlopen(url)
data = json.loads(response.read())
return data
2023-07-26 12:33:36 +00:00
def list_prs_to_merge(prs: list, allowed_users: list[str], bot_name: str) -> list:
2023-07-26 09:17:26 +00:00
prs_to_merge = []
for pr in prs:
2023-07-26 12:33:36 +00:00
if decide_merge(pr, allowed_users, bot_name) is True:
2023-07-26 09:17:26 +00:00
prs_to_merge.append(pr)
return prs_to_merge
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Merge PRs on clan.lol")
# parse a list of allowed users
parser.add_argument(
"--allowed-users",
nargs="+",
help="list of users allowed to merge",
required=True,
)
2023-07-26 12:33:36 +00:00
# option for bot-name
parser.add_argument(
"--bot-name",
help="name of the bot",
required=True,
)
2023-07-26 09:17:26 +00:00
# parse list of repository names for which to merge PRs
parser.add_argument(
"--repos",
nargs="+",
help="list of repositories for which to merge PRs",
required=True,
)
# option for dry run
parser.add_argument(
"--dry-run",
action="store_true",
help="dry run",
)
return parser.parse_args()
def clan_merge(
args: Optional[argparse.Namespace] = None, gitea_token: Optional[str] = None
) -> None:
if gitea_token is None:
gitea_token = load_token()
if args is None:
args = parse_args()
allowed_users = args.allowed_users
repos = args.repos
dry_run = args.dry_run
2023-07-26 12:33:36 +00:00
bot_name = args.bot_name
2023-07-26 09:17:26 +00:00
for repo in repos:
prs = list_prs(repo)
2023-07-26 12:33:36 +00:00
prs_to_merge = list_prs_to_merge(prs, allowed_users, bot_name)
2023-07-26 09:17:26 +00:00
for pr in prs_to_merge:
url = (
"https://git.clan.lol/api/v1/repos/clan/"
+ repo
+ "/pulls/"
+ str(pr["number"])
+ "/merge"
+ f"?token={gitea_token}"
)
if dry_run is True:
2023-08-07 08:40:37 +00:00
print(f"Would merge {pr_message(pr)}")
2023-07-26 09:17:26 +00:00
else:
2023-08-07 08:40:37 +00:00
print(f"Merging {pr_message(pr)}")
2023-07-26 09:17:26 +00:00
data = dict(
Do="merge",
)
data_encoded = json.dumps(data).encode("utf8")
req = urllib.request.Request(
url, data=data_encoded, headers={"Content-Type": "application/json"}
)
urllib.request.urlopen(req)
if __name__ == "__main__":
clan_merge()