repo/repo_settings.py

#!/usr/bin/env python3
#
# Copyright (C) 2023-2024 The LeafOS Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import argparse
import json
import os
import requests
import subprocess
import sys
import yaml
from xml.etree import ElementTree

PORT = "29418"
GERRIT = "review.leafos.org"
USER = None
leaf_devices = "leaf/devices/devices.yaml"
gerrit_structure = "leaf/gerrit-config/structure.yaml"


def parse_args():
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="subcommand")
    subparsers.required = True

    # create project
    parser_project = subparsers.add_parser("create_project")
    parser_project.add_argument("-b", "--branch", required=True)
    parser_project.add_argument("-d", "--device")
    parser_project.add_argument(
        "-f", "--project_file", default=".repo/manifests/snippets/leaf.xml"
    )
    parser_project.add_argument("--no-default-branch", action="store_true")

    # create branch
    parser_branch = subparsers.add_parser("create_branch")
    parser_branch.add_argument("-n", "--new_branch", required=True)
    parser_branch.add_argument("-b", "--base_branch", dest="branch", required=True)
    parser_branch.add_argument("-d", "--device", required=True)

    # update_groups
    parser_update_groups = subparsers.add_parser("update_groups")

    # fetch_structure_from_gerrit
    parser_fetch_structure = subparsers.add_parser("fetch_structure_from_gerrit")

    return parser.parse_args()


def check_gh_token():
    gh_token = os.environ.get("GH_TOKEN")
    if not gh_token:
        print("GH_TOKEN is not set!")
        sys.exit(1)
    return gh_token


def get_gh_user(token):
    global USER
    USER = requests.get(
        "https://api.github.com/user",
        headers={
            "Accept": "application/vnd.github+json",
            "Authorization": f"Bearer {token}",
        },
    ).json()["login"]


def create_github_repo(org, repo, token):
    url = f"https://api.github.com/orgs/{org}/repos"
    headers = {
        "Accept": "application/vnd.github+json",
        "Authorization": f"token {token}",
    }
    data = {
        "name": repo,
        "private": False,
        "has_issues": False,
        "has_projects": False,
        "has_wiki": False,
    }
    requests.post(url, headers=headers, json=data)


def set_github_repo_settings(project, branch, token, no_default_branch):
    url = f"https://api.github.com/repos/{project}"
    headers = {
        "Accept": "application/vnd.github+json",
        "Authorization": f"token {token}",
    }
    data = {
        "has_issues": False,
        "has_projects": False,
        "has_wiki": False,
    }
    if not no_default_branch:
        data["default_branch"] = branch
    requests.patch(url, headers=headers, json=data)


def create_gerrit_project(project, branch):
    projects = subprocess.run(
        ["ssh", "-n", "-p", PORT, f"{USER}@{GERRIT}", "gerrit", "ls-projects"],
        stdout=subprocess.PIPE,
        text=True,
    ).stdout.split("\n")
    if project not in projects:
        subprocess.run(
            [
                "ssh",
                "-n",
                "-p",
                PORT,
                f"{USER}@{GERRIT}",
                "gerrit",
                "create-project",
                project,
                "-b",
                branch,
            ],
            check=True,
        )


def create_gerrit_branch(project, new_branch, base_branch):
    try:
        subprocess.run(
            [
                "ssh",
                "-n",
                "-p",
                PORT,
                f"{USER}@{GERRIT}",
                "gerrit",
                "create-branch",
                project,
                new_branch,
                base_branch,
            ],
            check=True,
        )
    except subprocess.CalledProcessError:
        ...


def set_gerrit_project_head(project, branch):
    subprocess.run(
        [
            "ssh",
            "-n",
            "-p",
            PORT,
            f"{USER}@{GERRIT}",
            "gerrit",
            "set-head",
            project,
            "--new-head",
            branch,
        ],
        check=False,
    )


