path: root/linaropy/git/gitrepo.py

import unittest
import logging
import os

import uuid

from sh import git
from sh import ErrorReturnCode
from sh import ErrorReturnCode_128

import subprocess

from ..proj import Proj
from ..cd import cd

from contextlib import contextmanager

# This is called 'GitRepo' because a repository for a project doesn't only have
# to be a git repository.  Abstract base class used to define the operations
# that both Clone and Workdir share.  Notice that there is no actual repodir
# specified here.  You must instantiate a subclass in order to use/test these
# operations.  This is because 'clones' and 'workdirs' are created differently
# but you operate on them in the exact same way.


class GitRepo(object):

    def __init__(self, proj):
        self.repodir = u""
        if isinstance(proj, Proj):
            self.proj = proj
        else:
            raise TypeError('proj input parameter is not of type Proj')

    @staticmethod
    def is_valid_branch_name(branch):
        """Check if branch is a valid git branch name.

        Validity is determined based on git check-ref-format, but it is a bit
        more permissive, in the sense that branch names without any / are
        considered valid.
        """
        try:
            git("check-ref-format", "--allow-onelevel", branch)
            return True
        except ErrorReturnCode as exc:
            # TODO: maybe distinguish between invalid branch names and other
            # errors that may occur
            return False

    def branch_exists(self, branch):
        logging.info("Checking to see if branch %s exists" % branch)
        with cd(self.repodir):
            try:
                # Quote the branchname because it might have strange
                # characters in it.
                br = "%s" % branch
                git("rev-parse", "--verify", br)
            except ErrorReturnCode_128:
                logging.info("Couldn't find branch %s" % branch)
                return False
            return True

    # See if the specified branch exists as a remote.
    def remote_branch_exists(self, branch):
        return self.branch_exists(self.remote_branchname(branch))

    def getbranch(self):
        try:
            with cd(self.repodir):
                branch = git("rev-parse", "--abbrev-ref",
                             "HEAD").stdout.rstrip()
        except ErrorReturnCode:
            raise EnvironmentError("Unable to get the current branch")
        return branch

    # TODO make this a bit more sophisticated because the user might not be
    # using 'origin' as their remote name.
    def remote_branchname(self, branchname):
        return "remotes/origin/" + branchname

    # TODO: Fully test this.
    # Stash current changes and change to new branch.
    # Return to previous branch when context is exited.
    @contextmanager
    def checkoutbranch(self, branch, track=""):
        with cd(self.repodir):
            # The stashid will remain empty if there were no changes in the
            # directory to stash.
            stashid = u""

            try:
                self.prevbranch = git(
                    "rev-parse", "--abbrev-ref", "HEAD").stdout.rstrip()
            except ErrorReturnCode:
                raise EnvironmentError("Unable to get the current branch")

            try:
                # TODO: do we want to do a blanket git add *?
                stashid = git("stash", "create").stdout.rstrip()
                # If there's no stash then there's no reason to save.
                if stashid != "":
                    git("stash", "save")
            except ErrorReturnCode:
                raise EnvironmentError("Unable to get the current branch")

            try:
                # TODO This is a nop as it is.
                git("rev-parse", "--verify", branch)
                # if the branch exists the previous statement won't cause an
                # exception so we know we can just check it out.
                git("checkout", branch)
            except ErrorReturnCode:
                try:
                    # if it doesn't exist we need to checkout a new branch
                    git("checkout", "-b", branch, track).stdout.rstrip()
                except ErrorReturnCode as exc:
                    raise EnvironmentError(
                        "Unable to checkout -b %s %s" % (branch, track))
        try:
            yield
        finally:
            try:
                with cd(self.repodir):
                    git("checkout", self.prevbranch)
                    # If there's no stashid then there were no files stashed,
                    # so don't stash apply.
                    if stashid != "":
                        git("stash", "apply", stashid)
            except ErrorReturnCode as exc:
                raise EnvironmentError(
                    "Unable to return to the previous branch: %s" % exc.stderr)

    # TODO: This only works for local branches, make it work for remote ones
    # too
    def delete_branch(self, branch, force=False):
        """Delete the given branch.

        By default, the branch is only deleted if it has already been merged. If
        you wish to delete an unmerged branch, you will have to pass
        force=True.
        """
        deleteFlag = "-d"
        if force:
            deleteFlag = "-D"

        try:
            git("branch", deleteFlag, branch)
            logging.info("Deleted branch %s in %s" % (branch, self.repodir))
        except ErrorReturnCode as exc:
            raise EnvironmentError(
                'Unable to delete branch: %s' % str(exc))

    # TODO: Write a unit test for this.
    def add(self, filetogitadd):
        # TODO: Determine if filetogitadd is relative or absolute.
        # TODO: verify that filetogitadd is in self.repodir
        try:
            with cd(self.repodir):
                git("add", filetogitadd)
        except ErrorReturnCode:
            raise EnvironmentError("Unable to git add " + filetogitadd)

    # TODO: Write a unit test for this.
    # TODO: fix the default
    def commit(self, message="default"):
        logging.info("Attempting to commit changes to %s" % self.repodir)
        try:
            with cd(self.repodir):
                # Git commit first with a boiler plate message and then allow
                # the user to amend.
                if git("status", "--porcelain"):
                    # using python sh will suppress the git editor
                    subprocess.call(["git", "commit", "-q", "-m", message])
                    subprocess.call(["git", "commit", "-q", "--amend"])
                else:
                    logging.info("Nothing to commit.")
                    return False
        except ErrorReturnCode:
            raise EnvironmentError("Unable to git commit ")
            return False
        # Something was committed.
        return True

    # TODO: Write unit tests for this.
    def print_log(self, numcommits,oneline=False):
        try:
            with cd(self.repodir):
                if oneline:
                    print git("log", "--oneline", "-n %d" % numcommits)
                else:
                    print git("log", "-n %d" % numcommits)
        except ErrorReturnCode:
            raise EnvironmentError("Unable to git log -n %d" % numcommits)

    # versus is a branchname to compare against.  It will cause git rev-list to
    # list the differing commits.
    def rev_list(self, branch, numcommits=None, versus=None):
        try:
            with cd(self.repodir):
                if versus is not None:
                    return git("rev-list", "%s..%s" % (versus,
                        branch)).stdout.rstrip()
                else:
                    return git("rev-list", branch, "--max-count=%d" %
                         numcommits).stdout.rstrip()
        except ErrorReturnCode as exc:
            raise EnvironmentError("Unable to git rev-list because: %s" %
                    (numcommits, str(exc)))

    # TODO: Does this need to 'cd' first?
    def edit(self, toedit):
        editor = os.getenv('EDITOR')
        if not editor:
            editor = '/usr/bin/vim'
        os.system(editor + ' ' + self.repodir + '/' + toedit)

        self.add(toedit)

        return

    # TODO: Test this function with both dryrun=True and dryrun=False
    def pushToBranch(self, tobranch, dryrun=True):
        try:
            # TODO: Check for un-merged commits.
            with cd(self.repodir):
                branch = self.getbranch()
                if dryrun:
                    git("push", "--dry-run", branch, tobranch)
                else:
                    git("push", branch, tobranch)
        except ErrorReturnCode:
            raise EnvironmentError(
                "Unable to push branch %s to %s" % (branch, tobranch))

    def tag_exists(self, tag):
        try:
            with cd(self.repodir):
                tagref = "refs/tags/%s" % tag
                git("rev-parse", "-q", "--verify", tagref)
        except ErrorReturnCode_128:
            logging.info("Couldn't find tag %s" % tag)
            return False
        return True

    def __str__(self):
        return self.repodir

