modules/utils.py

from subprocess import PIPE
from subprocess import Popen
from subprocess import STDOUT
from subprocess import SubprocessError

import re

from linaropy.git.worktree import Worktree


class CommandPrinter(object):
    """Command consumer that just prints the commands that it receives."""

    def consume(self, command, directory):
        print("{}$ {}".format(directory, ' '.join(command)))


class CommandRunner(object):
    """Command consumer that runs the commands that it receives."""

    def consume(self, command, directory):
        """
        Run the given command in the given directory and print the stdout and
        stderr. If an exception is thrown while running the command, it will be
        rethrown as a RuntimeError.
        """
        try:
            with Popen(command, stdout=PIPE, stderr=STDOUT, cwd=directory) as process:
                for line in process.stdout:
                    print(str(line, 'utf-8'), end='')
            if process.returncode != 0:
                raise RuntimeError(
                    "Command failed with return code: {}\n".format(
                        process.returncode))
        except SubprocessError as exc:
            raise RuntimeError(
                "Error while running command\n{}".format(str(exc.output, 'utf-8'))) from exc


# FIXME: repo.pushToBranch doesn't work, because it doesn't handle remote
# branches properly. Furthermore, there's no support for getting the remote URL,
# so we need to resort to raw git commands. We may also consider moving the
# functionality for parsing info out of the remote URL into the repo object.
from sh import git
from linaropy.cd import cd


def get_user_from_remote(url):
    """Get the username used as part of the remote URL, or None.

    The remote URLs that we expect to see look like $protocol://$user@$location.
    If they look any different, we won't be able to parse them.
    """
    pattern = re.compile("(.*://)?(?P<user>.*)@(.*)\n?")
    match = pattern.match(str(url))
    if match is None:
        return None
    return match.group('user')


def get_remote_branch(repo, local_branch):
    """
    Get the name of the remote branch corresponding to the given local branch.
    """
    with cd(repo.repodir):
        remote = git("remote", "get-url", "--push", "origin").strip()
    user = get_user_from_remote(remote)
    if not user:
        raise EnvironmentError("Couldn't parse user from {}.".format(remote))

    remote_branch = "linaro-local/{}/{}".format(user, local_branch)
    if not repo.is_valid_branch_name(remote_branch):
        raise EnvironmentError(
            "{} is not a valid branch name.".format(remote_branch))

    return remote_branch


def push_branch(proj, local_branch, remote_branch, pathToRepo):
    """Push the given local branch into origin's given remote branch."""
    repo = Worktree(proj, pathToRepo)
    with cd(repo.repodir):
        git("push", "-u", "origin", "+{}:{}".format(local_branch, remote_branch))