tipbot/git.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL2.0
# Copyright Thomas Gleixner <tglx@linutronix.de>

from tipbot.util import FatalException, FetchException
import subprocess
import pygit2
import os

class git_repo(object):
    def __init__(self, repodir, logger):
        self.repodir = os.path.abspath(repodir)
        self.log = logger
        self.repo = pygit2.Repository(self.repodir)

    def fetch(self, remote='origin'):
        try:
            self.repo.remotes[remote].fetch(prune=pygit2.GIT_FETCH_PRUNE)
        except Exception as ex:
            self.log.log_exception(ex, 'Fetch failed\n')
            raise FetchException(ex)

    def set_base_ref(self, ref, sha):
        self.repo.references.create(ref, sha)

    def get_blob(self, ref, path):
        commitid = self.repo.lookup_reference(ref).target
        bentry = self.repo[commitid].tree[path]
        assert(bentry.type == 'blob')
        return self.repo[bentry.id].data.decode()

    def get_list_from_blob(self, ref, path):
        res = []
        for b in self.get_blob(ref, path).split('\n'):
            b = b.strip()
            if len(b) and not b.startswith('#'):
                res.append(b)
        return res

    def get_autobranch_refs(self, branch, basepath):
        bref = 'refs/heads/%s' %branch
        refs = []

        for fname in self.get_list_from_blob(bref, basepath):
            fname =  os.path.join(os.path.dirname(basepath), fname)
            for br in self.get_list_from_blob(bref, fname):
                if br in list(self.repo.branches):
                    refs.append('refs/heads/%s' %br)

        return refs

    def log_revs(self, args):
        '''
        Use git directly as pygit2 log is horribly slow

        Throws CalledProcessError if the return code is not 0
        '''
        oldpath = os.getcwd()
        try:
            os.chdir(self.repodir)
            res = subprocess.run(args, capture_output=True)
            res.check_returncode()
            os.chdir(oldpath)
            return res.stdout.decode().split()

        except Exception as ex:
            self.log.log_exception(ex, 'Log revisions failed\n')
            os.chdir(oldpath)
            raise FatalException

    def log_revs_from(self, base):
        '''
        Retrieve git log SHA1s from base to HEAD
        '''
        args = [ 'git', 'log', '--pretty=%H', '%s..' %base ]
        return self.log_revs(args)

    def log_revs_ref_from(self, base, ref):
        '''
        Retrieve git log SHA1s from base to head of branch
        '''
        args = [ 'git', 'log', '--no-merges', '--pretty=%H',
                 '%s..%s' %(base, ref) ]
        return self.log_revs(args)

    def get_branch_head(self, branch='master'):
        ref = 'refs/heads/%s' %branch
        return self.repo.lookup_reference(ref).target

    def shortlog(self, sha):
        subj = self.repo[sha].message.split('\n')[0]
        return '%s ("%s")' %(sha[:12], subj.strip())