# TestGitRepo is an abstract baseclass. Verify that the constructor
# will throw an exception if you try to run operations without
# defining a subclass.


class TestGitRepo(unittest.TestCase):
    repoprefix = "GitRepoUT"

    def __create_dummy_commit(self):
        filename = "file" + str(uuid.uuid4())
        open(filename, "a").close()
        git("add", filename)
        git("commit", "-m", "Branches without commits confuse git")

    def __get_current_branch(self):
        # Helper function used in some of the tests (e.g. for delete_branch). We
        # use this instead of the equivalent GitRepo::getbranch so we can test
        # different GitRepo functionality independently.
        branch = str(git("rev-parse", "--abbrev-ref", "HEAD"))
        # git rev-parse returns a trailing newline that we must get rid of
        return branch[:-1]

    # This class is only used to test the GitRepo functions since, as an,
    # abstract-baseclass, GitRepo doesn't define repodir, and we need an
    # actual repository to test git based functions.
    class DerivedGitRepo(GitRepo):

        def __init__(self, proj):
            super(TestGitRepo.DerivedGitRepo, self).__init__(proj)
            self.remote = u'http://git.linaro.org/toolchain/release-notes.git'
            dgr_dirname = "release-notes"
            self.repodir = self.proj.projdir + "/" + dgr_dirname
            try:
                with cd(self.proj.projdir):
                    git("clone", self.remote, dgr_dirname,
                        _err_to_out=True, _out="/dev/null")
            except ErrorReturnCode:
                raise EnvironmentError(
                    "Git unable to clone into " + self.repodir)

    @classmethod
    def setUpClass(cls):
        cls.proj = Proj(prefix=TestGitRepo.repoprefix, persist=False)
        cls.dgr = TestGitRepo.DerivedGitRepo(cls.proj)

    @classmethod
    def tearDownClass(cls):
        cls.proj.cleanup()

    # Test that the abstract baseclass throws an exception if one of the
    # functions are called from the baseclass without a derived class.
    def test_repo(self):
        self.repo = GitRepo(self.proj)
        with self.assertRaises(OSError):
            # We haven't set a repodir so this should throw an exception.
            branch = self.repo.getbranch()

    # TODO: Create a test-branch in the repo so it's always there.

    def test_tag_exists(self):
        with cd(self.dgr.repodir):
            git("tag", "-a", "linaro-99.9-2099.08-rc1", "-m", "This is a test tag")
        self.assertTrue(self.dgr.tag_exists("linaro-99.9-2099.08-rc1"))

    def test_is_valid_branch_name(self):
        self.assertTrue(GitRepo.is_valid_branch_name("mybranch"))
        self.assertTrue(GitRepo.is_valid_branch_name("mynamespace/mybranch"))
        self.assertFalse(GitRepo.is_valid_branch_name("some random string"))

    def test_not_bran_chexists(self):
        self.assertFalse(self.dgr.branch_exists("foobar"))

    def test_getbranch(self):
        with cd(self.dgr.repodir):
            try:
                git("checkout", "-b", "master_test",
                    "origin/master").stdout.rstrip()
            except ErrorReturnCode as exc:
                raise EnvironmentError(
                    "Unable to checkout -b %s %s" % (branch, track))
            self.assertEquals(self.dgr.getbranch(), "master_test")

    # TODO: Test checkoutbranch with various combinations of polluted
    # directories.

    def test_delete_merged_branch(self):
        with cd(self.dgr.repodir):
            branchToDelete = "delete_me"
            # Use a try block to separate failures in setting up the test from
            # failures in the test itself.
            try:
                previousBranch = self.__get_current_branch()

                # Create a new branch and don't commit anything to it
                git("checkout", "-b", branchToDelete)

                # Move to the previous branch since we can't remove the branch
                # that we're currently on.
                git("checkout", previousBranch)
            except ErrorReturnCode as exc:
                raise EnvironmentError("Failed to setup test: %s" % str(exc))

            # Delete the branch and check that it doesn't exist anymore.
            self.dgr.delete_branch(branchToDelete, False)

            with self.assertRaises(ErrorReturnCode) as context:
                git("rev-parse", branchToDelete)

    def test_delete_unmerged_branch(self):
        with cd(self.dgr.repodir):
            branchToDelete = "delete_me"
            # Use a try block to separate failures in setting up the test from
            # failures in the test itself.
            try:
                previousBranch = self.__get_current_branch()

                # Create a new branch and commit to it
                git("checkout", "-b", branchToDelete)
                self.__create_dummy_commit()

                # Move to the previous branch since we can't remove the branch
                # that we're currently on
                git("checkout", previousBranch)
            except ErrorReturnCode as exc:
                raise EnvironmentError("Failed to setup test: %s" % str(exc))

            # Try to delete the branch - this should fail because it has
            # unmerged commits
            with self.assertRaises(EnvironmentError) as context:
                self.dgr.delete_branch(branchToDelete, False)

            self.assertRegexpMatches(str(context.exception),
                                     "Unable to delete branch:")

            # Make sure the branch still exists (i.e. this doesn't throw an
            # exception)
            git("rev-parse", branchToDelete)

    def test_force_delete_branch(self):
        with cd(self.dgr.repodir):
            branchToDelete = "force_delete_me"
            # Use a try block to separate failures in setting up the test from
            # failures in the test itself.
            try:
                previousBranch = self.__get_current_branch()

                # Create a new branch and commit to it
                git("checkout", "-b", branchToDelete)
                self.__create_dummy_commit()

                # Move to the previous branch since we can't remove the branch
                # that we're currently on
                git("checkout", previousBranch)
            except ErrorReturnCode as exc:
                raise EnvironmentError("Failed to setup test: %s" % str(exc))

            # Delete the branch and check that it doesn't exist anymore.
            self.dgr.delete_branch(branchToDelete, True)

            with self.assertRaises(ErrorReturnCode) as context:
                git("rev-parse", branchToDelete)

    def test_delete_current_branch(self):
        with cd(self.dgr.repodir):
            try:
                currentBranch = self.__get_current_branch()

                # We can't delete the current branch
                with self.assertRaises(EnvironmentError) as context:
                    self.dgr.delete_branch(currentBranch, False)

                self.assertRegexpMatches(str(context.exception),
                                         "Unable to delete branch:")

                # Make sure the branch still exists (i.e. this doesn't throw an
                # exception)
                git("rev-parse", currentBranch)
            except ErrorReturnCode as exc:
                raise EnvironmentError("Failed to setup test: %s" % str(exc))

    def test_delete_nonexistent_branch(self):
        with cd(self.dgr.repodir):
            nonexistentBranch = "should_not_exist"

            # First, make sure that the branch really doesn't exist
            with self.assertRaises(ErrorReturnCode) as context:
                git("rev-parse", nonexistentBranch)

            with self.assertRaises(EnvironmentError) as context:
                self.dgr.delete_branch(nonexistentBranch, False)

            self.assertRegexpMatches(str(context.exception),
                                     "Unable to delete branch:")

if __name__ == '__main__':
    # logging.basicConfig(level="INFO")
    unittest.main()