# -*- coding: utf-8 -*- import functools import logging import os import re class RepositoryBase(object): def __init__(self, repo_path): import git self._repo = git.Repo(repo_path, search_parent_directories=(not repo_path)) # commit comparator def cmp(x, y): if x == y: return 0 if self._repo.is_ancestor(x, y): return -1 else: return 1 self.comparator = functools.cmp_to_key(cmp) def get_head_commit(self): return self._repo.commit(self._default) def iterate(self, begin, end): rev_range = '{}...{}'.format(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(Repository, self).__init__(repo_path) self._remote = self._repo.remotes[remote_name] self._remote.fetch() self._default = self._remote.refs[default_branch_name] 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(BareRepository, self).__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]))