path: root/linaropy/git/clone.py

import unittest
import logging
import os

from sh import git
from sh import ln
from sh import mkdir
from sh import ErrorReturnCode
from sh import chmod

# In order to set directory read/write permissions for test cases
from stat import *

from ..proj import Proj
from ..cd import cd
from linaropy.git.gitrepo import GitRepo

import time


class Clone(GitRepo):

    def __init__(self, proj, clonedir=None, remote=None):
        super(Clone, self).__init__(proj)

        # If the user passes a clonedir then they want to reuse an existing
        # clone.  This will often be the case with large repositories like
        # gcc, which take 20 minutes to clone.

        # Clone a new repo from the remote
        if clonedir is None or (isinstance(
                clonedir, str) and clonedir == ""):
            if remote == "" or remote is None:
                raise TypeError(
                    'Clone input parameter \'remote\' can not be empty')

            # TODO test that remote is a well formed URL?
            # TODO if there's a trailing / strip it.
            self.remote = remote

            # Strip the end off the repo url to find the clonedir name.
            clone_dirname = remote.rsplit('/', 1)[-1]

            logging.info('Cloning %s repository into %s/%s' %
                         (self.remote, self.proj.projdir, clone_dirname))

            self.repodir = self.proj.projdir + '/' + clone_dirname
            try:
                with cd(self.proj.projdir):
                    git("clone", self.remote, clone_dirname,
                        _err_to_out=True, _out="/dev/null")
            except ErrorReturnCode:
                raise EnvironmentError(
                    "Git unable to clone into " + self.repodir)

        # Reuse an existing git clone directory
        else:
            if not isinstance(clonedir, str):
                raise TypeError('clonedir must be of type basestring')
            elif not os.path.isdir(clonedir):
                raise EnvironmentError(
                    "Specified clonedir directory '%s' does not exist" %
                    clonedir)

            # Check to see if clonedir is a git repository.
            with cd(clonedir):
                # There might be a file called .git
                if not os.path.isdir(clonedir + '/.git'):
                    try:
                        git("rev-parse", _err_to_out=True, _out="/dev/null")
                    except ErrorReturnCode:
                        raise EnvironmentError(
                            'Specified clonedir is not a git repository.')

            # Strip off any trailing /
            clonedir = clonedir.rstrip('/')
            # We only want the final directory name, not the whole path.
            clone_dirname = clonedir.rsplit('/', 1)[-1]
            self.repodir = self.proj.projdir + '/' + clone_dirname

            # TODO: verify that we don't get another clone dir in the projdir.
            # It's possible the clonedir is already in the projdir.
            if os.path.exists(self.repodir):
                logging.info('The clonedir %s already exists in %s' %
                             (clonedir, self.repodir))
            else:
                logging.info(
                    'Reusing existing clonedir %s and symlinking as %s in %s' %
                    (clonedir, self.repodir, self.proj.projdir))
                try:
                    ln("-s", clonedir, self.repodir)
                except ErrorReturnCode:
                    raise EnvironmentError(
                        "Can't create symlink %s to %s." %
                        (clonedir, self.repodir))

        logging.info('Clone directory set as %s' % (self.repodir))
        return

    def clonedir(self):
        return self.repodir

    def __repr__(self):
        return "<Clone Class: remote = " + self.remote + \
            ", clonedir =" + self.repodir + ">"