def set_gerrit_project_parent(project, parent):
    subprocess.run(
        [
            "ssh",
            "-n",
            "-p",
            PORT,
            f"{USER}@{GERRIT}",
            "gerrit",
            "set-project-parent",
            project,
            "--parent",
            parent,
        ],
        check=False,
    )


def get_projects_from_devices(device, branch):
    projects = []
    with open(leaf_devices) as f:
        root = yaml.safe_load(f)

    for item in root:
        if "device" in item and "repositories" in item and device in item["device"]:
            for repo_list in item["repositories"]:
                if isinstance(repo_list, list):
                    for repository in repo_list:
                        projects.append(
                            {"name": repository["name"], "revision": branch}
                        )
                else:
                    projects.append({"name": repo_list["name"], "revision": branch})
    return projects


def get_projects_from_manifests(project_file, branch):
    projects = []

    with open(project_file) as xml_file:
        root = ElementTree.parse(xml_file).getroot()

        for project in root.findall("project"):
            name = project.get("name")
            revision = (
                branch
                or project.get("revision")
                or root.find("default").get("revision").split("/")[-1]
            )
            projects.append({"name": name, "revision": revision})

    return projects


def get_projects_from_gerrit_structure():
    with open(gerrit_structure, "r") as f:
        return yaml.safe_load(f)


def get_projects_from_gerrit(auth=None):
    url = f"https://{GERRIT}/a/projects/?t" if auth else f"https://{GERRIT}/projects/?t"
    resp = requests.get(url, auth=auth)
    if resp.status_code != 200:
        raise Exception(f"Error communicating with gerrit: {resp.text}")
    projects = json.loads(resp.text[5:])
    nodes = {}

    for name, project in projects.items():
        nodes[name] = []

    for name, project in projects.items():
        parent = project.get("parent")
        if parent:
            nodes[parent].append(name)
    for project in nodes.keys():
        nodes[project] = sorted(nodes[project])
    return nodes


def main():
    args = parse_args()

    gh_token = check_gh_token()
    get_gh_user(gh_token)

    if args.subcommand in ["create_project", "create_branch"]:
        if args.device:
            projects = get_projects_from_devices(args.device, args.branch)
        else:
            projects = get_projects_from_manifests(args.project_file, args.branch)

        for project in projects:
            name = project["name"]
            if any(
                org in name
                for org in ["LeafOS-Project", "LeafOS-Blobs", "LeafOS-Devices"]
            ):
                print(name)
                org, repo = name.split("/")
                branch = project["revision"]
                if args.subcommand == "create_branch":
                    create_gerrit_branch(name, args.new_branch, args.branch)
                else:
                    create_github_repo(org, repo, gh_token)
                    set_github_repo_settings(
                        name, branch, gh_token, args.no_default_branch
                    )
                    create_gerrit_project(name, branch)
                    if not args.no_default_branch:
                        set_gerrit_project_head(name, branch)
    elif args.subcommand == "update_groups":
        projects = get_projects_from_gerrit_structure()
        live_projects = get_projects_from_gerrit()
        changes = {}

        for parent, children in projects.items():
            if parent in live_projects:
                if not projects[parent] or set(live_projects[parent]) == set(
                    projects[parent]
                ):
                    continue
                else:
                    changes[parent] = list(
                        set(projects[parent]) - set(live_projects[parent])
                    )
                    if not changes[parent]:
                        del changes[parent]
            else:
                changes[parent] = children

        if changes:
            for parent, children in changes.items():
                for child in children:
                    print(f"Update parent of {child} to {parent}")
                    set_gerrit_project_parent(child, parent)
    elif args.subcommand == "fetch_structure_from_gerrit":
        projects = get_projects_from_gerrit()

        for node in sorted(projects.keys()):
            children = sorted(projects[node])
            if children:
                print(f"{node}:")
                for child in projects[node]:
                    print(f"  - {child}")


if __name__ == "__main__":
    main()