ClickHouse/tests/ci/cherry_pick_utils/local.py

# -*- coding: utf-8 -*-

import functools
import logging
import os
import re

import git


class RepositoryBase:
    def __init__(self, repo_path):

        self._repo = git.Repo(repo_path, search_parent_directories=(not repo_path))

        # comparator of commits
        def cmp(x, y):
            if str(x) == str(y):
                return 0
            if self._repo.is_ancestor(x, y):
                return -1
            else:
                return 1

        self.comparator = functools.cmp_to_key(cmp)

    def iterate(self, begin, end):
        rev_range = f"{begin}...{end}"
        for commit in self._repo.iter_commits(rev_range, first_parent=True):
            yield commit


class Repository(RepositoryBase):
    def __init__(self, repo_path, remote_name, default_branch_name):
        super().__init__(repo_path)
        self._remote = self._repo.remotes[remote_name]
        self._remote.fetch()
        self._default = self._remote.refs[default_branch_name]

    def get_head_commit(self):
        return self._repo.commit(self._default)

    def get_release_branches(self):
        """
        Returns sorted list of tuples:
         * remote branch (git.refs.remote.RemoteReference),
         * base commit (git.Commit),
         * head (git.Commit)).
        List is sorted by commits in ascending order.
        """
        release_branches = []

        RE_RELEASE_BRANCH_REF = re.compile(r"^refs/remotes/.+/\d+\.\d+$")

        for branch in [
            r for r in self._remote.refs if RE_RELEASE_BRANCH_REF.match(r.path)
        ]:
            base = self._repo.merge_base(self._default, self._repo.commit(branch))
            if not base:
                logging.info(
                    "Branch %s is not based on branch %s. Ignoring.",
                    branch.path,
                    self._default,
                )
            elif len(base) > 1:
                logging.info(
                    "Branch %s has more than one base commit. Ignoring.", branch.path
                )
            else:
                release_branches.append((os.path.basename(branch.name), base[0]))

        return sorted(release_branches, key=lambda x: self.comparator(x[1]))


class BareRepository(RepositoryBase):
    def __init__(self, repo_path, default_branch_name):
        super().__init__(repo_path)
        self._default = self._repo.branches[default_branch_name]

    def get_release_branches(self):
        """
        Returns sorted list of tuples:
         * branch (git.refs.head?),
         * base commit (git.Commit),
         * head (git.Commit)).
        List is sorted by commits in ascending order.
        """
        release_branches = []

        RE_RELEASE_BRANCH_REF = re.compile(r"^refs/heads/\d+\.\d+$")

        for branch in [
            r for r in self._repo.branches if RE_RELEASE_BRANCH_REF.match(r.path)
        ]:
            base = self._repo.merge_base(self._default, self._repo.commit(branch))
            if not base:
                logging.info(
                    "Branch %s is not based on branch %s. Ignoring.",
                    branch.path,
                    self._default,
                )
            elif len(base) > 1:
                logging.info(
                    "Branch %s has more than one base commit. Ignoring.", branch.path
                )
            else:
                release_branches.append((os.path.basename(branch.name), base[0]))

        return sorted(release_branches, key=lambda x: self.comparator(x[1]))