tests/unittests/testpush.py

import os
import unittest
import uuid

from sh import git

from linaropy.cd import cd
from linaropy.proj import Proj
from linaropy.git.clone import Clone
from linaropy.git.worktree import Worktree

from modules.llvm import get_user_from_remote, get_remote_branch, push_branch

# FIXME: Maybe move these helpers somewhere more public, or use LLVMTestCase for
# these tests as well...


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


def get_last_commit_message(branch=""):
    return str(git("rev-list", "-1", "--oneline", branch)).strip()


def create_dummy_repo(path):
    if not os.path.exists(path):
        os.makedirs(path)

    with cd(path):
        git("init")
        create_dummy_commit()


class TestGetUser(unittest.TestCase):
    testdirprefix = "GetUserUT"

    def test_full_remote(self):
        user = get_user_from_remote("https://user1@remote.url:1234")
        self.assertEqual(user, "user1")

        user = get_user_from_remote("ssh://user2@remote.url:1234")
        self.assertEqual(user, "user2")

    def test_no_protocol_remote(self):
        user = get_user_from_remote("user.name@remote.url:1234")
        self.assertEqual(user, "user.name")

    def test_no_user(self):
        user = get_user_from_remote("https://remote.url:1234:")
        self.assertEqual(user, None)


class TestGetRemoteBranch(unittest.TestCase):
    testdirprefix = "GetRemoteBranchUT"

    def setUp(self):
        # Create a git repo so we can play with its remotes.
        self.proj = Proj(prefix=self.testdirprefix)

        path = os.path.join(self.proj.projdir, "repo")
        create_dummy_repo(path)
        self.repo = Clone(self.proj, path)

    def tearDown(self):
        self.proj.cleanup()

    def test_valid_branch(self):
        """
        Test that we return the correct remote branch name if everything is
        valid.
        """
        user = "a.user"
        remote = "file://{}@{}".format(user, self.repo.repodir)
        with cd(self.repo.repodir):
            git("remote", "add", "origin", remote)

        local_branch = "a-branch"
        remote_branch = get_remote_branch(self.repo, local_branch)
        self.assertEqual(
            remote_branch, "linaro-local/{}/{}".format(user, local_branch))

    def test_no_user(self):
        """Test that we error out when we can't parse a user in the remote name."""
        remote = "file://" + self.repo.repodir
        with cd(self.repo.repodir):
            git("remote", "add", "origin", remote)

        with self.assertRaises(EnvironmentError) as context:
            get_remote_branch(self.repo, "a-branch")

        self.assertEqual(str(context.exception),
                         "Couldn't parse user from {}.".format(remote))

    def test_invalid_user(self):
        """
        Test that we error out when the value of the user parsed from the remote
        wouldn't look good in a branch name (e.g. if it contains spaces).
        """

        badUser = "LLVM Developer"
        with cd(self.repo.repodir):
            git("remote", "add", "origin",
                "file://{}@{}".format(badUser, self.repo.repodir))

        branch = "a-branch"
        with self.assertRaises(EnvironmentError) as context:
            get_remote_branch(self.repo, branch)

        self.assertEqual(str(context.exception),
                         "linaro-local/{}/{} is not a valid branch name.".format(badUser,
                                                                                 branch))


class TestPushBranch(unittest.TestCase):
    testdirprefix = "PushBranchUT"

    def setUp(self):
        # We're going to create a hierarchy with an origin repo, a clone repo
        # and a worktree of the clone. When pushing through the worktree, we
        # expect to see things in origin.
        self.proj = Proj(prefix=self.testdirprefix)
        self.user = "llvm-developer"

        path = os.path.join(self.proj.projdir, "origin")
        create_dummy_repo(path)
        self.origin = Clone(self.proj, path)

        path = os.path.join(self.proj.projdir, "clone")
        git("clone", "file://" + self.user + "@" + self.origin.repodir, path)
        self.clone = Clone(self.proj, path)

        self.worktreeBranch = "a-branch"
        self.remoteBranch = "remotely/a-branch"
        self.worktree = Worktree.create(
            self.proj, self.clone, os.path.join(
                self.proj.projdir, "worktree"), self.worktreeBranch)

    def tearDown(self):
        self.proj.cleanup()

    def test_push_new_branch(self):
        """Test that we can push a new branch to origin."""

        with cd(self.worktree.repodir):
            create_dummy_commit("This should make it to origin")

        push_branch(self.proj, self.worktreeBranch, self.remoteBranch,
                    self.worktree.repodir)

        with cd(self.origin.repodir):
            self.assertRegex(
                get_last_commit_message(self.remoteBranch),
                ".*This should make it to origin")

    def test_push_existing_branch(self):
        """Test that we can push to a branch that already exists in origin."""
        with cd(self.worktree.repodir):
            create_dummy_commit("This already exists in origin")
            git("push", "origin", "{}:{}".format(
                self.worktreeBranch, self.remoteBranch))

            create_dummy_commit("This should make it to origin too")

        push_branch(self.proj, self.worktreeBranch, self.remoteBranch,
                    self.worktree.repodir)

        with cd(self.origin.repodir):
            self.assertRegex(get_last_commit_message(self.remoteBranch),
                             ".*This should make it to origin too")

    def test_push_squashed_update(self):
        """
        Test that we can push again after squashing some fixes into a commit
        that has already been pushed to origin. This isn't a nice thing to do,
        but we need to support it because Gerrit requires squashes.
        """
        with cd(self.worktree.repodir):
            create_dummy_commit("First version of the patch")

        push_branch(self.proj, self.worktreeBranch, self.remoteBranch,
                    self.worktree.repodir)

        with cd(self.origin.repodir):
            self.assertRegex(
                get_last_commit_message(self.remoteBranch),
                ".*First version of the patch")

        with cd(self.worktree.repodir):
            git("commit", "--amend", "-m", "Second version of the patch")

        push_branch(self.proj, self.worktreeBranch, self.remoteBranch,
                    self.worktree.repodir)

        with cd(self.origin.repodir):
            self.assertRegex(
                get_last_commit_message(self.remoteBranch),
                "/*Second version of the patch")

    def test_push_rebased_branch(self):
        """ Test that we can push again with new updates after a rebase."""
        with cd(self.worktree.repodir):
            create_dummy_commit("First change")

        push_branch(self.proj, self.worktreeBranch, self.remoteBranch,
                    self.worktree.repodir)

        with cd(self.origin.repodir):
            self.assertRegex(
                get_last_commit_message(self.remoteBranch),
                ".*First change")

        with cd(self.origin.repodir):
            create_dummy_commit("Master moving forward")

        with cd(self.worktree.repodir):
            create_dummy_commit("Second change")
            git("fetch", "origin", "master")
            git("rebase", "origin/master")

        push_branch(self.proj, self.worktreeBranch, self.remoteBranch,
                    self.worktree.repodir)

        with cd(self.origin.repodir):
            self.assertRegex(get_last_commit_message(self.remoteBranch),
                             ".*Second change")

if __name__ == "__main__":
    unittest.main()