test_repo.py

# Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors
#
# This module is part of GitPython and is released under the
# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/

import glob
import io
from io import BytesIO
import itertools
import os
import pathlib
import pickle
import sys
import tempfile
from unittest import mock, skip

import pytest

from git import (
    InvalidGitRepositoryError,
    Repo,
    NoSuchPathError,
    Head,
    Commit,
    Object,
    Tree,
    IndexFile,
    Git,
    Reference,
    GitDB,
    Submodule,
    GitCmdObjectDB,
    Remote,
    BadName,
    GitCommandError,
)
from git.exc import (
    BadObject,
    UnsafeOptionError,
    UnsafeProtocolError,
)
from git.repo.fun import touch
from git.util import bin_to_hex, cygpath, join_path_native, rmfile, rmtree
from test.lib import TestBase, fixture, with_rw_directory, with_rw_repo

import os.path as osp


def iter_flatten(lol):
    for items in lol:
        for item in items:
            yield item


def flatten(lol):
    return list(iter_flatten(lol))


_tc_lock_fpaths = osp.join(osp.dirname(__file__), "../../.git/*.lock")


def _rm_lock_files():
    for lfp in glob.glob(_tc_lock_fpaths):
        rmfile(lfp)


class TestRepo(TestBase):
    def setUp(self):
        _rm_lock_files()

    def tearDown(self):
        for lfp in glob.glob(_tc_lock_fpaths):
            if osp.isfile(lfp):
                raise AssertionError("Previous TC left hanging git-lock file: {}".format(lfp))
        import gc

        gc.collect()

    def test_new_should_raise_on_invalid_repo_location(self):
        self.assertRaises(InvalidGitRepositoryError, Repo, tempfile.gettempdir())

    def test_new_should_raise_on_non_existent_path(self):
        self.assertRaises(NoSuchPathError, Repo, "repos/foobar")

    @with_rw_repo("0.3.2.1")
    def test_repo_creation_from_different_paths(self, rw_repo):
        r_from_gitdir = Repo(rw_repo.git_dir)
        self.assertEqual(r_from_gitdir.git_dir, rw_repo.git_dir)
        assert r_from_gitdir.git_dir.endswith(".git")
        assert not rw_repo.git.working_dir.endswith(".git")
        self.assertEqual(r_from_gitdir.git.working_dir, rw_repo.git.working_dir)

    @with_rw_repo("0.3.2.1")
    def test_repo_creation_pathlib(self, rw_repo):
        r_from_gitdir = Repo(pathlib.Path(rw_repo.git_dir))
        self.assertEqual(r_from_gitdir.git_dir, rw_repo.git_dir)

    def test_description(self):
        txt = "Test repository"
        self.rorepo.description = txt
        self.assertEqual(self.rorepo.description, txt)

    def test_heads_should_return_array_of_head_objects(self):
        for head in self.rorepo.heads:
            self.assertEqual(Head, head.__class__)

    def test_heads_should_populate_head_data(self):
        for head in self.rorepo.heads:
            assert head.name
            self.assertIsInstance(head.commit, Commit)
        # END for each head

        self.assertIsInstance(self.rorepo.heads.master, Head)
        self.assertIsInstance(self.rorepo.heads["master"], Head)

    def test_tree_from_revision(self):
        tree = self.rorepo.tree("0.1.6")
        self.assertEqual(len(tree.hexsha), 40)
        self.assertEqual(tree.type, "tree")
        self.assertEqual(self.rorepo.tree(tree), tree)

        # try from invalid revision that does not exist
        self.assertRaises(BadName, self.rorepo.tree, "hello world")

    def test_pickleable(self):
        pickle.loads(pickle.dumps(self.rorepo))

    def test_commit_from_revision(self):
        commit = self.rorepo.commit("0.1.4")
        self.assertEqual(commit.type, "commit")
        self.assertEqual(self.rorepo.commit(commit), commit)

    def test_commits(self):
        mc = 10
        commits = list(self.rorepo.iter_commits("0.1.6", max_count=mc))
        self.assertEqual(len(commits), mc)

        c = commits[0]
        self.assertEqual("9a4b1d4d11eee3c5362a4152216376e634bd14cf", c.hexsha)
        self.assertEqual(["c76852d0bff115720af3f27acdb084c59361e5f6"], [p.hexsha for p in c.parents])
        self.assertEqual("ce41fc29549042f1aa09cc03174896cf23f112e3", c.tree.hexsha)
        self.assertEqual("Michael Trier", c.author.name)
        self.assertEqual("mtrier@gmail.com", c.author.email)
        self.assertEqual(1232829715, c.authored_date)
        self.assertEqual(5 * 3600, c.author_tz_offset)
        self.assertEqual("Michael Trier", c.committer.name)
        self.assertEqual("mtrier@gmail.com", c.committer.email)
        self.assertEqual(1232829715, c.committed_date)
        self.assertEqual(5 * 3600, c.committer_tz_offset)
        self.assertEqual("Bumped version 0.1.6\n", c.message)

        c = commits[1]
        self.assertIsInstance(c.parents, tuple)

    def test_trees(self):
        mc = 30
        num_trees = 0
        for tree in self.rorepo.iter_trees("0.1.5", max_count=mc):
            num_trees += 1
            self.assertIsInstance(tree, Tree)
        # END for each tree
        self.assertEqual(num_trees, mc)

    def _assert_empty_repo(self, repo):
        """Test all kinds of things with an empty, freshly initialized repo.

        It should throw good errors.
        """
        # Entries should be empty.
        self.assertEqual(len(repo.index.entries), 0)

        # head is accessible.
        assert repo.head
        assert repo.head.ref
        assert not repo.head.is_valid()

        # We can change the head to some other ref.
        head_ref = Head.from_path(repo, Head.to_full_path("some_head"))
        assert not head_ref.is_valid()
        repo.head.ref = head_ref

        # is_dirty can handle all kwargs.
        for args in ((1, 0, 0), (0, 1, 0), (0, 0, 1)):
            assert not repo.is_dirty(*args)
        # END for each arg

        # We can add a file to the index (if we are not bare).
        if not repo.bare:
            pass
        # END test repos with working tree

    @with_rw_directory
    def test_clone_from_keeps_env(self, rw_dir):
        original_repo = Repo.init(osp.join(rw_dir, "repo"))
        environment = {"entry1": "value", "another_entry": "10"}

        cloned = Repo.clone_from(original_repo.git_dir, osp.join(rw_dir, "clone"), env=environment)

        self.assertEqual(environment, cloned.git.environment())

    @with_rw_directory
    def test_date_format(self, rw_dir):
        repo = Repo.init(osp.join(rw_dir, "repo"))
        # @-timestamp is the format used by git commit hooks.
        repo.index.commit("Commit messages", commit_date="@1400000000 +0000")

    @with_rw_directory
    def test_clone_from_pathlib(self, rw_dir):
        original_repo = Repo.init(osp.join(rw_dir, "repo"))

        Repo.clone_from(original_repo.git_dir, pathlib.Path(rw_dir) / "clone_pathlib")

    @with_rw_directory
    def test_clone_from_pathlib_withConfig(self, rw_dir):
        original_repo = Repo.init(osp.join(rw_dir, "repo"))

        cloned = Repo.clone_from(
            original_repo.git_dir,
            pathlib.Path(rw_dir) / "clone_pathlib_withConfig",
            multi_options=[
                "--recurse-submodules=repo",
                "--config core.filemode=false",
                "--config submodule.repo.update=checkout",
                "--config filter.lfs.clean='git-lfs clean -- %f'",
            ],
            allow_unsafe_options=True,
        )

        self.assertEqual(cloned.config_reader().get_value("submodule", "active"), "repo")
        self.assertEqual(cloned.config_reader().get_value("core", "filemode"), False)
        self.assertEqual(cloned.config_reader().get_value('submodule "repo"', "update"), "checkout")
        self.assertEqual(
            cloned.config_reader().get_value('filter "lfs"', "clean"),
            "git-lfs clean -- %f",
        )

    def test_clone_from_with_path_contains_unicode(self):
        with tempfile.TemporaryDirectory() as tmpdir:
            unicode_dir_name = "\u0394"
            path_with_unicode = os.path.join(tmpdir, unicode_dir_name)
            os.makedirs(path_with_unicode)

            try:
                Repo.clone_from(
                    url=self._small_repo_url(),
                    to_path=path_with_unicode,
                )
            except UnicodeEncodeError:
                self.fail("Raised UnicodeEncodeError")

    @with_rw_directory
    @skip(
        """The referenced repository was removed, and one needs to set up a new
        password controlled repo under the org's control."""
    )
    def test_leaking_password_in_clone_logs(self, rw_dir):
        password = "fakepassword1234"
        try:
            Repo.clone_from(
                url="https://fakeuser:{}@fakerepo.example.com/testrepo".format(password),
                to_path=rw_dir,
            )
        except GitCommandError as err:
            assert password not in str(err), "The error message '%s' should not contain the password" % err
        # Working example from a blank private project.
        Repo.clone_from(
            url="https://gitlab+deploy-token-392045:mLWhVus7bjLsy8xj8q2V@gitlab.com/mercierm/test_git_python",
            to_path=rw_dir,
        )

    @with_rw_repo("HEAD")
    def test_clone_unsafe_options(self, rw_repo):
        with tempfile.TemporaryDirectory() as tdir:
            tmp_dir = pathlib.Path(tdir)
            tmp_file = tmp_dir / "pwn"
            unsafe_options = [
                f"--upload-pack='touch {tmp_file}'",
                f"-u 'touch {tmp_file}'",
                "--config=protocol.ext.allow=always",
                "-c protocol.ext.allow=always",
            ]
            for unsafe_option in unsafe_options:
                with self.assertRaises(UnsafeOptionError):
                    rw_repo.clone(tmp_dir, multi_options=[unsafe_option])
                assert not tmp_file.exists()

            unsafe_options = [
                {"upload-pack": f"touch {tmp_file}"},
                {"u": f"touch {tmp_file}"},
                {"config": "protocol.ext.allow=always"},
                {"c": "protocol.ext.allow=always"},
            ]
            for unsafe_option in unsafe_options:
                with self.assertRaises(UnsafeOptionError):
                    rw_repo.clone(tmp_dir, **unsafe_option)
                assert not tmp_file.exists()

    @with_rw_repo("HEAD")
    def test_clone_unsafe_options_allowed(self, rw_repo):
        with tempfile.TemporaryDirectory() as tdir:
            tmp_dir = pathlib.Path(tdir)
            tmp_file = tmp_dir / "pwn"
            unsafe_options = [
                f"--upload-pack='touch {tmp_file}'",
                f"-u 'touch {tmp_file}'",
            ]
            for i, unsafe_option in enumerate(unsafe_options):
                destination = tmp_dir / str(i)
                assert not tmp_file.exists()
                # The options will be allowed, but the command will fail.
                with self.assertRaises(GitCommandError):
                    rw_repo.clone(destination, multi_options=[unsafe_option], allow_unsafe_options=True)
                assert tmp_file.exists()
                tmp_file.unlink()

            unsafe_options = [
                "--config=protocol.ext.allow=always",
                "-c protocol.ext.allow=always",
            ]
            for i, unsafe_option in enumerate(unsafe_options):
                destination = tmp_dir / str(i)
                assert not destination.exists()
                rw_repo.clone(destination, multi_options=[unsafe_option], allow_unsafe_options=True)
                assert destination.exists()

    @with_rw_repo("HEAD")
    def test_clone_safe_options(self, rw_repo):
        with tempfile.TemporaryDirectory() as tdir:
            tmp_dir = pathlib.Path(tdir)
            options = [
                "--depth=1",
                "--single-branch",
                "-q",
            ]
            for option in options:
                destination = tmp_dir / option
                assert not destination.exists()
                rw_repo.clone(destination, multi_options=[option])
                assert destination.exists()

    @with_rw_repo("HEAD")
    def test_clone_from_unsafe_options(self, rw_repo):
        with tempfile.TemporaryDirectory() as tdir:
            tmp_dir = pathlib.Path(tdir)
            tmp_file = tmp_dir / "pwn"
            unsafe_options = [
                f"--upload-pack='touch {tmp_file}'",
                f"-u 'touch {tmp_file}'",
                "--config=protocol.ext.allow=always",
                "-c protocol.ext.allow=always",
            ]
            for unsafe_option in unsafe_options:
                with self.assertRaises(UnsafeOptionError):
                    Repo.clone_from(rw_repo.working_dir, tmp_dir, multi_options=[unsafe_option])
                assert not tmp_file.exists()

            unsafe_options = [
                {"upload-pack": f"touch {tmp_file}"},
                {"u": f"touch {tmp_file}"},
                {"config": "protocol.ext.allow=always"},
                {"c": "protocol.ext.allow=always"},
            ]
            for unsafe_option in unsafe_options:
                with self.assertRaises(UnsafeOptionError):
                    Repo.clone_from(rw_repo.working_dir, tmp_dir, **unsafe_option)
                assert not tmp_file.exists()

    @with_rw_repo("HEAD")
    def test_clone_from_unsafe_options_allowed(self, rw_repo):
        with tempfile.TemporaryDirectory() as tdir:
            tmp_dir = pathlib.Path(tdir)
            tmp_file = tmp_dir / "pwn"
            unsafe_options = [
                f"--upload-pack='touch {tmp_file}'",
                f"-u 'touch {tmp_file}'",
            ]
            for i, unsafe_option in enumerate(unsafe_options):
                destination = tmp_dir / str(i)
                assert not tmp_file.exists()
                # The options will be allowed, but the command will fail.
                with self.assertRaises(GitCommandError):
                    Repo.clone_from(
                        rw_repo.working_dir, destination, multi_options=[unsafe_option], allow_unsafe_options=True
                    )
                assert tmp_file.exists()
                tmp_file.unlink()

            unsafe_options = [
                "--config=protocol.ext.allow=always",
                "-c protocol.ext.allow=always",
            ]
            for i, unsafe_option in enumerate(unsafe_options):
                destination = tmp_dir / str(i)
                assert not destination.exists()
                Repo.clone_from(
                    rw_repo.working_dir, destination, multi_options=[unsafe_option], allow_unsafe_options=True
                )
                assert destination.exists()

    @with_rw_repo("HEAD")
    def test_clone_from_safe_options(self, rw_repo):
        with tempfile.TemporaryDirectory() as tdir:
            tmp_dir = pathlib.Path(tdir)
            options = [
                "--depth=1",
                "--single-branch",
                "-q",
            ]
            for option in options:
                destination = tmp_dir / option
                assert not destination.exists()
                Repo.clone_from(rw_repo.common_dir, destination, multi_options=[option])
                assert destination.exists()

    def test_clone_from_unsafe_protocol(self):
        with tempfile.TemporaryDirectory() as tdir:
            tmp_dir = pathlib.Path(tdir)
            tmp_file = tmp_dir / "pwn"
            urls = [
                f"ext::sh -c touch% {tmp_file}",
                "fd::17/foo",
            ]
            for url in urls:
                with self.assertRaises(UnsafeProtocolError):
                    Repo.clone_from(url, tmp_dir / "repo")
                assert not tmp_file.exists()

    def test_clone_from_unsafe_protocol_allowed(self):
        with tempfile.TemporaryDirectory() as tdir:
            tmp_dir = pathlib.Path(tdir)
            tmp_file = tmp_dir / "pwn"
            urls = [
                f"ext::sh -c touch% {tmp_file}",
                "fd::/foo",
            ]
            for url in urls:
                # The URL will be allowed into the command, but the command will
                # fail since we don't have that protocol enabled in the Git config file.
                with self.assertRaises(GitCommandError):
                    Repo.clone_from(url, tmp_dir / "repo", allow_unsafe_protocols=True)
                assert not tmp_file.exists()

    def test_clone_from_unsafe_protocol_allowed_and_enabled(self):
        with tempfile.TemporaryDirectory() as tdir:
            tmp_dir = pathlib.Path(tdir)
            tmp_file = tmp_dir / "pwn"
            urls = [
                f"ext::sh -c touch% {tmp_file}",
            ]
            allow_ext = [
                "--config=protocol.ext.allow=always",
            ]
            for url in urls:
                # The URL will be allowed into the command, and the protocol is enabled,
                # but the command will fail since it can't read from the remote repo.
                assert not tmp_file.exists()
                with self.assertRaises(GitCommandError):
                    Repo.clone_from(
                        url,
                        tmp_dir / "repo",
                        multi_options=allow_ext,
                        allow_unsafe_protocols=True,
                        allow_unsafe_options=True,
                    )
                assert tmp_file.exists()
                tmp_file.unlink()

    @with_rw_repo("HEAD")
    def test_max_chunk_size(self, repo):
        class TestOutputStream(TestBase):
            def __init__(self, max_chunk_size):
                self.max_chunk_size = max_chunk_size

            def write(self, b):
                self.assertTrue(len(b) <= self.max_chunk_size)

        for chunk_size in [16, 128, 1024]:
            repo.git.status(output_stream=TestOutputStream(chunk_size), max_chunk_size=chunk_size)

        repo.git.log(
            n=100,
            output_stream=TestOutputStream(io.DEFAULT_BUFFER_SIZE),
            max_chunk_size=None,
        )
        repo.git.log(
            n=100,
            output_stream=TestOutputStream(io.DEFAULT_BUFFER_SIZE),
            max_chunk_size=-10,
        )
        repo.git.log(n=100, output_stream=TestOutputStream(io.DEFAULT_BUFFER_SIZE))

    def test_init(self):
        prev_cwd = os.getcwd()
        os.chdir(tempfile.gettempdir())
        git_dir_rela = "repos/foo/bar.git"
        del_dir_abs = osp.abspath("repos")
        git_dir_abs = osp.abspath(git_dir_rela)
        try:
            # with specific path
            for path in (git_dir_rela, git_dir_abs):
                r = Repo.init(path=path, bare=True)
                self.assertIsInstance(r, Repo)
                assert r.bare is True
                assert not r.has_separate_working_tree()
                assert osp.isdir(r.git_dir)

                self._assert_empty_repo(r)

                # test clone
                clone_path = path + "_clone"
                rc = r.clone(clone_path)
                self._assert_empty_repo(rc)

                try:
                    rmtree(clone_path)
                except OSError:
                    # When relative paths are used, the clone may actually be inside
                    # of the parent directory.
                    pass
                # END exception handling

                # Try again, this time with the absolute version.
                rc = Repo.clone_from(r.git_dir, clone_path)
                self._assert_empty_repo(rc)

                rmtree(git_dir_abs)
                try:
                    rmtree(clone_path)
                except OSError:
                    # When relative paths are used, the clone may actually be inside
                    # of the parent directory.
                    pass
                # END exception handling

            # END for each path

            os.makedirs(git_dir_rela)
            os.chdir(git_dir_rela)
            r = Repo.init(bare=False)
            assert r.bare is False
            assert not r.has_separate_working_tree()

            self._assert_empty_repo(r)
        finally:
            try:
                rmtree(del_dir_abs)
            except OSError:
                pass
            os.chdir(prev_cwd)
        # END restore previous state

    def test_bare_property(self):
        self.rorepo.bare

    def test_daemon_export(self):
        orig_val = self.rorepo.daemon_export
        self.rorepo.daemon_export = not orig_val
        self.assertEqual(self.rorepo.daemon_export, (not orig_val))
        self.rorepo.daemon_export = orig_val
        self.assertEqual(self.rorepo.daemon_export, orig_val)

    def test_alternates(self):
        cur_alternates = self.rorepo.alternates
        # empty alternates
        self.rorepo.alternates = []
        self.assertEqual(self.rorepo.alternates, [])
        alts = ["other/location", "this/location"]
        self.rorepo.alternates = alts
        self.assertEqual(alts, self.rorepo.alternates)
        self.rorepo.alternates = cur_alternates

    def test_repr(self):
        assert repr(self.rorepo).startswith("<git.repo.base.Repo ")

    def test_is_dirty_with_bare_repository(self):
        orig_value = self.rorepo._bare
        self.rorepo._bare = True
        self.assertFalse(self.rorepo.is_dirty())
        self.rorepo._bare = orig_value

    def test_is_dirty(self):
        self.rorepo._bare = False
        for index in (0, 1):
            for working_tree in (0, 1):
                for untracked_files in (0, 1):
                    assert self.rorepo.is_dirty(index, working_tree, untracked_files) in (True, False)
                # END untracked files
            # END working tree
        # END index
        orig_val = self.rorepo._bare
        self.rorepo._bare = True
        assert self.rorepo.is_dirty() is False
        self.rorepo._bare = orig_val

    def test_is_dirty_pathspec(self):
        self.rorepo._bare = False
        for index in (0, 1):
            for working_tree in (0, 1):
                for untracked_files in (0, 1):
                    assert self.rorepo.is_dirty(index, working_tree, untracked_files, path=":!foo") in (True, False)
                # END untracked files
            # END working tree
        # END index
        orig_val = self.rorepo._bare
        self.rorepo._bare = True
        assert self.rorepo.is_dirty() is False
        self.rorepo._bare = orig_val

    @with_rw_repo("HEAD")
    def test_is_dirty_with_path(self, rwrepo):
        assert rwrepo.is_dirty(path="git") is False

        with open(osp.join(rwrepo.working_dir, "git", "util.py"), "at") as f:
            f.write("junk")
        assert rwrepo.is_dirty(path="git") is True
        assert rwrepo.is_dirty(path="doc") is False

        rwrepo.git.add(Git.polish_url(osp.join("git", "util.py")))
        assert rwrepo.is_dirty(index=False, path="git") is False
        assert rwrepo.is_dirty(path="git") is True

        with open(osp.join(rwrepo.working_dir, "doc", "no-such-file.txt"), "wt") as f:
            f.write("junk")
        assert rwrepo.is_dirty(path="doc") is False
        assert rwrepo.is_dirty(untracked_files=True, path="doc") is True

    def test_head(self):
        self.assertEqual(self.rorepo.head.reference.object, self.rorepo.active_branch.object)

    def test_index(self):
        index = self.rorepo.index
        self.assertIsInstance(index, IndexFile)

    def test_tag(self):
        assert self.rorepo.tag("refs/tags/0.1.5").commit

    def test_tag_to_full_tag_path(self):
        tags = ["0.1.5", "tags/0.1.5", "refs/tags/0.1.5"]
        value_errors = []
        for tag in tags:
            try:
                self.rorepo.tag(tag)
            except ValueError as valueError:
                value_errors.append(valueError.args[0])
        self.assertEqual(value_errors, [])

    def test_archive(self):
        tmpfile = tempfile.mktemp(suffix="archive-test")
        with open(tmpfile, "wb") as stream:
            self.rorepo.archive(stream, "0.1.6", path="doc")
            assert stream.tell()
        os.remove(tmpfile)

    @mock.patch.object(Git, "_call_process")
    def test_should_display_blame_information(self, git):
        git.return_value = fixture("blame")
        b = self.rorepo.blame("master", "lib/git.py")
        self.assertEqual(13, len(b))
        self.assertEqual(2, len(b[0]))
        # self.assertEqual(25, reduce(lambda acc, x: acc + len(x[-1]), b))
        self.assertEqual(hash(b[0][0]), hash(b[9][0]))
        c = b[0][0]
        self.assertTrue(git.called)

        self.assertEqual("634396b2f541a9f2d58b00be1a07f0c358b999b3", c.hexsha)
        self.assertEqual("Tom Preston-Werner", c.author.name)
        self.assertEqual("tom@mojombo.com", c.author.email)
        self.assertEqual(1191997100, c.authored_date)
        self.assertEqual("Tom Preston-Werner", c.committer.name)
        self.assertEqual("tom@mojombo.com", c.committer.email)
        self.assertEqual(1191997100, c.committed_date)
        self.assertRaisesRegex(
            ValueError,
            "634396b2f541a9f2d58b00be1a07f0c358b999b3 missing",
            lambda: c.message,
        )

        # Test the 'lines per commit' entries.
        tlist = b[0][1]
        self.assertTrue(tlist)
        self.assertTrue(isinstance(tlist[0], str))
        self.assertTrue(len(tlist) < sum(len(t) for t in tlist))  # Test for single-char bug.

        # BINARY BLAME
        git.return_value = fixture("blame_binary")
        blames = self.rorepo.blame("master", "rps")
        self.assertEqual(len(blames), 2)

    def test_blame_real(self):
        c = 0
        nml = 0  # Amount of multi-lines per blame.
        for item in self.rorepo.head.commit.tree.traverse(
            predicate=lambda i, d: i.type == "blob" and i.path.endswith(".py")
        ):
            c += 1

            for b in self.rorepo.blame(self.rorepo.head, item.path):
                nml += int(len(b[1]) > 1)
        # END for each item to traverse
        assert c, "Should have executed at least one blame command"
        assert nml, "There should at least be one blame commit that contains multiple lines"

    @mock.patch.object(Git, "_call_process")
    def test_blame_incremental(self, git):
        # Loop over two fixtures, create a test fixture for 2.11.1+ syntax.
        for git_fixture in ("blame_incremental", "blame_incremental_2.11.1_plus"):
            git.return_value = fixture(git_fixture)
            blame_output = self.rorepo.blame_incremental("9debf6b0aafb6f7781ea9d1383c86939a1aacde3", "AUTHORS")
            blame_output = list(blame_output)
            self.assertEqual(len(blame_output), 5)

            # Check all outputted line numbers.
            ranges = flatten([entry.linenos for entry in blame_output])
            self.assertEqual(
                ranges,
                flatten(
                    [
                        range(2, 3),
                        range(14, 15),
                        range(1, 2),
                        range(3, 14),
                        range(15, 17),
                    ]
                ),
            )

            commits = [entry.commit.hexsha[:7] for entry in blame_output]
            self.assertEqual(commits, ["82b8902", "82b8902", "c76852d", "c76852d", "c76852d"])

            # Original filenames.
            self.assertSequenceEqual(
                [entry.orig_path for entry in blame_output],
                ["AUTHORS"] * len(blame_output),
            )

            # Original line numbers.
            orig_ranges = flatten([entry.orig_linenos for entry in blame_output])
            self.assertEqual(
                orig_ranges,
                flatten(
                    [
                        range(2, 3),
                        range(14, 15),
                        range(1, 2),
                        range(2, 13),
                        range(13, 15),
                    ]
                ),
            )  # noqa E501

    @mock.patch.object(Git, "_call_process")
    def test_blame_complex_revision(self, git):
        git.return_value = fixture("blame_complex_revision")
        res = self.rorepo.blame("HEAD~10..HEAD", "README.md")
        self.assertEqual(len(res), 1)
        self.assertEqual(len(res[0][1]), 83, "Unexpected amount of parsed blame lines")

    @mock.patch.object(Git, "_call_process")
    def test_blame_accepts_rev_opts(self, git):
        expected_args = ["blame", "HEAD", "-M", "-C", "-C", "--", "README.md"]
        boilerplate_kwargs = {"p": True, "stdout_as_string": False}
        self.rorepo.blame("HEAD", "README.md", rev_opts=["-M", "-C", "-C"])
        git.assert_called_once_with(*expected_args, **boilerplate_kwargs)

    @with_rw_repo("HEAD", bare=False)
    def test_untracked_files(self, rwrepo):
        for run, repo_add in enumerate((rwrepo.index.add, rwrepo.git.add)):
            base = rwrepo.working_tree_dir
            files = (
                join_path_native(base, "%i_test _myfile" % run),
                join_path_native(base, "%i_test_other_file" % run),
                join_path_native(base, "%i__çava verböten" % run),
                join_path_native(base, "%i_çava-----verböten" % run),
            )

            num_recently_untracked = 0
            for fpath in files:
                with open(fpath, "wb"):
                    pass
            untracked_files = rwrepo.untracked_files
            num_recently_untracked = len(untracked_files)

            # Ensure we have all names - they are relative to the git-dir.
            num_test_untracked = 0
            for utfile in untracked_files:
                num_test_untracked += join_path_native(base, utfile) in files
            self.assertEqual(len(files), num_test_untracked)

            repo_add(untracked_files)
            self.assertEqual(len(rwrepo.untracked_files), (num_recently_untracked - len(files)))
        # END for each run

    def test_config_reader(self):
        reader = self.rorepo.config_reader()  # All config files.
        assert reader.read_only
        reader = self.rorepo.config_reader("repository")  # Single config file.
        assert reader.read_only

    def test_config_writer(self):
        for config_level in self.rorepo.config_level:
            try:
                with self.rorepo.config_writer(config_level) as writer:
                    self.assertFalse(writer.read_only)
            except IOError:
                # It's okay not to get a writer for some configuration files if we
                # have no permissions.
                pass

    def test_config_level_paths(self):
        for config_level in self.rorepo.config_level:
            assert self.rorepo._get_config_path(config_level)

    def test_creation_deletion(self):
        # Just a very quick test to assure it generally works. There are
        # specialized cases in the test_refs module.
        head = self.rorepo.create_head("new_head", "HEAD~1")
        self.rorepo.delete_head(head)

        try:
            tag = self.rorepo.create_tag("new_tag", "HEAD~2")
        finally:
            self.rorepo.delete_tag(tag)
        with self.rorepo.config_writer():
            pass
        try:
            remote = self.rorepo.create_remote("new_remote", "git@server:repo.git")
        finally:
            self.rorepo.delete_remote(remote)

    def test_comparison_and_hash(self):
        # This is only a preliminary test, more testing done in test_index.
        self.assertEqual(self.rorepo, self.rorepo)
        self.assertFalse(self.rorepo != self.rorepo)
        self.assertEqual(len({self.rorepo, self.rorepo}), 1)

    @with_rw_directory
    def test_tilde_and_env_vars_in_repo_path(self, rw_dir):
        with mock.patch.dict(os.environ, {"HOME": rw_dir}):
            os.environ["HOME"] = rw_dir
            Repo.init(osp.join("~", "test.git"), bare=True)

        with mock.patch.dict(os.environ, {"FOO": rw_dir}):
            os.environ["FOO"] = rw_dir
            Repo.init(osp.join("$FOO", "test.git"), bare=True)

    def test_git_cmd(self):
        # Test CatFileContentStream, just to be very sure we have no fencepost errors.
        # The last \n is the terminating newline that it expects.
        l1 = b"0123456789\n"
        l2 = b"abcdefghijklmnopqrstxy\n"
        l3 = b"z\n"
        d = l1 + l2 + l3 + b"\n"

        l1p = l1[:5]

        # Full size.
        # Size is without terminating newline.
        def mkfull():
            return Git.CatFileContentStream(len(d) - 1, BytesIO(d))

        ts = 5

        def mktiny():
            return Git.CatFileContentStream(ts, BytesIO(d))

        # readlines no limit
        s = mkfull()
        lines = s.readlines()
        self.assertEqual(len(lines), 3)
        self.assertTrue(lines[-1].endswith(b"\n"), lines[-1])
        self.assertEqual(s._stream.tell(), len(d))  # Must have scrubbed to the end.

        # realines line limit
        s = mkfull()
        lines = s.readlines(5)
        self.assertEqual(len(lines), 1)

        # readlines on tiny sections
        s = mktiny()
        lines = s.readlines()
        self.assertEqual(len(lines), 1)
        self.assertEqual(lines[0], l1p)
        self.assertEqual(s._stream.tell(), ts + 1)

        # readline no limit
        s = mkfull()
        self.assertEqual(s.readline(), l1)
        self.assertEqual(s.readline(), l2)
        self.assertEqual(s.readline(), l3)
        self.assertEqual(s.readline(), b"")
        self.assertEqual(s._stream.tell(), len(d))

        # readline limit
        s = mkfull()
        self.assertEqual(s.readline(5), l1p)
        self.assertEqual(s.readline(), l1[5:])

        # readline on tiny section
        s = mktiny()
        self.assertEqual(s.readline(), l1p)
        self.assertEqual(s.readline(), b"")
        self.assertEqual(s._stream.tell(), ts + 1)

        # read no limit
        s = mkfull()
        self.assertEqual(s.read(), d[:-1])
        self.assertEqual(s.read(), b"")
        self.assertEqual(s._stream.tell(), len(d))

        # read limit
        s = mkfull()
        self.assertEqual(s.read(5), l1p)
        self.assertEqual(s.read(6), l1[5:])
        self.assertEqual(s._stream.tell(), 5 + 6)  # It's not yet done.

        # read tiny
        s = mktiny()
        self.assertEqual(s.read(2), l1[:2])
        self.assertEqual(s._stream.tell(), 2)
        self.assertEqual(s.read(), l1[2:ts])
        self.assertEqual(s._stream.tell(), ts + 1)

    def _assert_rev_parse_types(self, name, rev_obj):
        rev_parse = self.rorepo.rev_parse

        if rev_obj.type == "tag":
            rev_obj = rev_obj.object

        # Tree and blob type.
        obj = rev_parse(name + "^{tree}")
        self.assertEqual(obj, rev_obj.tree)

        obj = rev_parse(name + ":CHANGES")
        self.assertEqual(obj.type, "blob")
        self.assertEqual(obj.path, "CHANGES")
        self.assertEqual(rev_obj.tree["CHANGES"], obj)

    def _assert_rev_parse(self, name):
        """tries multiple different rev-parse syntaxes with the given name
        :return: parsed object"""
        rev_parse = self.rorepo.rev_parse
        orig_obj = rev_parse(name)
        if orig_obj.type == "tag":
            obj = orig_obj.object
        else:
            obj = orig_obj
        # END deref tags by default

        # Try history
        rev = name + "~"
        obj2 = rev_parse(rev)
        self.assertEqual(obj2, obj.parents[0])
        self._assert_rev_parse_types(rev, obj2)

        # History with number
        ni = 11
        history = [obj.parents[0]]
        for _ in range(ni):
            history.append(history[-1].parents[0])
        # END get given amount of commits

        for pn in range(11):
            rev = name + "~%i" % (pn + 1)
            obj2 = rev_parse(rev)
            self.assertEqual(obj2, history[pn])
            self._assert_rev_parse_types(rev, obj2)
        # END history check

        # Parent (default)
        rev = name + "^"
        obj2 = rev_parse(rev)
        self.assertEqual(obj2, obj.parents[0])
        self._assert_rev_parse_types(rev, obj2)

        # Parent with number
        for pn, parent in enumerate(obj.parents):
            rev = name + "^%i" % (pn + 1)
            self.assertEqual(rev_parse(rev), parent)
            self._assert_rev_parse_types(rev, parent)
        # END for each parent

        return orig_obj

    @with_rw_repo("HEAD", bare=False)
    def test_rw_rev_parse(self, rwrepo):
        # Verify it does not confuse branches with hexsha ids.
        ahead = rwrepo.create_head("aaaaaaaa")
        assert rwrepo.rev_parse(str(ahead)) == ahead.commit

    def test_rev_parse(self):
        rev_parse = self.rorepo.rev_parse

        # Try special case: This one failed at some point, make sure its fixed.
        self.assertEqual(rev_parse("33ebe").hexsha, "33ebe7acec14b25c5f84f35a664803fcab2f7781")

        # Start from reference.
        num_resolved = 0

        for ref_no, ref in enumerate(Reference.iter_items(self.rorepo)):
            path_tokens = ref.path.split("/")