ClickHouse/tests/ci/auto_release.py

import argparse
import dataclasses
import json
import os
import sys
from typing import List

from get_robot_token import get_best_robot_token
from github_helper import GitHub
from ci_utils import Shell
from env_helper import GITHUB_REPOSITORY
from report import SUCCESS
from ci_buddy import CIBuddy
from ci_config import CI


def parse_args():
    parser = argparse.ArgumentParser(
        "Checks if enough days elapsed since the last release on each release "
        "branches and do a release in case for green builds."
    )
    parser.add_argument("--token", help="GitHub token, if not set, used from smm")
    parser.add_argument(
        "--post-status",
        action="store_true",
        help="Post release branch statuses",
    )
    parser.add_argument(
        "--post-auto-release-complete",
        action="store_true",
        help="Post autorelease completion status",
    )
    parser.add_argument(
        "--prepare",
        action="store_true",
        help="Prepare autorelease info",
    )
    parser.add_argument(
        "--wf-status",
        type=str,
        default="",
        help="overall workflow status [success|failure]",
    )
    return parser.parse_args(), parser


MAX_NUMBER_OF_COMMITS_TO_CONSIDER_FOR_RELEASE = 5
AUTORELEASE_INFO_FILE = "/tmp/autorelease_info.json"


@dataclasses.dataclass
class ReleaseParams:
    ready: bool
    ci_status: str
    num_patches: int
    release_branch: str
    commit_sha: str
    commits_to_branch_head: int
    latest: bool

    def to_dict(self):
        return dataclasses.asdict(self)


@dataclasses.dataclass
class AutoReleaseInfo:
    releases: List[ReleaseParams]

    def add_release(self, release_params: ReleaseParams) -> None:
        self.releases.append(release_params)

    def dump(self):
        print(f"Dump release info into [{AUTORELEASE_INFO_FILE}]")
        with open(AUTORELEASE_INFO_FILE, "w", encoding="utf-8") as f:
            print(json.dumps(dataclasses.asdict(self), indent=2), file=f)

    @staticmethod
    def from_file() -> "AutoReleaseInfo":
        with open(AUTORELEASE_INFO_FILE, "r", encoding="utf-8") as json_file:
            res = json.load(json_file)
        releases = [ReleaseParams(**release) for release in res["releases"]]
        return AutoReleaseInfo(releases=releases)


def _prepare(token):
    assert len(token) > 10
    os.environ["GH_TOKEN"] = token
    Shell.run("gh auth status", check=True)

    gh = GitHub(token)
    prs = gh.get_release_pulls(GITHUB_REPOSITORY)
    prs.sort(key=lambda x: x.head.ref)
    branch_names = [pr.head.ref for pr in prs]
    print(f"Found release branches [{branch_names}]")

    repo = gh.get_repo(GITHUB_REPOSITORY)
    autoRelease_info = AutoReleaseInfo(releases=[])

    for pr in prs:
        print(f"\nChecking PR [{pr.head.ref}]")

        refs = list(repo.get_git_matching_refs(f"tags/v{pr.head.ref}"))
        assert refs

        refs.sort(key=lambda ref: ref.ref)
        latest_release_tag_ref = refs[-1]
        latest_release_tag = repo.get_git_tag(latest_release_tag_ref.object.sha)

        commits = Shell.run(
            f"git rev-list --first-parent {latest_release_tag.tag}..origin/{pr.head.ref}",
            check=True,
        ).split("\n")
        commit_num = len(commits)
        print(
            f"Previous release [{latest_release_tag.tag}] was [{commit_num}] commits ago, date [{latest_release_tag.tagger.date}]"
        )

        commits_to_check = commits[:-1]  # Exclude the version bump commit
        commit_sha = ""
        commit_ci_status = ""
        commits_to_branch_head = 0

        for idx, commit in enumerate(
            commits_to_check[:MAX_NUMBER_OF_COMMITS_TO_CONSIDER_FOR_RELEASE]
        ):
            print(
                f"Check commit [{commit}] [{pr.head.ref}~{idx+1}] as release candidate"
            )
            commit_num -= 1

            is_completed = CI.GHActions.check_wf_completed(
                token=token, commit_sha=commit
            )
            if not is_completed:
                print(f"CI is in progress for [{commit}] - check previous commit")
                commits_to_branch_head += 1
                continue

            commit_ci_status = CI.GHActions.get_commit_status_by_name(
                token=token,
                commit_sha=commit,
                status_name=(CI.JobNames.BUILD_CHECK, "ClickHouse build check"),
            )
            commit_sha = commit
            if commit_ci_status == SUCCESS:
                break

            print(f"CI status [{commit_ci_status}] - skip")
            commits_to_branch_head += 1

        ready = False
        if commit_ci_status == SUCCESS and commit_sha:
            print(
                f"Add release ready info for commit [{commit_sha}] and release branch [{pr.head.ref}]"
            )
            ready = True
        else:
            print(f"WARNING: No ready commits found for release branch [{pr.head.ref}]")

        autoRelease_info.add_release(
            ReleaseParams(
                release_branch=pr.head.ref,
                commit_sha=commit_sha,
                ready=ready,
                ci_status=commit_ci_status,
                num_patches=commit_num,
                commits_to_branch_head=commits_to_branch_head,
                latest=False,
            )
        )

    if autoRelease_info.releases:
        autoRelease_info.releases[-1].latest = True

    autoRelease_info.dump()


def main():
    args, parser = parse_args()

    if args.post_status:
        info = AutoReleaseInfo.from_file()
        for release_info in info.releases:
            if release_info.ready:
                CIBuddy(dry_run=False).post_info(
                    title=f"Auto Release Status for {release_info.release_branch}",
                    body=release_info.to_dict(),
                )
            else:
                CIBuddy(dry_run=False).post_warning(
                    title=f"Auto Release Status for {release_info.release_branch}",
                    body=release_info.to_dict(),
                )
    elif args.post_auto_release_complete:
        assert args.wf_status, "--wf-status Required with --post-auto-release-complete"
        if args.wf_status != SUCCESS:
            CIBuddy(dry_run=False).post_job_error(
                error_description="Autorelease workflow failed",
                job_name="Autorelease",
                with_instance_info=False,
                with_wf_link=True,
                critical=True,
            )
        else:
            CIBuddy(dry_run=False).post_info(
                title=f"Autorelease completed",
                body="",
                with_wf_link=True,
            )
    elif args.prepare:
        _prepare(token=args.token or get_best_robot_token())
    else:
        parser.print_help()
        sys.exit(2)


if __name__ == "__main__":
    main()