path: root/ganarchy/git.py

# This file is part of GAnarchy - decentralized project hub
# Copyright (C) 2020  Soni L.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

"""Git abstraction.
"""

# Errors are raised when we can't provide an otherwise valid result.
# For example, we return 0 for counts instead of raising, but raise
# instead of returning empty strings for commit hashes and messages.

import shutil
import subprocess

class GitError(Exception):
    """Raised when a git operation fails, generally due to a
    missing commit or branch, or network connection issues.
    """
    pass

class Git:
    """A git repo.

    Takes a ``pathlib.Path`` as argument.
    """

    def __init__(self, path):
        self.path = path

    #########################################
    # Operations supported on any git repo. #
    #########################################

    def check_branchname(self, branchname):
        """Checks if the given branchname is a valid branch name.
        Raises if it isn't.

        Args:
            branchname (str): Name of branch.

        Raises:
            GitError: If an error occurs.
        """
        try:
            if branchname.startswith("-"):
                raise GitError("check branchname", branchname)
            out = self._cmd(
                "check-ref-format", "--branch", branchname
            ).stdout.decode("utf-8")
            # protect against @{-1}/@{-n} ("previous checkout operation")
            # is also fairly future-proofed, I hope?
            if (not out.startswith(branchname)) or (
                out.removeprefix(branchname) not in ('\r\n', '\n', '')
            ):
                raise GitError("check branchname", out, branchname)
        except subprocess.CalledProcessError as e:
            raise GitError("check branchname") from e

    def get_hash(self, target):
        """Returns the commit hash for a given target.

        Args:
            target (str): a refspec.

        Raises:
            GitError: If an error occurs.
        """
        try:
            return self._cmd(
                "show", target, "-s", "--format=format:%H", "--"
            ).stdout.decode("utf-8")
        except subprocess.CalledProcessError as e:
            raise GitError("get hash") from e

    def get_commit_message(self, target):
        """Returns the commit message for a given target.

        Args:
            target (str): a refspec.

        Raises:
            GitError: If an error occurs.
        """
        try:
            return self._cmd(
                "show", target, "-s", "--format=format:%B", "--"
            ).stdout.decode("utf-8", "replace")
        except subprocess.CalledProcessError as e:
            raise GitError("get commit message") from e

    ########################
    # Low-level operations #
    ########################

    def _cmd_init(self, *args):
        """Runs a command for initializing this git repo.

        Always uses ``--bare``.

        Returns:
            subprocess.CompletedProcess: The results of running the command.

        Raises:
            subprocess.CalledProcessError: If the command exited with a non-zero
            status.
        """
        return self._cmd_common('init', '--bare', *args, self.path)

    def _cmd_clone_from(self, from_, *args):
        """Runs a command for cloning into this git repo.

        Always uses ``--bare``.

        Returns:
            subprocess.CompletedProcess: The results of running the command.

        Raises:
            subprocess.CalledProcessError: If the command exited with a non-zero
            status.
        """
        return self._cmd_common('clone', '--bare', *args, from_, self.path)

    def _cmd(self, *args):
        """Runs a command for operating on this git repo.

        Note: Doesn't work for git init and git clone operations. Use
        ``_cmd_init`` and ``_cmd_clone_from`` instead.

        Always uses ``--bare``.

        Returns:
            subprocess.CompletedProcess: The results of running the command.

        Raises:
            subprocess.CalledProcessError: If the command exited with a non-zero
            status.
        """
        return self._cmd_common('-C', self.path, '--bare', *args)

    def _cmd_common(self, *args):
        """Runs a git command with the given args.

        This is a simple wrapper around ``subprocess.run``.

        Returns:
            subprocess.CompletedProcess: The results of running the command.

        Raises:
            subprocess.CalledProcessError: If the command exited with a non-zero
            status.
        """
        return subprocess.run(
            ('git',) + args,
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

class GitCache(Git):
    """A permanent repository used to cache remote objects.
    """

    #####################
    # Public operations #
    #####################

    def create(self):
        """Creates the local repo.

        Can safely be called on an existing repo.
        """
        try:
            return self._cmd_init()
        except subprocess.CalledProcessError as e:
            raise GitError("create") from e

    def with_work_repos(self, count):
        """Creates a context manager for managing work repos.

        Args:
            count (int): The number of work repos.
        """
        """From Rust:
        /// Creates the given number of work repos, and calls the closure to run
        /// operations on them.
        ///
        /// The operations can be done on the individual repos, and they'll be
        /// merged into the main repo as this function returns.
        ///
        /// If the callback fails, the work repos will be deleted. If the function
        /// succeeds, the work repos will be merged back into the main repo.
        ///
        /// # Panics
        ///
        /// Panics if a merge conflict is detected. Specifically, if two work repos
        /// modify the same work branch.
        ///
        /// # "Poisoning"
        ///
        /// If this method unwinds, the underlying git repos, if any, will not be
        /// deleted. Instead, future calls to this method will return a GitError.
        """
        work_repos = []
        for i in range(0, count):
            new_path = self.path.with_name('ganarchy-fetch-{}.git'.format(i))
            work_repos.append(GitFetch(new_path))
        physical_work_repos = []
        for repo in work_repos:
            self._fork(repo)
            physical_work_repos.append(repo)
        return _WithWorkRepos(self, physical_work_repos)

    def check_history(self, local_head, commit):
        """Checks if the local head contains commit in its history.
        Raises if it doesn't.

        Args:
            local_head (str): Name of local head.
            commit (str): Commit hash.

        Raises:
            GitError: If an error occurs.
        """
        try:
            self._cmd("merge-base", "--is-ancestor", commit, local_head)
        except subprocess.CalledProcessError as e:
            raise GitError("check history") from e

    #######################
    # Internal operations #
    #######################

    def _fetch_work(self, from_, branch, from_branch):
        try:
            self._cmd(
                "fetch", from_.path, "+{}:{}".format(from_branch, branch)
            )
        except subprocess.CalledProcessError as e:
            raise GitError("fetch work") from e

    def _replace(self, old_name, new_name):
        try:
            self._cmd(
                "branch", "-M", old_name, new_name
            )
        except subprocess.CalledProcessError as e:
            raise GitError("replace") from e

    def _fork(self, into):
        """Makes a shared clone of this local repo into the given work repo.

        Equivalent to ``git clone --bare --shared``, which is very dangerous!
        """
        try:
            return into._cmd_clone_from(self.path, '--shared')
        except subprocess.CalledProcessError as e:
            raise GitError("fork") from e

class GitFetch(Git):
    """A temporary repository used to fetch remote objects.
    """
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.pending_branches = set()

    #####################
    # Public operations #
    #####################

    def force_fetch(self, url, remote_head, local_head):
        """Fetches a remote head into a local head.
        
        If the local head already exists, it is replaced.

        Args:
            url (str): Remote url.
            remote_head (str): Name of remote head.
            local_head (str): Name of local head.

        Raises:
            GitError: If an error occurs.
        """
        try:
            self._cmd(
                "fetch", url, "+" + remote_head + ":" + local_head
            )
            self.pending_branches.add(local_head)
        except subprocess.CalledProcessError as e:
            raise GitError("fetch source") from e

    def get_count(self, first_hash, last_hash):
        """Returns a count of the commits added since ``first_hash``
        up to ``last_hash``.

        Args:
            first_hash (str): A commit.
            last_hash (str): Another commit.

        Returns:
            int: A count of commits added between the hashes, or 0
            if an error occurs.
        """
        try:
            res = self._cmd(
                "rev-list", "--count", first_hash + ".." + last_hash, "--"
            ).stdout.decode("utf-8").strip()
            return int(res)
        except subprocess.CalledProcessError as e:
            return 0

    #######################
    # Internal operations #
    #######################

    def _rm_branch(self, branch):
        try:
            self._cmd("branch", "-D", branch)
        except subprocess.CalledProcessError as e:
            raise GitError("rm branch") from e

    def _delete(self):
        try:
            shutil.rmtree(self.path)
        except IOError as e:
            raise GitError("delete", self.path) from e

class _WithWorkRepos:
    """Context manager for merging forked repos in ``with_work_repos``.
    """
    def __init__(self, cache_repo, work_repos):
        self.cache_repo = cache_repo
        self.work_repos = work_repos

    def __enter__(self):
        return self.work_repos

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            branches = set()
            for work in self.work_repos:
                for branch in work.pending_branches:
                    if branch in branches:
                        raise GitError("Branch {} is in conflict!".format(branch))
                    branches.add(branch)
            del branches
            del work
            del branch

            for i, repo in enumerate(self.work_repos):
                for branch in repo.pending_branches:
                    fetch_head = "{}-{}".format(branch, i)
                    # First collect the work branch into a fetch head
                    self.cache_repo._fetch_work(repo, fetch_head, branch)
                    # If that succeeds, delete the work branch to free up disk
                    repo._rm_branch(branch)
                    # We have all the objects in the main repo and we probably
                    # have enough disk, so just replace the fetch head into
                    # the main branch and hope nothing errors.
                    self.cache_repo._replace(fetch_head, branch)
                repo._delete()