class TestClone(unittest.TestCase):
    testdirprefix = "CloneUT"

    @classmethod
    def setUpClass(cls):
        # We need a Proj dir to store the mother clone in.
        cls.proj = Proj(prefix=TestClone.testdirprefix)

        # Setup a single remote clone in a Proj directory and then all of the
        # other tests in this module should clone from this resulting local
        # clone as if it were a remote.  This will reduce the network burden on
        # the remote git server being tested. If this fails then the entire
        # TestClone will be marked as not executed.

        # TODO: Perform a connectivity test to determine online or offline
        #      testing mode.

        cls.rnclone = Clone(
            cls.proj,
            remote='http://git.linaro.org/toolchain/release-notes.git')
        # OFFLINE:
        # cls.rnclone=Clone(cls.proj,remote=u'/var/run/media/ryanarn/sidecar/reldir/release-notes')
        cls.startbranch = cls.rnclone.getbranch()

        cls.rtremote = 'http://git.linaro.org/toolchain/tcwg-release-tools.git'
        # OFFLINE:
        # cls.rtremote=u'/var/run/media/ryanarn/sidecar/releases/tcwg-release-tools'

        # All further tests will use this as the remote.
        cls.rnremote = cls.rnclone.repodir

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

    # Every instance of TestClass requires its own proj directory.
    def setUp(self):
        # TODO: Is this redundant with the setUpClass()?
        self.proj = Proj(prefix=TestClone.testdirprefix)

    def tearDown(self):
        # TODO: Is this redundant with the tearDownClass()?
        self.proj.cleanup()

    # These immediate tests clone from a remote repository
    def test_clone(self):
        self.clone = Clone(self.proj, remote=TestClone.rnremote)
        self.assertTrue(os.path.isdir(self.clone.clonedir()))

    def test_two_different_clones_in_one_projdir(self):
        self.clonern = Clone(self.proj, remote=TestClone.rnremote)
        self.clonert = Clone(self.proj, remote=TestClone.rtremote)
        self.assertTrue(os.path.isdir(self.clonern.clonedir())
                        and os.path.isdir(self.clonert.clonedir()))

    # All clones after this point should use TestClone.rnremote as the remote.

    def test_str_function(self):
        self.clone = Clone(self.proj, remote=TestClone.rnremote)
        self.assertEqual(str(self.clone), self.clone.clonedir())

    def test_empty_string_clonedir_with_remote(self):
        # This will use the remote because clonedir is an empty string.
        self.clone = Clone(self.proj, remote=TestClone.rnremote, clonedir="")
        self.assertTrue(os.path.isdir(self.clone.clonedir()))

    def test_none_clonedir_with_remote(self):
        # This will use the remote because there is no clonedir
        self.clone = Clone(self.proj, remote=TestClone.rnremote)
        self.assertTrue(os.path.isdir(self.clone.clonedir()))

    def test_empty_remote_empty_clonedir(self):
        # A bogus directory will fail.
        with self.assertRaises(TypeError):
            self.clone = Clone(self.proj, remote="", clonedir="")

    def test_non_string_clonedir(self):
        self.foo = Clone(self.proj, remote=TestClone.rnremote, clonedir="")
        with self.assertRaises(TypeError):
            self.clone = Clone(self.proj, clonedir=self.foo)

    def test_no_string_clonedir_empty_remote(self):
        # remote can't be be empty if clonedir is not set.
        with self.assertRaises(TypeError):
            self.clone = Clone(self.proj, remote="")

    def test_existing_clonedir(self):
        # Clone's clonedir parameter allows a user to specify an existing
        # directory use for the clone.  It will setup a symlink in the proj
        # directory pointing to the tree.  This is a useful feature for
        # very large repositories.

        # Use the clonedir from TestClone.rnclone
        self.clone = Clone(self.proj, clonedir=TestClone.rnclone.clonedir())

        # Verify that a symlink is used for the clone.
        self.assertTrue(os.path.islink(self.clone.clonedir()))

    # This was a bug at some point, as it would strip off the repository name.
    def test_existing_clonedir_with_trailing_slash(self):
        self.clone = Clone(
            self.proj, clonedir=TestClone.rnclone.clonedir() + "/")

        # Verify that a symlink is used for the clone.
        self.assertTrue(os.path.islink(self.clone.clonedir()))

    def test_failed_symlink(self):
        # We'll make the projdir read-only so we can force symlink creation
        # to fail.
        with cd(self.proj.projdir):
            # Set to read-only to force symlink to fail.
            os.chmod(self.proj.projdir, S_IRUSR | S_IXUSR)

        # Use the clonedir from TestClone.rnclone
        with self.assertRaises(EnvironmentError):
            # Verify that this throws an exception.
            self.clone = Clone(
                self.proj, clonedir=TestClone.rnclone.clonedir())

        with cd(self.proj.projdir):
            # Reset to writable so we can remove the projdir
            os.chmod(self.proj.projdir, S_IWUSR | S_IXUSR)

    def test_clone_failure_due_to_write_permissions(self):
        with cd(self.proj.projdir):
            # Set to read-only to force the git clone to fail.
            os.chmod(self.proj.projdir, S_IRUSR | S_IXUSR)

        with self.assertRaises(EnvironmentError):
            self.clone = Clone(self.proj, remote=TestClone.rnremote)

        with cd(self.proj.projdir):
            # Reset to writable so we can remove the projdir
            os.chmod(self.proj.projdir, S_IWUSR | S_IXUSR)

    def test_nonexistant_clonedir(self):
        with self.assertRaises(EnvironmentError):
            self.clone_one = Clone(
                self.proj, clonedir="/a/nonexistant/directory/")

    def test_not_gitdir(self):
        # Verify that an exception is raised when attempting to clone something
        # that's not a git repository.
        nongitdir = self.proj.projdir + "/nongitdir"
        with cd(self.proj.projdir):
            os.mkdir(nongitdir)

        with self.assertRaises(EnvironmentError):
            self.clone = Clone(self.proj, remote="", clonedir=nongitdir)

    def test_not_gitdir_but_with_dot_git(self):
        # Verify that an exception is raised when attempting to clone a
        # directory that has a .git file but is NOT a git repository.
        nongitdir = self.proj.projdir + "/nongitdir"
        with cd(self.proj.projdir):
            os.mkdir(nongitdir)
            # Create and empty .git file.
            with open(".git", 'a'):
                os.utime(".git", None)

        with self.assertRaises(EnvironmentError):
            self.clone = Clone(self.proj, remote="", clonedir=nongitdir)

    def test_incorrect_input_proj_type(self):
        # Clone requires a valid Proj type as an input parameter.  Verify that
        # it throws an exception.
        proj = 'stringtypeonpurpose'
        with self.assertRaises(TypeError):
            self.clone = Clone(proj, remote=TestClone.rnremote)

    def test_non_destructive_cleanup(self):
        # Make sure that the file removal method in proj.cleanup() doesn't
        # follow and remove symlinks.
        self.clone_one = Clone(self.proj, remote=TestClone.rnremote)
        self.proj_two = Proj(prefix=TestClone.testdirprefix)
        # Use the clonedir from the first project as the clonedir for the
        # second project.
        self.clone_two = Clone(
            self.proj_two, clonedir=self.clone_one.clonedir())
        self.assertTrue(os.path.islink(self.clone_two.clonedir()))

        # Cleanup proj_two
        self.proj_two.cleanup()

        # Make sure clone_one.clonedir wasn't removed when proj_two was removed
        # which is a concern since proj_two used a clone directory from
        # proj.
        self.assertTrue(os.path.isdir(self.clone_one.clonedir()))

    def test_get_branch_clone(self):
        self.clone = Clone(self.proj, remote=TestClone.rnremote)
        self.assertEqual(TestClone.startbranch, self.clone.getbranch())

    def test_checkoutbranch_clone(self):
        self.clone = Clone(self.proj, remote=TestClone.rnremote)

        # Test a new branch
        with self.clone.checkoutbranch("TestClone", "origin/HEAD"):
            self.assertEqual("TestClone", self.clone.getbranch())

        self.assertEqual(TestClone.startbranch, self.clone.getbranch())

        # Test being able to switch to a branch that already exists.
        with self.clone.checkoutbranch("TestClone"):
            self.assertEqual("TestClone", self.clone.getbranch())

        self.assertEqual(TestClone.startbranch, self.clone.getbranch())

        # TODO test a non-default 'track' passed to checkoutbranch.

    def test_checkoutbranch_clone_with_local_changes(self):
        self.clone = Clone(self.proj, remote=TestClone.rnremote)

        # create an empty file and add it to the repository.
        with cd(self.clone.repodir):
            # Create and empty file.
            with open("checkouttest", 'a'):
                os.utime("checkouttest", None)

            self.clone.add(self.clone.repodir + '/checkouttest')

        with self.clone.checkoutbranch("TestClone", "origin/HEAD"):
            self.assertEqual("TestClone", self.clone.getbranch())

            # Verify there's no file name 'checkouttest' when we switch branch.
            with cd(self.clone.repodir):
                self.assertFalse(os.path.isfile(
                    self.clone.repodir + '/checkouttest'))

        # Verify that we've returned to the correct branch.
        self.assertEqual(TestClone.startbranch, self.clone.getbranch())

        # Show that checkouttest does exist once were out of the
        # checkoutbranch context.
        with cd(self.clone.repodir):
            self.assertTrue(os.path.isfile(
                self.clone.repodir + '/checkouttest'))

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